import threading

from fastapi import HTTPException
from fastapi.testclient import TestClient
from sqlalchemy import select

from services.interaction_service.app import app as interaction_app
from services.shared.db import get_session
from services.shared.sql_models import Customer, CustomerExternalIdentity, Interaction, TelegramMessageRow, TelegramThreadRow
from services.telegram_adapter_service import app as telegram_module
from services.telegram_adapter_service.app import app as telegram_app


def admin_headers():
    return {"X-User": "admin", "X-Role": "admin"}


def operator_headers(user="operator"):
    return {"X-User": user, "X-Role": "operator"}


def patch_interaction_request(monkeypatch):
    def fake_request(method: str, path: str, *, payload: dict | None = None) -> dict:
        session = get_session()
        try:
            interaction_id = path.split("/")[2]
            interaction = session.execute(
                select(Interaction).where(Interaction.interaction_id == interaction_id)
            ).scalar_one()
            if path.endswith("/assign"):
                interaction.assigned_to = payload["assignee"]
                interaction.status = "in_progress"
            elif path.endswith("/status"):
                interaction.status = payload["status"]
            elif path.endswith("/escalate"):
                interaction.status = "escalated"
                interaction.queue_id = payload["target_queue_id"]
            interaction.updated_at = telegram_module.utc_now_iso()
            session.commit()
            return {
                "interaction_id": interaction.interaction_id,
                "status": interaction.status,
                "assigned_to": interaction.assigned_to,
                "queue_id": interaction.queue_id,
            }
        finally:
            session.close()

    monkeypatch.setattr(telegram_module, "_interaction_request", fake_request)


def test_manual_webhook_creates_thread_and_linked_interaction():
    telegram_client = TestClient(telegram_app)
    interaction_client = TestClient(interaction_app)

    response = telegram_client.post(
        "/integrations/telegram/webhook",
        json={
            "chat_id": "chat_1",
            "text": "Привет из Telegram",
            "customer_external_id": "telegram:user_1",
            "payload": {
                "username": "lead_one",
                "first_name": "Lead",
                "last_name": "One",
                "telegram_user_id": "tg_user_1",
            },
        },
    )
    assert response.status_code == 200
    payload = response.json()
    assert payload["thread_id"].startswith("tgt_")
    assert payload["interaction_id"].startswith("int_")
    assert payload["direction"] == "inbound"

    thread_list = telegram_client.get("/integrations/telegram/threads", headers=admin_headers())
    assert thread_list.status_code == 200
    listed_thread = next(item for item in thread_list.json() if item["thread_id"] == payload["thread_id"])
    assert listed_thread["chat_id"] == "chat_1"
    assert listed_thread["interaction_id"] == payload["interaction_id"]

    interaction = interaction_client.get(f"/interactions/{payload['interaction_id']}")
    assert interaction.status_code == 200
    assert interaction.json()["channel"] == "telegram"
    assert interaction.json()["customer_id"].startswith("cus_")

    session = get_session()
    try:
        identity = session.execute(
            select(CustomerExternalIdentity).where(
                CustomerExternalIdentity.channel == "telegram",
                CustomerExternalIdentity.external_subject == "tg_user_1",
            )
        ).scalar_one_or_none()
        assert identity is not None
        assert identity.customer_id == interaction.json()["customer_id"]
    finally:
        session.close()


def test_repeated_inbound_reuses_same_thread_and_interaction():
    telegram_client = TestClient(telegram_app)

    first = telegram_client.post(
        "/integrations/telegram/webhook",
        json={"chat_id": "chat_reuse", "text": "Первое", "payload": {"username": "reuse_user"}},
    )
    second = telegram_client.post(
        "/integrations/telegram/webhook",
        json={"chat_id": "chat_reuse", "text": "Второе", "payload": {"username": "reuse_user"}},
    )

    assert first.status_code == 200
    assert second.status_code == 200
    assert first.json()["thread_id"] == second.json()["thread_id"]
    assert first.json()["interaction_id"] == second.json()["interaction_id"]


def test_list_messages_requires_roles():
    telegram_client = TestClient(telegram_app)
    created = telegram_client.post(
        "/integrations/telegram/webhook",
        json={"chat_id": "chat_messages_auth", "text": "Auth check"},
    )
    assert created.status_code == 200

    anonymous = telegram_client.get(
        "/integrations/telegram/messages",
        params={"chat_id": "chat_messages_auth"},
    )
    assert anonymous.status_code == 403

    admin = telegram_client.get(
        "/integrations/telegram/messages",
        params={"chat_id": "chat_messages_auth"},
        headers=admin_headers(),
    )
    assert admin.status_code == 200
    assert len(admin.json()) == 1

    operator = telegram_client.get(
        "/integrations/telegram/messages",
        params={"chat_id": "chat_messages_auth"},
        headers=operator_headers(),
    )
    assert operator.status_code == 200
    assert len(operator.json()) == 1


def test_closed_interaction_is_reactivated_on_new_inbound():
    telegram_client = TestClient(telegram_app)
    interaction_client = TestClient(interaction_app)

    created = telegram_client.post(
        "/integrations/telegram/webhook",
        json={"chat_id": "chat_reactivate", "text": "Первое сообщение"},
    )
    assert created.status_code == 200
    interaction_id = created.json()["interaction_id"]
    thread_id = created.json()["thread_id"]

    session = get_session()
    try:
        thread = session.execute(select(TelegramThreadRow).where(TelegramThreadRow.thread_id == thread_id)).scalar_one()
        interaction = session.execute(select(Interaction).where(Interaction.interaction_id == interaction_id)).scalar_one()
        thread.status = "closed"
        interaction.status = "closed"
        session.commit()
    finally:
        session.close()

    reopened = telegram_client.post(
        "/integrations/telegram/webhook",
        json={"chat_id": "chat_reactivate", "text": "Новое сообщение после close"},
    )
    assert reopened.status_code == 200
    assert reopened.json()["thread_id"] == thread_id
    assert reopened.json()["interaction_id"] == interaction_id

    interaction = interaction_client.get(f"/interactions/{interaction_id}")
    assert interaction.status_code == 200
    assert interaction.json()["status"] == "new"


def test_operator_can_claim_thread_and_foreign_operator_cannot_reply(monkeypatch):
    patch_interaction_request(monkeypatch)
    telegram_client = TestClient(telegram_app)

    created = telegram_client.post(
        "/integrations/telegram/webhook",
        json={"chat_id": "chat_claim", "text": "Нужна помощь"},
    )
    thread_id = created.json()["thread_id"]

    claimed = telegram_client.post(
        f"/integrations/telegram/threads/{thread_id}/claim",
        headers=operator_headers("operator_a"),
    )
    assert claimed.status_code == 200
    assert claimed.json()["claimed_by_user"] == "operator_a"
    assert claimed.json()["status"] == "in_progress"

    forbidden = telegram_client.post(
        f"/integrations/telegram/threads/{thread_id}/messages",
        headers=operator_headers("operator_b"),
        json={"text": "Чужой reply"},
    )
    assert forbidden.status_code == 403


def test_operator_reply_sends_outbound_message_and_persists_row(monkeypatch):
    patch_interaction_request(monkeypatch)
    started: list[str] = []
    monkeypatch.setattr(telegram_module, "_start_telegram_reply_delivery", lambda message_id: started.append(message_id))

    telegram_client = TestClient(telegram_app)
    created = telegram_client.post(
        "/integrations/telegram/webhook",
        json={"chat_id": "chat_reply", "text": "Вопрос клиента"},
    )
    thread_id = created.json()["thread_id"]

    claimed = telegram_client.post(
        f"/integrations/telegram/threads/{thread_id}/claim",
        headers=operator_headers("operator"),
    )
    assert claimed.status_code == 200

    sent = telegram_client.post(
        f"/integrations/telegram/threads/{thread_id}/messages",
        headers=operator_headers("operator"),
        json={"text": "Ответ оператора"},
    )
    assert sent.status_code == 200
    payload = sent.json()
    assert payload["direction"] == "outbound"
    assert payload["delivery_status"] == "pending"
    assert payload["telegram_message_id_external"] is None
    assert payload["operator_user"] == "operator"
    assert payload["author_type"] == "human"
    assert payload["author_id"] == "operator"
    assert started == [payload["message_id"]]

    messages = telegram_client.get(
        f"/integrations/telegram/threads/{thread_id}/messages",
        headers=admin_headers(),
    )
    assert messages.status_code == 200
    assert len(messages.json()) == 2
    assert messages.json()[-1]["delivery_status"] == "pending"
    assert messages.json()[-1]["text"] == "Ответ оператора"
    monkeypatch.setattr(
        telegram_module,
        "_send_telegram_message",
        lambda chat_id, text: {"ok": True, "result": {"message_id": 7788, "chat": {"id": chat_id}, "text": text}},
    )
    telegram_module._deliver_pending_telegram_reply(payload["message_id"])

    delivered = telegram_client.get(
        f"/integrations/telegram/threads/{thread_id}/messages",
        headers=admin_headers(),
    )
    assert delivered.status_code == 200
    assert delivered.json()[-1]["delivery_status"] == "sent"
    assert delivered.json()[-1]["telegram_message_id_external"] == "7788"


def test_returning_human_owned_thread_to_ai_reopens_ai_flow(monkeypatch):
    monkeypatch.setenv("AI_TELEGRAM_ENABLED", "1")
    enqueue_calls: list[tuple[str, str | None]] = []
    monkeypatch.setattr(
        telegram_module,
        "_ai_enqueue_request",
        lambda thread_id, trigger_message_id: enqueue_calls.append((thread_id, trigger_message_id)),
    )
    patch_interaction_request(monkeypatch)

    telegram_client = TestClient(telegram_app)
    created = telegram_client.post(
        "/integrations/telegram/webhook",
        json={
            "chat_id": "chat_return_ai",
            "text": "Первое сообщение",
            "payload": {"telegram_user_id": "tg_return_ai", "username": "return_ai_user"},
        },
    )
    assert created.status_code == 200
    thread_id = created.json()["thread_id"]
    first_message_id = created.json()["message_id"]
    enqueue_calls.clear()

    claimed = telegram_client.post(
        f"/integrations/telegram/threads/{thread_id}/claim",
        headers=operator_headers("operator_ai"),
    )
    assert claimed.status_code == 200
    assert claimed.json()["ai_state"] == "human_owned"

    returned = telegram_client.post(
        f"/integrations/telegram/threads/{thread_id}/return-to-ai",
        headers=operator_headers("operator_ai"),
    )
    assert returned.status_code == 200
    assert returned.json()["ai_state"] == "queued"
    assert returned.json()["claimed_by_user"] is None
    assert returned.json()["status"] == "new"
    assert enqueue_calls == []

    follow_up = telegram_client.post(
        "/integrations/telegram/webhook",
        json={
            "chat_id": "chat_return_ai",
            "text": "Новое сообщение после возврата AI",
            "payload": {"telegram_user_id": "tg_return_ai", "username": "return_ai_user"},
        },
    )
    assert follow_up.status_code == 200
    assert follow_up.json()["thread_id"] == thread_id
    assert follow_up.json()["message_id"] != first_message_id
    assert enqueue_calls == [(thread_id, follow_up.json()["message_id"])]

    session = get_session()
    try:
        thread = session.execute(
            select(TelegramThreadRow).where(TelegramThreadRow.thread_id == thread_id)
        ).scalar_one()
        interaction = session.execute(
            select(Interaction).where(Interaction.interaction_id == thread.interaction_id)
        ).scalar_one()
        assert thread.ai_state == "queued"
        assert thread.claimed_by_user is None
        assert thread.status == "new"
        assert interaction.status == "new"
        assert interaction.assigned_to is None
    finally:
        session.close()


def test_failed_reply_is_scheduled_for_retry_and_then_sent(monkeypatch):
    patch_interaction_request(monkeypatch)
    monkeypatch.setenv("TELEGRAM_REPLY_DELIVERY_POLL_SECONDS", "3600")
    monkeypatch.setattr(telegram_module, "_start_telegram_reply_delivery", lambda message_id: None)

    telegram_client = TestClient(telegram_app)
    created = telegram_client.post(
        "/integrations/telegram/webhook",
        json={"chat_id": "chat_retry", "text": "Retry me"},
    )
    thread_id = created.json()["thread_id"]

    claimed = telegram_client.post(
        f"/integrations/telegram/threads/{thread_id}/claim",
        headers=operator_headers("operator_retry"),
    )
    assert claimed.status_code == 200

    reply = telegram_client.post(
        f"/integrations/telegram/threads/{thread_id}/messages",
        headers=operator_headers("operator_retry"),
        json={"text": "Please retry"},
    )
    assert reply.status_code == 200
    message_id = reply.json()["message_id"]

    attempts: list[int] = []

    def flaky_send(chat_id: str, text: str) -> dict:
        attempts.append(len(attempts) + 1)
        if len(attempts) == 1:
            raise HTTPException(status_code=502, detail="temporary send failure")
        return {"ok": True, "result": {"message_id": 8899, "chat": {"id": chat_id}, "text": text}}

    monkeypatch.setattr(telegram_module, "_send_telegram_message", flaky_send)

    assert telegram_module._deliver_pending_telegram_reply(message_id) is False

    session = get_session()
    try:
        row = session.execute(
            select(TelegramMessageRow).where(TelegramMessageRow.message_id == message_id)
        ).scalar_one()
        assert row.delivery_status == "retrying"
        assert row.delivery_attempts == 1
        assert row.next_delivery_attempt_at is not None
        assert row.delivery_locked_until is None
        assert row.last_delivery_error == "temporary send failure"
        row.next_delivery_attempt_at = telegram_module.utc_now_iso()
        session.commit()
    finally:
        session.close()

    assert telegram_module._deliver_pending_telegram_reply(message_id) is True
    assert attempts == [1, 2]

    session = get_session()
    try:
        row = session.execute(
            select(TelegramMessageRow).where(TelegramMessageRow.message_id == message_id)
        ).scalar_one()
        assert row.delivery_status == "sent"
        assert row.delivery_attempts == 2
        assert row.telegram_message_id_external == "8899"
        assert row.next_delivery_attempt_at is None
        assert row.delivery_locked_until is None
        assert row.last_delivery_error is None
    finally:
        session.close()


def test_worker_batch_recovers_stale_sending_reply(monkeypatch):
    patch_interaction_request(monkeypatch)
    monkeypatch.setenv("TELEGRAM_REPLY_DELIVERY_POLL_SECONDS", "3600")
    monkeypatch.setattr(telegram_module, "_start_telegram_reply_delivery", lambda message_id: None)
    monkeypatch.setattr(
        telegram_module,
        "_send_telegram_message",
        lambda chat_id, text: {"ok": True, "result": {"message_id": 9901, "chat": {"id": chat_id}, "text": text}},
    )

    telegram_client = TestClient(telegram_app)
    created = telegram_client.post(
        "/integrations/telegram/webhook",
        json={"chat_id": "chat_stale_worker", "text": "Recover me"},
    )
    thread_id = created.json()["thread_id"]

    claimed = telegram_client.post(
        f"/integrations/telegram/threads/{thread_id}/claim",
        headers=operator_headers("operator_stale"),
    )
    assert claimed.status_code == 200

    reply = telegram_client.post(
        f"/integrations/telegram/threads/{thread_id}/messages",
        headers=operator_headers("operator_stale"),
        json={"text": "Stale sending"},
    )
    assert reply.status_code == 200
    message_id = reply.json()["message_id"]

    session = get_session()
    try:
        row = session.execute(
            select(TelegramMessageRow).where(TelegramMessageRow.message_id == message_id)
        ).scalar_one()
        row.delivery_status = "sending"
        row.delivery_locked_until = "2000-01-01T00:00:00+00:00"
        row.next_delivery_attempt_at = telegram_module.utc_now_iso()
        session.commit()
    finally:
        session.close()

    processed = telegram_module._process_due_telegram_reply_batch(batch_size=1)
    assert processed == 1

    session = get_session()
    try:
        row = session.execute(
            select(TelegramMessageRow).where(TelegramMessageRow.message_id == message_id)
        ).scalar_one()
        assert row.delivery_status == "sent"
        assert row.delivery_attempts == 1
        assert row.telegram_message_id_external == "9901"
        assert row.delivery_locked_until is None
    finally:
        session.close()


def test_bot_webhook_secret_and_unsupported_content_placeholder(monkeypatch):
    monkeypatch.setenv("TELEGRAM_WEBHOOK_SECRET", "secret-15")
    telegram_client = TestClient(telegram_app)

    forbidden = telegram_client.post(
        "/integrations/telegram/bot/webhook",
        json={"message": {"message_id": 1, "chat": {"id": 44}}},
    )
    assert forbidden.status_code == 403

    accepted = telegram_client.post(
        "/integrations/telegram/bot/webhook",
        headers={"X-Telegram-Bot-Api-Secret-Token": "secret-15"},
        json={
            "message": {
                "message_id": 9,
                "chat": {"id": 555},
                "from": {"id": 77, "username": "photo_user"},
                "photo": [{"file_id": "abc"}],
            }
        },
    )
    assert accepted.status_code == 200
    thread_id = accepted.json()["thread_id"]

    messages = telegram_client.get(
        f"/integrations/telegram/threads/{thread_id}/messages",
        headers=admin_headers(),
    )
    assert messages.status_code == 200
    assert messages.json()[0]["direction"] == "system"
    assert "[Unsupported Telegram content: photo]" in messages.json()[0]["text"]


def test_bot_webhook_unsupported_content_does_not_enqueue_ai(monkeypatch):
    monkeypatch.setenv("AI_TELEGRAM_ENABLED", "1")
    enqueue_calls: list[tuple[str, str | None]] = []
    monkeypatch.setattr(
        telegram_module,
        "_ai_enqueue_request",
        lambda thread_id, trigger_message_id: enqueue_calls.append((thread_id, trigger_message_id)),
    )
    telegram_client = TestClient(telegram_app)

    accepted = telegram_client.post(
        "/integrations/telegram/bot/webhook",
        json={
            "message": {
                "message_id": 19,
                "chat": {"id": 559},
                "from": {"id": 79, "username": "photo_user"},
                "photo": [{"file_id": "abc"}],
            }
        },
    )

    assert accepted.status_code == 200
    assert enqueue_calls == []

    session = get_session()
    try:
        thread = session.execute(
            select(TelegramThreadRow).where(TelegramThreadRow.thread_id == accepted.json()["thread_id"])
        ).scalar_one()
        assert thread.ai_state is None
        assert thread.ai_handoff_reason is None
    finally:
        session.close()


def test_bot_webhook_contact_saves_customer_phone(monkeypatch):
    monkeypatch.setenv("AI_TELEGRAM_ENABLED", "1")
    enqueue_calls: list[tuple[str, str | None]] = []
    monkeypatch.setattr(
        telegram_module,
        "_ai_enqueue_request",
        lambda thread_id, trigger_message_id: enqueue_calls.append((thread_id, trigger_message_id)),
    )
    telegram_client = TestClient(telegram_app)

    accepted = telegram_client.post(
        "/integrations/telegram/bot/webhook",
        json={
            "message": {
                "message_id": 21,
                "chat": {"id": 562},
                "from": {"id": 81, "username": "contact_user", "first_name": "Contact"},
                "contact": {"phone_number": "+7 (777) 123-45-67", "user_id": 81},
            }
        },
    )

    assert accepted.status_code == 200
    assert enqueue_calls == []

    session = get_session()
    try:
        thread = session.execute(
            select(TelegramThreadRow).where(TelegramThreadRow.thread_id == accepted.json()["thread_id"])
        ).scalar_one()
        interaction = session.execute(
            select(Interaction).where(Interaction.interaction_id == thread.interaction_id)
        ).scalar_one()
        customer = session.execute(
            select(Customer).where(Customer.customer_id == interaction.customer_id)
        ).scalar_one_or_none()
    finally:
        session.close()

    assert thread.ai_state is None
    assert customer is not None
    assert customer.preferred_phone == "+77771234567"
    assert "+77771234567" in customer.phones_json

    messages = telegram_client.get(
        f"/integrations/telegram/threads/{accepted.json()['thread_id']}/messages",
        headers=admin_headers(),
    )
    assert messages.status_code == 200
    assert messages.json()[0]["direction"] == "system"
    assert messages.json()[0]["text"] == "[Shared Telegram contact]"


def test_system_message_preserves_existing_handoff_state(monkeypatch):
    monkeypatch.setenv("AI_TELEGRAM_ENABLED", "1")
    enqueue_calls: list[tuple[str, str | None]] = []
    monkeypatch.setattr(
        telegram_module,
        "_ai_enqueue_request",
        lambda thread_id, trigger_message_id: enqueue_calls.append((thread_id, trigger_message_id)),
    )
    telegram_client = TestClient(telegram_app)

    created = telegram_client.post(
        "/integrations/telegram/webhook",
        json={"chat_id": "chat_system_handoff", "text": "Need human"},
    )
    assert created.status_code == 200
    thread_id = created.json()["thread_id"]

    session = get_session()
    try:
        thread = session.execute(select(TelegramThreadRow).where(TelegramThreadRow.thread_id == thread_id)).scalar_one()
        thread.ai_state = "handoff_required"
        thread.ai_handoff_reason = "Existing handoff"
        session.commit()
    finally:
        session.close()

    enqueue_calls.clear()
    accepted = telegram_client.post(
        "/integrations/telegram/bot/webhook",
        json={
            "message": {
                "message_id": 29,
                "chat": {"id": "chat_system_handoff"},
                "from": {"id": 93, "username": "photo_user"},
                "photo": [{"file_id": "abc"}],
            }
        },
    )
    assert accepted.status_code == 200
    assert enqueue_calls == []

    session = get_session()
    try:
        thread = session.execute(select(TelegramThreadRow).where(TelegramThreadRow.thread_id == thread_id)).scalar_one()
        assert thread.ai_state == "handoff_required"
        assert thread.ai_handoff_reason == "Existing handoff"
    finally:
        session.close()


def test_bot_webhook_duplicate_message_id_is_idempotent():
    telegram_client = TestClient(telegram_app)
    payload = {
        "update_id": 9015,
        "message": {
            "message_id": 5150,
            "chat": {"id": 55150},
            "from": {"id": 7715, "username": "duplicate_user"},
            "text": "Duplicate payload",
        },
    }

    first = telegram_client.post("/integrations/telegram/bot/webhook", json=payload)
    second = telegram_client.post("/integrations/telegram/bot/webhook", json=payload)

    assert first.status_code == 200
    assert second.status_code == 200
    assert first.json()["thread_id"] == second.json()["thread_id"]
    assert first.json()["interaction_id"] == second.json()["interaction_id"]
    assert first.json()["message_id"] == second.json()["message_id"]
    assert first.json()["thread_created"] is True
    assert second.json()["thread_created"] is False

    messages = telegram_client.get(
        f"/integrations/telegram/threads/{first.json()['thread_id']}/messages",
        headers=admin_headers(),
    )
    assert messages.status_code == 200
    assert len(messages.json()) == 1


def test_reply_after_close_is_conflict_and_close_clears_claim(monkeypatch):
    patch_interaction_request(monkeypatch)
    monkeypatch.setattr(
        telegram_module,
        "_send_telegram_message",
        lambda chat_id, text: (_ for _ in ()).throw(AssertionError("sendMessage should not be called")),
    )

    telegram_client = TestClient(telegram_app)
    created = telegram_client.post(
        "/integrations/telegram/webhook",
        json={"chat_id": "chat_close_reply", "text": "Need closure"},
    )
    thread_id = created.json()["thread_id"]

    claimed = telegram_client.post(
        f"/integrations/telegram/threads/{thread_id}/claim",
        headers=operator_headers("operator_close"),
    )
    assert claimed.status_code == 200

    closed = telegram_client.post(
        f"/integrations/telegram/threads/{thread_id}/close",
        headers=operator_headers("operator_close"),
    )
    assert closed.status_code == 200
    assert closed.json()["status"] == "closed"
    assert closed.json()["claimed_by_user"] is None
    assert closed.json()["claimed_at"] is None

    reply = telegram_client.post(
        f"/integrations/telegram/threads/{thread_id}/messages",
        headers=operator_headers("operator_close"),
        json={"text": "Should fail"},
    )
    assert reply.status_code == 409

    session = get_session()
    try:
        thread = session.execute(select(TelegramThreadRow).where(TelegramThreadRow.thread_id == thread_id)).scalar_one()
        messages = session.execute(
            select(TelegramMessageRow).where(TelegramMessageRow.thread_id == thread_id)
        ).scalars().all()
        assert thread.status == "closed"
        assert thread.claimed_by_user is None
        assert thread.claimed_at is None
        assert len(messages) == 1
    finally:
        session.close()


def test_ai_handoff_is_rejected_for_human_owned_thread(monkeypatch):
    patch_interaction_request(monkeypatch)
    telegram_client = TestClient(telegram_app)

    created = telegram_client.post(
        "/integrations/telegram/webhook",
        json={"chat_id": "chat_ai_handoff_locked", "text": "Need operator"},
    )
    assert created.status_code == 200
    thread_id = created.json()["thread_id"]

    claimed = telegram_client.post(
        f"/integrations/telegram/threads/{thread_id}/claim",
        headers=operator_headers("operator_handoff_lock"),
    )
    assert claimed.status_code == 200
    assert claimed.json()["ai_state"] == "human_owned"

    handoff = telegram_client.post(
        f"/integrations/telegram/threads/{thread_id}/ai/handoff",
        headers=admin_headers(),
        json={"reason": "Late AI handoff"},
    )
    assert handoff.status_code == 409
    assert handoff.json()["detail"] == "Telegram thread is owned by a human operator"

    session = get_session()
    try:
        thread = session.execute(select(TelegramThreadRow).where(TelegramThreadRow.thread_id == thread_id)).scalar_one()
        assert thread.ai_state == "human_owned"
        assert thread.claimed_by_user == "operator_handoff_lock"
    finally:
        session.close()


def test_claim_race_returns_single_winner(monkeypatch):
    entered_assign = threading.Event()
    release_assign = threading.Event()
    assign_calls: list[str] = []
    assign_lock = threading.Lock()

    def fake_request(method: str, path: str, *, payload: dict | None = None) -> dict:
        if path.endswith("/assign"):
            with assign_lock:
                assign_calls.append(payload["assignee"])
                if len(assign_calls) == 1:
                    entered_assign.set()
                    assert release_assign.wait(5)

        session = get_session()
        try:
            interaction_id = path.split("/")[2]
            interaction = session.execute(
                select(Interaction).where(Interaction.interaction_id == interaction_id)
            ).scalar_one()
            if path.endswith("/assign"):
                interaction.assigned_to = payload["assignee"]
                interaction.status = "in_progress"
            elif path.endswith("/status"):
                interaction.status = payload["status"]
            elif path.endswith("/escalate"):
                interaction.status = "escalated"
                interaction.queue_id = payload["target_queue_id"]
            interaction.updated_at = telegram_module.utc_now_iso()
            session.commit()
            return {
                "interaction_id": interaction.interaction_id,
                "status": interaction.status,
                "assigned_to": interaction.assigned_to,
                "queue_id": interaction.queue_id,
            }
        finally:
            session.close()

    monkeypatch.setattr(telegram_module, "_interaction_request", fake_request)

    telegram_client = TestClient(telegram_app)
    created = telegram_client.post(
        "/integrations/telegram/webhook",
        json={"chat_id": "chat_claim_race", "text": "Race me"},
    )
    thread_id = created.json()["thread_id"]

    winner: dict[str, object] = {}

    def run_first_claim() -> None:
        first_client = TestClient(telegram_app)
        try:
            winner["response"] = first_client.post(
                f"/integrations/telegram/threads/{thread_id}/claim",
                headers=operator_headers("operator_a"),
            )
        finally:
            first_client.close()

    claim_thread = threading.Thread(target=run_first_claim)
    claim_thread.start()

    assert entered_assign.wait(5)

    loser_client = TestClient(telegram_app)
    try:
        loser = loser_client.post(
            f"/integrations/telegram/threads/{thread_id}/claim",
            headers=operator_headers("operator_b"),
        )
    finally:
        loser_client.close()

    release_assign.set()
    claim_thread.join(timeout=5)

    first = winner["response"]
    assert getattr(first, "status_code") == 200
    assert loser.status_code == 409
    assert assign_calls == ["operator_a"]

    session = get_session()
    try:
        thread = session.execute(select(TelegramThreadRow).where(TelegramThreadRow.thread_id == thread_id)).scalar_one()
        assert thread.claimed_by_user == "operator_a"
        assert thread.status == "in_progress"
    finally:
        session.close()