import pytest

from src.services import ai_agent_autonomous_runtime_control as runtime_control_module
from src.services.ai_agent_autonomous_runtime_control import (
    _RUNTIME_ALERT_OPERATION_COUNTS_SQL,
    _RUNTIME_ALERTMANAGER_EVENT_COUNTS_SQL,
    _RUNTIME_GROUPED_ALERT_EVENT_COUNTS_SQL,
    _RUNTIME_PLAYBOOK_TRUST_COUNTS_FALLBACK_SQL,
    _RUNTIME_TELEGRAM_COUNTS_SQL,
    _RUNTIME_TELEGRAM_LATEST_SQL,
    _RUNTIME_TIMELINE_COUNTS_SQL,
    build_ai_agent_autonomous_runtime_control,
    build_runtime_receipt_readback_from_rows,
    classify_deploy_control_plane_observation,
)


class _FailingRuntimeDb:
    async def execute(self, *_args, **_kwargs):
        raise RuntimeError("runtime trace query unavailable")


class _FailingRuntimeDbContext:
    async def __aenter__(self) -> _FailingRuntimeDb:
        return _FailingRuntimeDb()

    async def __aexit__(self, _exc_type, _exc, _tb) -> bool:
        return False


def _assert_log_controlled_writeback_executor(payload: dict):
    assert payload["schema_version"] == "ai_agent_log_controlled_writeback_executor_readback_v1"
    assert payload["priority"] == "P1-LOG-KM-RAG-MCP-PLAYBOOK"
    assert payload["status"] == "controlled_writeback_executor_ready"
    assert payload["active_blockers"] == []

    policy = payload["executor_policy"]
    assert policy["low_medium_high_controlled_apply_enabled"] is True
    assert policy["owner_review_required_for_low_medium_high"] is False
    assert policy["critical_break_glass_required"] is True
    assert policy["target_selector_required"] is True
    assert policy["source_of_truth_diff_required"] is True
    assert policy["check_mode_required"] is True
    assert policy["rollback_required"] is True
    assert policy["post_apply_verifier_required"] is True

    assert payload["rollups"]["source_writeback_plan_count"] == 18
    assert payload["rollups"]["execution_batch_count"] == 6
    assert payload["rollups"]["ready_execution_batch_count"] == 6
    assert payload["rollups"]["target_selector_count"] == 18
    assert payload["rollups"]["source_of_truth_diff_count"] == 18
    assert payload["rollups"]["controlled_executor_dispatch_ready"] is True
    assert payload["rollups"]["controlled_apply_enabled_by_policy"] is True
    assert payload["rollups"]["current_blocker_execution_queue_count"] == 1
    assert payload["rollups"]["current_blocker_control_path_blocked_count"] == 0
    assert (
        payload["rollups"]["current_blocker_control_path_pressure_blocked_count"]
        == 0
    )
    assert payload["rollups"]["current_blocker_local_recovery_package_count"] == 1
    assert payload["rollups"]["runtime_dispatch_performed"] is False

    context = payload["agent_consumption_context"]
    assert "ai_agent_autonomous_runtime_control" in context["consumable_by"]
    assert "km_rag_playbook_learning_loop" in context["consumable_by"]
    assert len(context["next_action_queue"]) == 6
    assert len(context["current_blocker_execution_queue"]) == 1
    current_queue = context["current_blocker_execution_queue"][0]
    assert current_queue["queue_item_id"] == (
        "current-p0-blocker::controlled_cd_lane_guardrails_blocked"
    )
    assert current_queue["controlled_recovery_package"] == (
        "check-awoooi-110-controlled-cd-lane-readiness.sh"
    )
    assert current_queue["post_apply_verifier"] == (
        "harbor-registry-controlled-recovery-receipt validator"
    )
    assert [phase["phase_id"] for phase in current_queue[
        "controlled_local_console_execution_plan"
    ]] == [
        "verify_ssh_control_path_restored",
        "verify_harbor_v2_ready",
        "verify_controlled_cd_lane_guardrails",
    ]
    assert current_queue["post_recovery_readback_commands"] == [
        "check-awoooi-110-controlled-cd-lane-readiness.sh",
        "curl -k https://registry.wooo.work/v2/",
        "curl http://192.168.0.110:5000/v2/",
    ]
    assert "workflow_dispatch_from_this_endpoint" in current_queue[
        "forbidden_runtime_actions"
    ]
    assert current_queue["external_control_path_blocker"] == ""
    assert current_queue["node_load_classifier"] == "load_not_high"
    assert current_queue["node_load_high"] is False
    assert current_queue["control_path_pressure_blocker"] == ""
    assert context["raw_payload_required"] is False

    boundaries = payload["operation_boundaries"]
    assert boundaries["executor_readback_only"] is True
    assert boundaries["executor_dispatch_performed"] is False
    assert boundaries["km_write_performed"] is False
    assert boundaries["rag_index_write_performed"] is False
    assert boundaries["playbook_trust_write_performed"] is False
    assert boundaries["mcp_tool_call_performed"] is False
    assert boundaries["agent_runtime_action_performed"] is False
    assert boundaries["raw_log_payload_persisted"] is False
    assert boundaries["secret_value_collection_allowed"] is False
    assert boundaries["github_api_used"] is False


def _log_controlled_writeback_consumer_readback() -> dict:
    targets = ("km", "rag", "playbook", "mcp", "verifier", "ai_agent")
    bindings = [
        {
            "consumer_binding_id": f"consumer::receipt::{target}",
            "dispatch_receipt_id": f"receipt::{target}",
            "ledger_op_id": f"op::{target}",
            "target": target,
            "target_surface": f"{target}_metadata_feedback_binding",
            "consumer_surface": f"{target}_consumer_context",
            "status": "ready_for_consumer_context",
            "raw_payload_included": False,
            "post_apply_verifier_refs": [f"verifier://{target}"],
            "target_write_performed": False,
        }
        for target in targets
    ]
    return {
        "schema_version": "ai_agent_log_controlled_writeback_consumer_readback_v1",
        "priority": "P1-LOG-KM-RAG-MCP-PLAYBOOK",
        "scope": "ai_agent_log_controlled_writeback_consumer_readback",
        "status": "controlled_writeback_consumer_readback_ready",
        "readback": {
            "workplan_id": "P1-LOG-CONTROLLED-WRITEBACK-CONSUMER-READBACK",
            "source_operation_type": "log_controlled_writeback_dispatched",
            "source_executor_route": "ai_agent_metadata_writeback_executor",
        },
        "controlled_consume": {
            "controlled_consume_allowed": True,
            "runtime_target_write_performed": False,
        },
        "consumer_bindings": bindings,
        "target_rollups": [
            {
                "target": target,
                "consumer_surface": f"{target}_consumer_context",
                "binding_count": 1,
                "ready_binding_count": 1,
                "metadata_only": True,
            }
            for target in targets
        ],
        "rollups": {
            "target_count": 6,
            "dispatch_ledger_row_count": 6,
            "consumer_binding_count": 6,
            "ready_consumer_binding_count": 6,
            "ready_target_count": 6,
            "metadata_only_receipt_count": 6,
            "post_apply_verifier_ref_count": 6,
            "controlled_consumer_readback_ready": True,
            "km_consumer_binding_count": 1,
            "rag_consumer_binding_count": 1,
            "playbook_consumer_binding_count": 1,
            "mcp_consumer_binding_count": 1,
            "verifier_consumer_binding_count": 1,
            "ai_agent_consumer_binding_count": 1,
            "runtime_target_write_performed": False,
        },
        "active_blockers": [],
        "operation_boundaries": {
            "consumer_readback_only": True,
            "metadata_ledger_read_performed": True,
            "km_write_performed": False,
            "rag_index_write_performed": False,
            "playbook_trust_write_performed": False,
            "mcp_tool_call_performed": False,
            "agent_runtime_action_performed": False,
            "telegram_send_performed": False,
            "workflow_trigger_performed": False,
            "raw_log_payload_persisted": False,
            "secret_value_collection_allowed": False,
            "github_api_used": False,
        },
    }


def _assert_log_controlled_writeback_consumer(payload: dict):
    assert payload["schema_version"] == (
        "ai_agent_log_controlled_writeback_consumer_readback_v1"
    )
    assert payload["status"] == "controlled_writeback_consumer_readback_ready"
    assert payload["active_blockers"] == []
    assert payload["controlled_consume"]["controlled_consume_allowed"] is True
    assert payload["controlled_consume"]["runtime_target_write_performed"] is False
    assert payload["rollups"]["consumer_binding_count"] == 6
    assert payload["rollups"]["ready_consumer_binding_count"] == 6
    assert payload["rollups"]["ready_target_count"] == 6
    assert payload["rollups"]["controlled_consumer_readback_ready"] is True
    assert payload["operation_boundaries"]["consumer_readback_only"] is True
    assert payload["operation_boundaries"]["mcp_tool_call_performed"] is False
    assert payload["operation_boundaries"]["raw_log_payload_persisted"] is False
    assert payload["operation_boundaries"]["github_api_used"] is False


@pytest.mark.asyncio
async def test_live_runtime_receipt_keeps_consumer_readback_when_trace_query_fails(
    monkeypatch,
):
    runtime_control_module._clear_runtime_receipt_readback_cache()

    async def _fake_consumer_readback(*, project_id: str):
        assert project_id == "awoooi"
        return _log_controlled_writeback_consumer_readback()

    monkeypatch.setattr(
        runtime_control_module,
        "get_db_context",
        lambda project_id: _FailingRuntimeDbContext(),
    )
    monkeypatch.setattr(
        runtime_control_module,
        "_load_log_controlled_writeback_consumer_readback",
        _fake_consumer_readback,
    )

    try:
        readback = (
            await runtime_control_module.load_ai_agent_autonomous_runtime_receipt_readback()
        )
    finally:
        runtime_control_module._clear_runtime_receipt_readback_cache()

    assert readback["db_read_status"] == "unavailable"
    assert readback["error"]["type"] == "RuntimeError"
    _assert_log_controlled_writeback_consumer(
        readback["log_controlled_writeback_consumer"]
    )


@pytest.mark.asyncio
async def test_live_runtime_receipt_readback_uses_short_cache_to_protect_db_pool(
    monkeypatch,
):
    runtime_control_module._clear_runtime_receipt_readback_cache()
    calls = 0

    async def _fake_uncached_loader(
        *,
        project_id: str,
        lookback_hours: int,
        limit: int,
    ) -> dict:
        nonlocal calls
        calls += 1
        return build_runtime_receipt_readback_from_rows(
            project_id=project_id,
            lookback_hours=lookback_hours,
            db_read_status="ok",
            service_log_count_rows=[
                {"status": "sanitized_recent_logs", "total": calls, "recent": calls},
            ],
        )

    monkeypatch.setattr(
        runtime_control_module,
        "_load_ai_agent_autonomous_runtime_receipt_readback_uncached",
        _fake_uncached_loader,
    )

    try:
        first = await runtime_control_module.load_ai_agent_autonomous_runtime_receipt_readback()
        second = await runtime_control_module.load_ai_agent_autonomous_runtime_receipt_readback()
    finally:
        runtime_control_module._clear_runtime_receipt_readback_cache()

    assert calls == 1
    assert first["db_read_status"] == "ok"
    assert second["db_read_status"] == "ok"
    assert (
        second["log_integration_taxonomy"]["rollups"]["classified_event_total"]
        == first["log_integration_taxonomy"]["rollups"]["classified_event_total"]
    )


@pytest.mark.asyncio
async def test_live_runtime_receipt_readback_does_not_cache_pool_unavailable(
    monkeypatch,
):
    runtime_control_module._clear_runtime_receipt_readback_cache()
    calls = 0

    async def _fake_uncached_loader(
        *,
        project_id: str,
        lookback_hours: int,
        limit: int,
    ) -> dict:
        nonlocal calls
        calls += 1
        if calls == 1:
            return build_runtime_receipt_readback_from_rows(
                project_id=project_id,
                lookback_hours=lookback_hours,
                db_read_status="unavailable",
                error_type="TimeoutError",
            )
        return build_runtime_receipt_readback_from_rows(
            project_id=project_id,
            lookback_hours=lookback_hours,
            db_read_status="ok",
            service_log_count_rows=[
                {"status": "sanitized_recent_logs", "total": 3, "recent": 1},
            ],
        )

    monkeypatch.setattr(
        runtime_control_module,
        "_load_ai_agent_autonomous_runtime_receipt_readback_uncached",
        _fake_uncached_loader,
    )

    try:
        first = await runtime_control_module.load_ai_agent_autonomous_runtime_receipt_readback()
        second = await runtime_control_module.load_ai_agent_autonomous_runtime_receipt_readback()
    finally:
        runtime_control_module._clear_runtime_receipt_readback_cache()

    assert calls == 2
    assert first["db_read_status"] == "unavailable"
    assert second["db_read_status"] == "ok"


def test_runtime_control_rollups_project_consumer_dispatch_ledger_if_trace_missing():
    payload = build_ai_agent_autonomous_runtime_control()
    readback = build_runtime_receipt_readback_from_rows(
        project_id="awoooi",
        db_read_status="unavailable",
        log_controlled_writeback_consumer=_log_controlled_writeback_consumer_readback(),
    )

    runtime_control_module._attach_runtime_receipt_readback(payload, readback)

    assert payload["rollups"]["live_log_controlled_writeback_dispatch_count"] == 6
    assert (
        payload["rollups"][
            "live_log_controlled_writeback_consumer_dispatch_ledger_count"
        ]
        == 6
    )
    assert payload["rollups"]["live_log_controlled_writeback_consumer_binding_count"] == 6
    assert (
        payload["rollups"]["live_log_controlled_writeback_consumer_ready_binding_count"]
        == 6
    )
    assert payload["rollups"]["live_log_controlled_writeback_consumer_ready_count"] == 1
    assert payload["rollups"]["live_log_controlled_writeback_consumer_blocker_count"] == 0


def test_runtime_receipt_auxiliary_sql_keeps_source_family_counts_schema_safe():
    from src.services.ai_agent_autonomous_runtime_control import (
        _RUNTIME_OPERATION_COUNTS_SQL,
        _RUNTIME_OPERATION_LATEST_SQL,
    )

    assert "semantic_operation_type" in _RUNTIME_OPERATION_COUNTS_SQL
    assert "km_linked" in _RUNTIME_OPERATION_COUNTS_SQL
    assert "log_controlled_writeback_dispatched" in _RUNTIME_OPERATION_COUNTS_SQL
    assert "semantic_operation_type" in _RUNTIME_OPERATION_LATEST_SQL
    assert "km_linked" in _RUNTIME_OPERATION_LATEST_SQL
    assert "GROUP BY coalesce(status, 'unknown')" in _RUNTIME_TIMELINE_COUNTS_SQL
    assert "FROM timeline_events" in _RUNTIME_TIMELINE_COUNTS_SQL
    assert "count(*) AS total" in _RUNTIME_PLAYBOOK_TRUST_COUNTS_FALLBACK_SQL
    assert "FROM playbooks" in _RUNTIME_PLAYBOOK_TRUST_COUNTS_FALLBACK_SQL
    assert "updated_at" not in _RUNTIME_PLAYBOOK_TRUST_COUNTS_FALLBACK_SQL
    assert "trust_score" not in _RUNTIME_PLAYBOOK_TRUST_COUNTS_FALLBACK_SQL
    assert "FROM alert_operation_log" in _RUNTIME_ALERT_OPERATION_COUNTS_SQL
    assert "ALERT_RECEIVED" in _RUNTIME_ALERT_OPERATION_COUNTS_SQL
    assert "FROM awooop_conversation_event" in _RUNTIME_ALERTMANAGER_EVENT_COUNTS_SQL
    assert "source_envelope ->> 'stage'" in _RUNTIME_ALERTMANAGER_EVENT_COUNTS_SQL
    assert "channel_chat_id LIKE 'alert-group:%'" in _RUNTIME_GROUPED_ALERT_EVENT_COUNTS_SQL


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_internal_control_plane_readback_retry_20260628_2"
    )
    assert data["program_status"]["legacy_no_send_no_live_rules_overridden"] is True
    assert data["program_status"]["implementation_completion_percent"] == 100
    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",
        "log_controlled_writeback_dispatched",
    }.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"
    recovery = data["runtime_receipt_readback"]["runtime_receipt_readback_recovery"]
    assert recovery["status"] == "blocked_runtime_receipt_db_readback_unavailable"
    assert recovery["safe_next_action_id"] == (
        "repair_runtime_receipt_db_readback_then_rerun_log_integration_taxonomy"
    )
    assert data["rollups"]["live_runtime_receipt_recovery_status"] == (
        "blocked_runtime_receipt_db_readback_unavailable"
    )
    assert data["rollups"]["live_runtime_receipt_safe_next_action_id"] == (
        "repair_runtime_receipt_db_readback_then_rerun_log_integration_taxonomy"
    )
    _assert_log_controlled_writeback_executor(
        data["runtime_receipt_readback"]["log_controlled_writeback_executor"]
    )
    consumer = data["runtime_receipt_readback"]["log_controlled_writeback_consumer"]
    assert consumer["schema_version"] == (
        "ai_agent_log_controlled_writeback_consumer_readback_v1"
    )
    assert consumer["status"] == "blocked_waiting_controlled_writeback_consumer_receipts"
    assert data["rollups"]["live_log_controlled_writeback_executor_batch_count"] == 6
    assert data["rollups"]["live_log_controlled_writeback_executor_ready_batch_count"] == 6
    assert data["rollups"]["live_log_controlled_writeback_executor_ready_count"] == 1
    assert data["rollups"]["live_log_controlled_writeback_executor_blocker_count"] == 0
    assert data["rollups"]["live_log_controlled_writeback_next_action_queue_count"] == 6
    assert (
        data["rollups"]["live_log_controlled_writeback_current_blocker_queue_count"]
        == 1
    )
    assert (
        data["rollups"][
            "live_log_controlled_writeback_current_blocker_control_path_blocked_count"
        ]
        == 0
    )
    assert (
        data["rollups"][
            "live_log_controlled_writeback_current_blocker_local_recovery_package_count"
        ]
        == 1
    )
    assert data["rollups"]["live_log_controlled_writeback_dispatch_count"] == 0
    assert data["rollups"]["live_log_controlled_writeback_recent_dispatch_count"] == 0
    assert data["rollups"]["live_log_controlled_writeback_consumer_binding_count"] == 0
    assert data["rollups"]["live_log_controlled_writeback_consumer_ready_count"] == 0
    assert data["rollups"]["live_log_controlled_writeback_consumer_blocker_count"] == 1
    assert data["runtime_receipt_readback"]["learning_loop"]["status"] == "in_progress"
    assert (
        data["runtime_receipt_readback"]["learning_loop"]["rollups"][
            "required_stage_missing_count"
        ]
        == 7
    )
    progress_items = {
        item["work_item_id"]: item
        for item in data["runtime_receipt_readback"]["work_item_progress"]["ordered_items"]
    }
    assert progress_items["P0-A-runtime-truth"]["status"] == "blocked"
    assert progress_items["P0-A-runtime-truth"]["blocker"] == (
        "runtime_receipt_db_readback_unavailable"
    )


def test_ai_agent_autonomous_runtime_control_exposes_internal_control_loop():
    data = build_ai_agent_autonomous_runtime_control()

    integration = data["control_plane_integration"]
    assert integration["schema_version"] == "ai_agent_autonomous_runtime_internal_loop_v1"
    assert integration["status"] == "mcp_rag_km_playbook_log_control_loop_declared"
    assert {sensor["normalized_event"] for sensor in integration["mcp_sensors"]} == {
        "RunObservation",
        "RunnerLaneState",
        "ProductionTruthSnapshot",
        "FrontendTruthSnapshot",
    }
    assert "controlled_cd_lane_capacity_label_guardrails" in integration["rag_context_queries"]
    assert "running_no_container_stale_ui" in integration["playbook_decision_classes"]
    assert integration["km_writeback_contract"]["stores_raw_logs"] is False
    assert integration["km_writeback_contract"]["stores_secret_values"] is False
    assert integration["log_projection_contract"]["raw_html_or_long_log_allowed"] is False
    assert data["rollups"]["mcp_sensor_count"] == 4
    assert data["rollups"]["playbook_decision_class_count"] == 7


def test_deploy_control_plane_classifier_separates_stale_spinner_from_real_failure():
    stale = classify_deploy_control_plane_observation(
        run_status="running",
        is_latest_deploy_intent=True,
        active_task_container_count=0,
        production_marker_hit=True,
        latest_flow_closed=True,
        runner_capacity_ok=True,
        runner_forbidden_label_count=0,
    )
    assert stale["classification"] == "running_no_container_stale_ui"
    assert stale["action"] == "treat_gitea_spinner_as_stale_and_keep_production_truth"
    assert stale["safety_boundary"]["writes_runtime_state"] is False
    assert stale["internal_writeback"]["km_writeback_required"] is True

    failure = classify_deploy_control_plane_observation(
        run_status="failure",
        is_latest_deploy_intent=True,
        active_task_container_count=0,
        production_marker_hit=False,
        latest_flow_closed=False,
        runner_capacity_ok=True,
        runner_forbidden_label_count=0,
    )
    assert failure["classification"] == "real_failure_requires_playbook_repair"
    assert failure["action"] == "open_cd_repair_playbook_with_target_selector_and_verifier"
    assert failure["safety_boundary"]["opens_legacy_runner"] is False
    assert failure["internal_writeback"]["playbook_route_required"] is True


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_candidate_matched",
                "status": "dry_run",
                "total": 1,
                "recent": 1,
            },
            {
                "operation_type": "ansible_apply_executed",
                "status": "success",
                "total": 1,
                "recent": 1,
            },
            {
                "operation_type": "ansible_check_mode_executed",
                "status": "success",
                "total": 1,
                "recent": 1,
            },
            {
                "operation_type": "ansible_learning_writeback_recorded",
                "status": "success",
                "total": 1,
                "recent": 1,
            },
            {
                "operation_type": "log_controlled_writeback_dispatched",
                "status": "success",
                "total": 6,
                "recent": 6,
            },
        ],
        operation_latest_rows=[
            {
                "op_id": "candidate-op",
                "parent_op_id": None,
                "operation_type": "ansible_candidate_matched",
                "status": "dry_run",
                "actor": "decision_manager",
                "incident_id": incident_id,
                "catalog_id": "ansible:188-momo-backup-user",
                "playbook_path": "infra/ansible/playbooks/188-momo-backup-user.yml",
                "execution_mode": "check_mode",
            },
            {
                "op_id": "check-mode-op",
                "parent_op_id": "candidate-op",
                "operation_type": "ansible_check_mode_executed",
                "status": "success",
                "actor": "ansible_check_mode_worker",
                "incident_id": incident_id,
                "catalog_id": "ansible:188-momo-backup-user",
                "playbook_path": "infra/ansible/playbooks/188-momo-backup-user.yml",
                "execution_mode": "check_mode",
                "returncode": "0",
            },
            {
                "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",
                "source_candidate_op_id": "candidate-op",
                "check_mode_op_id": "check-mode-op",
                "returncode": "0",
                "duration_ms": 7727,
            },
            {
                "op_id": "learning-op",
                "parent_op_id": apply_op_id,
                "operation_type": "ansible_learning_writeback_recorded",
                "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": "learning_writeback",
                "returncode": "0",
            },
        ],
        auto_repair_count_rows=[
            {"result_status": "success", "total": 1, "recent": 1},
        ],
        auto_repair_latest_rows=[
            {
                "id": "auto-repair-1",
                "incident_id": incident_id,
                "catalog_id": "ansible:188-momo-backup-user",
                "playbook_name": "infra/ansible/playbooks/188-momo-backup-user.yml",
                "result_status": "success",
                "executed_steps_text": f'["apply:{apply_op_id}"]',
                "triggered_by": "ansible_controlled_apply",
                "risk_level": "low",
                "execution_time_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",
            },
        ],
        mcp_gateway_count_rows=[
            {"status": "success", "total": 3, "recent": 1},
        ],
        legacy_mcp_count_rows=[
            {"status": "success", "total": 2, "recent": 2},
        ],
        service_log_count_rows=[
            {"status": "sanitized_recent_logs", "total": 2, "recent": 2},
        ],
        executor_log_count_rows=[
            {"status": "success", "total": 2, "recent": 1},
        ],
        timeline_count_rows=[
            {"status": "success", "total": 1, "recent": 1},
        ],
        playbook_trust_count_rows=[
            {"status": "learning_active", "total": 4, "recent": 1},
        ],
        alert_operation_count_rows=[
            {"event_type": "ALERT_RECEIVED", "total": 5, "recent": 2},
            {"event_type": "AUTO_REPAIR_TRIGGERED", "total": 2, "recent": 1},
            {"event_type": "EXECUTION_STARTED", "total": 1, "recent": 1},
            {"event_type": "EXECUTION_COMPLETED", "total": 1, "recent": 1},
            {"event_type": "NOTIFICATION_CLASSIFIED", "total": 1, "recent": 1},
            {"event_type": "GUARDRAIL_BLOCKED", "total": 1, "recent": 0},
        ],
        alertmanager_event_count_rows=[
            {"stage": "received", "total": 5, "recent": 2},
            {"stage": "converged", "total": 3, "recent": 1},
            {"stage": "llm_inflight_suppressed", "total": 1, "recent": 0},
        ],
        grouped_alert_event_count_rows=[
            {"status": "grouped_child_alert", "total": 4, "recent": 1},
        ],
        log_controlled_writeback_consumer=_log_controlled_writeback_consumer_readback(),
    )

    assert readback["db_read_status"] == "ok"
    assert readback["writes_on_read"] is False
    recovery = readback["runtime_receipt_readback_recovery"]
    assert recovery["status"] == "completed_live_runtime_receipts_observed"
    assert recovery["classified_event_total"] > 0
    assert recovery["active_source_family_count"] == 10
    assert recovery["inactive_source_family_ids"] == []
    assert readback["ansible_apply_executed"]["total"] == 1
    assert (
        readback["ansible_operations"]["counts"]["log_controlled_writeback_dispatched"][
            "total"
        ]
        == 6
    )
    assert readback["auto_repair_execution_receipt"]["by_status"]["success"] == 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": [],
    }
    assert readback["latest_failure_classification"]["classification"] == (
        "latest_controlled_apply_closed_success"
    )
    assert readback["controlled_retry_package"]["status"] == "not_required_for_latest_apply"
    ledger = readback["autonomous_execution_loop_ledger"]
    assert ledger["schema_version"] == "ai_agent_autonomous_execution_loop_ledger_v1"
    assert ledger["operation_id"] == apply_op_id
    assert ledger["root_candidate_op_id"] == "candidate-op"
    assert ledger["check_mode_op_id"] == "check-mode-op"
    assert ledger["apply_op_id"] == apply_op_id
    assert ledger["execution_state"] == "closed_success"
    assert ledger["closed"] is True
    assert ledger["missing_stage_ids"] == []
    assert ledger["next_executor_action"] == "keep_receipt_chain_closed"
    assert [stage["stage_id"] for stage in ledger["stages"]] == [
        "candidate",
        "check_mode",
        "controlled_apply",
        "auto_repair_execution_receipt",
        "post_apply_verifier",
        "km_playbook_writeback",
        "telegram_receipt",
    ]
    assert {stage["present"] for stage in ledger["stages"]} == {True}
    assert ledger["safety_contract"]["backfill_may_send_telegram"] is False
    assert ledger["safety_contract"]["live_apply_may_send_telegram_gateway_receipt"] is True
    trace = readback["trace_ledger"]
    assert trace["schema_version"] == "ai_agent_autonomous_trace_ledger_v1"
    trace_stage_ids = {stage["stage_id"] for stage in trace["stages"]}
    assert {
        "mcp_context",
        "service_log_evidence",
        "candidate",
        "check_mode",
        "executor_log_projection",
        "controlled_apply",
        "auto_repair_execution_receipt",
        "post_apply_verifier",
        "rag_km_learning",
        "playbook_trust",
        "timeline_projection",
        "telegram_receipt",
    } == trace_stage_ids
    assert trace["missing_required_stage_ids"] == []
    assert trace["recorded_stage_count"] == trace["stage_count"]
    assert "rag_km_learning" in trace["learning_source_stage_ids"]
    assert "playbook_trust" in trace["learning_source_stage_ids"]
    assert trace["public_safety"] == {
        "reads_raw_sessions": False,
        "stores_secret_values": False,
        "stores_unredacted_telegram_payload": False,
        "stores_internal_reasoning": False,
    }
    assert readback["mcp_context"]["total"] == 5
    assert readback["service_log_evidence"]["total"] == 2
    assert readback["executor_log_projection"]["total"] == 2
    assert readback["playbook_trust"]["by_status"]["learning_active"] == 4
    taxonomy = readback["log_integration_taxonomy"]
    assert taxonomy["schema_version"] == "ai_agent_log_integration_taxonomy_v1"
    assert taxonomy["normalized_event_flow"] == [
        "collect_source_log_or_receipt",
        "redact_and_hash_sensitive_fields",
        "assign_labels",
        "correlate_incident_operation_playbook",
        "write_trace_ledger",
        "retrieve_similar_context_via_rag",
        "select_or_repair_playbook",
        "run_check_mode_then_controlled_apply",
        "verify_and_write_learning_back",
    ]
    assert {
        "project",
        "product",
        "website",
        "service",
        "package",
        "tool",
        "incident",
        "operation",
        "playbook",
    }.issubset(set(taxonomy["label_dimensions"]))
    source_family_ids = {item["source_family_id"] for item in taxonomy["source_families"]}
    assert {
        "mcp_gateway_tool_calls",
        "legacy_mcp_tool_calls",
        "service_package_logs",
        "executor_operation_logs",
        "auto_repair_receipts",
        "post_apply_verifier",
        "rag_km_entries",
        "playbook_trust_signals",
        "operator_timeline_projection",
        "telegram_delivery_receipts",
    } == source_family_ids
    assert taxonomy["rollups"]["source_family_count"] == 10
    assert taxonomy["rollups"]["active_source_family_count"] == 10
    assert taxonomy["rollups"]["classified_event_total"] > 0
    assert {
        "project",
        "product",
        "website",
        "service",
        "package",
        "tool",
        "source_family",
    }.issubset(set(taxonomy["required_label_dimensions"]))
    assert taxonomy["public_safety"]["raw_secret_collection_allowed"] is False
    assert taxonomy["public_safety"]["unredacted_payload_storage_allowed"] is False
    _assert_log_controlled_writeback_executor(
        readback["log_controlled_writeback_executor"]
    )
    _assert_log_controlled_writeback_consumer(
        readback["log_controlled_writeback_consumer"]
    )
    decision_wiring = readback["agent_decision_wiring"]
    assert decision_wiring["schema_version"] == "ai_agent_decision_wiring_readback_v1"
    assert decision_wiring["status"] == "completed"
    assert decision_wiring["missing_required_stage_ids"] == []
    assert decision_wiring["closed_loop_observed"] is True
    assert {
        stage["stage_id"]
        for stage in decision_wiring["stages"]
        if stage["required_for_decision_wiring"]
    } == {
        "labeled_evidence_sources",
        "rag_context_retrieval",
        "playbook_candidate_selection",
        "check_mode_dry_run",
        "controlled_apply_boundary",
        "post_apply_verifier",
    }
    assert decision_wiring["rollups"]["candidate_total"] == 1
    assert decision_wiring["rollups"]["check_mode_total"] == 1
    assert decision_wiring["rollups"]["controlled_apply_total"] == 1
    assert decision_wiring["rollups"]["required_stage_missing_count"] == 0
    learning_loop = readback["learning_loop"]
    assert learning_loop["schema_version"] == "ai_agent_learning_loop_readback_v1"
    assert learning_loop["status"] == "completed"
    assert learning_loop["missing_required_stage_ids"] == []
    assert {
        stage["stage_id"]
        for stage in learning_loop["stages"]
        if stage["required_for_learning_loop"]
    } == {
        "verified_execution_outcome",
        "km_learning_writeback",
        "learning_repair_record",
        "playbook_trust_delta",
        "similar_case_context",
        "repair_candidate_feedback",
        "next_decision_context",
    }
    assert learning_loop["rollups"]["required_stage_present_count"] == 7
    assert learning_loop["rollups"]["required_stage_missing_count"] == 0
    assert learning_loop["rollups"]["learning_writeback_total"] == 1
    assert learning_loop["rollups"]["learning_writeback_recent"] == 1
    assert learning_loop["rollups"]["similar_case_source_total"] > 0
    assert learning_loop["rollups"]["repair_feedback_ready_count"] == 1
    assert learning_loop["public_safety"]["stores_secret_values"] is False
    assert learning_loop["public_safety"]["executes_on_read"] is False
    alert_noise = readback["alert_noise_reduction"]
    assert alert_noise["schema_version"] == "ai_agent_alert_noise_reduction_readback_v1"
    assert alert_noise["status"] == "completed"
    assert alert_noise["missing_required_stage_ids"] == []
    assert {
        stage["stage_id"]
        for stage in alert_noise["stages"]
        if stage["required_for_noise_reduction"]
    } == {
        "alert_intake_receipts",
        "duplicate_convergence",
        "notification_suppression",
        "telegram_outbound_receipts",
        "ai_controlled_routing",
        "learning_feedback",
    }
    assert alert_noise["rollups"]["required_stage_present_count"] == 6
    assert alert_noise["rollups"]["required_stage_missing_count"] == 0
    assert alert_noise["rollups"]["alert_received_total"] == 10
    assert alert_noise["rollups"]["suppressed_alert_total"] == 8
    assert alert_noise["rollups"]["telegram_outbound_receipt_total"] == 1
    assert alert_noise["rollups"]["controlled_route_total"] == 8
    assert alert_noise["routing_policy"]["manual_default_route_allowed"] is False
    assert alert_noise["routing_policy"]["low_medium_high_alerts_route_to_ai_controlled_queue"] is True
    assert alert_noise["public_safety"]["stores_raw_alert_payload"] is False
    host_load = readback["host_sustained_load_automation"]
    assert host_load["schema_version"] == "host_sustained_load_controlled_automation_readback_v1"
    assert host_load["status"] == "completed"
    assert host_load["current_work_item_id"] == (
        "P1-D2-host-sustained-load-controlled-automation"
    )
    assert host_load["rollups"]["action_class_count"] == 5
    assert host_load["rollups"]["controlled_apply_class_count"] == 2
    assert host_load["rollups"]["required_asset_count"] == 6
    assert host_load["rollups"]["ready_asset_count"] == 6
    assert {
        "orphan_browser_smoke_runaway_process",
        "ci_runner_load_saturation",
        "memory_or_swap_pressure",
        "gitea_queue_or_hook_backlog",
        "unknown_sustained_load",
    } == {item["class_id"] for item in host_load["action_classes"]}
    assert host_load["operation_boundaries"]["executes_on_read"] is False
    assert host_load["operation_boundaries"]["legacy_runner_restore_allowed"] is False
    ui_productization = readback["ui_productization"]
    assert ui_productization["schema_version"] == "ai_agent_ui_productization_readback_v1"
    assert ui_productization["status"] == "completed"
    assert ui_productization["missing_required_surface_ids"] == []
    assert ui_productization["public_safety"]["manual_default_outcome_allowed"] is False
    assert ui_productization["rollups"]["required_surface_present_count"] == 5
    assert ui_productization["rollups"]["segmented_filter_count"] == 5
    multi_product = readback["multi_product_taxonomy"]
    assert multi_product["schema_version"] == "ai_agent_multi_product_taxonomy_contract_v1"
    assert multi_product["status"] == "completed"
    assert multi_product["missing_product_scope_ids"] == []
    assert multi_product["missing_required_dimension_ids"] == []
    assert {
        "awoooi",
        "stockplatform",
        "vibework",
        "momo",
        "awooogo",
        "tsenyang",
        "agent_bounty_protocol",
        "public_websites",
    } == {scope["product_id"] for scope in multi_product["product_scopes"]}
    assert multi_product["rollups"]["product_scope_count"] == 8
    assert multi_product["rollups"]["contract_ready_product_scope_count"] == 8
    assert multi_product["rollups"]["runtime_adapter_active_count"] == 1
    assert multi_product["public_safety"]["external_product_runtime_write_enabled_on_read"] is False
    progress = readback["work_item_progress"]
    assert progress["schema_version"] == "ai_agent_automation_work_item_progress_v1"
    ordered_ids = [item["work_item_id"] for item in progress["ordered_items"]]
    assert ordered_ids == [
        "P0-A-runtime-truth",
        "P0-B-trace-ledger",
        "P0-C-log-taxonomy",
        "P0-D-ui-visibility",
        "P0-E-verification-deploy",
        "P1-A-ingestion-coverage",
        "P1-B-agent-decision-wiring",
        "P1-C-learning-loop",
        "P1-D-alert-noise-reduction",
        "P1-D2-host-sustained-load-controlled-automation",
        "P1-E-log-controlled-writeback-executor",
        "P1-F-log-controlled-writeback-consumer",
        "P2-A-ui-ux-productization",
        "P2-B-multi-product-expansion",
    ]
    assert progress["ordered_items"][4]["status"] == "completed"
    assert progress["ordered_items"][0]["status"] == "completed"
    assert progress["ordered_items"][0]["classified_event_total"] > 0
    assert progress["ordered_items"][5]["status"] == "completed"
    assert progress["ordered_items"][6]["status"] == "completed"
    assert progress["ordered_items"][7]["status"] == "completed"
    assert progress["ordered_items"][7]["remaining_learning_loop_stage_count"] == 0
    assert progress["ordered_items"][8]["status"] == "completed"
    assert progress["ordered_items"][8]["remaining_alert_noise_stage_count"] == 0
    assert progress["ordered_items"][9]["status"] == "completed"
    assert progress["ordered_items"][9]["controlled_action_class_count"] == 2
    assert progress["ordered_items"][9]["ready_asset_count"] == 6
    assert progress["ordered_items"][10]["status"] == "completed"
    assert progress["ordered_items"][10]["remaining_executor_batch_count"] == 0
    assert progress["ordered_items"][10]["active_blocker_count"] == 0
    assert progress["ordered_items"][11]["status"] == "completed"
    assert progress["ordered_items"][11]["remaining_consumer_binding_count"] == 0
    assert progress["ordered_items"][11]["active_blocker_count"] == 0
    assert progress["ordered_items"][12]["status"] == "completed"
    assert progress["ordered_items"][12]["remaining_ui_surface_count"] == 0
    assert progress["ordered_items"][13]["status"] == "completed"
    assert progress["ordered_items"][13]["remaining_product_scope_count"] == 0
    assert progress["source_family_items"]
    assert {item["status"] for item in progress["source_family_items"]} == {"completed"}
    assert progress["rollups"]["source_family_work_item_count"] == 10
    assert progress["rollups"]["completed_count"] == 24
    assert progress["rollups"]["pending_count"] == 0


def test_runtime_receipt_recovery_flags_zero_live_log_events():
    readback = build_runtime_receipt_readback_from_rows(
        project_id="awoooi",
        db_read_status="ok",
    )

    recovery = readback["runtime_receipt_readback_recovery"]
    assert recovery["status"] == "blocked_live_log_source_family_events_missing"
    assert recovery["safe_next_action_id"] == (
        "connect_service_tool_package_logs_to_trace_ledger_then_rerun_taxonomy"
    )
    assert recovery["active_source_family_count"] == 0
    assert len(recovery["inactive_source_family_ids"]) == 10

    progress_items = {
        item["work_item_id"]: item
        for item in readback["work_item_progress"]["ordered_items"]
    }
    assert progress_items["P0-A-runtime-truth"]["status"] == "in_progress"
    assert progress_items["P0-A-runtime-truth"]["blocker"] == (
        "live_log_classified_event_total_zero"
    )
    assert progress_items["P1-A-ingestion-coverage"]["status"] == "in_progress"
    assert readback["work_item_progress"]["rollups"]["not_started_count"] == 10


def test_runtime_receipt_recovery_keeps_partial_live_log_events_visible():
    readback = build_runtime_receipt_readback_from_rows(
        project_id="awoooi",
        db_read_status="partial",
        service_log_count_rows=[
            {"status": "sanitized_recent_logs", "total": 12, "recent": 3},
        ],
        partial_query_failures=[
            {"query_name": "km_counts", "error_type": "TimeoutError"},
        ],
        error_type="partial_query_failures",
    )

    recovery = readback["runtime_receipt_readback_recovery"]
    assert recovery["status"] == "degraded_runtime_receipt_partial_query_failures"
    assert recovery["safe_next_action_id"] == (
        "repair_partial_runtime_receipt_queries_then_rerun_log_integration_taxonomy"
    )
    assert recovery["partial_query_failures"] == [
        {"query_name": "km_counts", "error_type": "TimeoutError"}
    ]
    assert recovery["classified_event_total"] == 12
    assert recovery["active_source_family_count"] == 1

    progress_items = {
        item["work_item_id"]: item
        for item in readback["work_item_progress"]["ordered_items"]
    }
    assert progress_items["P0-A-runtime-truth"]["status"] == "in_progress"
    assert progress_items["P0-A-runtime-truth"]["blocker"] == (
        "runtime_receipt_partial_query_failures"
    )


def test_runtime_receipt_work_items_use_learning_receipts_without_latest_telegram_gate():
    apply_op_id = "2f8ef5c8-fd4e-4950-99e9-dc9e61150cab"
    incident_id = "INC-20260629-LEARNING"

    readback = build_runtime_receipt_readback_from_rows(
        project_id="awoooi",
        db_read_status="ok",
        operation_count_rows=[
            {
                "operation_type": "ansible_candidate_matched",
                "status": "dry_run",
                "total": 1,
                "recent": 1,
            },
            {
                "operation_type": "ansible_check_mode_executed",
                "status": "success",
                "total": 1,
                "recent": 1,
            },
            {
                "operation_type": "ansible_apply_executed",
                "status": "success",
                "total": 1,
                "recent": 1,
            },
            {
                "operation_type": "ansible_learning_writeback_recorded",
                "status": "success",
                "total": 1,
                "recent": 1,
            },
        ],
        operation_latest_rows=[
            {
                "op_id": "candidate-op-learning",
                "parent_op_id": None,
                "operation_type": "ansible_candidate_matched",
                "status": "dry_run",
                "actor": "decision_manager",
                "incident_id": incident_id,
                "catalog_id": "ansible:188-ai-web",
                "playbook_path": "infra/ansible/playbooks/188-ai-web.yml",
                "execution_mode": "check_mode",
            },
            {
                "op_id": "check-mode-op-learning",
                "parent_op_id": "candidate-op-learning",
                "operation_type": "ansible_check_mode_executed",
                "status": "success",
                "actor": "ansible_check_mode_worker",
                "incident_id": incident_id,
                "catalog_id": "ansible:188-ai-web",
                "playbook_path": "infra/ansible/playbooks/188-ai-web.yml",
                "execution_mode": "check_mode",
                "returncode": "0",
            },
            {
                "op_id": apply_op_id,
                "parent_op_id": "check-mode-op-learning",
                "operation_type": "ansible_apply_executed",
                "status": "success",
                "actor": "ansible_controlled_apply_worker",
                "incident_id": incident_id,
                "catalog_id": "ansible:188-ai-web",
                "playbook_path": "infra/ansible/playbooks/188-ai-web.yml",
                "execution_mode": "controlled_apply",
                "source_candidate_op_id": "candidate-op-learning",
                "check_mode_op_id": "check-mode-op-learning",
                "returncode": "0",
            },
            {
                "op_id": "learning-op",
                "parent_op_id": apply_op_id,
                "operation_type": "ansible_learning_writeback_recorded",
                "status": "success",
                "actor": "ansible_controlled_apply_worker",
                "incident_id": incident_id,
                "catalog_id": "ansible:188-ai-web",
                "playbook_path": "infra/ansible/playbooks/188-ai-web.yml",
                "execution_mode": "learning_writeback",
                "returncode": "0",
            },
        ],
        auto_repair_count_rows=[
            {"result_status": "success", "total": 1, "recent": 1},
        ],
        auto_repair_latest_rows=[
            {
                "id": "auto-repair-learning",
                "incident_id": incident_id,
                "catalog_id": "ansible:188-ai-web",
                "playbook_name": "infra/ansible/playbooks/188-ai-web.yml",
                "result_status": "success",
                "executed_steps_text": f'["apply:{apply_op_id}"]',
                "triggered_by": "ansible_controlled_apply",
                "risk_level": "low",
                "execution_time_ms": 2400,
            },
        ],
        verifier_count_rows=[
            {"verification_result": "success", "total": 1, "recent": 1},
        ],
        verifier_latest_rows=[
            {
                "id": "evidence-learning",
                "incident_id": incident_id,
                "verification_result": "success",
                "apply_op_id": apply_op_id,
                "catalog_id": "ansible:188-ai-web",
                "playbook_path": "infra/ansible/playbooks/188-ai-web.yml",
                "returncode": "0",
            },
        ],
        km_count_rows=[
            {"status": "review", "total": 1, "recent": 1},
        ],
        km_latest_rows=[
            {
                "id": "km-learning",
                "title": "AI 自動修復沉澱：INC-20260629-LEARNING",
                "related_incident_id": incident_id,
                "related_playbook_id": "ansible:188-ai-web",
                "path_type": "ansible_apply_receipt:2f8ef5c8",
                "status": "review",
                "created_by": "ai_agent_ansible_worker",
            },
        ],
        telegram_count_rows=[
            {"send_status": "sent", "total": 1, "recent": 1},
        ],
        telegram_latest_rows=[],
        mcp_gateway_count_rows=[
            {"status": "success", "total": 1, "recent": 1},
        ],
        legacy_mcp_count_rows=[
            {"status": "success", "total": 1, "recent": 1},
        ],
        service_log_count_rows=[
            {"status": "sanitized_recent_logs", "total": 1, "recent": 1},
        ],
        executor_log_count_rows=[
            {"status": "success", "total": 1, "recent": 1},
        ],
        timeline_count_rows=[
            {"status": "success", "total": 1, "recent": 1},
        ],
        playbook_trust_count_rows=[
            {"status": "learning_active", "total": 1, "recent": 1},
        ],
        alert_operation_count_rows=[
            {"event_type": "ALERT_RECEIVED", "total": 5, "recent": 2},
            {"event_type": "AUTO_REPAIR_TRIGGERED", "total": 2, "recent": 1},
            {"event_type": "EXECUTION_STARTED", "total": 1, "recent": 1},
            {"event_type": "EXECUTION_COMPLETED", "total": 1, "recent": 1},
        ],
        alertmanager_event_count_rows=[
            {"stage": "received", "total": 5, "recent": 2},
            {"stage": "converged", "total": 3, "recent": 1},
            {"stage": "llm_inflight_suppressed", "total": 1, "recent": 1},
        ],
        grouped_alert_event_count_rows=[
            {"status": "grouped_child_alert", "total": 4, "recent": 1},
        ],
        log_controlled_writeback_consumer=_log_controlled_writeback_consumer_readback(),
    )

    assert readback["latest_flow_closure"]["closed"] is False
    assert readback["latest_flow_closure"]["missing"] == ["telegram_receipt"]
    assert readback["autonomous_execution_loop_ledger"]["closed"] is False
    assert readback["learning_loop"]["status"] == "completed"
    assert readback["learning_loop"]["missing_required_stage_ids"] == []
    assert readback["learning_loop"]["rollups"]["next_decision_ready_count"] == 1
    assert readback["alert_noise_reduction"]["status"] == "completed"
    assert readback["alert_noise_reduction"]["missing_required_stage_ids"] == []
    progress = readback["work_item_progress"]
    statuses = {
        item["work_item_id"]: item["status"]
        for item in progress["ordered_items"]
    }
    assert statuses["P1-C-learning-loop"] == "completed"
    assert statuses["P1-D-alert-noise-reduction"] == "completed"
    assert statuses["P1-D2-host-sustained-load-controlled-automation"] == "completed"
    assert statuses["P1-E-log-controlled-writeback-executor"] == "completed"
    assert statuses["P1-F-log-controlled-writeback-consumer"] == "completed"
    assert statuses["P2-A-ui-ux-productization"] == "completed"
    assert statuses["P2-B-multi-product-expansion"] == "completed"
    assert {item["status"] for item in progress["source_family_items"]} == {"completed"}
    assert progress["rollups"]["completed_count"] == 24
    assert progress["rollups"]["pending_count"] == 0


def test_runtime_receipt_readback_classifies_closed_failed_apply_as_ai_repair():
    apply_op_id = "94925d5e-6fdc-49c3-90e8-f0a0d57a6a58"
    incident_id = "INC-20260628-A40A9A"

    readback = build_runtime_receipt_readback_from_rows(
        project_id="awoooi",
        db_read_status="ok",
        operation_count_rows=[
            {
                "operation_type": "ansible_apply_executed",
                "status": "failed",
                "total": 1,
                "recent": 1,
            },
        ],
        operation_latest_rows=[
            {
                "op_id": "8b555f41-e81f-4d8e-956b-fb20d358db63",
                "parent_op_id": "candidate-op-2",
                "operation_type": "ansible_check_mode_executed",
                "status": "success",
                "actor": "ansible_check_mode_worker",
                "incident_id": incident_id,
                "catalog_id": "ansible:188-ai-web",
                "playbook_path": "infra/ansible/playbooks/188-ai-web-readonly.yml",
                "execution_mode": "check_mode",
                "returncode": "0",
            },
            {
                "op_id": apply_op_id,
                "parent_op_id": "8b555f41-e81f-4d8e-956b-fb20d358db63",
                "operation_type": "ansible_apply_executed",
                "status": "failed",
                "actor": "ansible_controlled_apply_worker",
                "incident_id": incident_id,
                "catalog_id": "ansible:188-ai-web",
                "playbook_path": "infra/ansible/playbooks/188-ai-web.yml",
                "execution_mode": "controlled_apply",
                "source_candidate_op_id": "candidate-op-2",
                "check_mode_op_id": "8b555f41-e81f-4d8e-956b-fb20d358db63",
                "returncode": "2",
                "duration_ms": 4797,
            },
        ],
        auto_repair_count_rows=[
            {"result_status": "failed", "total": 1, "recent": 1},
        ],
        auto_repair_latest_rows=[
            {
                "id": "auto-repair-2",
                "incident_id": incident_id,
                "catalog_id": "ansible:188-ai-web",
                "playbook_name": "infra/ansible/playbooks/188-ai-web.yml",
                "result_status": "failed",
                "executed_steps_text": f'["apply:{apply_op_id}"]',
                "triggered_by": "ansible_controlled_apply",
                "risk_level": "medium",
                "execution_time_ms": 4797,
            },
        ],
        verifier_count_rows=[
            {"verification_result": "failed", "total": 1, "recent": 1},
        ],
        verifier_latest_rows=[
            {
                "id": "evidence-1",
                "incident_id": incident_id,
                "verification_result": "failed",
                "apply_op_id": apply_op_id,
                "catalog_id": "ansible:188-ai-web",
                "playbook_path": "infra/ansible/playbooks/188-ai-web.yml",
                "returncode": "2",
            },
        ],
        km_count_rows=[
            {"status": "REVIEW", "total": 1, "recent": 1},
        ],
        km_latest_rows=[
            {
                "id": "km-1",
                "title": "AI 自動修復沉澱：INC-20260628-A40A9A",
                "related_incident_id": incident_id,
                "related_playbook_id": "ansible:188-ai-web",
                "path_type": "ansible_apply_receipt:94925d5e",
                "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": "32016",
                "incident_id": incident_id,
                "action": "controlled_apply_result",
            },
        ],
    )

    classification = readback["latest_failure_classification"]
    assert classification["classification"] == "closed_failed_apply_requires_ai_repair"
    assert classification["action"] == "queue_check_mode_replay_and_playbook_repair_candidate"
    assert classification["target_selector"] == {
        "incident_id": incident_id,
        "apply_op_id": apply_op_id,
        "parent_op_id": "8b555f41-e81f-4d8e-956b-fb20d358db63",
        "catalog_id": "ansible:188-ai-web",
        "playbook_path": "infra/ansible/playbooks/188-ai-web.yml",
        "execution_mode": "controlled_apply",
    }
    assert classification["evidence"]["returncode"] == "2"
    assert classification["evidence"]["verification_result"] == "failed"
    assert classification["evidence"]["latest_flow_closed"] is True
    assert classification["evidence"]["output_tail_in_readback"] is False
    assert classification["evidence"]["unredacted_output_required"] is False
    assert classification["safe_next_steps"] == [
        "run_no_write_check_mode_replay",
        "extract_sanitized_failed_task_summary",
        "write_km_playbook_repair_candidate",
        "retry_controlled_apply_only_after_check_mode_passes",
    ]

    retry = readback["controlled_retry_package"]
    assert retry["package_id"] == "ansible_retry:94925d5e"
    assert retry["status"] == "ready_for_no_write_check_mode_replay"
    assert retry["source_of_truth"] == {
        "catalog_id": "ansible:188-ai-web",
        "playbook_path": "infra/ansible/playbooks/188-ai-web.yml",
        "source_diff_required_before_retry": True,
        "failed_task_summary_required": True,
    }
    assert retry["preflight"]["no_write_check_mode_replay_required"] is True
    assert retry["preflight"]["reuse_parent_check_mode_op_id"] == (
        "8b555f41-e81f-4d8e-956b-fb20d358db63"
    )
    assert retry["apply_gate"]["controlled_apply_retry_allowed_now"] is False
    assert retry["apply_gate"]["requires_check_mode_success_before_apply"] is True
    assert retry["rollback"]["destructive_rollback_allowed"] is False
    assert retry["post_apply"] == {
        "post_apply_verifier_required": True,
        "km_playbook_trust_writeback_required": True,
        "telegram_receipt_required": True,
    }
    assert retry["next_ai_action"] == "run_no_write_check_mode_replay"
    ledger = readback["autonomous_execution_loop_ledger"]
    assert ledger["closed"] is True
    assert ledger["execution_state"] == "closed_failed_apply_repair_ready"
    assert ledger["next_executor_action"] == "run_no_write_check_mode_replay"
    assert ledger["missing_stage_ids"] == []
    assert ledger["stages"][3]["stage_id"] == "auto_repair_execution_receipt"
    assert ledger["stages"][3]["status"] == "failed"


def test_runtime_telegram_receipt_queries_include_alert_notifications() -> None:
    """Runtime readback must count alert notification receipts, not only callbacks."""
    assert "alert_notification_sent" in _RUNTIME_TELEGRAM_COUNTS_SQL
    assert "alert_notification_sent" in _RUNTIME_TELEGRAM_LATEST_SQL
    assert "controlled_apply_result" in _RUNTIME_TELEGRAM_COUNTS_SQL
    assert "controlled_apply_result" in _RUNTIME_TELEGRAM_LATEST_SQL


def test_runtime_execution_loop_ledger_does_not_mix_unrelated_check_mode_rows():
    apply_op_id = "db3f12ce-08fc-4289-8c93-338305d5850c"
    readback = build_runtime_receipt_readback_from_rows(
        project_id="awoooi",
        db_read_status="ok",
        operation_latest_rows=[
            {
                "op_id": "unrelated-check-mode-op",
                "parent_op_id": "unrelated-candidate",
                "operation_type": "ansible_check_mode_executed",
                "status": "success",
                "actor": "ansible_check_mode_worker",
                "incident_id": "INC-OTHER",
                "catalog_id": "ansible:110-devops",
                "playbook_path": "infra/ansible/playbooks/110-devops.yml",
                "execution_mode": "check_mode",
                "returncode": "0",
            },
            {
                "op_id": apply_op_id,
                "parent_op_id": "expected-check-mode-op",
                "operation_type": "ansible_apply_executed",
                "status": "success",
                "actor": "ansible_controlled_apply_worker",
                "incident_id": "INC-20260629-LEDGER",
                "catalog_id": "ansible:188-momo-backup-user",
                "playbook_path": "infra/ansible/playbooks/188-momo-backup-user.yml",
                "execution_mode": "controlled_apply",
                "returncode": "0",
            },
        ],
    )

    ledger = readback["autonomous_execution_loop_ledger"]
    assert ledger["operation_id"] == apply_op_id
    assert ledger["check_mode_op_id"] == "expected-check-mode-op"
    assert ledger["root_candidate_op_id"] is None
    assert ledger["incident_id"] == "INC-20260629-LEDGER"
    assert ledger["stages"][1]["stage_id"] == "check_mode"
    assert ledger["stages"][1]["ref_id"] == "expected-check-mode-op"
    assert ledger["stages"][1]["status"] == "inferred_from_apply_parent"
    assert "candidate" in ledger["missing_stage_ids"]
