from __future__ import annotations

import json
import logging
import os
import threading
from datetime import datetime, timedelta, timezone
from typing import Any

import httpx
from fastapi import Depends, FastAPI, Header, HTTPException
from sqlalchemy import and_, case, inspect, or_, select, text, update
from sqlalchemy.exc import IntegrityError

from services.shared.core import Role, new_id, utc_now_iso
from services.shared.db import engine, get_session
from services.shared.models import (
    AITelegramEnqueueIn,
    EscalateRequest,
    HealthResponse,
    InteractionStatus,
    TelegramThreadAISummaryOut,
    TelegramThreadAIHandoffIn,
    TelegramThreadAIReplyIn,
    TelegramThreadEscalateIn,
    TelegramThreadMessageOut,
    TelegramThreadOut,
    TelegramThreadReplyIn,
    TelegramWebhookIn,
    TelegramWebhookOut,
)
from services.shared.security import issue_app_token, require_roles
from services.shared.sql_init import init_sql_schema
from services.shared.sql_models import (
    AISessionRow,
    Customer,
    CustomerExternalIdentity,
    Interaction,
    InteractionTimeline,
    TelegramMessageRow,
    TelegramThreadRow,
)

app = FastAPI(title="telegram-adapter-service", version="2.0.0")

init_sql_schema()

logger = logging.getLogger(__name__)

TELEGRAM_BOT_DEDUP_INDEX = "ix_telegram_messages_chat_external_unique"
TELEGRAM_REPLY_NEXT_ATTEMPT_INDEX = "ix_telegram_messages_next_delivery_attempt_at"
TELEGRAM_REPLY_LOCKED_UNTIL_INDEX = "ix_telegram_messages_delivery_locked_until"

_reply_delivery_worker_lock = threading.Lock()
_reply_delivery_worker_thread: threading.Thread | None = None
_reply_delivery_worker_stop: threading.Event | None = None
_reply_delivery_worker_wakeup: threading.Event | None = None
_reply_delivery_worker_clients = 0
_HANDOFF_REASON_UNSET = object()


def _utc_now() -> datetime:
    return datetime.now(timezone.utc).replace(microsecond=0)


def _utc_after_seconds(seconds: int) -> str:
    return (_utc_now() + timedelta(seconds=max(seconds, 0))).isoformat()


def _text_column_type() -> str:
    return "TEXT"


def _string_column_type(length: int = 64) -> str:
    return f"VARCHAR({length})"


def _integer_env(name: str, default: int) -> int:
    raw = os.getenv(name)
    if raw is None:
        return default
    try:
        return int(raw.strip())
    except ValueError:
        return default


def _ensure_telegram_message_indexes() -> None:
    inspector = inspect(engine)
    if "telegram_messages" not in inspector.get_table_names():
        return

    columns = {item["name"] for item in inspector.get_columns("telegram_messages")}
    indexes = {item["name"] for item in inspector.get_indexes("telegram_messages")}

    with engine.begin() as conn:
        if "delivery_attempts" not in columns:
            conn.execute(text("ALTER TABLE telegram_messages ADD COLUMN delivery_attempts INTEGER DEFAULT 0"))
        if "next_delivery_attempt_at" not in columns:
            conn.execute(
                text(
                    f"ALTER TABLE telegram_messages ADD COLUMN next_delivery_attempt_at {_string_column_type()}"
                )
            )
        if "delivery_locked_until" not in columns:
            conn.execute(
                text(
                    f"ALTER TABLE telegram_messages ADD COLUMN delivery_locked_until {_string_column_type()}"
                )
            )
        if "last_delivery_error" not in columns:
            conn.execute(
                text(f"ALTER TABLE telegram_messages ADD COLUMN last_delivery_error {_text_column_type()}")
            )

        duplicates = conn.execute(
            text(
                """
                SELECT chat_id, telegram_message_id_external, MIN(id) AS keep_id
                FROM telegram_messages
                WHERE telegram_message_id_external IS NOT NULL
                GROUP BY chat_id, telegram_message_id_external
                HAVING COUNT(*) > 1
                """
            )
        ).mappings()
        for row in duplicates:
            conn.execute(
                text(
                    """
                    DELETE FROM telegram_messages
                    WHERE chat_id = :chat_id
                      AND telegram_message_id_external = :external_message_id
                      AND id <> :keep_id
                    """
                ),
                {
                    "chat_id": row["chat_id"],
                    "external_message_id": row["telegram_message_id_external"],
                    "keep_id": row["keep_id"],
                },
            )
        if TELEGRAM_BOT_DEDUP_INDEX not in indexes:
            conn.execute(
                text(
                    f"""
                    CREATE UNIQUE INDEX IF NOT EXISTS {TELEGRAM_BOT_DEDUP_INDEX}
                    ON telegram_messages (chat_id, telegram_message_id_external)
                    """
                )
            )
        if TELEGRAM_REPLY_NEXT_ATTEMPT_INDEX not in indexes:
            conn.execute(
                text(
                    f"""
                    CREATE INDEX IF NOT EXISTS {TELEGRAM_REPLY_NEXT_ATTEMPT_INDEX}
                    ON telegram_messages (next_delivery_attempt_at)
                    """
                )
            )
        if TELEGRAM_REPLY_LOCKED_UNTIL_INDEX not in indexes:
            conn.execute(
                text(
                    f"""
                    CREATE INDEX IF NOT EXISTS {TELEGRAM_REPLY_LOCKED_UNTIL_INDEX}
                    ON telegram_messages (delivery_locked_until)
                    """
                )
            )


_ensure_telegram_message_indexes()


def _bool_env(name: str, default: bool) -> bool:
    raw = os.getenv(name)
    if raw is None:
        return default
    return raw.strip().lower() in {"1", "true", "yes", "on"}


def _bot_enabled() -> bool:
    return _bool_env("TELEGRAM_BOT_ENABLED", False)


def _bot_token() -> str:
    return os.getenv("TELEGRAM_BOT_TOKEN", "").strip()


def _webhook_secret() -> str:
    return os.getenv("TELEGRAM_WEBHOOK_SECRET", "").strip()


def _default_queue_id() -> str:
    return os.getenv("TELEGRAM_DEFAULT_QUEUE_ID", "q_telegram").strip() or "q_telegram"


def _bot_api_base() -> str:
    return os.getenv("TELEGRAM_BOT_API_BASE", "https://api.telegram.org").rstrip("/")


def _reply_delivery_worker_enabled() -> bool:
    return _bool_env("TELEGRAM_REPLY_DELIVERY_WORKER_ENABLED", True)


def _reply_delivery_poll_seconds() -> int:
    return max(1, _integer_env("TELEGRAM_REPLY_DELIVERY_POLL_SECONDS", 5))


def _reply_delivery_lease_seconds() -> int:
    return max(5, _integer_env("TELEGRAM_REPLY_DELIVERY_LEASE_SECONDS", 30))


def _reply_delivery_retry_base_seconds() -> int:
    return max(1, _integer_env("TELEGRAM_REPLY_DELIVERY_RETRY_BASE_SECONDS", 5))


def _reply_delivery_retry_cap_seconds() -> int:
    return max(_reply_delivery_retry_base_seconds(), _integer_env("TELEGRAM_REPLY_DELIVERY_RETRY_CAP_SECONDS", 300))


def _reply_delivery_max_attempts() -> int:
    return max(1, _integer_env("TELEGRAM_REPLY_DELIVERY_MAX_ATTEMPTS", 5))


def _interaction_service_url() -> str:
    return os.getenv("INTERACTION_SERVICE_URL", "http://localhost:8004").rstrip("/")


def _ai_service_url() -> str:
    return os.getenv("AI_ORCHESTRATOR_SERVICE_URL", "http://localhost:8017").rstrip("/")


def _sales_service_url() -> str:
    return os.getenv("SALES_SERVICE_URL", "http://localhost:8020").rstrip("/")


def _sales_telegram_sync_tenant_id() -> str | None:
    return os.getenv("SALES_TELEGRAM_SYNC_TENANT_ID", "").strip() or None


def _ai_telegram_enabled() -> bool:
    return _bool_env("AI_TELEGRAM_ENABLED", False)


def _service_auth_headers() -> dict[str, str]:
    token = issue_app_token(
        subject="svc:telegram-adapter",
        username="telegram-adapter",
        role="admin",
        auth_source="service",
        provider="telegram-adapter",
        ttl_seconds=300,
    )
    return {"Authorization": f"Bearer {token}"}


def _internal_service_headers(
    *,
    subject: str,
    username: str,
    provider: str,
    tenant_id: str | None = None,
) -> dict[str, str]:
    token = issue_app_token(
        subject=subject,
        username=username,
        role="admin",
        auth_source="service",
        provider=provider,
        tenant_id=tenant_id,
        ttl_seconds=300,
    )
    return {"Authorization": f"Bearer {token}"}


def _interaction_request(method: str, path: str, *, payload: dict | None = None) -> dict:
    with httpx.Client(timeout=5.0) as client:
        response = client.request(
            method,
            f"{_interaction_service_url()}{path}",
            json=payload,
            headers=_service_auth_headers(),
        )
    response.raise_for_status()
    return response.json()


def _sales_sync_request(payload: dict[str, Any]) -> None:
    try:
        with httpx.Client(timeout=5.0) as client:
            response = client.post(
                f"{_sales_service_url()}/internal/sales-sync/telegram",
                json=payload,
                headers=_internal_service_headers(
                    subject="svc:telegram-adapter",
                    username="telegram-adapter",
                    provider="telegram-adapter",
                    tenant_id=_sales_telegram_sync_tenant_id(),
                ),
            )
        response.raise_for_status()
    except Exception:
        logger.exception(
            "Telegram sales sync failed",
            extra={
                "thread_id": payload.get("thread_id"),
                "message_id": payload.get("message_id"),
                "direction": payload.get("direction"),
            },
        )


def _sync_sales_thread(thread: TelegramThreadRow, row: TelegramMessageRow | None = None) -> None:
    payload = {
        "thread_id": thread.thread_id,
        "chat_id": thread.chat_id,
        "interaction_id": thread.interaction_id,
        "customer_id": getattr(row, "customer_id", None),
        "phone_number": None,
        "display_name": thread.display_name,
        "queue_id": thread.queue_id,
        "status": thread.status,
        "ai_state": thread.ai_state,
        "ai_handoff_reason": thread.ai_handoff_reason,
        "message_id": getattr(row, "message_id", None),
        "external_message_id": getattr(row, "telegram_message_id_external", None),
        "text": getattr(row, "text", None),
        "direction": getattr(row, "direction", "inbound"),
        "author_type": getattr(row, "author_type", None),
        "author_id": getattr(row, "author_id", None),
        "happened_at": getattr(row, "created_at", None) or thread.updated_at,
        "metadata": {
            "telegram_user_id": thread.telegram_user_id,
            "username": thread.username,
            "claimed_by_user": thread.claimed_by_user,
            "claimed_at": thread.claimed_at,
            "ai_session_id": thread.ai_session_id,
            "last_message_at": thread.last_message_at,
            "last_message_preview": thread.last_message_preview,
        },
    }
    _sales_sync_request(payload)


def _ai_enqueue_request(thread_id: str, trigger_message_id: str | None) -> None:
    if not _ai_telegram_enabled():
        return
    try:
        with httpx.Client(timeout=10.0) as client:
            response = client.post(
                f"{_ai_service_url()}/ai/telegram/threads/{thread_id}/enqueue",
                json=AITelegramEnqueueIn(trigger_message_id=trigger_message_id).model_dump(),
                headers=_internal_service_headers(
                    subject="svc:telegram-adapter",
                    username="telegram-adapter",
                    provider="telegram-adapter",
                ),
            )
        response.raise_for_status()
    except Exception:  # noqa: BLE001
        logger.exception("Telegram AI enqueue failed", extra={"thread_id": thread_id})


def _preview_text(text: str, limit: int = 120) -> str:
    value = str(text or "").strip()
    if len(value) <= limit:
        return value
    return f"{value[: limit - 3]}..."


def _normalize_external_subject(value: str | None) -> str | None:
    raw = str(value or "").strip()
    if not raw:
        return None
    if raw.startswith("telegram:"):
        return raw.split(":", 1)[1].strip() or None
    return raw


def _first_external_subject(
    telegram_user_id: str | None,
    chat_id: str,
    explicit: str | None = None,
) -> str:
    return (
        _normalize_external_subject(telegram_user_id)
        or _normalize_external_subject(explicit)
        or _normalize_external_subject(chat_id)
        or chat_id
    )


def _customer_external_id(telegram_user_id: str | None, chat_id: str, explicit: str | None = None) -> str:
    return f"telegram:{_first_external_subject(telegram_user_id, chat_id, explicit)}"


def _customer_id_is_real(customer_id: str | None) -> bool:
    return str(customer_id or "").startswith("cus_")


def _telegram_identity_subjects(
    telegram_user_id: str | None,
    chat_id: str,
    explicit: str | None = None,
) -> list[str]:
    ordered = [
        _normalize_external_subject(telegram_user_id),
        _normalize_external_subject(explicit),
        _normalize_external_subject(chat_id),
    ]
    seen: set[str] = set()
    result: list[str] = []
    for item in ordered:
        if not item or item in seen:
            continue
        seen.add(item)
        result.append(item)
    return result


def _ensure_customer_external_identity(
    session,
    *,
    customer_id: str,
    channel: str,
    external_subject: str,
    display_name_snapshot: str | None,
    now: str,
) -> None:
    row = session.execute(
        select(CustomerExternalIdentity).where(
            CustomerExternalIdentity.channel == channel,
            CustomerExternalIdentity.external_subject == external_subject,
        )
    ).scalar_one_or_none()
    if row:
        row.customer_id = customer_id
        if display_name_snapshot:
            row.display_name_snapshot = display_name_snapshot
        row.updated_at = now
        return
    session.add(
        CustomerExternalIdentity(
            identity_id=new_id("cei"),
            customer_id=customer_id,
            channel=channel,
            external_subject=external_subject,
            display_name_snapshot=display_name_snapshot,
            created_at=now,
            updated_at=now,
        )
    )


def _resolve_or_create_customer_id(
    session,
    *,
    telegram_user_id: str | None,
    chat_id: str,
    display_name: str,
    explicit_customer_external_id: str | None = None,
) -> str:
    now = utc_now_iso()
    explicit_value = str(explicit_customer_external_id or "").strip()
    subjects = _telegram_identity_subjects(telegram_user_id, chat_id, explicit_value)

    for subject in subjects:
        identity = session.execute(
            select(CustomerExternalIdentity).where(
                CustomerExternalIdentity.channel == "telegram",
                CustomerExternalIdentity.external_subject == subject,
            )
        ).scalar_one_or_none()
        if identity:
            if display_name and identity.display_name_snapshot != display_name:
                identity.display_name_snapshot = display_name
                identity.updated_at = now
            return identity.customer_id

    if _customer_id_is_real(explicit_value):
        customer = session.execute(
            select(Customer).where(Customer.customer_id == explicit_value)
        ).scalar_one_or_none()
        if not customer:
            customer = Customer(
                customer_id=explicit_value,
                display_name=display_name,
                phones_json="[]",
                preferred_phone=None,
                tags_json=json.dumps(["telegram"], ensure_ascii=False),
                created_at=now,
            )
            session.add(customer)
        elif display_name and customer.display_name != display_name:
            customer.display_name = display_name
        customer_id = customer.customer_id
    else:
        customer = Customer(
            customer_id=new_id("cus"),
            display_name=display_name,
            phones_json="[]",
            preferred_phone=None,
            tags_json=json.dumps(["telegram"], ensure_ascii=False),
            created_at=now,
        )
        session.add(customer)
        customer_id = customer.customer_id

    for subject in subjects or [_normalize_external_subject(chat_id) or chat_id]:
        _ensure_customer_external_identity(
            session,
            customer_id=customer_id,
            channel="telegram",
            external_subject=subject,
            display_name_snapshot=display_name,
            now=now,
        )
    return customer_id


def _display_name(
    *,
    username: str | None = None,
    first_name: str | None = None,
    last_name: str | None = None,
    fallback: str | None = None,
) -> str:
    parts = [str(first_name or "").strip(), str(last_name or "").strip()]
    full_name = " ".join(part for part in parts if part).strip()
    if full_name:
        return full_name
    if username:
        return username
    if fallback:
        return fallback
    return "Telegram user"


def _inbound_author_type(direction: str) -> str:
    return "system" if direction == "system" else "customer"


def _message_can_trigger_ai(author_type: str | None) -> bool:
    return _ai_telegram_enabled() and author_type == "customer"


def _normalize_phone_number(value: str | None) -> str | None:
    raw = str(value or "").strip()
    if not raw:
        return None
    digits = "".join(ch for ch in raw if ch.isdigit())
    if not digits:
        return None
    if raw.startswith("+"):
        return f"+{digits}"
    return digits


def _contact_phone_number(contact: dict | None, *, sender_user_id: str | None) -> str | None:
    if not isinstance(contact, dict):
        return None
    phone_number = _normalize_phone_number(contact.get("phone_number"))
    if not phone_number:
        return None
    contact_user_id = str(contact.get("user_id") or "").strip() or None
    if sender_user_id and contact_user_id and contact_user_id != sender_user_id:
        return None
    return phone_number


def _customer_phone_list(raw: str | None) -> list[str]:
    try:
        value = json.loads(raw or "[]")
    except Exception:  # noqa: BLE001
        return []
    if not isinstance(value, list):
        return []
    result: list[str] = []
    for item in value:
        normalized = _normalize_phone_number(item)
        if normalized and normalized not in result:
            result.append(normalized)
    return result


def _ensure_customer_phone(session, *, customer_id: str, phone_number: str | None) -> None:
    normalized_phone = _normalize_phone_number(phone_number)
    if not normalized_phone or not _customer_id_is_real(customer_id):
        return
    session.flush()
    customer = session.execute(
        select(Customer).where(Customer.customer_id == customer_id)
    ).scalar_one_or_none()
    if not customer:
        return
    phones = _customer_phone_list(customer.phones_json)
    if normalized_phone not in phones:
        phones.append(normalized_phone)
        customer.phones_json = json.dumps(phones, ensure_ascii=False)
    if not customer.preferred_phone:
        customer.preferred_phone = normalized_phone


def _push_timeline(session, interaction_id: str, action: str, metadata: dict | None = None) -> None:
    session.add(
        InteractionTimeline(
            interaction_id=interaction_id,
            timestamp=utc_now_iso(),
            action=action,
            metadata_json=json.dumps(metadata or {}, ensure_ascii=False),
        )
    )


def _get_thread(session, thread_id: str) -> TelegramThreadRow:
    row = session.execute(
        select(TelegramThreadRow).where(TelegramThreadRow.thread_id == thread_id)
    ).scalar_one_or_none()
    if not row:
        raise HTTPException(status_code=404, detail="Telegram thread not found")
    return row


def _to_thread_out(row: TelegramThreadRow) -> TelegramThreadOut:
    return TelegramThreadOut(
        thread_id=row.thread_id,
        chat_id=row.chat_id,
        interaction_id=row.interaction_id,
        telegram_user_id=row.telegram_user_id,
        username=row.username,
        display_name=row.display_name,
        queue_id=row.queue_id,
        status=row.status,  # type: ignore[arg-type]
        claimed_by_user=row.claimed_by_user,
        claimed_at=row.claimed_at,
        ai_session_id=row.ai_session_id,
        ai_state=row.ai_state,  # type: ignore[arg-type]
        ai_handoff_reason=row.ai_handoff_reason,
        ai_last_model_at=row.ai_last_model_at,
        last_message_at=row.last_message_at,
        last_message_preview=row.last_message_preview,
        created_at=row.created_at,
        updated_at=row.updated_at,
    )


def _to_message_out(row: TelegramMessageRow) -> TelegramThreadMessageOut:
    return TelegramThreadMessageOut(
        message_id=row.message_id,
        thread_id=row.thread_id or "",
        interaction_id=row.interaction_id or "",
        chat_id=row.chat_id,
        direction=row.direction,  # type: ignore[arg-type]
        text=row.text,
        telegram_message_id_external=row.telegram_message_id_external,
        operator_user=row.operator_user,
        author_type=row.author_type,  # type: ignore[arg-type]
        author_id=row.author_id,
        delivery_status=row.delivery_status,
        payload=json.loads(row.payload_json or "{}"),
        created_at=row.created_at,
    )


def _loads_payload(raw: str | None) -> dict[str, Any]:
    try:
        payload = json.loads(raw or "{}")
    except Exception:  # noqa: BLE001
        return {}
    return payload if isinstance(payload, dict) else {}


def _to_webhook_out(row: TelegramMessageRow) -> TelegramWebhookOut:
    return TelegramWebhookOut(
        message_id=row.message_id,
        chat_id=row.chat_id,
        text=row.text,
        customer_external_id=row.customer_external_id,
        payload=_loads_payload(row.payload_json),
        thread_id=row.thread_id,
        interaction_id=row.interaction_id,
        direction=row.direction,  # type: ignore[arg-type]
        created_at=row.created_at,
    )


def _to_bot_webhook_result(row: TelegramMessageRow, *, thread_created: bool) -> dict[str, Any]:
    return {
        "ok": True,
        "thread_id": row.thread_id or "",
        "interaction_id": row.interaction_id or "",
        "message_id": row.message_id,
        "thread_created": thread_created,
    }


def _existing_bot_message(
    session,
    *,
    chat_id: str,
    external_message_id: str | None,
) -> TelegramMessageRow | None:
    if not external_message_id:
        return None
    return session.execute(
        select(TelegramMessageRow).where(
            TelegramMessageRow.chat_id == chat_id,
            TelegramMessageRow.telegram_message_id_external == external_message_id,
        )
    ).scalar_one_or_none()


def _latest_ai_session_for_thread(session, thread: TelegramThreadRow) -> AISessionRow | None:
    if thread.ai_session_id:
        ai_session = session.execute(
            select(AISessionRow).where(AISessionRow.session_id == thread.ai_session_id)
        ).scalar_one_or_none()
        if ai_session:
            return ai_session
    return session.execute(
        select(AISessionRow)
        .where(AISessionRow.thread_id == thread.thread_id)
        .order_by(AISessionRow.updated_at.desc(), AISessionRow.id.desc())
        .limit(1)
    ).scalar_one_or_none()


def _summary_customer_request_text(
    session,
    *,
    thread: TelegramThreadRow,
    ai_session: AISessionRow,
    generated_at: str,
) -> str:
    if ai_session.last_user_message_id:
        customer_message = session.execute(
            select(TelegramMessageRow).where(TelegramMessageRow.message_id == ai_session.last_user_message_id)
        ).scalar_one_or_none()
        if customer_message and customer_message.text.strip():
            return customer_message.text.strip()
    customer_message = session.execute(
        select(TelegramMessageRow)
        .where(
            TelegramMessageRow.thread_id == thread.thread_id,
            TelegramMessageRow.direction == "inbound",
            TelegramMessageRow.author_type == "customer",
            TelegramMessageRow.created_at <= generated_at,
        )
        .order_by(TelegramMessageRow.created_at.desc(), TelegramMessageRow.id.desc())
        .limit(1)
    ).scalar_one_or_none()
    if customer_message and customer_message.text.strip():
        return customer_message.text.strip()
    return "Последний запрос клиента недоступен."


def _summary_ai_outcome_text(
    session,
    *,
    thread: TelegramThreadRow,
    ai_session: AISessionRow,
    generated_at: str,
) -> str:
    if ai_session.last_ai_message_id:
        ai_message = session.execute(
            select(TelegramMessageRow).where(TelegramMessageRow.message_id == ai_session.last_ai_message_id)
        ).scalar_one_or_none()
        if ai_message and ai_message.text.strip():
            return ai_message.text.strip()
    ai_message = session.execute(
        select(TelegramMessageRow)
        .where(
            TelegramMessageRow.thread_id == thread.thread_id,
            TelegramMessageRow.direction == "outbound",
            TelegramMessageRow.author_type == "ai",
            TelegramMessageRow.created_at <= generated_at,
        )
        .order_by(TelegramMessageRow.created_at.desc(), TelegramMessageRow.id.desc())
        .limit(1)
    ).scalar_one_or_none()
    if ai_message and ai_message.text.strip():
        return ai_message.text.strip()
    return "AI передал диалог оператору без ответа клиенту."


def _summary_status(
    session,
    *,
    thread: TelegramThreadRow,
    ai_session: AISessionRow,
    generated_at: str,
) -> tuple[str, str]:
    if ai_session.last_ai_message_id:
        ai_message = session.execute(
            select(TelegramMessageRow).where(TelegramMessageRow.message_id == ai_session.last_ai_message_id)
        ).scalar_one_or_none()
        if ai_message and ai_message.text.strip():
            return ("AI ответил клиенту", "answered")
    ai_message = session.execute(
        select(TelegramMessageRow.message_id)
        .where(
            TelegramMessageRow.thread_id == thread.thread_id,
            TelegramMessageRow.direction == "outbound",
            TelegramMessageRow.author_type == "ai",
            TelegramMessageRow.created_at <= generated_at,
        )
        .order_by(TelegramMessageRow.created_at.desc(), TelegramMessageRow.id.desc())
        .limit(1)
    ).scalar_one_or_none()
    if ai_message:
        return ("AI ответил клиенту", "answered")
    return ("AI передал без ответа", "handoff")


def _summary_recommended_next_step(thread: TelegramThreadRow) -> str:
    if thread.ai_state == "handoff_required":
        return "Заберите чат и ответьте клиенту вручную."
    if thread.ai_state == "human_owned":
        return "Продолжайте диалог вручную; AI больше не отвечает в этот thread."
    return ""


def _thread_supports_ai_summary(thread: TelegramThreadRow, ai_session: AISessionRow | None) -> bool:
    if not ai_session:
        return False
    if thread.ai_state == "handoff_required":
        return True
    if thread.ai_state == "human_owned":
        return bool(thread.ai_handoff_reason or ai_session.handoff_reason)
    return False


def _build_ai_summary(session, thread: TelegramThreadRow) -> TelegramThreadAISummaryOut | None:
    ai_session = _latest_ai_session_for_thread(session, thread)
    if not _thread_supports_ai_summary(thread, ai_session):
        return None
    assert ai_session is not None
    generated_at = thread.ai_last_model_at or ai_session.updated_at or thread.updated_at
    handoff_reason = (thread.ai_handoff_reason or ai_session.handoff_reason or "").strip()
    if not handoff_reason:
        return None
    status_label, status_tone = _summary_status(
        session,
        thread=thread,
        ai_session=ai_session,
        generated_at=generated_at,
    )
    return TelegramThreadAISummaryOut(
        thread_id=thread.thread_id,
        session_id=ai_session.session_id,
        status_label=status_label,
        status_tone=status_tone,  # type: ignore[arg-type]
        customer_request_text=_summary_customer_request_text(
            session,
            thread=thread,
            ai_session=ai_session,
            generated_at=generated_at,
        ),
        ai_outcome_text=_summary_ai_outcome_text(
            session,
            thread=thread,
            ai_session=ai_session,
            generated_at=generated_at,
        ),
        handoff_reason=handoff_reason,
        recommended_next_step=_summary_recommended_next_step(thread),
        generated_at=generated_at,
    )


def _reactivate_thread_interaction(
    session,
    thread: TelegramThreadRow,
    now: str,
    *,
    requeue_ai: bool,
) -> None:
    interaction = session.execute(
        select(Interaction).where(Interaction.interaction_id == thread.interaction_id)
    ).scalar_one_or_none()
    if not interaction:
        return
    interaction.status = "new"
    interaction.assigned_to = None
    interaction.updated_at = now
    thread.status = "new"
    thread.claimed_by_user = None
    thread.claimed_at = None
    thread.ai_handoff_reason = None
    if thread.ai_state != "human_owned":
        thread.ai_state = "queued" if requeue_ai else None
    thread.updated_at = now
    _push_timeline(
        session,
        interaction.interaction_id,
        "interaction.status_changed",
        {"status": "new", "source": "telegram-reactivation"},
    )
    _push_timeline(
        session,
        interaction.interaction_id,
        "telegram.thread_reactivated",
        {"thread_id": thread.thread_id, "chat_id": thread.chat_id},
    )


def _create_interaction(session, *, subject: str, customer_id: str, queue_id: str, created_at: str) -> str:
    interaction_id = new_id("int")
    session.add(
        Interaction(
            interaction_id=interaction_id,
            channel="telegram",
            subject=_preview_text(subject, 120),
            customer_id=customer_id,
            queue_id=queue_id,
            priority=3,
            status="new",
            assigned_to=None,
            created_at=created_at,
            updated_at=created_at,
        )
    )
    _push_timeline(session, interaction_id, "interaction.created", {"channel": "telegram"})
    return interaction_id


def _persist_message(
    session,
    *,
    thread: TelegramThreadRow,
    text: str,
    payload: dict,
    direction: str,
    created_at: str,
    customer_external_id: str | None = None,
    telegram_message_id_external: str | None = None,
    operator_user: str | None = None,
    author_type: str | None = None,
    author_id: str | None = None,
    delivery_status: str | None = None,
) -> TelegramMessageRow:
    resolved_author_type = author_type or ("human" if direction == "outbound" else ("system" if direction == "system" else "customer"))
    row = TelegramMessageRow(
        message_id=new_id("tgm"),
        thread_id=thread.thread_id,
        interaction_id=thread.interaction_id,
        chat_id=thread.chat_id,
        text=text,
        customer_external_id=customer_external_id,
        direction=direction,
        telegram_message_id_external=telegram_message_id_external,
        operator_user=operator_user,
        author_type=resolved_author_type,
        author_id=author_id,
        delivery_status=delivery_status,
        payload_json=json.dumps(payload, ensure_ascii=False),
        created_at=created_at,
    )
    session.add(row)
    thread.last_message_at = created_at
    thread.last_message_preview = _preview_text(text, 140)
    thread.updated_at = created_at
    return row


def _mark_thread_ai_state(
    thread: TelegramThreadRow,
    *,
    ai_state: str | None,
    ai_handoff_reason: str | None = None,
    ai_last_model_at: str | None = None,
) -> None:
    thread.ai_state = ai_state
    thread.ai_handoff_reason = ai_handoff_reason
    if ai_last_model_at is not None:
        thread.ai_last_model_at = ai_last_model_at


def _update_ai_session_state(
    session,
    *,
    session_id: str | None,
    status: str,
    updated_at: str,
    handoff_reason: object = _HANDOFF_REASON_UNSET,
    closed: bool = False,
) -> None:
    if not session_id:
        return
    ai_session = session.execute(
        select(AISessionRow).where(AISessionRow.session_id == session_id)
    ).scalar_one_or_none()
    if not ai_session:
        return
    ai_session.status = status
    ai_session.updated_at = updated_at
    if handoff_reason is not _HANDOFF_REASON_UNSET:
        ai_session.handoff_reason = handoff_reason
    if closed:
        ai_session.closed_at = updated_at


def _maybe_enqueue_ai_for_thread(
    thread: TelegramThreadRow,
    trigger_message_id: str | None,
    *,
    author_type: str | None = None,
) -> None:
    if not _ai_telegram_enabled():
        return
    if thread.ai_state == "human_owned":
        return
    if author_type and author_type != "customer":
        return
    threading.Thread(
        target=_ai_enqueue_request,
        args=(thread.thread_id, trigger_message_id),
        daemon=True,
        name=f"telegram-ai-enqueue-{thread.thread_id}",
    ).start()


def _upsert_thread_for_inbound(
    session,
    *,
    chat_id: str,
    telegram_user_id: str | None,
    username: str | None,
    display_name: str,
    text: str,
    payload: dict,
    customer_external_id: str | None = None,
    external_message_id: str | None = None,
    phone_number: str | None = None,
    queue_id: str | None = None,
    direction: str = "inbound",
) -> tuple[TelegramThreadRow, TelegramMessageRow, bool]:
    now = utc_now_iso()
    author_type = _inbound_author_type(direction)
    ai_eligible = _message_can_trigger_ai(author_type)
    thread = session.execute(
        select(TelegramThreadRow).where(TelegramThreadRow.chat_id == chat_id)
    ).scalar_one_or_none()
    created = False
    assigned_customer_id = _resolve_or_create_customer_id(
        session,
        telegram_user_id=telegram_user_id,
        chat_id=chat_id,
        display_name=display_name,
        explicit_customer_external_id=customer_external_id,
    )
    _ensure_customer_phone(session, customer_id=assigned_customer_id, phone_number=phone_number)
    resolved_queue_id = queue_id or (thread.queue_id if thread else None) or _default_queue_id()

    if not thread:
        interaction_id = _create_interaction(
            session,
            subject=text,
            customer_id=assigned_customer_id,
            queue_id=resolved_queue_id,
            created_at=now,
        )
        thread = TelegramThreadRow(
            thread_id=new_id("tgt"),
            chat_id=chat_id,
            interaction_id=interaction_id,
            telegram_user_id=telegram_user_id,
            username=username,
            display_name=display_name,
            queue_id=resolved_queue_id,
            status="new",
            claimed_by_user=None,
            claimed_at=None,
            ai_session_id=None,
            ai_state="queued" if ai_eligible else None,
            ai_handoff_reason=None,
            ai_last_model_at=None,
            last_message_at=now,
            last_message_preview=_preview_text(text, 140),
            created_at=now,
            updated_at=now,
        )
        session.add(thread)
        created = True
    else:
        if thread.status == "closed":
            _reactivate_thread_interaction(session, thread, now, requeue_ai=ai_eligible)
        thread.telegram_user_id = telegram_user_id or thread.telegram_user_id
        thread.username = username or thread.username
        thread.display_name = display_name or thread.display_name
        thread.queue_id = thread.queue_id or resolved_queue_id
        interaction = session.execute(
            select(Interaction).where(Interaction.interaction_id == thread.interaction_id)
        ).scalar_one_or_none()
        if interaction and interaction.customer_id != assigned_customer_id:
            interaction.customer_id = assigned_customer_id
            interaction.updated_at = now
        if ai_eligible and thread.ai_state != "human_owned":
            _mark_thread_ai_state(thread, ai_state="queued", ai_handoff_reason=None)

    message_row = _persist_message(
        session,
        thread=thread,
        text=text,
        payload=payload,
        direction=direction,
        created_at=now,
        customer_external_id=assigned_customer_id,
        telegram_message_id_external=external_message_id,
        author_type=author_type,
        author_id=_first_external_subject(telegram_user_id, chat_id, customer_external_id),
        delivery_status="received",
    )
    _push_timeline(
        session,
        thread.interaction_id,
        "telegram.message_received",
        {
            "thread_id": thread.thread_id,
            "chat_id": chat_id,
            "direction": direction,
            "username": username,
        },
    )
    session.flush()
    return thread, message_row, created


def _manual_webhook_metadata(payload: TelegramWebhookIn) -> dict[str, Any]:
    raw_payload = payload.payload if isinstance(payload.payload, dict) else {}
    sender = raw_payload.get("from") if isinstance(raw_payload.get("from"), dict) else {}
    username = str(raw_payload.get("username") or sender.get("username") or "").strip() or None
    first_name = str(raw_payload.get("first_name") or sender.get("first_name") or "").strip() or None
    last_name = str(raw_payload.get("last_name") or sender.get("last_name") or "").strip() or None
    telegram_user_id = str(raw_payload.get("telegram_user_id") or sender.get("id") or "").strip() or None
    display_name = _display_name(
        username=username,
        first_name=first_name,
        last_name=last_name,
        fallback=f"Telegram {payload.chat_id}",
    )
    contact = raw_payload.get("contact") if isinstance(raw_payload.get("contact"), dict) else {}
    phone_number = _contact_phone_number(contact, sender_user_id=telegram_user_id) or _normalize_phone_number(
        raw_payload.get("phone_number")
    )
    queue_id = str(raw_payload.get("queue_id") or "").strip() or None
    return {
        "telegram_user_id": telegram_user_id,
        "username": username,
        "display_name": display_name,
        "phone_number": phone_number,
        "queue_id": queue_id,
    }


def _parse_bot_update(update: dict) -> dict:
    message = update.get("message") or update.get("edited_message")
    if not isinstance(message, dict):
        raise HTTPException(status_code=400, detail="Unsupported Telegram update")

    chat = message.get("chat") if isinstance(message.get("chat"), dict) else {}
    sender = message.get("from") if isinstance(message.get("from"), dict) else {}
    chat_id = str(chat.get("id") or "").strip()
    if not chat_id:
        raise HTTPException(status_code=400, detail="Telegram chat_id is missing")

    external_message_id = str(message.get("message_id") or "").strip() or None
    username = str(sender.get("username") or chat.get("username") or "").strip() or None
    first_name = str(sender.get("first_name") or "").strip() or None
    last_name = str(sender.get("last_name") or "").strip() or None
    telegram_user_id = str(sender.get("id") or chat.get("id") or "").strip() or None
    contact = message.get("contact") if isinstance(message.get("contact"), dict) else {}
    phone_number = _contact_phone_number(contact, sender_user_id=telegram_user_id)
    text = str(message.get("text") or message.get("caption") or "").strip()
    direction = "inbound"
    if not text:
        if phone_number:
            text = "[Shared Telegram contact]"
            direction = "system"
        else:
            content_kind = "message"
            for candidate in ("photo", "document", "audio", "video", "voice", "sticker", "location", "contact"):
                if candidate in message:
                    content_kind = candidate
                    break
            text = f"[Unsupported Telegram content: {content_kind}]"
            direction = "system"

    display_name = _display_name(
        username=username,
        first_name=first_name,
        last_name=last_name,
        fallback=str(chat.get("title") or f"Telegram {chat_id}"),
    )
    return {
        "chat_id": chat_id,
        "telegram_user_id": telegram_user_id,
        "username": username,
        "display_name": display_name,
        "text": text,
        "payload": update,
        "external_message_id": external_message_id,
        "phone_number": phone_number,
        "direction": direction,
    }


def _send_telegram_message(chat_id: str, text: str) -> dict:
    if not _bot_enabled() or not _bot_token():
        raise HTTPException(status_code=503, detail="Telegram bot sending is not configured")
    with httpx.Client(timeout=10.0) as client:
        response = client.post(
            f"{_bot_api_base()}/bot{_bot_token()}/sendMessage",
            json={"chat_id": chat_id, "text": text},
        )
    if response.status_code >= 400:
        raise HTTPException(status_code=502, detail=f"Telegram sendMessage failed: {response.text}")
    payload = response.json()
    if not payload.get("ok"):
        raise HTTPException(status_code=502, detail=f"Telegram sendMessage failed: {payload}")
    return payload


def _telegram_reply_delivery_filter(now: str):
    return or_(
        TelegramMessageRow.delivery_status == "pending",
        TelegramMessageRow.delivery_status == "retrying",
        and_(
            TelegramMessageRow.delivery_status == "sending",
            or_(
                TelegramMessageRow.delivery_locked_until.is_(None),
                TelegramMessageRow.delivery_locked_until < now,
            ),
        ),
    )


def _telegram_reply_delivery_due_filter(now: str):
    return or_(
        TelegramMessageRow.next_delivery_attempt_at.is_(None),
        TelegramMessageRow.next_delivery_attempt_at <= now,
    )


def _delivery_error_text(error: Exception) -> str:
    if isinstance(error, HTTPException):
        detail = error.detail
        if isinstance(detail, str):
            return detail[:1000]
        return str(detail)[:1000]
    return str(error or error.__class__.__name__)[:1000]


def _telegram_reply_backoff_seconds(next_attempt: int) -> int:
    base = _reply_delivery_retry_base_seconds()
    cap = _reply_delivery_retry_cap_seconds()
    return min(cap, base * (2 ** max(next_attempt - 1, 0)))


def _claim_telegram_reply_delivery(session, message_id: str) -> TelegramMessageRow | None:
    now = utc_now_iso()
    leased_until = _utc_after_seconds(_reply_delivery_lease_seconds())
    claim = session.execute(
        update(TelegramMessageRow)
        .where(
            TelegramMessageRow.message_id == message_id,
            TelegramMessageRow.direction == "outbound",
            TelegramMessageRow.thread_id.is_not(None),
            TelegramMessageRow.interaction_id.is_not(None),
            _telegram_reply_delivery_filter(now),
            _telegram_reply_delivery_due_filter(now),
        )
        .values(
            delivery_status="sending",
            delivery_locked_until=leased_until,
        )
    )
    session.commit()
    if claim.rowcount != 1:
        return None
    return session.execute(
        select(TelegramMessageRow).where(TelegramMessageRow.message_id == message_id)
    ).scalar_one_or_none()


def _mark_telegram_reply_delivery_failure(
    session,
    row: TelegramMessageRow,
    error_text: str,
    *,
    allow_retry: bool = True,
) -> None:
    attempts = int(row.delivery_attempts or 0) + 1
    row.delivery_attempts = attempts
    row.delivery_locked_until = None
    row.last_delivery_error = error_text
    should_retry = allow_retry and attempts < _reply_delivery_max_attempts()
    if should_retry:
        row.delivery_status = "retrying"
        row.next_delivery_attempt_at = _utc_after_seconds(_telegram_reply_backoff_seconds(attempts))
    else:
        row.delivery_status = "failed"
        row.next_delivery_attempt_at = None
        if row.interaction_id and row.thread_id:
            _push_timeline(
                session,
                row.interaction_id,
                "telegram.message_failed",
                {
                    "thread_id": row.thread_id,
                    "operator_user": row.operator_user,
                    "attempts": attempts,
                    "error": error_text,
                },
            )
    session.commit()


def _mark_telegram_reply_delivery_success(session, row: TelegramMessageRow, telegram_payload: dict) -> None:
    row.delivery_attempts = int(row.delivery_attempts or 0) + 1
    row.delivery_status = "sent"
    row.delivery_locked_until = None
    row.next_delivery_attempt_at = None
    row.last_delivery_error = None
    result_message = telegram_payload.get("result") if isinstance(telegram_payload, dict) else {}
    if isinstance(result_message, dict) and result_message.get("message_id") is not None:
        row.telegram_message_id_external = str(result_message.get("message_id"))
    _push_timeline(
        session,
        row.interaction_id,
        "telegram.message_sent",
        {"thread_id": row.thread_id, "operator_user": row.operator_user},
    )
    session.commit()


def _deliver_pending_telegram_reply(message_id: str) -> bool:
    session = get_session()
    try:
        row = _claim_telegram_reply_delivery(session, message_id)
        if not row or not row.thread_id or not row.interaction_id:
            return False

        thread = session.execute(
            select(TelegramThreadRow).where(TelegramThreadRow.thread_id == row.thread_id)
        ).scalar_one_or_none()
        if not thread:
            _mark_telegram_reply_delivery_failure(
                session,
                row,
                "Telegram thread not found during reply delivery",
                allow_retry=False,
            )
            return False

        try:
            telegram_payload = _send_telegram_message(row.chat_id, row.text)
        except Exception as exc:  # noqa: BLE001
            _mark_telegram_reply_delivery_failure(session, row, _delivery_error_text(exc))
            return False

        _mark_telegram_reply_delivery_success(session, row, telegram_payload)
        return True
    finally:
        session.close()


def _claim_next_due_telegram_reply_message_id(session) -> str | None:
    now = utc_now_iso()
    return session.execute(
        select(TelegramMessageRow.message_id)
        .where(
            TelegramMessageRow.direction == "outbound",
            TelegramMessageRow.thread_id.is_not(None),
            TelegramMessageRow.interaction_id.is_not(None),
            _telegram_reply_delivery_filter(now),
            _telegram_reply_delivery_due_filter(now),
        )
        .order_by(
            case(
                (TelegramMessageRow.delivery_status == "sending", 0),
                (TelegramMessageRow.delivery_status == "retrying", 1),
                else_=2,
            ),
            TelegramMessageRow.created_at.asc(),
            TelegramMessageRow.id.asc(),
        )
        .limit(1)
    ).scalar_one_or_none()


def _process_due_telegram_reply_batch(batch_size: int = 20) -> int:
    processed = 0
    attempts = 0
    while processed < batch_size and attempts < batch_size * 2:
        attempts += 1
        session = get_session()
        try:
            message_id = _claim_next_due_telegram_reply_message_id(session)
        finally:
            session.close()
        if not message_id:
            break
        if _deliver_pending_telegram_reply(message_id):
            processed += 1
    return processed


def _reply_delivery_worker_loop(stop_event: threading.Event, wakeup_event: threading.Event) -> None:
    while not stop_event.is_set():
        wakeup_event.wait(timeout=_reply_delivery_poll_seconds())
        wakeup_event.clear()
        if stop_event.is_set():
            break
        try:
            _process_due_telegram_reply_batch()
        except Exception:  # noqa: BLE001
            logger.exception("Telegram reply delivery worker cycle failed")


def _wake_telegram_reply_delivery_worker() -> None:
    with _reply_delivery_worker_lock:
        wakeup_event = _reply_delivery_worker_wakeup
    if wakeup_event:
        wakeup_event.set()


def _start_reply_delivery_worker() -> None:
    global _reply_delivery_worker_clients
    global _reply_delivery_worker_stop
    global _reply_delivery_worker_thread
    global _reply_delivery_worker_wakeup

    if not _reply_delivery_worker_enabled():
        return

    with _reply_delivery_worker_lock:
        _reply_delivery_worker_clients += 1
        if _reply_delivery_worker_thread and _reply_delivery_worker_thread.is_alive():
            wakeup_event = _reply_delivery_worker_wakeup
        else:
            stop_event = threading.Event()
            wakeup_event = threading.Event()
            worker = threading.Thread(
                target=_reply_delivery_worker_loop,
                args=(stop_event, wakeup_event),
                name="telegram-reply-delivery-worker",
                daemon=True,
            )
            _reply_delivery_worker_stop = stop_event
            _reply_delivery_worker_wakeup = wakeup_event
            _reply_delivery_worker_thread = worker
            worker.start()
    if wakeup_event:
        wakeup_event.set()


def _stop_reply_delivery_worker() -> None:
    global _reply_delivery_worker_clients
    global _reply_delivery_worker_stop
    global _reply_delivery_worker_thread
    global _reply_delivery_worker_wakeup

    thread: threading.Thread | None = None
    stop_event: threading.Event | None = None
    wakeup_event: threading.Event | None = None

    with _reply_delivery_worker_lock:
        if _reply_delivery_worker_clients > 0:
            _reply_delivery_worker_clients -= 1
        if _reply_delivery_worker_clients > 0:
            return
        thread = _reply_delivery_worker_thread
        stop_event = _reply_delivery_worker_stop
        wakeup_event = _reply_delivery_worker_wakeup
        _reply_delivery_worker_thread = None
        _reply_delivery_worker_stop = None
        _reply_delivery_worker_wakeup = None

    if stop_event:
        stop_event.set()
    if wakeup_event:
        wakeup_event.set()
    if thread and thread.is_alive():
        thread.join(timeout=2)


def _start_telegram_reply_delivery(message_id: str) -> None:
    if _reply_delivery_worker_enabled():
        _wake_telegram_reply_delivery_worker()
        return
    threading.Thread(
        target=_deliver_pending_telegram_reply,
        args=(message_id,),
        daemon=True,
    ).start()


@app.on_event("startup")
def _start_delivery_worker_on_startup() -> None:
    _start_reply_delivery_worker()


@app.on_event("shutdown")
def _stop_delivery_worker_on_shutdown() -> None:
    _stop_reply_delivery_worker()


def _ensure_operator_can_manage(
    actor: dict,
    thread: TelegramThreadRow,
    *,
    for_reply: bool = False,
    require_claim: bool = False,
) -> None:
    if actor["role"] in {Role.ADMIN.value, Role.SUPERVISOR.value}:
        return
    if actor["role"] != Role.OPERATOR.value:
        raise HTTPException(status_code=403, detail="Insufficient role")
    if for_reply and thread.claimed_by_user != actor["user"]:
        raise HTTPException(status_code=403, detail="Thread must be claimed by the current operator")
    if require_claim and thread.claimed_by_user != actor["user"]:
        raise HTTPException(status_code=403, detail="Thread must be claimed by the current operator")
    if not for_reply and thread.claimed_by_user and thread.claimed_by_user != actor["user"]:
        raise HTTPException(status_code=403, detail="Thread is already claimed by another operator")


def _raise_claim_conflict(thread: TelegramThreadRow, actor_user: str) -> None:
    if thread.status == "closed":
        raise HTTPException(status_code=409, detail="Telegram thread is closed")
    if thread.claimed_by_user and thread.claimed_by_user != actor_user:
        raise HTTPException(status_code=409, detail="Telegram thread is already claimed by another operator")
    raise HTTPException(status_code=409, detail="Telegram thread claim conflict")


def _ensure_thread_available_for_ai(thread: TelegramThreadRow) -> None:
    if thread.status == "closed":
        raise HTTPException(status_code=409, detail="Telegram thread is closed")
    if thread.ai_state == "human_owned" or thread.claimed_by_user:
        raise HTTPException(status_code=409, detail="Telegram thread is owned by a human operator")


def _claim_thread_locally(
    session,
    *,
    thread_id: str,
    actor_user: str,
    claimed_at: str,
) -> tuple[TelegramThreadRow, str, bool]:
    thread = _get_thread(session, thread_id)
    if thread.claimed_by_user == actor_user and thread.status == "in_progress":
        return thread, thread.status, False

    previous_status = thread.status
    result = session.execute(
        update(TelegramThreadRow)
        .where(
            TelegramThreadRow.thread_id == thread_id,
            TelegramThreadRow.status != "closed",
            or_(
                TelegramThreadRow.claimed_by_user.is_(None),
                TelegramThreadRow.claimed_by_user == actor_user,
            ),
        )
        .values(
            claimed_by_user=actor_user,
            claimed_at=claimed_at,
            updated_at=claimed_at,
        )
    )
    if not result.rowcount:
        session.rollback()
        _raise_claim_conflict(_get_thread(session, thread_id), actor_user)

    session.commit()
    return _get_thread(session, thread_id), previous_status, True


def _rollback_thread_claim(
    session,
    *,
    thread_id: str,
    actor_user: str,
    claimed_at: str,
    previous_status: str,
) -> None:
    try:
        session.execute(
            update(TelegramThreadRow)
            .where(
                TelegramThreadRow.thread_id == thread_id,
                TelegramThreadRow.claimed_by_user == actor_user,
                TelegramThreadRow.claimed_at == claimed_at,
            )
            .values(
                claimed_by_user=None,
                claimed_at=None,
                status=previous_status,
                updated_at=utc_now_iso(),
            )
        )
        session.commit()
    except Exception:
        session.rollback()


@app.get("/health", response_model=HealthResponse)
def health() -> HealthResponse:
    return HealthResponse(status="ok", service="telegram-adapter-service", version="v2")


@app.post("/integrations/telegram/webhook", response_model=TelegramWebhookOut)
def webhook(payload: TelegramWebhookIn) -> TelegramWebhookOut:
    session = get_session()
    try:
        metadata = _manual_webhook_metadata(payload)
        thread, row, _ = _upsert_thread_for_inbound(
            session,
            chat_id=payload.chat_id,
            telegram_user_id=metadata["telegram_user_id"],
            username=metadata["username"],
            display_name=metadata["display_name"],
            text=payload.text,
            payload=payload.payload,
            customer_external_id=payload.customer_external_id,
            phone_number=metadata["phone_number"],
            queue_id=metadata["queue_id"],
        )
        session.commit()
        session.refresh(row)
        session.refresh(thread)
        _sync_sales_thread(thread, row)
        _maybe_enqueue_ai_for_thread(thread, row.message_id, author_type=row.author_type)
        return _to_webhook_out(row)
    finally:
        session.close()


@app.post("/integrations/telegram/bot/webhook")
def bot_webhook(
    payload: dict,
    x_telegram_bot_api_secret_token: str | None = Header(
        default=None,
        alias="X-Telegram-Bot-Api-Secret-Token",
    ),
) -> dict:
    expected_secret = _webhook_secret()
    if expected_secret and x_telegram_bot_api_secret_token != expected_secret:
        raise HTTPException(status_code=403, detail="Invalid Telegram webhook secret")
    session = get_session()
    try:
        parsed = _parse_bot_update(payload)
        existing = _existing_bot_message(
            session,
            chat_id=parsed["chat_id"],
            external_message_id=parsed["external_message_id"],
        )
        if existing:
            return _to_bot_webhook_result(existing, thread_created=False)
        thread, row, created = _upsert_thread_for_inbound(
            session,
            chat_id=parsed["chat_id"],
            telegram_user_id=parsed["telegram_user_id"],
            username=parsed["username"],
            display_name=parsed["display_name"],
            text=parsed["text"],
            payload=parsed["payload"],
            external_message_id=parsed["external_message_id"],
            phone_number=parsed["phone_number"],
            direction=parsed["direction"],
        )
        try:
            session.commit()
        except IntegrityError:
            session.rollback()
            existing = _existing_bot_message(
                session,
                chat_id=parsed["chat_id"],
                external_message_id=parsed["external_message_id"],
            )
            if existing:
                return _to_bot_webhook_result(existing, thread_created=False)
            raise
        session.refresh(row)
        session.refresh(thread)
        _sync_sales_thread(thread, row)
        _maybe_enqueue_ai_for_thread(thread, row.message_id, author_type=row.author_type)
        return _to_bot_webhook_result(row, thread_created=created)
    finally:
        session.close()


@app.get(
    "/integrations/telegram/messages",
    response_model=list[TelegramWebhookOut],
)
def list_messages(
    limit: int = 100,
    chat_id: str | None = None,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> list[TelegramWebhookOut]:
    session = get_session()
    try:
        stmt = select(TelegramMessageRow).order_by(TelegramMessageRow.id.desc())
        if chat_id:
            stmt = stmt.where(TelegramMessageRow.chat_id == chat_id)
        rows = session.execute(stmt.limit(max(limit, 1))).scalars().all()
        return [_to_webhook_out(r) for r in reversed(rows)]
    finally:
        session.close()


@app.get(
    "/integrations/telegram/threads",
    response_model=list[TelegramThreadOut],
)
def list_threads(
    limit: int = 100,
    status: InteractionStatus | None = None,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> list[TelegramThreadOut]:
    session = get_session()
    try:
        stmt = select(TelegramThreadRow).order_by(TelegramThreadRow.last_message_at.desc())
        if status:
            stmt = stmt.where(TelegramThreadRow.status == status)
        rows = session.execute(stmt.limit(max(limit, 1))).scalars().all()
        return [_to_thread_out(row) for row in rows]
    finally:
        session.close()


@app.get(
    "/integrations/telegram/threads/{thread_id}",
    response_model=TelegramThreadOut,
)
def get_thread(
    thread_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> TelegramThreadOut:
    session = get_session()
    try:
        return _to_thread_out(_get_thread(session, thread_id))
    finally:
        session.close()


@app.get(
    "/integrations/telegram/threads/{thread_id}/ai-summary",
    response_model=TelegramThreadAISummaryOut | None,
)
def get_thread_ai_summary(
    thread_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> TelegramThreadAISummaryOut | None:
    session = get_session()
    try:
        thread = _get_thread(session, thread_id)
        if actor["role"] == Role.OPERATOR.value and thread.claimed_by_user and thread.claimed_by_user != actor["user"]:
            raise HTTPException(status_code=403, detail="Thread is already claimed by another operator")
        return _build_ai_summary(session, thread)
    finally:
        session.close()


@app.get(
    "/integrations/telegram/threads/{thread_id}/messages",
    response_model=list[TelegramThreadMessageOut],
)
def get_thread_messages(
    thread_id: str,
    limit: int = 200,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> list[TelegramThreadMessageOut]:
    session = get_session()
    try:
        thread = _get_thread(session, thread_id)
        rows = session.execute(
            select(TelegramMessageRow)
            .where(TelegramMessageRow.thread_id == thread.thread_id)
            .order_by(TelegramMessageRow.id.desc())
            .limit(max(limit, 1))
        ).scalars().all()
        return [_to_message_out(row) for row in reversed(rows)]
    finally:
        session.close()


@app.post(
    "/integrations/telegram/threads/{thread_id}/claim",
    response_model=TelegramThreadOut,
)
def claim_thread(
    thread_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> TelegramThreadOut:
    session = get_session()
    try:
        claimed_at = utc_now_iso()
        thread, previous_status, local_claimed = _claim_thread_locally(
            session,
            thread_id=thread_id,
            actor_user=actor["user"],
            claimed_at=claimed_at,
        )
        if not local_claimed:
            return _to_thread_out(thread)
        try:
            result = _interaction_request(
                "PATCH",
                f"/interactions/{thread.interaction_id}/assign",
                payload={"assignee": actor["user"]},
            )
        except Exception:
            _rollback_thread_claim(
                session,
                thread_id=thread_id,
                actor_user=actor["user"],
                claimed_at=claimed_at,
                previous_status=previous_status,
            )
            raise
        thread = _get_thread(session, thread_id)
        thread.status = result.get("status", "in_progress")
        thread.updated_at = utc_now_iso()
        _mark_thread_ai_state(thread, ai_state="human_owned", ai_handoff_reason=None)
        _update_ai_session_state(
            session,
            session_id=thread.ai_session_id,
            status="human_owned",
            updated_at=thread.updated_at,
        )
        _push_timeline(
            session,
            thread.interaction_id,
            "ai.human_takeover",
            {"thread_id": thread.thread_id, "actor_user": actor["user"]},
        )
        session.commit()
        session.refresh(thread)
        _sync_sales_thread(thread)
        return _to_thread_out(thread)
    finally:
        session.close()


@app.post(
    "/integrations/telegram/threads/{thread_id}/return-to-ai",
    response_model=TelegramThreadOut,
)
def return_thread_to_ai(
    thread_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> TelegramThreadOut:
    if not _ai_telegram_enabled():
        raise HTTPException(status_code=409, detail="Telegram AI is disabled")

    session = get_session()
    try:
        thread = _get_thread(session, thread_id)
        if thread.status == "closed":
            raise HTTPException(status_code=409, detail="Telegram thread is closed")
        _ensure_operator_can_manage(actor, thread, require_claim=True)

        now = utc_now_iso()
        interaction = session.execute(
            select(Interaction).where(Interaction.interaction_id == thread.interaction_id)
        ).scalar_one_or_none()
        if interaction:
            interaction.status = "new"
            interaction.assigned_to = None
            interaction.updated_at = now
            _push_timeline(
                session,
                interaction.interaction_id,
                "interaction.status_changed",
                {"status": "new", "source": "telegram-return-to-ai"},
            )

        thread.status = "new"
        thread.claimed_by_user = None
        thread.claimed_at = None
        thread.updated_at = now
        _mark_thread_ai_state(thread, ai_state="queued", ai_handoff_reason=None)
        _update_ai_session_state(
            session,
            session_id=thread.ai_session_id,
            status="active",
            updated_at=now,
            handoff_reason=None,
        )
        _push_timeline(
            session,
            thread.interaction_id,
            "ai.returned_to_ai",
            {"thread_id": thread.thread_id, "actor_user": actor["user"]},
        )
        session.commit()
        session.refresh(thread)
        _sync_sales_thread(thread)
        return _to_thread_out(thread)
    finally:
        session.close()


@app.post(
    "/integrations/telegram/threads/{thread_id}/messages",
    response_model=TelegramThreadMessageOut,
)
def reply_thread(
    thread_id: str,
    payload: TelegramThreadReplyIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> TelegramThreadMessageOut:
    session = get_session()
    try:
        thread = _get_thread(session, thread_id)
        if thread.status == "closed":
            raise HTTPException(status_code=409, detail="Telegram thread is closed")
        _ensure_operator_can_manage(actor, thread, for_reply=True)
        now = utc_now_iso()
        row = _persist_message(
            session,
            thread=thread,
            text=payload.text,
            payload={"operator_user": actor["user"]},
            direction="outbound",
            created_at=now,
            operator_user=actor["user"],
            author_type="human",
            author_id=actor["user"],
            delivery_status="pending",
        )
        row.delivery_attempts = 0
        row.next_delivery_attempt_at = now
        row.delivery_locked_until = None
        row.last_delivery_error = None
        _mark_thread_ai_state(thread, ai_state="human_owned", ai_handoff_reason=None)
        _update_ai_session_state(
            session,
            session_id=thread.ai_session_id,
            status="human_owned",
            updated_at=now,
        )
        session.commit()
        session.refresh(row)
        session.refresh(thread)
        _sync_sales_thread(thread, row)
        _start_telegram_reply_delivery(row.message_id)
        return _to_message_out(row)
    finally:
        session.close()


@app.post(
    "/integrations/telegram/threads/{thread_id}/ai/reply",
    response_model=TelegramThreadMessageOut,
)
def ai_reply_thread(
    thread_id: str,
    payload: TelegramThreadAIReplyIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR)),
) -> TelegramThreadMessageOut:
    session = get_session()
    try:
        thread = _get_thread(session, thread_id)
        _ensure_thread_available_for_ai(thread)
        now = utc_now_iso()
        row = _persist_message(
            session,
            thread=thread,
            text=payload.text,
            payload={
                "agent_profile": payload.agent_profile,
                "model": payload.model,
                "trigger_message_id": payload.trigger_message_id,
                "language": payload.language,
                "confidence": payload.confidence,
                "kb_refs": payload.kb_refs,
                **(payload.payload or {}),
            },
            direction="outbound",
            created_at=now,
            author_type="ai",
            author_id=payload.agent_profile,
            delivery_status="pending",
        )
        row.delivery_attempts = 0
        row.next_delivery_attempt_at = now
        row.delivery_locked_until = None
        row.last_delivery_error = None
        _mark_thread_ai_state(thread, ai_state="active", ai_handoff_reason=None, ai_last_model_at=now)
        _update_ai_session_state(
            session,
            session_id=thread.ai_session_id,
            status="active",
            updated_at=now,
            handoff_reason=None,
        )
        if thread.ai_session_id:
            ai_session = session.execute(
                select(AISessionRow).where(AISessionRow.session_id == thread.ai_session_id)
            ).scalar_one_or_none()
            if ai_session:
                ai_session.last_ai_message_id = row.message_id
                if payload.language:
                    ai_session.language = payload.language
        _push_timeline(
            session,
            thread.interaction_id,
            "ai.reply_generated",
            {
                "thread_id": thread.thread_id,
                "message_id": row.message_id,
                "agent_profile": payload.agent_profile,
                "model": payload.model,
                "confidence": payload.confidence,
                "kb_refs": payload.kb_refs,
                "actor_user": actor["user"],
            },
        )
        session.commit()
        session.refresh(row)
        session.refresh(thread)
        _sync_sales_thread(thread, row)
        _start_telegram_reply_delivery(row.message_id)
        return _to_message_out(row)
    finally:
        session.close()


@app.post(
    "/integrations/telegram/threads/{thread_id}/ai/handoff",
    response_model=TelegramThreadOut,
)
def ai_handoff_thread(
    thread_id: str,
    payload: TelegramThreadAIHandoffIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR)),
) -> TelegramThreadOut:
    session = get_session()
    try:
        thread = _get_thread(session, thread_id)
        _ensure_thread_available_for_ai(thread)
        now = utc_now_iso()
        _mark_thread_ai_state(
            thread,
            ai_state="handoff_required",
            ai_handoff_reason=payload.reason,
            ai_last_model_at=now,
        )
        thread.updated_at = now
        _update_ai_session_state(
            session,
            session_id=thread.ai_session_id,
            status="handoff_required",
            updated_at=now,
            handoff_reason=payload.reason,
        )
        _push_timeline(
            session,
            thread.interaction_id,
            "ai.handoff_requested",
            {
                "thread_id": thread.thread_id,
                "reason": payload.reason,
                "agent_profile": payload.agent_profile,
                "trigger_message_id": payload.trigger_message_id,
                "confidence": payload.confidence,
                "actor_user": actor["user"],
            },
        )
        session.commit()
        session.refresh(thread)
        _sync_sales_thread(thread)
        return _to_thread_out(thread)
    finally:
        session.close()


@app.post(
    "/integrations/telegram/threads/{thread_id}/close",
    response_model=TelegramThreadOut,
)
def close_thread(
    thread_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> TelegramThreadOut:
    session = get_session()
    try:
        thread = _get_thread(session, thread_id)
        _ensure_operator_can_manage(actor, thread, require_claim=True)
        _interaction_request(
            "PATCH",
            f"/interactions/{thread.interaction_id}/status",
            payload={"status": "closed"},
        )
        thread.status = "closed"
        thread.claimed_by_user = None
        thread.claimed_at = None
        thread.updated_at = utc_now_iso()
        _mark_thread_ai_state(thread, ai_state="closed", ai_handoff_reason=None, ai_last_model_at=thread.ai_last_model_at)
        _update_ai_session_state(
            session,
            session_id=thread.ai_session_id,
            status="closed",
            updated_at=thread.updated_at,
            closed=True,
        )
        session.commit()
        session.refresh(thread)
        _sync_sales_thread(thread)
        return _to_thread_out(thread)
    finally:
        session.close()


@app.post(
    "/integrations/telegram/threads/{thread_id}/escalate",
    response_model=TelegramThreadOut,
)
def escalate_thread(
    thread_id: str,
    payload: TelegramThreadEscalateIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> TelegramThreadOut:
    session = get_session()
    try:
        thread = _get_thread(session, thread_id)
        _ensure_operator_can_manage(actor, thread, require_claim=True)
        result = _interaction_request(
            "POST",
            f"/interactions/{thread.interaction_id}/escalate",
            payload=EscalateRequest(target_queue_id=payload.target_queue_id).model_dump(),
        )
        thread.status = result.get("status", "escalated")
        thread.queue_id = result.get("queue_id", payload.target_queue_id)
        thread.updated_at = utc_now_iso()
        _mark_thread_ai_state(thread, ai_state="human_owned", ai_handoff_reason=None)
        _update_ai_session_state(
            session,
            session_id=thread.ai_session_id,
            status="human_owned",
            updated_at=thread.updated_at,
        )
        session.commit()
        session.refresh(thread)
        _sync_sales_thread(thread)
        return _to_thread_out(thread)
    finally:
        session.close()
