from __future__ import annotations

# ruff: noqa: E402
import base64
import hashlib
import os
import struct
from pathlib import Path
from unittest.mock import AsyncMock

os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://test:test@localhost/test")

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from pydantic import ValidationError

from src.api.v1 import agents as agents_api
from src.core.config import settings
from src.models.agent99_completion import Agent99TelegramLifecycleRequest
from src.services import agent99_telegram_lifecycle as lifecycle_service
from src.services import telegram_gateway as gateway_service


def payload() -> dict:
    return {
        "schema_version": "agent99_telegram_lifecycle_v1",
        "delivery_id": "agent99-lifecycle-0123456789abcdef0123456789abcdef01234567",
        "project_id": "awoooi",
        "incident_id": "INC-20260715-AG9901",
        "fingerprint": "0123456789abcdef0123456789abcdef",
        "event_type": "performance_warning",
        "severity": "warning",
        "lifecycle": "investigating",
        "state_key": "investigating|warning",
        "run_id": "run-agent99-20260715-01",
        "trace_id": "trace-agent99-20260715-01",
        "work_item_id": "work-item-agent99-20260715-01",
        "executor_name": "Agent99:Recover",
        "verifier_name": "agent99-recover-post-condition-v1",
        "received_status": "passed",
        "check_status": "passed",
        "apply_status": "running",
        "verifier_status": "pending",
        "closure_status": "pending",
        "title": "Host 110 主機效能異常",
        "target": "192.168.0.110",
        "impact": "資源壓力可能降低服務回應速度。",
        "action": "Agent99 正在執行 allowlisted 診斷與降載。",
        "verification": "等待 CPU、記憶體與磁碟 post-verifier。",
        "duration_text": "12.4 秒",
        "occurrences": 2,
        "next_update": "狀態改變時更新，相同狀態不重複推播。",
        "occurred_at": "2026-07-15T04:20:00+08:00",
    }


def visual_payload() -> dict:
    raw = (
        b"\x89PNG\r\n\x1a\n"
        + struct.pack(">I", 13)
        + b"IHDR"
        + struct.pack(">II", 1200, 760)
        + b"\x08\x02\x00\x00\x00"
    )
    return {
        "schema_version": "agent99_telegram_visual_v1",
        "media_type": "image/png",
        "generated_by": "agent99_system_drawing_v1",
        "width": 1200,
        "height": 760,
        "sha256": hashlib.sha256(raw).hexdigest(),
        "content_base64": base64.b64encode(raw).decode(),
    }


def app_client() -> TestClient:
    app = FastAPI()
    app.include_router(agents_api.router, prefix="/api/v1")
    return TestClient(app)


def test_agent99_client_retries_one_202_with_exact_db_only_receipt() -> None:
    source = (
        Path(__file__).resolve().parents[3] / "agent99-control-plane.ps1"
    ).read_text(encoding="utf-8-sig")
    send_section = source.split("function Send-AgentTelegram {", 1)[1].split(
        "function Record-AgentEvent {",
        1,
    )[0]

    assert send_section.count("Invoke-AgentCanonicalLifecycleIngress ") == 2
    assert 'if ([int]$ingress.statusCode -eq 202)' in send_section
    assert '"agent99_telegram_lifecycle_reconcile_v1"' in send_section
    assert '$payload["reconcile_receipt"] = $reconcileReceipt' in send_section
    assert "Start-Sleep -Seconds ([int]$ingress.retryAfterSeconds)" in send_section
    assert "[int]$ingress.retryAfterSeconds -le 2" in send_section
    assert "canonicalIngressAttemptCount = 2" in send_section
    assert "reconcileProviderSendPerformed" in send_section
    assert "provider_message_id" in send_section
    assert "destination_binding" in send_section
    for lifecycle_field in (
        "trace_id",
        "work_item_id",
        "executor_name",
        "verifier_name",
        "received_status",
        "check_status",
        "apply_status",
        "verifier_status",
        "closure_status",
    ):
        assert f"{lifecycle_field} =" in send_section
    for card_field in (
        "receivedStatus",
        "checkStatus",
        "applyStatus",
        "verifierStatus",
        "closureStatus",
        "traceKey",
        "workItemKey",
    ):
        assert f"$card.{card_field}" in send_section
    assert "X-Agent99-Lifecycle-Token" not in send_section
    selftest_section = source.split(
        "function Invoke-AgentTelegramDeliverySelfTest {",
        1,
    )[1].split("function Invoke-AgentSensorGateSelfTest {", 1)[0]
    assert 'id = $testId' in selftest_section
    assert selftest_section.count(
        'Send-AgentTelegram "warning" "telegram_delivery_test"'
    ) == 1


def test_lifecycle_payload_rejects_unbounded_content() -> None:
    with pytest.raises(ValidationError):
        Agent99TelegramLifecycleRequest.model_validate(
            {**payload(), "raw_log": "must-not-enter-lifecycle-ingress"}
        )
    with pytest.raises(ValidationError):
        Agent99TelegramLifecycleRequest.model_validate(
            {**payload(), "incident_id": r"C:\Wooo\Agent99\evidence\raw.json"}
        )
    with pytest.raises(ValidationError):
        Agent99TelegramLifecycleRequest.model_validate(
            {**payload(), "project_id": "unregistered-product"}
        )
    with pytest.raises(ValidationError):
        Agent99TelegramLifecycleRequest.model_validate(
            {
                **payload(),
                "visual": {**visual_payload(), "sha256": "0" * 64},
            }
        )
    with pytest.raises(ValidationError, match="reconcile_identity_mismatch"):
        Agent99TelegramLifecycleRequest.model_validate(
            {
                **payload(),
                "reconcile_receipt": {
                    "schema_version": (
                        "agent99_telegram_lifecycle_reconcile_v1"
                    ),
                    "delivery_id": payload()["delivery_id"],
                    "incident_id": "INC-WRONG",
                    "state_key": payload()["state_key"],
                    "provider_message_id": "8123",
                    "destination_binding": "a" * 64,
                },
            }
        )
    with pytest.raises(ValidationError):
        Agent99TelegramLifecycleRequest.model_validate(
            {**payload(), "closure_status": "resolved_without_receipt"}
        )


def test_lifecycle_card_exposes_agent99_stage_and_identity_receipts() -> None:
    request = Agent99TelegramLifecycleRequest.model_validate(
        {
            **payload(),
            "check_status": "failed",
            "apply_status": "passed",
            "verifier_status": "failed",
            "closure_status": "failed",
        }
    )

    card = lifecycle_service._format_lifecycle_card(request)
    caption = lifecycle_service._format_lifecycle_caption(request)

    assert "<b>AI Agent 執行鏈</b>" in card
    assert "Received：完成 / PASSED" in card
    assert "Check：失敗 / FAILED" in card
    assert "Apply：完成 / PASSED" in card
    assert "Verifier：失敗 / FAILED" in card
    assert "Closure：失敗 / FAILED" in card
    assert "<code>run-agent99-20260715-01</code>" in card
    assert "<code>trace-agent99-20260715-01</code>" in card
    assert "<code>work-item-agent99-20260715-01</code>" in card
    assert "<code>Agent99:Recover</code>" in card
    assert "<code>agent99-recover-post-condition-v1</code>" in card
    assert "Received:PASSED" in caption
    assert "Check:FAILED" in caption
    assert "Closure:FAILED" in caption
    assert "Trace:trace-agent99-20260715-01" in caption


def test_lifecycle_card_missing_stage_evidence_stays_visibly_pending() -> None:
    optional_receipt_fields = {
        "run_id",
        "trace_id",
        "work_item_id",
        "executor_name",
        "verifier_name",
        "received_status",
        "check_status",
        "apply_status",
        "verifier_status",
        "closure_status",
    }
    request = Agent99TelegramLifecycleRequest.model_validate(
        {
            key: value
            for key, value in payload().items()
            if key not in optional_receipt_fields
        }
    )

    card = lifecycle_service._format_lifecycle_card(request)
    caption = lifecycle_service._format_lifecycle_caption(request)

    assert card.count("待回報 / PENDING") >= 5
    assert card.count("未提供 / PENDING") >= 5
    assert "Received:PENDING" in caption
    assert "Closure:PENDING" in caption
    assert "ID｜Run:未提供 / PENDING" in caption


def test_visual_caption_stays_within_telegram_limit_without_broken_html() -> None:
    request = Agent99TelegramLifecycleRequest.model_validate(
        {
            **payload(),
            "title": "&" * 120,
            "target": "&" * 120,
            "action": "&" * 360,
            "verification": "&" * 300,
            "duration_text": "&" * 80,
        }
    )

    caption = lifecycle_service._format_lifecycle_caption(request)

    assert len(caption) <= 1024
    assert caption.count("<b>") == caption.count("</b>") == 2
    assert caption.count("<code>") == caption.count("</code>") == 1
    assert "&amp" not in caption.replace("&amp;", "")


def test_lifecycle_endpoint_rejects_wrong_token(monkeypatch) -> None:
    monkeypatch.setattr(settings, "AGENT99_SRE_ALERT_RELAY_TOKEN", "expected")

    response = app_client().post(
        "/api/v1/agents/agent99/telegram-lifecycle",
        headers={"X-Agent99-Lifecycle-Token": "wrong"},
        json=payload(),
    )

    assert response.status_code == 401
    assert response.json()["detail"] == "agent99_lifecycle_auth_failed"


def test_lifecycle_endpoint_returns_structured_pending_receipt(monkeypatch) -> None:
    monkeypatch.setattr(settings, "AGENT99_SRE_ALERT_RELAY_TOKEN", "expected")

    async def failed_delivery(request):
        return {
            "ok": False,
            "delivery_id": request.delivery_id,
            "delivery_status": "pending_unknown",
        }

    monkeypatch.setattr(
        agents_api,
        "deliver_agent99_telegram_lifecycle",
        failed_delivery,
    )
    response = app_client().post(
        "/api/v1/agents/agent99/telegram-lifecycle",
        headers={"X-Agent99-Lifecycle-Token": "expected"},
        json=payload(),
    )

    assert response.status_code == 202
    assert response.headers["retry-after"] == "2"
    assert response.json()["ok"] is False
    assert response.json()["delivery_status"] == "pending_unknown"
    assert response.json()["delivery_id"] == payload()["delivery_id"]


def test_lifecycle_endpoint_fails_closed_for_non_pending_delivery(monkeypatch) -> None:
    monkeypatch.setattr(settings, "AGENT99_SRE_ALERT_RELAY_TOKEN", "expected")

    async def failed_delivery(request):
        return {
            "ok": False,
            "delivery_id": request.delivery_id,
            "delivery_status": "durable_readback_failed",
        }

    monkeypatch.setattr(
        agents_api,
        "deliver_agent99_telegram_lifecycle",
        failed_delivery,
    )
    response = app_client().post(
        "/api/v1/agents/agent99/telegram-lifecycle",
        headers={"X-Agent99-Lifecycle-Token": "expected"},
        json=payload(),
    )

    assert response.status_code == 503
    assert "durable_readback_failed" in response.json()["detail"]


def test_lifecycle_endpoint_returns_same_delivery_receipt(monkeypatch) -> None:
    monkeypatch.setattr(settings, "AGENT99_SRE_ALERT_RELAY_TOKEN", "expected")

    async def successful_delivery(request):
        return {
            "schema_version": "agent99_telegram_lifecycle_receipt_v1",
            "ok": True,
            "delivery_id": request.delivery_id,
            "incident_id": request.incident_id,
            "delivery_status": "sent",
            "provider_message_id": "8123",
            "durable_outbound_acknowledged": True,
        }

    monkeypatch.setattr(
        agents_api,
        "deliver_agent99_telegram_lifecycle",
        successful_delivery,
    )
    response = app_client().post(
        "/api/v1/agents/agent99/telegram-lifecycle",
        headers={"X-Agent99-Lifecycle-Token": "expected"},
        json=payload(),
    )

    assert response.status_code == 200
    assert response.json()["delivery_id"] == payload()["delivery_id"]
    assert response.json()["provider_message_id"] == "8123"


def test_lifecycle_endpoint_202_retry_is_db_only_and_returns_200(
    monkeypatch,
) -> None:
    monkeypatch.setattr(settings, "AGENT99_SRE_ALERT_RELAY_TOKEN", "expected")
    destination_binding = "a" * 64
    calls = {"deliver": 0, "reconcile": 0}

    async def pending_delivery(request):
        calls["deliver"] += 1
        assert request.reconcile_receipt is None
        return {
            "schema_version": "agent99_telegram_lifecycle_receipt_v1",
            "ok": False,
            "delivery_id": request.delivery_id,
            "incident_id": request.incident_id,
            "delivery_status": "pending_unknown",
            "provider_message_id": "8124",
            "destination_binding": destination_binding,
            "provider_send_performed": True,
            "reconcile_receipt": {
                "schema_version": "agent99_telegram_lifecycle_reconcile_v1",
                "delivery_id": request.delivery_id,
                "incident_id": request.incident_id,
                "state_key": request.state_key,
                "provider_message_id": "8124",
                "destination_binding": destination_binding,
            },
        }

    async def durable_reconcile(request):
        calls["reconcile"] += 1
        assert request.reconcile_receipt is not None
        assert request.reconcile_receipt.provider_message_id == "8124"
        assert (
            request.reconcile_receipt.destination_binding
            == destination_binding
        )
        return {
            "schema_version": "agent99_telegram_lifecycle_receipt_v1",
            "ok": True,
            "delivery_id": request.delivery_id,
            "incident_id": request.incident_id,
            "delivery_status": "sent_reused",
            "provider_message_id": "8124",
            "destination_binding": destination_binding,
            "durable_outbound_acknowledged": True,
            "destination_binding_verified": True,
            "provider_send_performed": False,
        }

    monkeypatch.setattr(
        agents_api,
        "deliver_agent99_telegram_lifecycle",
        pending_delivery,
    )
    monkeypatch.setattr(
        agents_api,
        "reconcile_agent99_telegram_lifecycle",
        durable_reconcile,
    )
    client = app_client()
    first = client.post(
        "/api/v1/agents/agent99/telegram-lifecycle",
        headers={"X-Agent99-Lifecycle-Token": "expected"},
        json=payload(),
    )

    assert first.status_code == 202
    assert first.headers["retry-after"] == "2"
    retry_payload = {
        **payload(),
        "reconcile_receipt": first.json()["reconcile_receipt"],
    }
    second = client.post(
        "/api/v1/agents/agent99/telegram-lifecycle",
        headers={"X-Agent99-Lifecycle-Token": "expected"},
        json=retry_payload,
    )

    assert second.status_code == 200
    assert second.json()["delivery_status"] == "sent_reused"
    assert second.json()["provider_send_performed"] is False
    assert calls == {"deliver": 1, "reconcile": 1}


def test_lifecycle_outbox_identity_is_stable_and_separate() -> None:
    base = {
        "callback_reply": {
            "project_id": "awoooi",
            "automation_run_id": "agent99-lifecycle-0123456789",
            "incident_id": "INC-20260715-AG9901",
            "apply_op_id": "investigating|warning",
        }
    }
    lifecycle_source = {
        "callback_reply": {
            **base["callback_reply"],
            "action": "agent99_lifecycle",
        }
    }
    controlled_source = {
        "callback_reply": {
            **base["callback_reply"],
            "action": "controlled_apply_result",
        }
    }

    lifecycle_identity = gateway_service._controlled_apply_result_delivery_identity(
        lifecycle_source
    )
    controlled_identity = gateway_service._controlled_apply_result_delivery_identity(
        controlled_source
    )

    assert lifecycle_identity is not None
    assert controlled_identity is not None
    assert lifecycle_identity["delivery_kind"] == "agent99_lifecycle"
    assert controlled_identity["delivery_kind"] == "controlled_apply_result"
    assert gateway_service._controlled_apply_result_delivery_run_id(
        lifecycle_identity
    ) == gateway_service._controlled_apply_result_delivery_run_id(lifecycle_identity)
    assert gateway_service._controlled_apply_result_delivery_run_id(
        lifecycle_identity
    ) != gateway_service._controlled_apply_result_delivery_run_id(controlled_identity)


def test_lifecycle_outbox_persists_destination_and_visual_bindings() -> None:
    destination_binding = "a" * 64

    envelope = gateway_service._merge_outbound_source_envelope_extra(
        {"adapter": "legacy_telegram_gateway"},
        {
            "canonical_route_receipt": {
                "destination_binding": destination_binding,
            },
            "visual_delivery": {
                "requested": True,
                "raw_visual_stored": False,
            },
        },
    )

    assert envelope["canonical_route_receipt"] == {
        "destination_binding": destination_binding,
    }
    assert envelope["visual_delivery"] == {
        "requested": True,
        "raw_visual_stored": False,
    }


@pytest.mark.asyncio
async def test_lifecycle_durable_readback_binds_exact_identity_without_egress(
    monkeypatch,
) -> None:
    gateway = object.__new__(gateway_service.TelegramGateway)
    destination_binding = "a" * 64
    reconcile = AsyncMock(
        return_value={
            "status": "sent",
            "provider_message_id": "8124",
            "destination_binding": destination_binding,
            "visual_requested": True,
        }
    )
    monkeypatch.setattr(
        gateway,
        "_reconcile_controlled_apply_result_outbound",
        reconcile,
    )

    result = await gateway.reconcile_agent99_lifecycle_receipt(
        delivery_id=payload()["delivery_id"],
        incident_id=payload()["incident_id"],
        state_key=payload()["state_key"],
        expected_provider_message_id="8124",
        expected_destination_binding=destination_binding,
    )

    assert result["ok"] is True
    assert result["_awooop_delivery_status"] == "sent_reused"
    assert result["_awooop_provider_send_performed"] is False
    assert result["_awooop_durable_readback_performed"] is True
    assert result["_awooop_delivery_context"] == {
        "destination_binding": destination_binding,
        "destination_binding_verified": True,
        "durable_exact_row_readback": True,
    }
    reconcile.assert_awaited_once_with(
        identity={
            "project_id": "awoooi",
            "automation_run_id": payload()["delivery_id"],
            "incident_id": payload()["incident_id"],
            "apply_op_id": payload()["state_key"],
            "delivery_kind": "agent99_lifecycle",
        },
        expected_provider_message_id="8124",
        expected_destination_binding=destination_binding,
    )


@pytest.mark.asyncio
async def test_lifecycle_reconcile_destination_mismatch_never_finalizes(
    monkeypatch,
) -> None:
    gateway = object.__new__(gateway_service.TelegramGateway)
    readback = AsyncMock(
        return_value={
            "status": "pending_unknown",
            "run_id": "00000000-0000-0000-0000-000000000201",
            "message_id": "00000000-0000-0000-0000-000000000202",
            "destination_binding": "a" * 64,
        }
    )
    finalize = AsyncMock(return_value=True)
    monkeypatch.setattr(
        gateway,
        "_read_controlled_apply_result_outbound",
        readback,
    )
    monkeypatch.setattr(
        gateway,
        "_finalize_controlled_apply_result_outbound",
        finalize,
    )

    result = await gateway._reconcile_controlled_apply_result_outbound(
        identity={
            "project_id": "awoooi",
            "automation_run_id": payload()["delivery_id"],
            "incident_id": payload()["incident_id"],
            "apply_op_id": payload()["state_key"],
            "delivery_kind": "agent99_lifecycle",
        },
        expected_provider_message_id="8124",
        expected_destination_binding="b" * 64,
    )

    assert result["status"] == "destination_binding_mismatch"
    finalize.assert_not_awaited()


@pytest.mark.asyncio
async def test_lifecycle_reconcile_service_never_enters_provider_sender(
    monkeypatch,
) -> None:
    destination_binding = "a" * 64
    calls: list[dict] = []

    class Gateway:
        async def send_agent99_lifecycle_receipt(self, **_kwargs):
            raise AssertionError("provider sender must not run during reconcile")

        async def reconcile_agent99_lifecycle_receipt(self, **kwargs):
            calls.append(kwargs)
            return {
                "ok": True,
                "result": {"message_id": "8124"},
                "_awooop_outbound_mirror_acknowledged": True,
                "_awooop_delivery_status": "sent_reused",
                "_awooop_provider_send_performed": False,
                "_awooop_delivery_context": {
                    "destination_binding": destination_binding,
                    "destination_binding_verified": True,
                    "durable_exact_row_readback": True,
                },
            }

    monkeypatch.setattr(
        lifecycle_service,
        "get_telegram_gateway",
        lambda: Gateway(),
    )
    request = Agent99TelegramLifecycleRequest.model_validate(
        {
            **payload(),
            "reconcile_receipt": {
                "schema_version": "agent99_telegram_lifecycle_reconcile_v1",
                "delivery_id": payload()["delivery_id"],
                "incident_id": payload()["incident_id"],
                "state_key": payload()["state_key"],
                "provider_message_id": "8124",
                "destination_binding": destination_binding,
            },
        }
    )

    receipt = await lifecycle_service.reconcile_agent99_telegram_lifecycle(
        request
    )

    assert receipt["ok"] is True
    assert receipt["delivery_status"] == "sent_reused"
    assert receipt["provider_send_performed"] is False
    assert receipt["destination_binding_verified"] is True
    assert calls == [
        {
            "delivery_id": payload()["delivery_id"],
            "incident_id": payload()["incident_id"],
            "state_key": payload()["state_key"],
            "project_id": "awoooi",
            "expected_provider_message_id": "8124",
            "expected_destination_binding": destination_binding,
        }
    ]


@pytest.mark.asyncio
async def test_lifecycle_service_uses_durable_canonical_sender(monkeypatch) -> None:
    calls: list[dict] = []

    class Gateway:
        async def send_agent99_lifecycle_receipt(self, **kwargs):
            calls.append(kwargs)
            return {
                "ok": True,
                "result": {"message_id": 9123},
                "_awooop_outbound_mirror_acknowledged": True,
                "_awooop_delivery_status": "sent",
                "_awooop_provider_send_performed": True,
                "_awooop_delivery_context": {
                    "destination_binding_verified": True,
                },
            }

    monkeypatch.setattr(
        lifecycle_service,
        "get_telegram_gateway",
        lambda: Gateway(),
    )
    request = Agent99TelegramLifecycleRequest.model_validate(payload())

    receipt = await lifecycle_service.deliver_agent99_telegram_lifecycle(request)

    assert receipt["ok"] is True
    assert receipt["durable_outbound_acknowledged"] is True
    assert receipt["destination_binding_verified"] is True
    assert receipt["provider_message_id"] == "9123"
    assert calls[0]["priority"] == "P1"
    assert "<b>Agent99 事件" in calls[0]["text"]
    assert "AI 處置" in calls[0]["text"]
    assert "raw_log" not in calls[0]["text"]


@pytest.mark.asyncio
async def test_lifecycle_reconciles_pending_send_without_duplicate_provider_message(
    monkeypatch,
) -> None:
    send_calls: list[dict] = []
    reconcile_calls: list[dict] = []
    sleeps: list[float] = []
    initial_response = {
        "ok": True,
        "result": {"message_id": 9126},
        "_awooop_outbound_mirror_acknowledged": False,
        "_awooop_delivery_status": "pending_unknown",
        "_awooop_provider_send_performed": True,
        "_awooop_visual_delivery_verified": True,
        "_awooop_delivery_context": {
            "destination_binding_verified": True,
            "destination_binding": "a" * 64,
        },
    }
    reconciled_response = {
        "ok": True,
        "result": {"message_id": 9126},
        "_awooop_outbound_mirror_acknowledged": True,
        "_awooop_delivery_status": "sent_reused",
        "_awooop_provider_send_performed": False,
        "_awooop_visual_delivery_verified": True,
    }

    class Gateway:
        async def send_agent99_lifecycle_receipt(self, **kwargs):
            send_calls.append(kwargs)
            return initial_response

        async def reconcile_agent99_lifecycle_receipt(self, **kwargs):
            reconcile_calls.append(kwargs)
            return reconciled_response

    async def no_sleep(delay: float) -> None:
        sleeps.append(delay)

    monkeypatch.setattr(
        lifecycle_service,
        "get_telegram_gateway",
        lambda: Gateway(),
    )
    monkeypatch.setattr(lifecycle_service.asyncio, "sleep", no_sleep)
    request = Agent99TelegramLifecycleRequest.model_validate(
        {**payload(), "visual": visual_payload()}
    )

    receipt = await lifecycle_service.deliver_agent99_telegram_lifecycle(request)

    assert receipt["ok"] is True
    assert receipt["delivery_status"] == "sent_reused"
    assert receipt["durable_outbound_acknowledged"] is True
    assert receipt["provider_message_id"] == "9126"
    assert receipt["provider_send_performed"] is True
    assert receipt["durable_reconciliation_attempts"] == 1
    assert receipt["visual_sent"] is True
    assert sleeps == [0.25]
    assert len(send_calls) == 1
    assert len(reconcile_calls) == 1
    assert send_calls[0]["delivery_id"] == payload()["delivery_id"]
    assert reconcile_calls[0] == {
        "delivery_id": payload()["delivery_id"],
        "incident_id": payload()["incident_id"],
        "state_key": payload()["state_key"],
        "project_id": "awoooi",
        "expected_provider_message_id": "9126",
        "expected_destination_binding": "a" * 64,
    }


@pytest.mark.asyncio
async def test_lifecycle_follower_only_reads_inflight_durable_delivery(
    monkeypatch,
) -> None:
    send_calls: list[dict] = []
    reconcile_calls: list[dict] = []

    class Gateway:
        async def send_agent99_lifecycle_receipt(self, **kwargs):
            send_calls.append(kwargs)
            return {
                "ok": False,
                "_awooop_outbound_mirror_acknowledged": False,
                "_awooop_delivery_status": "pending_unknown",
                "_awooop_provider_send_performed": False,
            }

        async def reconcile_agent99_lifecycle_receipt(self, **kwargs):
            reconcile_calls.append(kwargs)
            return {
                "ok": True,
                "result": {"message_id": 9128},
                "_awooop_outbound_mirror_acknowledged": True,
                "_awooop_delivery_status": "sent_reused",
                "_awooop_provider_send_performed": False,
                "_awooop_delivery_context": {
                    "destination_binding_verified": True,
                    "durable_exact_row_readback": True,
                },
            }

    async def no_sleep(_delay: float) -> None:
        return None

    monkeypatch.setattr(
        lifecycle_service,
        "get_telegram_gateway",
        lambda: Gateway(),
    )
    monkeypatch.setattr(lifecycle_service.asyncio, "sleep", no_sleep)

    receipt = await lifecycle_service.deliver_agent99_telegram_lifecycle(
        Agent99TelegramLifecycleRequest.model_validate(payload())
    )

    assert receipt["ok"] is True
    assert receipt["provider_message_id"] == "9128"
    assert receipt["provider_send_performed"] is False
    assert len(send_calls) == 1
    assert len(reconcile_calls) == 1
    assert reconcile_calls[0]["expected_provider_message_id"] is None
    assert reconcile_calls[0]["expected_destination_binding"] is None


@pytest.mark.asyncio
async def test_lifecycle_reconciliation_fails_closed_after_bounded_readbacks(
    monkeypatch,
) -> None:
    send_calls: list[dict] = []
    reconcile_calls: list[dict] = []
    sleeps: list[float] = []

    class Gateway:
        async def send_agent99_lifecycle_receipt(self, **kwargs):
            send_calls.append(kwargs)
            return {
                "ok": True,
                "result": {"message_id": 9127},
                "_awooop_outbound_mirror_acknowledged": False,
                "_awooop_delivery_status": "pending_unknown",
                "_awooop_provider_send_performed": True,
                "_awooop_delivery_context": {
                    "destination_binding_verified": True,
                    "destination_binding": "a" * 64,
                },
            }

        async def reconcile_agent99_lifecycle_receipt(self, **kwargs):
            reconcile_calls.append(kwargs)
            return {
                "ok": False,
                "_awooop_outbound_mirror_acknowledged": False,
                "_awooop_delivery_status": "pending_unknown",
                "_awooop_provider_send_performed": False,
                "_awooop_delivery_context": {
                    "destination_binding_verified": False,
                    "durable_exact_row_readback": True,
                },
            }

    async def no_sleep(delay: float) -> None:
        sleeps.append(delay)

    monkeypatch.setattr(
        lifecycle_service,
        "get_telegram_gateway",
        lambda: Gateway(),
    )
    monkeypatch.setattr(lifecycle_service.asyncio, "sleep", no_sleep)
    request = Agent99TelegramLifecycleRequest.model_validate(payload())

    receipt = await lifecycle_service.deliver_agent99_telegram_lifecycle(request)

    assert receipt["ok"] is False
    assert receipt["delivery_status"] == "pending_unknown"
    assert receipt["durable_outbound_acknowledged"] is False
    assert receipt["provider_send_performed"] is True
    assert receipt["durable_reconciliation_attempts"] == 3
    assert sleeps == [0.25, 0.5, 1.0]
    assert len(send_calls) == 1
    assert len(reconcile_calls) == 3
    assert all(
        call["expected_provider_message_id"] == "9127"
        for call in reconcile_calls
    )
    assert all(
        call["expected_destination_binding"] == "a" * 64
        for call in reconcile_calls
    )


@pytest.mark.asyncio
async def test_visual_lifecycle_uses_one_durable_send_photo_receipt(monkeypatch) -> None:
    calls: list[dict] = []

    class Gateway:
        async def send_agent99_lifecycle_receipt(self, **kwargs):
            calls.append(kwargs)
            return {
                "ok": True,
                "result": {"message_id": 9124},
                "_awooop_outbound_mirror_acknowledged": True,
                "_awooop_delivery_status": "sent",
                "_awooop_provider_send_performed": True,
                "_awooop_visual_delivery_verified": True,
                "_awooop_delivery_context": {
                    "destination_binding_verified": True,
                },
            }

    monkeypatch.setattr(
        lifecycle_service,
        "get_telegram_gateway",
        lambda: Gateway(),
    )
    request = Agent99TelegramLifecycleRequest.model_validate(
        {**payload(), "visual": visual_payload()}
    )

    receipt = await lifecycle_service.deliver_agent99_telegram_lifecycle(request)

    assert receipt["ok"] is True
    assert receipt["visual_requested"] is True
    assert receipt["visual_sent"] is True
    assert receipt["raw_visual_stored"] is False
    assert calls[0]["visual_png"].startswith(b"\x89PNG")
    assert calls[0]["visual_sha256"] == visual_payload()["sha256"]
    assert len(calls[0]["text"]) <= 1024
    assert "Agent99：" in calls[0]["text"]


@pytest.mark.asyncio
async def test_reused_legacy_text_receipt_does_not_claim_visual_delivery(
    monkeypatch,
) -> None:
    class Gateway:
        async def send_agent99_lifecycle_receipt(self, **kwargs):
            return {
                "ok": True,
                "result": {"message_id": 7001},
                "_awooop_outbound_mirror_acknowledged": True,
                "_awooop_delivery_status": "sent_reused",
                "_awooop_provider_send_performed": False,
            }

    monkeypatch.setattr(
        lifecycle_service,
        "get_telegram_gateway",
        lambda: Gateway(),
    )
    request = Agent99TelegramLifecycleRequest.model_validate(
        {**payload(), "visual": visual_payload()}
    )

    receipt = await lifecycle_service.deliver_agent99_telegram_lifecycle(request)

    assert receipt["ok"] is True
    assert receipt["visual_requested"] is True
    assert receipt["visual_sent"] is False
    assert receipt["visual_delivery_status"] == "sent_reused"


@pytest.mark.asyncio
async def test_gateway_builds_primary_photo_with_same_send_once_identity() -> None:
    calls: list[tuple[str, dict]] = []
    gateway = object.__new__(gateway_service.TelegramGateway)

    async def capture(method: str, outgoing: dict) -> dict:
        calls.append((method, outgoing))
        return {"ok": True}

    gateway._send_request = capture
    image = Agent99TelegramLifecycleRequest.model_validate(
        {**payload(), "visual": visual_payload()}
    ).visual
    assert image is not None

    await gateway.send_agent99_lifecycle_receipt(
        delivery_id=payload()["delivery_id"],
        incident_id=payload()["incident_id"],
        state_key=payload()["state_key"],
        text="short caption",
        priority="P1",
        visual_png=image.png_bytes(),
        visual_sha256=image.sha256,
    )

    method, outgoing = calls[0]
    assert method == "sendPhoto"
    assert outgoing["photo"] == "attach://photo"
    assert outgoing["caption"] == "short caption"
    assert outgoing[gateway_service._BINARY_UPLOAD_KEY]["content"].startswith(
        b"\x89PNG"
    )
    source = outgoing[gateway_service._AWOOOP_SOURCE_ENVELOPE_EXTRA_KEY]
    assert source["callback_reply"]["automation_run_id"] == payload()["delivery_id"]
    assert source["visual_delivery"]["raw_visual_stored"] is False


@pytest.mark.asyncio
async def test_gateway_posts_visual_as_multipart_without_binary_json(monkeypatch) -> None:
    requests: list[tuple[str, dict]] = []

    class Response:
        def raise_for_status(self) -> None:
            return None

        def json(self) -> dict:
            return {
                "ok": True,
                "result": {"message_id": 9125, "chat": {"id": "-10099"}},
            }

    class Client:
        async def post(self, url: str, **kwargs):
            requests.append((url, kwargs))
            return Response()

    monkeypatch.setattr(settings, "OPENCLAW_TG_BOT_TOKEN", "test-token")
    gateway = object.__new__(gateway_service.TelegramGateway)
    gateway._initialized = True
    gateway._http_client = Client()
    binding = gateway_service._telegram_destination_binding("-10099")
    gateway._authorize_canonical_send_message = lambda **_: (
        "-10099",
        {
            "schema_version": "telegram_canonical_egress_receipt_v1",
            "decision": "allow",
            "classifier": "canonical_route",
            "product_id": "awoooi",
            "signal_family": "incident_lifecycle",
            "severity": "P1",
            "destination_alias": "telegram_chat_alias:awoooi_sre_war_room",
            "destination_binding": binding,
            "sender_bot_alias": "tsenyang_bot",
            "provider_send_performed": False,
        },
    )
    image = Agent99TelegramLifecycleRequest.model_validate(
        {**payload(), "visual": visual_payload()}
    ).visual
    assert image is not None

    receipt = await gateway._send_request(
        "sendPhoto",
        {
            "photo": "attach://photo",
            "caption": "Agent99 visual test",
            gateway_service._CANONICAL_ROUTE_CONTEXT_KEY: {
                "product_id": "awoooi",
                "signal_family": "incident_lifecycle",
                "severity": "P1",
            },
            gateway_service._BINARY_UPLOAD_KEY: {
                "field_name": "photo",
                "filename": "agent99-card.png",
                "content_type": "image/png",
                "content": image.png_bytes(),
            },
        },
    )

    assert receipt["_awooop_delivery_status"] == "sent"
    _, request_kwargs = requests[0]
    assert "json" not in request_kwargs
    assert request_kwargs["data"]["photo"] == "attach://photo"
    assert request_kwargs["files"]["photo"][0] == "agent99-card.png"
    assert request_kwargs["files"]["photo"][1].startswith(b"\x89PNG")