from __future__ import annotations

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

import httpx
from fastapi import Depends, FastAPI, Header, HTTPException, Query, Request
from fastapi.responses import JSONResponse
from sqlalchemy import func, or_, select
from sqlalchemy.exc import IntegrityError

from services.shared.core import Role, new_id, utc_now_iso
from services.shared.db import get_session
from services.shared.models import HealthResponse
from services.shared.sales_models import (
    SalesAutomationTaskOut,
    SalesCallCompleteIn,
    SalesCallOut,
    SalesCallWebhookIn,
    SalesCommunicationBindExternalIn,
    SalesChannelSwitchOut,
    SalesCommunicationOut,
    SalesCommunicationStartIn,
    SalesCommunicationSummaryIn,
    SalesCommunicationSwitchChannelIn,
    SalesConditionOut,
    SalesConditionUpsertIn,
    SalesCounterpartyOut,
    SalesCounterpartyUpsertIn,
    SalesCustomerOut,
    SalesCustomerUpdate,
    SalesDashboardStageOut,
    SalesDashboardSummaryOut,
    SalesDealCloseIn,
    SalesDealCreate,
    SalesDealNextActionIn,
    SalesDealOut,
    SalesDealScenarioIn,
    SalesDealStageChangeIn,
    SalesDealUpdate,
    SalesDocumentCreate,
    SalesDocumentOut,
    SalesEscalationCreateIn,
    SalesEscalationAssignIn,
    SalesEscalationCancelIn,
    SalesEscalationOut,
    SalesEscalationResolveIn,
    SalesInvoiceCreate,
    SalesInvoiceOut,
    SalesLeadCreate,
    SalesLeadEnrichIn,
    SalesLeadOut,
    SalesLeadUpdate,
    SalesMessageOut,
    SalesMessageSendIn,
    SalesMessageWebhookIn,
    SalesNoteCreateIn,
    SalesNoteOut,
    SalesNoteUpdateIn,
    SalesOfferCreate,
    SalesOfferOut,
    SalesPaymentOut,
    SalesPaymentReconcileIn,
    SalesPaymentWebhookIn,
    SalesPipelineCreate,
    SalesPipelineOut,
    SalesPipelineStageCreate,
    SalesPipelineStageOut,
    SalesPipelineStageUpdate,
    SalesPipelineUpdate,
    SalesStageHistoryOut,
    SalesStageRefOut,
    SalesTelegramSyncIn,
    SalesTimelineEventOut,
    SalesTranscriptIn,
    SalesTranscriptOut,
    SalesVoiceSyncIn,
    SalesWorkspaceOut,
)
from services.shared.security import get_actor, issue_app_token, require_roles
from services.shared.sql_init import init_sql_schema
from services.shared.sql_models import Customer
from services.shared.sales_sql_models import (
    SalesAutomationTaskRow,
    SalesCallRow,
    SalesChannelSwitchRow,
    SalesCommunicationSessionRow,
    SalesConditionRow,
    SalesCounterpartyRow,
    SalesDealRow,
    SalesDocumentRow,
    SalesEscalationRow,
    SalesExternalLinkRow,
    SalesInvoiceRow,
    SalesLeadRow,
    SalesMessageRow,
    SalesNoteRow,
    SalesOfferRow,
    SalesPaymentRow,
    SalesPaymentWebhookEventRow,
    SalesPipelineRow,
    SalesPipelineStageRow,
    SalesStageHistoryRow,
    SalesTranscriptRow,
    TenantIntegrationRow,
    TenantSalesSettingsRow,
)
from services.sales_service import sales_events as sales_event_types
from services.sales_service.deal_state_machine import DealStateMachineError, transition_deal_stage
from services.sales_service.event_publisher import SalesEventPublisher
from services.sales_service.payment_webhooks import (
    PaymentWebhookVerificationError,
    extract_event_type,
    extract_external_event_id,
    extract_external_payment_id,
    extract_payment_provider,
    extract_provider_account_id,
    get_payment_provider_adapter,
    header_value,
    normalize_headers,
    normalize_payment_event_type,
    normalize_payment_status,
    raw_payload_hash,
    safe_json_loads,
    verify_payment_webhook_signature,
)

# Import the models module so sales tables are registered on the shared Base metadata.
from services.shared import sales_sql_models as _sales_sql_models  # noqa: F401

app = FastAPI(title="sales-service", version="1.0.0")


@app.exception_handler(DealStateMachineError)
async def _deal_state_machine_error_handler(_request, exc: DealStateMachineError) -> JSONResponse:
    return JSONResponse(status_code=exc.status_code, content=exc.response())

init_sql_schema()

DEFAULT_SALES_CURRENCY = os.getenv("DEFAULT_SALES_CURRENCY", "KZT").strip() or "KZT"
DEFAULT_PIPELINE_CODE = "default_sales"
DEFAULT_PIPELINE_NAME = "Базовая воронка продаж"
DEFAULT_STAGE_CODE = "new_qualified_lead"
LEGACY_STAGE_ALIASES = {
    "new": "new_qualified_lead",
    "active_text": "active_text_communication",
    "invoice": "invoice_sent",
    "paid": "paid",
    "won": "won",
}
DEFAULT_PIPELINE_STAGES = [
    {"code": "new_qualified_lead", "name": "Новый квалифицированный лид", "category": "entry", "sort_order": 10},
    {"code": "warm_lead", "name": "Теплый лид", "category": "entry", "sort_order": 20},
    {"code": "hot_lead", "name": "Горячий лид", "category": "entry", "sort_order": 30},
    {"code": "enrichment_required", "name": "Нужно дообогащение", "category": "entry", "sort_order": 40},
    {"code": "active_text_communication", "name": "Текстовая коммуникация", "category": "communication", "sort_order": 50},
    {"code": "active_voice_communication", "name": "Голосовая коммуникация", "category": "communication", "sort_order": 60},
    {"code": "waiting_customer_reply", "name": "Ждем ответ клиента", "category": "communication", "sort_order": 70},
    {"code": "need_clarification", "name": "Нужно уточнение", "category": "communication", "sort_order": 80},
    {"code": "need_confirmed", "name": "Потребность подтверждена", "category": "communication", "sort_order": 90},
    {"code": "offer_selection", "name": "Подбор предложения", "category": "commercial", "sort_order": 100},
    {"code": "offer_preparing", "name": "Готовим предложение", "category": "commercial", "sort_order": 110},
    {"code": "offer_sent", "name": "Предложение отправлено", "category": "commercial", "sort_order": 120},
    {"code": "conditions_negotiation", "name": "Согласование условий", "category": "commercial", "sort_order": 130},
    {"code": "counterparty_data_requested", "name": "Запрос реквизитов", "category": "paperwork", "sort_order": 140},
    {"code": "counterparty_data_received", "name": "Реквизиты получены", "category": "paperwork", "sort_order": 150},
    {"code": "document_preparing", "name": "Готовим документ", "category": "paperwork", "sort_order": 160},
    {"code": "document_sent", "name": "Документ отправлен", "category": "paperwork", "sort_order": 170},
    {"code": "document_under_review", "name": "Документ на согласовании", "category": "paperwork", "sort_order": 180},
    {"code": "document_confirmed", "name": "Документ подтвержден", "category": "paperwork", "sort_order": 190},
    {"code": "invoice_preparing", "name": "Готовим счет", "category": "finance", "sort_order": 200},
    {"code": "invoice_sent", "name": "Счет отправлен", "category": "finance", "sort_order": 210},
    {"code": "payment_expected", "name": "Ожидаем оплату", "category": "finance", "sort_order": 220},
    {"code": "partially_paid", "name": "Оплачено частично", "category": "finance", "sort_order": 230},
    {"code": "paid", "name": "Оплачено", "category": "finance", "sort_order": 240},
    {"code": "payment_overdue", "name": "Оплата просрочена", "category": "finance", "sort_order": 250},
    {"code": "won", "name": "Сделка выиграна", "category": "closing", "sort_order": 260, "is_terminal": True},
    {"code": "lost", "name": "Сделка проиграна", "category": "closing", "sort_order": 270, "is_terminal": True},
    {"code": "postponed", "name": "Отложено", "category": "closing", "sort_order": 280, "is_terminal": True},
    {"code": "follow_up_scheduled", "name": "Запланирован follow-up", "category": "closing", "sort_order": 290},
    {"code": "transferred_to_execution", "name": "Передано в исполнение", "category": "closing", "sort_order": 300, "is_terminal": True},
    {"code": "transferred_to_support", "name": "Передано человеку", "category": "closing", "sort_order": 310},
]


def _tenant_id(actor: dict) -> str:
    tenant_id = str(actor.get("tenant_id") or "").strip()
    if not tenant_id:
        raise HTTPException(status_code=400, detail="Tenant context is required")
    return tenant_id


def _optional_tenant_id(actor: dict) -> str | None:
    tenant_id = str(actor.get("tenant_id") or "").strip()
    return tenant_id or None


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


def _json_dict(raw: str | None) -> dict:
    try:
        payload = json.loads(raw or "{}")
    except json.JSONDecodeError:
        return {}
    return payload if isinstance(payload, dict) else {}


def _json_list(raw: str | None) -> list:
    try:
        payload = json.loads(raw or "[]")
    except json.JSONDecodeError:
        return []
    return payload if isinstance(payload, list) else []


def _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 _bool_env(name: str, default: bool = False) -> bool:
    raw = os.getenv(name)
    if raw is None:
        return default
    return raw.strip().lower() in {"1", "true", "yes", "on"}


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 _voice_runtime_enabled() -> bool:
    return _bool_env("SALES_VOICE_RUNTIME_ENABLED", False)


def _telegram_bridge_enabled() -> bool:
    return _bool_env("SALES_TELEGRAM_BRIDGE_ENABLED", False)


def _telegram_autostage_message_interval() -> int:
    return max(1, _integer_env("SALES_TELEGRAM_STAGE_EVERY_MESSAGES", 3))


def _voice_runtime_url() -> str:
    return os.getenv("AI_VOICE_RUNTIME_SERVICE_URL", "http://localhost:8018").rstrip("/")


def _telegram_adapter_url() -> str:
    return os.getenv("TELEGRAM_ADAPTER_SERVICE_URL", "http://localhost:8007").rstrip("/")


def _default_handoff_queue_id() -> str:
    return os.getenv("SALES_DEFAULT_HANDOFF_QUEUE_ID", "sales_line").strip() or "sales_line"


def _infer_text_channel(preferred_channel: str | None) -> str:
    return "voice" if (preferred_channel or "").strip().lower() == "voice" else "text"


def _infer_stage_for_channel(channel: str) -> str:
    return "active_voice_communication" if channel == "voice" else "active_text_communication"


def _normalize_stage_code(value: str | None) -> str:
    code = str(value or DEFAULT_STAGE_CODE).strip()
    return LEGACY_STAGE_ALIASES.get(code, code)


def _pipeline_to_out(row: SalesPipelineRow) -> SalesPipelineOut:
    return SalesPipelineOut(
        pipeline_id=row.pipeline_id,
        tenant_id=row.tenant_id,
        code=row.code,
        name=row.name,
        description=row.description,
        is_default=row.is_default,
        is_active=row.is_active,
        created_at=row.created_at,
        updated_at=row.updated_at,
    )


def _stage_to_ref(row: SalesPipelineStageRow | None) -> SalesStageRefOut | None:
    if row is None:
        return None
    return SalesStageRefOut(
        id=row.stage_id,
        code=row.code,
        name=row.name,
        category=row.category,  # type: ignore[arg-type]
        sort_order=row.sort_order,
    )


def _stage_to_out(row: SalesPipelineStageRow) -> SalesPipelineStageOut:
    return SalesPipelineStageOut(
        stage_id=row.stage_id,
        tenant_id=row.tenant_id,
        pipeline_id=row.pipeline_id,
        code=row.code,
        name=row.name,
        category=row.category,  # type: ignore[arg-type]
        sort_order=row.sort_order,
        is_terminal=row.is_terminal,
        is_system=row.is_system,
        is_active=row.is_active,
        created_at=row.created_at,
        updated_at=row.updated_at,
    )


def _get_pipeline(session, pipeline_id: str, tenant_id: str) -> SalesPipelineRow:
    row = session.execute(
        select(SalesPipelineRow).where(
            SalesPipelineRow.pipeline_id == pipeline_id,
            SalesPipelineRow.tenant_id == tenant_id,
        )
    ).scalar_one_or_none()
    if row is None:
        raise HTTPException(status_code=404, detail="Pipeline not found")
    return row


def _get_stage(session, stage_id: str, tenant_id: str, pipeline_id: str | None = None) -> SalesPipelineStageRow:
    stmt = select(SalesPipelineStageRow).where(
        SalesPipelineStageRow.stage_id == stage_id,
        SalesPipelineStageRow.tenant_id == tenant_id,
    )
    row = session.execute(stmt).scalar_one_or_none()
    if row is None:
        raise HTTPException(status_code=404, detail="Stage not found")
    if pipeline_id and row.pipeline_id != pipeline_id:
        raise HTTPException(status_code=400, detail="Stage does not belong to pipeline")
    return row


def _set_default_pipeline(session, pipeline: SalesPipelineRow) -> None:
    now = utc_now_iso()
    rows = session.execute(
        select(SalesPipelineRow).where(
            SalesPipelineRow.tenant_id == pipeline.tenant_id,
            SalesPipelineRow.is_default == True,  # noqa: E712
            SalesPipelineRow.pipeline_id != pipeline.pipeline_id,
        )
    ).scalars().all()
    for row in rows:
        row.is_default = False
        row.updated_at = now
    pipeline.is_default = True
    pipeline.is_active = True
    pipeline.updated_at = now
    settings = session.execute(
        select(TenantSalesSettingsRow).where(TenantSalesSettingsRow.tenant_id == pipeline.tenant_id)
    ).scalar_one_or_none()
    if settings is None:
        settings = TenantSalesSettingsRow(
            tenant_id=pipeline.tenant_id,
            default_pipeline_id=pipeline.pipeline_id,
            default_language="ru",
            default_currency=DEFAULT_SALES_CURRENCY,
            enabled_text_channels_json="[]",
            enabled_voice_channels_json="[]",
            payment_provider_settings_json="{}",
            document_settings_json="{}",
            created_at=now,
            updated_at=now,
        )
        session.add(settings)
    else:
        settings.default_pipeline_id = pipeline.pipeline_id
        settings.updated_at = now


def ensure_default_stages(session, tenant_id: str, pipeline_id: str) -> list[SalesPipelineStageRow]:
    now = utc_now_iso()
    existing = {
        row.code: row
        for row in session.execute(
            select(SalesPipelineStageRow).where(
                SalesPipelineStageRow.tenant_id == tenant_id,
                SalesPipelineStageRow.pipeline_id == pipeline_id,
            )
        ).scalars().all()
    }
    for spec in DEFAULT_PIPELINE_STAGES:
        if spec["code"] in existing:
            continue
        row = SalesPipelineStageRow(
            stage_id=new_id("pst"),
            tenant_id=tenant_id,
            pipeline_id=pipeline_id,
            code=spec["code"],
            name=spec["name"],
            category=spec["category"],
            sort_order=int(spec["sort_order"]),
            is_terminal=bool(spec.get("is_terminal", False)),
            is_system=True,
            is_active=True,
            created_at=now,
            updated_at=now,
        )
        session.add(row)
        existing[row.code] = row
    session.flush()
    return list(existing.values())


def _backfill_legacy_stage_values(
    session,
    *,
    tenant_id: str,
    pipeline: SalesPipelineRow,
    stages: list[SalesPipelineStageRow],
) -> None:
    stage_by_code = {row.code: row for row in stages}
    tenant_stages_by_id = {
        row.stage_id: row
        for row in session.execute(
            select(SalesPipelineStageRow).where(SalesPipelineStageRow.tenant_id == tenant_id)
        ).scalars().all()
    }
    tenant_stage_ids = set(tenant_stages_by_id)
    default_stage = stage_by_code[DEFAULT_STAGE_CODE]
    legacy_pipeline_values = {"", "sales_default", DEFAULT_PIPELINE_CODE}

    for deal in session.execute(select(SalesDealRow).where(SalesDealRow.tenant_id == tenant_id)).scalars().all():
        if deal.stage_id in tenant_stage_ids:
            stage = tenant_stages_by_id[deal.stage_id]
            if deal.pipeline_id != stage.pipeline_id:
                deal.pipeline_id = stage.pipeline_id
                deal.updated_at = utc_now_iso()
            continue
        code = _normalize_stage_code(deal.stage_id)
        stage = stage_by_code.get(code, default_stage)
        deal.stage_id = stage.stage_id
        if deal.pipeline_id in legacy_pipeline_values or deal.pipeline_id != pipeline.pipeline_id:
            deal.pipeline_id = pipeline.pipeline_id
        deal.updated_at = utc_now_iso()

    for history in session.execute(select(SalesStageHistoryRow).where(SalesStageHistoryRow.tenant_id == tenant_id)).scalars().all():
        if history.from_stage_id and history.from_stage_id not in tenant_stage_ids:
            from_code = _normalize_stage_code(history.from_stage_id)
            history.from_stage_id = stage_by_code.get(from_code, default_stage).stage_id
        if history.to_stage_id not in tenant_stage_ids:
            to_code = _normalize_stage_code(history.to_stage_id)
            history.to_stage_id = stage_by_code.get(to_code, default_stage).stage_id


def ensure_default_pipeline(session, tenant_id: str) -> SalesPipelineRow:
    pipeline = session.execute(
        select(SalesPipelineRow)
        .where(
            SalesPipelineRow.tenant_id == tenant_id,
            SalesPipelineRow.is_default == True,  # noqa: E712
            SalesPipelineRow.is_active == True,  # noqa: E712
        )
        .order_by(SalesPipelineRow.id.asc())
    ).scalars().first()
    if pipeline is None:
        pipeline = session.execute(
            select(SalesPipelineRow).where(
                SalesPipelineRow.tenant_id == tenant_id,
                SalesPipelineRow.code == DEFAULT_PIPELINE_CODE,
            )
        ).scalar_one_or_none()
    if pipeline is None:
        now = utc_now_iso()
        pipeline = SalesPipelineRow(
            pipeline_id=new_id("pip"),
            tenant_id=tenant_id,
            code=DEFAULT_PIPELINE_CODE,
            name=DEFAULT_PIPELINE_NAME,
            description=None,
            is_default=True,
            is_active=True,
            created_at=now,
            updated_at=now,
        )
        session.add(pipeline)
        session.flush()
    if not pipeline.is_default or not pipeline.is_active:
        _set_default_pipeline(session, pipeline)
    stages = ensure_default_stages(session, tenant_id, pipeline.pipeline_id)
    _backfill_legacy_stage_values(session, tenant_id=tenant_id, pipeline=pipeline, stages=stages)
    return pipeline


def get_default_pipeline(session, tenant_id: str) -> SalesPipelineRow:
    return ensure_default_pipeline(session, tenant_id)


def resolve_stage_by_code_or_id(
    session,
    *,
    tenant_id: str,
    pipeline_id: str,
    stage_id: str | None = None,
    stage_code: str | None = None,
    require_active: bool = True,
) -> SalesPipelineStageRow:
    ensure_default_pipeline(session, tenant_id)

    def resolve_one(raw: str | None) -> SalesPipelineStageRow:
        value = str(raw or DEFAULT_STAGE_CODE).strip()
        stage = session.execute(
            select(SalesPipelineStageRow).where(
                SalesPipelineStageRow.tenant_id == tenant_id,
                SalesPipelineStageRow.stage_id == value,
            )
        ).scalar_one_or_none()
        if stage is not None:
            if stage.pipeline_id != pipeline_id:
                raise HTTPException(status_code=400, detail="Stage does not belong to pipeline")
            if require_active and not stage.is_active:
                raise HTTPException(status_code=400, detail="Stage is inactive")
            return stage

        code = _normalize_stage_code(value)
        stage = session.execute(
            select(SalesPipelineStageRow).where(
                SalesPipelineStageRow.tenant_id == tenant_id,
                SalesPipelineStageRow.pipeline_id == pipeline_id,
                SalesPipelineStageRow.code == code,
            )
        ).scalar_one_or_none()
        if stage is None:
            raise HTTPException(status_code=404, detail="Stage not found")
        if require_active and not stage.is_active:
            raise HTTPException(status_code=400, detail="Stage is inactive")
        return stage

    by_id_or_code = resolve_one(stage_id or stage_code or DEFAULT_STAGE_CODE)
    if stage_id and stage_code:
        by_code = resolve_one(stage_code)
        if by_id_or_code.stage_id != by_code.stage_id:
            raise HTTPException(status_code=400, detail="stage_id and stage_code point to different stages")
    return by_id_or_code


def _stage_filter_ids(session, tenant_id: str, value: str) -> list[str]:
    normalized = str(value or "").strip()
    if not normalized:
        return []
    direct = session.execute(
        select(SalesPipelineStageRow.stage_id).where(
            SalesPipelineStageRow.tenant_id == tenant_id,
            SalesPipelineStageRow.stage_id == normalized,
        )
    ).scalars().all()
    if direct:
        return list(direct)
    code = _normalize_stage_code(normalized)
    return list(
        session.execute(
            select(SalesPipelineStageRow.stage_id).where(
                SalesPipelineStageRow.tenant_id == tenant_id,
                SalesPipelineStageRow.code == code,
            )
        ).scalars().all()
    )


def _stage_refs_by_id(session, tenant_id: str, stage_ids: list[str | None]) -> dict[str, SalesPipelineStageRow]:
    ids = [stage_id for stage_id in stage_ids if stage_id]
    if not ids:
        return {}
    rows = session.execute(
        select(SalesPipelineStageRow).where(
            SalesPipelineStageRow.tenant_id == tenant_id,
            SalesPipelineStageRow.stage_id.in_(ids),
        )
    ).scalars().all()
    return {row.stage_id: row for row in rows}


def _deal_pipeline_stage(session, deal: SalesDealRow) -> tuple[SalesPipelineRow | None, SalesPipelineStageRow | None]:
    pipeline = session.execute(
        select(SalesPipelineRow).where(
            SalesPipelineRow.tenant_id == deal.tenant_id,
            SalesPipelineRow.pipeline_id == deal.pipeline_id,
        )
    ).scalar_one_or_none()
    stage = session.execute(
        select(SalesPipelineStageRow).where(
            SalesPipelineStageRow.tenant_id == deal.tenant_id,
            SalesPipelineStageRow.stage_id == deal.stage_id,
        )
    ).scalar_one_or_none()
    return pipeline, stage


def _deal_to_out_with_refs(session, deal: SalesDealRow) -> SalesDealOut:
    pipeline, stage = _deal_pipeline_stage(session, deal)
    return _deal_to_out(deal, pipeline=pipeline, stage=stage)


def _publish_sales_event(
    session,
    *,
    tenant_id: str,
    event_type: str,
    aggregate_type: str,
    aggregate_id: str,
    payload: dict[str, Any],
    actor_type: str | None = None,
    actor_id: str | None = None,
    correlation_id: str | None = None,
    causation_id: str | None = None,
) -> None:
    SalesEventPublisher.publish_sales_event(
        session,
        tenant_id=tenant_id,
        event_type=event_type,
        aggregate_type=aggregate_type,
        aggregate_id=aggregate_id,
        payload=payload,
        actor_type=actor_type,
        actor_id=actor_id,
        correlation_id=correlation_id,
        causation_id=causation_id,
    )


def _stage_by_id_or_none(session, tenant_id: str, stage_id: str | None) -> SalesPipelineStageRow | None:
    if not stage_id:
        return None
    return session.execute(
        select(SalesPipelineStageRow).where(
            SalesPipelineStageRow.tenant_id == tenant_id,
            SalesPipelineStageRow.stage_id == stage_id,
        )
    ).scalar_one_or_none()


def _publish_lead_entered_event(
    session,
    *,
    lead: SalesLeadRow,
    deal: SalesDealRow,
    stage: SalesPipelineStageRow,
    actor_type: str,
    actor_id: str | None,
) -> None:
    _publish_sales_event(
        session,
        tenant_id=lead.tenant_id,
        event_type=sales_event_types.LEAD_ENTERED_CRM,
        aggregate_type="lead",
        aggregate_id=lead.lead_id,
        actor_type=actor_type,
        actor_id=actor_id,
        payload={
            "lead_id": lead.lead_id,
            "deal_id": deal.deal_id,
            "source_type": lead.source_type,
            "source_channel": lead.source_channel,
            "lead_temperature": lead.lead_temperature,
            "initial_stage_id": stage.stage_id,
            "initial_stage_code": stage.code,
        },
    )


def _lead_to_out(row: SalesLeadRow, deal: SalesDealRow | None = None) -> SalesLeadOut:
    return SalesLeadOut(
        lead_id=row.lead_id,
        tenant_id=row.tenant_id,
        deal_id=deal.deal_id if deal is not None else None,
        source_type=row.source_type,
        source_channel=row.source_channel,
        source_campaign_id=row.source_campaign_id,
        full_name=row.full_name,
        company_name=row.company_name,
        phone=row.phone,
        email=row.email,
        messenger_handles=_json_dict(row.messenger_handles_json),
        lead_temperature=row.lead_temperature,  # type: ignore[arg-type]
        lead_score=row.lead_score,
        customer_type=row.customer_type,
        segment_type=row.segment_type,
        initial_need_summary=row.initial_need_summary,
        preferred_channel=row.preferred_channel,  # type: ignore[arg-type]
        assigned_agent_type=row.assigned_agent_type,  # type: ignore[arg-type]
        status=row.status,  # type: ignore[arg-type]
        crm_customer_id=row.crm_customer_id,
        created_at=row.created_at,
        updated_at=row.updated_at,
    )


def _customer_to_out(row: Customer, deal_count: int = 0) -> SalesCustomerOut:
    return SalesCustomerOut(
        customer_id=row.customer_id,
        display_name=row.display_name,
        phones=_json_list(row.phones_json),
        preferred_phone=row.preferred_phone,
        tags=_json_list(row.tags_json),
        deal_count=deal_count,
        created_at=row.created_at,
    )


def _deal_to_out(
    row: SalesDealRow,
    *,
    pipeline: SalesPipelineRow | None = None,
    stage: SalesPipelineStageRow | None = None,
) -> SalesDealOut:
    return SalesDealOut(
        deal_id=row.deal_id,
        tenant_id=row.tenant_id,
        lead_id=row.lead_id,
        customer_id=row.customer_id,
        pipeline_id=row.pipeline_id,
        stage_id=row.stage_id,
        pipeline=_pipeline_to_out(pipeline) if pipeline else None,
        stage=_stage_to_ref(stage),
        scenario_type=row.scenario_type,  # type: ignore[arg-type]
        priority=row.priority,
        title=row.title,
        need_summary=row.need_summary,
        product_context=_json_dict(row.product_context_json),
        estimated_amount=row.estimated_amount,
        final_amount=row.final_amount,
        currency=row.currency,
        payment_model=row.payment_model,
        document_required=row.document_required,
        payment_required=row.payment_required,
        assigned_human_user_id=row.assigned_human_user_id,
        assigned_ai_orchestrator_id=row.assigned_ai_orchestrator_id,
        preferred_channel=row.preferred_channel,
        current_channel=row.current_channel,
        status=row.status,  # type: ignore[arg-type]
        won_reason=row.won_reason,
        lost_reason=row.lost_reason,
        close_reason=row.close_reason,
        next_action_type=row.next_action_type,
        next_action_at=row.next_action_at,
        last_contact_at=row.last_contact_at,
        closed_at=row.closed_at,
        created_at=row.created_at,
        updated_at=row.updated_at,
    )


def _communication_to_out(row: SalesCommunicationSessionRow) -> SalesCommunicationOut:
    metadata = _json_dict(row.metadata_json)
    channel_provider = row.channel_provider or metadata.get("channel_provider")
    if not channel_provider:
        channel_provider = "voice" if row.channel_type == "voice" else None
    return SalesCommunicationOut(
        communication_id=row.communication_id,
        tenant_id=row.tenant_id,
        deal_id=row.deal_id,
        lead_id=row.lead_id,
        customer_id=row.customer_id,
        channel_type=row.channel_type,  # type: ignore[arg-type]
        channel_provider=channel_provider,
        direction=row.direction,  # type: ignore[arg-type]
        agent_type=row.agent_type,  # type: ignore[arg-type]
        started_at=row.started_at,
        ended_at=row.ended_at,
        duration_sec=row.duration_sec,
        subject=row.subject,
        status=row.status,
        summary=row.summary,
        transcript_id=row.transcript_id,
        next_action_type=row.next_action_type,
        next_action_at=row.next_action_at,
        sentiment=row.sentiment,
        result_code=row.result_code,
        metadata=metadata,
        created_at=row.created_at,
        updated_at=row.updated_at,
    )


def _message_to_out(row: SalesMessageRow) -> SalesMessageOut:
    return SalesMessageOut(
        message_id=row.message_id,
        deal_id=row.deal_id,
        communication_id=row.communication_id,
        sender_type=row.sender_type,
        sender_id=row.sender_id,
        channel_provider=row.channel_provider,
        external_message_id=row.external_message_id,
        body=row.body,
        attachments=_json_list(row.attachments_json),
        delivery_status=row.delivery_status,
        read_status=row.read_status,
        metadata=_json_dict(row.message_metadata_json),
        sent_at=row.sent_at,
        created_at=row.created_at,
    )


def _call_to_out(row: SalesCallRow) -> SalesCallOut:
    return SalesCallOut(
        call_id=row.call_id,
        deal_id=row.deal_id,
        communication_id=row.communication_id,
        phone_number=row.phone_number,
        direction=row.direction,  # type: ignore[arg-type]
        provider=row.provider,
        external_call_id=row.external_call_id,
        recording_url=row.recording_url,
        transcript_status=row.transcript_status,  # type: ignore[arg-type]
        transcript_id=row.transcript_id,
        call_status=row.call_status,
        started_at=row.started_at,
        ended_at=row.ended_at,
        duration_sec=row.duration_sec,
        summary=row.summary,
        created_at=row.created_at,
        updated_at=row.updated_at,
    )


def _transcript_to_out(row: SalesTranscriptRow) -> SalesTranscriptOut:
    return SalesTranscriptOut(
        transcript_id=row.transcript_id,
        call_id=row.call_id,
        language=row.language,
        transcript_text=row.transcript_text,
        diarization=_json_dict(row.diarization_json),
        extracted_entities=_json_dict(row.extracted_entities_json),
        created_at=row.created_at,
        updated_at=row.updated_at,
    )


def _offer_to_out(row: SalesOfferRow) -> SalesOfferOut:
    return SalesOfferOut(
        offer_id=row.offer_id,
        deal_id=row.deal_id,
        offer_type=row.offer_type,  # type: ignore[arg-type]
        title=row.title,
        description=row.description,
        line_items=_json_list(row.line_items_json),
        pricing=_json_dict(row.pricing_json),
        total_amount=row.total_amount,
        currency=row.currency,
        validity_until=row.validity_until,
        status=row.status,  # type: ignore[arg-type]
        rendered_document_url=row.rendered_document_url,
        created_by_type=row.created_by_type,
        created_at=row.created_at,
        updated_at=row.updated_at,
    )


def _condition_to_out(row: SalesConditionRow) -> SalesConditionOut:
    return SalesConditionOut(
        condition_id=row.condition_id,
        deal_id=row.deal_id,
        product_name=row.product_name,
        service_name=row.service_name,
        quantity=row.quantity,
        unit=row.unit,
        delivery_mode=row.delivery_mode,
        execution_date=row.execution_date,
        start_date=row.start_date,
        end_date=row.end_date,
        payment_terms=row.payment_terms,
        custom_terms=_json_dict(row.custom_terms_json),
        agreed_price=row.agreed_price,
        currency=row.currency,
        confirmed_at=row.confirmed_at,
        created_at=row.created_at,
        updated_at=row.updated_at,
    )


def _counterparty_to_out(row: SalesCounterpartyRow) -> SalesCounterpartyOut:
    return SalesCounterpartyOut(
        counterparty_id=row.counterparty_id,
        deal_id=row.deal_id,
        customer_id=row.customer_id,
        company_name=row.company_name,
        full_name=row.full_name,
        bin_iin=row.bin_iin,
        address=row.address,
        bank_details=_json_dict(row.bank_details_json),
        signer_name=row.signer_name,
        signer_role=row.signer_role,
        signer_basis=row.signer_basis,
        email_for_docs=row.email_for_docs,
        phone_for_docs=row.phone_for_docs,
        completeness_status=row.completeness_status,
        created_at=row.created_at,
        updated_at=row.updated_at,
    )


def _document_to_out(row: SalesDocumentRow) -> SalesDocumentOut:
    return SalesDocumentOut(
        document_id=row.document_id,
        deal_id=row.deal_id,
        customer_id=row.customer_id,
        document_type=row.document_type,  # type: ignore[arg-type]
        template_id=row.template_id,
        version=row.version,
        status=row.status,  # type: ignore[arg-type]
        file_url=row.file_url,
        rendered_payload=_json_dict(row.rendered_payload_json),
        external_sign_provider_id=row.external_sign_provider_id,
        signed_at=row.signed_at,
        created_at=row.created_at,
        updated_at=row.updated_at,
    )


def _invoice_to_out(row: SalesInvoiceRow) -> SalesInvoiceOut:
    return SalesInvoiceOut(
        invoice_id=row.invoice_id,
        deal_id=row.deal_id,
        customer_id=row.customer_id,
        invoice_number=row.invoice_number,
        basis_document_id=row.basis_document_id,
        amount=row.amount,
        currency=row.currency,
        due_date=row.due_date,
        status=row.status,  # type: ignore[arg-type]
        payment_link=row.payment_link,
        line_items=_json_list(row.line_items_json),
        metadata=_json_dict(row.metadata_json),
        issued_at=row.issued_at,
        paid_at=row.paid_at,
        created_at=row.created_at,
        updated_at=row.updated_at,
    )


def _payment_to_out(row: SalesPaymentRow) -> SalesPaymentOut:
    return SalesPaymentOut(
        payment_id=row.payment_id,
        deal_id=row.deal_id,
        invoice_id=row.invoice_id,
        payment_provider=row.payment_provider,
        external_payment_id=row.external_payment_id,
        amount=row.amount,
        currency=row.currency,
        status=row.status,  # type: ignore[arg-type]
        paid_at=row.paid_at,
        payment_method=row.payment_method,
        failure_reason=row.failure_reason,
        metadata=_json_dict(row.metadata_json),
        created_at=row.created_at,
        updated_at=row.updated_at,
    )


def _escalation_to_out(row: SalesEscalationRow) -> SalesEscalationOut:
    return SalesEscalationOut(
        escalation_id=row.escalation_id,
        deal_id=row.deal_id,
        escalation_type=row.escalation_type,
        reason=row.reason,
        severity=row.severity,  # type: ignore[arg-type]
        status=row.status,  # type: ignore[arg-type]
        assigned_to_user_id=row.assigned_to_user_id,
        assigned_at=row.assigned_at,
        started_at=row.started_at,
        created_at=row.created_at,
        resolved_at=row.resolved_at,
        canceled_at=row.canceled_at,
        resolution_code=row.resolution_code,
        resolution_summary=row.resolution_summary,
        sla_due_at=row.sla_due_at,
        source_channel=row.source_channel,
        source_communication_session_id=row.source_communication_session_id,
        source_automation_task_id=row.source_automation_task_id,
        updated_at=row.updated_at,
    )


def _task_to_out(row: SalesAutomationTaskRow, deal: SalesDealRow | None = None) -> SalesAutomationTaskOut:
    payload = _json_dict(row.payload_json)
    return SalesAutomationTaskOut(
        task_id=row.task_id,
        deal_id=row.deal_id,
        deal_title=deal.title if deal is not None else None,
        deal_status=deal.status if deal is not None else None,  # type: ignore[arg-type]
        deal_stage_id=deal.stage_id if deal is not None else None,
        deal_pipeline_id=deal.pipeline_id if deal is not None else None,
        task_type=row.task_type,
        payload=payload,
        run_at=row.run_at,
        status=row.status,  # type: ignore[arg-type]
        retry_count=row.retry_count,
        max_retries=row.max_retries,
        locked_at=row.locked_at,
        locked_by=row.locked_by,
        completed_at=row.completed_at,
        failed_at=row.failed_at,
        last_error=row.last_error,
        recommended_to_channel=payload.get("recommended_to_channel") or payload.get("to_channel"),
        reason_code=payload.get("reason_code") or payload.get("reason"),
        created_at=row.created_at,
        updated_at=row.updated_at,
    )


def _stage_history_to_out(row: SalesStageHistoryRow, stage_refs: dict[str, SalesPipelineStageRow] | None = None) -> SalesStageHistoryOut:
    refs = stage_refs or {}
    return SalesStageHistoryOut(
        history_id=row.history_id,
        deal_id=row.deal_id,
        from_stage_id=row.from_stage_id,
        to_stage_id=row.to_stage_id,
        from_stage=_stage_to_ref(refs.get(row.from_stage_id or "")),
        to_stage=_stage_to_ref(refs.get(row.to_stage_id)),
        changed_by_type=row.changed_by_type,
        changed_by_id=row.changed_by_id,
        reason=row.reason,
        changed_at=row.changed_at,
    )


def _channel_switch_to_out(row: SalesChannelSwitchRow) -> SalesChannelSwitchOut:
    return SalesChannelSwitchOut(
        switch_id=row.switch_id,
        deal_id=row.deal_id,
        communication_id=row.communication_id,
        communication_session_id=row.communication_session_id,
        previous_communication_session_id=row.previous_communication_session_id,
        new_communication_session_id=row.new_communication_session_id,
        from_channel=row.from_channel,
        to_channel=row.to_channel,
        reason_code=row.reason_code,
        reason_text=row.reason_text,
        reason_for_channel_switch=row.reason_for_channel_switch,
        initiated_by_type=row.initiated_by_type,
        initiated_by_id=row.initiated_by_id,
        source_type=row.source_type,
        source_id=row.source_id,
        recommended_by_task_id=row.recommended_by_task_id,
        switched_at=row.switched_at,
        created_at=row.created_at,
    )


def _channel_switch_to_out_with_session(
    row: SalesChannelSwitchRow,
    new_communication: SalesCommunicationSessionRow | None = None,
) -> SalesChannelSwitchOut:
    out = _channel_switch_to_out(row)
    if new_communication is not None:
        out.new_communication = _communication_to_out(new_communication)
    return out


def _note_to_out(row: SalesNoteRow) -> SalesNoteOut:
    return SalesNoteOut(
        note_id=row.note_id,
        deal_id=row.deal_id,
        author_type=row.author_type,
        author_id=row.author_id,
        note_type=row.note_type,
        content=row.content,
        source_type=row.source_type,
        source_id=row.source_id,
        created_at=row.created_at,
        updated_at=row.updated_at,
        archived_at=row.archived_at,
    )


def _get_lead(session, lead_id: str, tenant_id: str) -> SalesLeadRow:
    row = session.execute(
        select(SalesLeadRow).where(SalesLeadRow.lead_id == lead_id, SalesLeadRow.tenant_id == tenant_id)
    ).scalar_one_or_none()
    if row is None:
        raise HTTPException(status_code=404, detail="Lead not found")
    return row


def _latest_deal_for_lead(session, lead_id: str, tenant_id: str) -> SalesDealRow | None:
    return session.execute(
        select(SalesDealRow)
        .where(SalesDealRow.lead_id == lead_id, SalesDealRow.tenant_id == tenant_id)
        .order_by(SalesDealRow.id.desc())
    ).scalars().first()


def _latest_deals_by_lead(session, lead_ids: list[str], tenant_id: str) -> dict[str, SalesDealRow]:
    if not lead_ids:
        return {}
    rows = session.execute(
        select(SalesDealRow)
        .where(SalesDealRow.lead_id.in_(lead_ids), SalesDealRow.tenant_id == tenant_id)
        .order_by(SalesDealRow.id.desc())
    ).scalars().all()
    latest: dict[str, SalesDealRow] = {}
    for row in rows:
        if row.lead_id and row.lead_id not in latest:
            latest[row.lead_id] = row
    return latest


def _get_deal(session, deal_id: str, tenant_id: str) -> SalesDealRow:
    row = session.execute(
        select(SalesDealRow).where(SalesDealRow.deal_id == deal_id, SalesDealRow.tenant_id == tenant_id)
    ).scalar_one_or_none()
    if row is None:
        raise HTTPException(status_code=404, detail="Deal not found")
    return row


def _get_communication(session, communication_id: str, tenant_id: str) -> SalesCommunicationSessionRow:
    row = session.execute(
        select(SalesCommunicationSessionRow).where(
            SalesCommunicationSessionRow.communication_id == communication_id,
            SalesCommunicationSessionRow.tenant_id == tenant_id,
        )
    ).scalar_one_or_none()
    if row is None:
        raise HTTPException(status_code=404, detail="Communication not found")
    return row


def _find_external_link(
    session,
    *,
    tenant_id: str,
    thread_id: str | None = None,
    voice_session_id: str | None = None,
    external_call_id: str | None = None,
    interaction_id: str | None = None,
) -> SalesExternalLinkRow | None:
    conditions = []
    if str(thread_id or "").strip():
        conditions.append(SalesExternalLinkRow.external_thread_id == str(thread_id).strip())
    if str(voice_session_id or "").strip():
        conditions.append(SalesExternalLinkRow.voice_session_id == str(voice_session_id).strip())
    if str(external_call_id or "").strip():
        conditions.append(SalesExternalLinkRow.external_call_id == str(external_call_id).strip())
    if str(interaction_id or "").strip():
        conditions.append(SalesExternalLinkRow.interaction_id == str(interaction_id).strip())
    if not conditions:
        return None
    return session.execute(
        select(SalesExternalLinkRow)
        .where(SalesExternalLinkRow.tenant_id == tenant_id)
        .where(or_(*conditions))
        .order_by(SalesExternalLinkRow.id.desc())
    ).scalars().first()


def _metadata_value(metadata: dict | None, *keys: str) -> str | None:
    payload = metadata or {}
    for key in keys:
        value = payload.get(key)
        normalized = str(value or "").strip()
        if normalized:
            return normalized
    return None


def _resolve_provider_tenant_id(
    session,
    actor: dict,
    *,
    provider_type: str,
    provider_name: str,
    metadata: dict | None = None,
    provider_account_id: str | None = None,
    external_identifier: str | None = None,
) -> str:
    tenant_id = _optional_tenant_id(actor)
    if tenant_id:
        return tenant_id

    normalized_provider_name = str(provider_name or "").strip().lower()
    normalized_provider_type = str(provider_type or "").strip().lower()
    normalized_account = str(provider_account_id or "").strip() or _metadata_value(
        metadata,
        "provider_account_id",
        "payment_provider_account_id",
        "merchant_id",
        "account_id",
        "integration_id",
    )
    normalized_external = str(external_identifier or "").strip() or _metadata_value(
        metadata,
        "external_identifier",
        "channel_id",
        "phone_number",
        "merchant_id",
    )

    base_stmt = select(TenantIntegrationRow).where(
        func.lower(TenantIntegrationRow.provider_type) == normalized_provider_type,
        func.lower(TenantIntegrationRow.provider_name) == normalized_provider_name,
        TenantIntegrationRow.is_active == True,  # noqa: E712
    )
    if normalized_account:
        row = session.execute(
            base_stmt.where(TenantIntegrationRow.provider_account_id == normalized_account).order_by(TenantIntegrationRow.id.desc())
        ).scalars().first()
        if row is not None:
            return row.tenant_id
    if normalized_external:
        row = session.execute(
            base_stmt.where(TenantIntegrationRow.external_identifier == normalized_external).order_by(TenantIntegrationRow.id.desc())
        ).scalars().first()
        if row is not None:
            return row.tenant_id

    raise HTTPException(status_code=400, detail="Tenant context is required")


def _find_customer_by_customer_id(session, customer_id: str | None) -> Customer | None:
    normalized = str(customer_id or "").strip()
    if not normalized:
        return None
    return session.execute(select(Customer).where(Customer.customer_id == normalized)).scalar_one_or_none()


def _sales_customer_ids_for_tenant(session, tenant_id: str) -> set[str]:
    deal_ids = session.execute(
        select(SalesDealRow.customer_id).where(
            SalesDealRow.tenant_id == tenant_id,
            SalesDealRow.customer_id.is_not(None),
        )
    ).scalars().all()
    lead_ids = session.execute(
        select(SalesLeadRow.crm_customer_id).where(
            SalesLeadRow.tenant_id == tenant_id,
            SalesLeadRow.crm_customer_id.is_not(None),
        )
    ).scalars().all()
    return {str(customer_id) for customer_id in [*deal_ids, *lead_ids] if customer_id}


def _get_sales_customer(session, customer_id: str, tenant_id: str) -> Customer:
    normalized = str(customer_id or "").strip()
    if not normalized or normalized not in _sales_customer_ids_for_tenant(session, tenant_id):
        raise HTTPException(status_code=404, detail="Customer not found")
    row = _find_customer_by_customer_id(session, normalized)
    if row is None:
        raise HTTPException(status_code=404, detail="Customer not found")
    return row


def _customer_deal_count(session, customer_id: str, tenant_id: str) -> int:
    return int(
        session.execute(
            select(func.count()).select_from(SalesDealRow).where(
                SalesDealRow.tenant_id == tenant_id,
                SalesDealRow.customer_id == customer_id,
            )
        ).scalar_one()
        or 0
    )


def _create_lead_and_deal_for_contact(
    session,
    *,
    tenant_id: str,
    source_channel: str,
    preferred_channel: str,
    full_name: str,
    phone: str | None,
    customer_id: str | None = None,
    subject: str | None = None,
) -> tuple[SalesLeadRow, SalesDealRow]:
    now = _now().isoformat()
    pipeline = get_default_pipeline(session, tenant_id)
    stage = resolve_stage_by_code_or_id(
        session,
        tenant_id=tenant_id,
        pipeline_id=pipeline.pipeline_id,
        stage_id=DEFAULT_STAGE_CODE,
    )
    lead = SalesLeadRow(
        lead_id=new_id("sld"),
        tenant_id=tenant_id,
        source_type="autosync",
        source_channel=source_channel,
        source_campaign_id=None,
        full_name=full_name,
        company_name=None,
        phone=phone,
        email=None,
        messenger_handles_json="{}",
        lead_temperature="warm",
        lead_score=60.0,
        customer_type=None,
        segment_type=None,
        initial_need_summary=subject,
        preferred_channel=preferred_channel,
        assigned_agent_type="voice_ai" if preferred_channel == "voice" else "text_ai",
        status="new_qualified_lead",
        crm_customer_id=customer_id,
        created_at=now,
        updated_at=now,
    )
    deal = SalesDealRow(
        deal_id=new_id("sde"),
        tenant_id=tenant_id,
        lead_id=lead.lead_id,
        customer_id=customer_id,
        pipeline_id=pipeline.pipeline_id,
        stage_id=stage.stage_id,
        scenario_type="quick_sale",
        priority=3,
        title=subject or f"Сделка {full_name}",
        need_summary=subject,
        product_context_json="{}",
        estimated_amount=None,
        final_amount=None,
        currency=DEFAULT_SALES_CURRENCY,
        payment_model=None,
        document_required=True,
        payment_required=True,
        assigned_human_user_id=None,
        assigned_ai_orchestrator_id=None,
        preferred_channel=preferred_channel,
        current_channel=preferred_channel,
        status="active",
        won_reason=None,
        lost_reason=None,
        close_reason=None,
        next_action_type=None,
        next_action_at=None,
        last_contact_at=None,
        closed_at=None,
        created_at=now,
        updated_at=now,
    )
    session.add(lead)
    session.add(deal)
    _record_stage_change(
        session,
        deal=deal,
        from_stage_id=None,
        to_stage_id=stage.stage_id,
        changed_by_type="system",
        changed_by_id=None,
        reason="autosync.created",
    )
    _publish_lead_entered_event(session, lead=lead, deal=deal, stage=stage, actor_type="system", actor_id=None)
    return lead, deal


def _resolve_or_create_sync_deal(
    session,
    *,
    tenant_id: str,
    source_channel: str,
    preferred_channel: str,
    customer_id: str | None = None,
    phone: str | None = None,
    display_name: str | None = None,
    interaction_id: str | None = None,
    thread_id: str | None = None,
    voice_session_id: str | None = None,
    external_call_id: str | None = None,
    subject: str | None = None,
) -> tuple[SalesLeadRow | None, SalesDealRow]:
    link = _find_external_link(
        session,
        tenant_id=tenant_id,
        thread_id=thread_id,
        voice_session_id=voice_session_id,
        external_call_id=external_call_id,
        interaction_id=interaction_id,
    )
    if link is not None:
        deal = _get_deal(session, link.deal_id, tenant_id)
        lead = _get_lead(session, deal.lead_id, tenant_id) if deal.lead_id else None
        return lead, deal

    normalized_customer_id = str(customer_id or "").strip() or None
    if normalized_customer_id:
        deal = session.execute(
            select(SalesDealRow)
            .where(SalesDealRow.tenant_id == tenant_id)
            .where(SalesDealRow.customer_id == normalized_customer_id)
            .where(SalesDealRow.status == "active")
            .order_by(SalesDealRow.updated_at.desc(), SalesDealRow.id.desc())
        ).scalars().first()
        if deal is not None:
            lead = _get_lead(session, deal.lead_id, tenant_id) if deal.lead_id else None
            return lead, deal

    normalized_phone = str(phone or "").strip() or None
    if normalized_phone:
        lead, deal = _resolve_deal_for_inbound(
            session,
            tenant_id=tenant_id,
            deal_id=None,
            lead_id=None,
            phone=normalized_phone,
            subject=subject,
            source_channel=source_channel,
            preferred_channel=preferred_channel,
        )
        if normalized_customer_id and deal.customer_id != normalized_customer_id:
            deal.customer_id = normalized_customer_id
            deal.updated_at = utc_now_iso()
        return lead, deal

    customer = _find_customer_by_customer_id(session, normalized_customer_id)
    full_name = str(display_name or (customer.display_name if customer else "") or "").strip() or (
        "Telegram customer" if source_channel == "telegram" else "Voice customer"
    )
    lead, deal = _create_lead_and_deal_for_contact(
        session,
        tenant_id=tenant_id,
        source_channel=source_channel,
        preferred_channel=preferred_channel,
        full_name=full_name,
        phone=normalized_phone,
        customer_id=normalized_customer_id,
        subject=subject,
    )
    return lead, deal


def _find_customer_by_phone(session, phone: str | None) -> Customer | None:
    normalized = str(phone or "").strip()
    if not normalized:
        return None
    direct = session.execute(select(Customer).where(Customer.preferred_phone == normalized)).scalar_one_or_none()
    if direct is not None:
        return direct
    rows = session.execute(select(Customer).order_by(Customer.id.desc())).scalars().all()
    for row in rows:
        phones = _json_list(row.phones_json)
        if normalized in phones:
            return row
    return None


def _ensure_crm_customer(session, *, lead: SalesLeadRow | None, deal: SalesDealRow, counterparty: SalesCounterpartyRow | None) -> str | None:
    if deal.customer_id:
        return deal.customer_id

    phone = None
    display_name = None
    tags = ["sales"]
    if counterparty:
        phone = counterparty.phone_for_docs
        display_name = counterparty.full_name or counterparty.company_name
        if counterparty.company_name:
            tags.append("b2b")
    if not display_name and lead:
        display_name = lead.full_name
        phone = phone or lead.phone
    if not display_name:
        return None

    existing = _find_customer_by_phone(session, phone)
    if existing is not None:
        deal.customer_id = existing.customer_id
        if lead and not lead.crm_customer_id:
            lead.crm_customer_id = existing.customer_id
            lead.updated_at = utc_now_iso()
        return existing.customer_id

    now = utc_now_iso()
    customer = Customer(
        customer_id=new_id("cus"),
        display_name=display_name,
        phones_json=json.dumps([phone] if phone else [], ensure_ascii=False),
        preferred_phone=phone,
        tags_json=json.dumps(tags, ensure_ascii=False),
        created_at=now,
    )
    session.add(customer)
    deal.customer_id = customer.customer_id
    if lead:
        lead.crm_customer_id = customer.customer_id
        lead.updated_at = now
    return customer.customer_id


def _record_stage_change(session, *, deal: SalesDealRow, from_stage_id: str | None, to_stage_id: str, changed_by_type: str, changed_by_id: str | None, reason: str | None = None) -> None:
    session.add(
        SalesStageHistoryRow(
            history_id=new_id("sth"),
            tenant_id=deal.tenant_id,
            deal_id=deal.deal_id,
            from_stage_id=from_stage_id,
            to_stage_id=to_stage_id,
            changed_by_type=changed_by_type,
            changed_by_id=changed_by_id,
            reason=reason,
            changed_at=utc_now_iso(),
        )
    )


def _schedule_task(session, *, deal: SalesDealRow, task_type: str, run_at: str | None, payload: dict | None = None) -> SalesAutomationTaskRow:
    now = utc_now_iso()
    task = SalesAutomationTaskRow(
        task_id=new_id("tsk"),
        tenant_id=deal.tenant_id,
        deal_id=deal.deal_id,
        task_type=task_type,
        payload_json=json.dumps(payload or {}, ensure_ascii=False),
        run_at=run_at or now,
        status="pending",
        retry_count=0,
        max_retries=3,
        locked_at=None,
        locked_by=None,
        completed_at=None,
        failed_at=None,
        last_error=None,
        created_at=now,
        updated_at=now,
    )
    session.add(task)
    return task


def _automation_run_at_after(*, minutes: int = 0, days: int = 0) -> str:
    return (_now() + timedelta(days=days, minutes=minutes)).isoformat()


def _parse_due_date(value: str | None) -> datetime | None:
    raw = str(value or "").strip()
    if not raw:
        return None
    try:
        parsed = datetime.fromisoformat(raw.replace("Z", "+00:00"))
    except ValueError:
        return None
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    return parsed.astimezone(timezone.utc).replace(microsecond=0)


def _invoice_reminder_run_at(due_date: str | None) -> str:
    parsed = _parse_due_date(due_date)
    if parsed is None:
        return _automation_run_at_after(days=1)
    return _iso_at_or_now(parsed - timedelta(days=1))


def _invoice_overdue_run_at(due_date: str | None) -> str:
    parsed = _parse_due_date(due_date)
    if parsed is None:
        return _automation_run_at_after(days=1)
    return _iso_at_or_now(parsed + timedelta(days=1))


def _iso_at_or_now(value: datetime) -> str:
    now = _now()
    return (value if value > now else now).isoformat()


def _schedule_unique_task(
    session,
    *,
    deal: SalesDealRow,
    task_type: str,
    run_at: str | None,
    payload: dict | None = None,
    match_payload: dict | None = None,
) -> SalesAutomationTaskRow:
    payload_data = payload or {}
    match = match_payload or payload_data
    rows = session.execute(
        select(SalesAutomationTaskRow)
        .where(
            SalesAutomationTaskRow.tenant_id == deal.tenant_id,
            SalesAutomationTaskRow.deal_id == deal.deal_id,
            SalesAutomationTaskRow.task_type == task_type,
            SalesAutomationTaskRow.status.in_(["pending", "running"]),
        )
        .order_by(SalesAutomationTaskRow.id.desc())
    ).scalars().all()
    for row in rows:
        existing_payload = _json_dict(row.payload_json)
        if all(existing_payload.get(key) == value for key, value in match.items()):
            return row
    return _schedule_task(session, deal=deal, task_type=task_type, run_at=run_at, payload=payload_data)


def _cancel_pending_automation_tasks(
    session,
    *,
    deal: SalesDealRow,
    task_types: set[str],
    reason: str,
) -> None:
    now = utc_now_iso()
    rows = session.execute(
        select(SalesAutomationTaskRow).where(
            SalesAutomationTaskRow.tenant_id == deal.tenant_id,
            SalesAutomationTaskRow.deal_id == deal.deal_id,
            SalesAutomationTaskRow.task_type.in_(task_types),
            SalesAutomationTaskRow.status == "pending",
        )
    ).scalars().all()
    for row in rows:
        payload = _json_dict(row.payload_json)
        payload["canceled_reason"] = reason
        row.payload_json = json.dumps(payload, ensure_ascii=False)
        row.status = "canceled"
        row.updated_at = now


def apply_stage_by_id(
    session,
    deal: SalesDealRow,
    stage_id: str,
    *,
    actor_type: str,
    actor_id: str | None,
    reason: str | None = None,
    metadata: dict | None = None,
    force: bool = False,
) -> SalesPipelineStageRow:
    stage = _get_stage(session, stage_id, deal.tenant_id, deal.pipeline_id)
    return transition_deal_stage(
        session,
        tenant_id=deal.tenant_id,
        deal_id=deal.deal_id,
        target_stage_code=stage.code,
        actor_type=actor_type,
        actor_id=actor_id,
        reason=reason,
        metadata=metadata,
        force=force,
    )


def apply_stage_by_code(
    session,
    deal: SalesDealRow,
    stage_code: str,
    *,
    actor_type: str,
    actor_id: str | None,
    reason: str | None = None,
    metadata: dict | None = None,
    force: bool = False,
) -> SalesPipelineStageRow:
    ensure_default_pipeline(session, deal.tenant_id)
    resolve_stage_by_code_or_id(
        session,
        tenant_id=deal.tenant_id,
        pipeline_id=deal.pipeline_id,
        stage_id=stage_code,
    )
    return transition_deal_stage(
        session,
        tenant_id=deal.tenant_id,
        deal_id=deal.deal_id,
        target_stage_code=stage_code,
        actor_type=actor_type,
        actor_id=actor_id,
        reason=reason,
        metadata=metadata,
        force=force,
    )


def _apply_stage(
    session,
    deal: SalesDealRow,
    stage_code: str,
    *,
    actor_type: str,
    actor_id: str | None,
    reason: str | None = None,
    metadata: dict | None = None,
    force: bool = False,
) -> None:
    apply_stage_by_code(
        session,
        deal,
        stage_code,
        actor_type=actor_type,
        actor_id=actor_id,
        reason=reason,
        metadata=metadata,
        force=force,
    )


def _apply_stage_if_needed(
    session,
    deal: SalesDealRow,
    stage_code: str,
    *,
    actor_type: str,
    actor_id: str | None,
    reason: str | None = None,
    metadata: dict | None = None,
    force: bool = False,
) -> None:
    _, current_stage = _deal_pipeline_stage(session, deal)
    if current_stage is not None and current_stage.code == stage_code:
        deal.updated_at = utc_now_iso()
        return
    _apply_stage(
        session,
        deal,
        stage_code,
        actor_type=actor_type,
        actor_id=actor_id,
        reason=reason,
        metadata=metadata,
        force=force,
    )


def _is_customer_inbound_telegram_message(payload: SalesTelegramSyncIn) -> bool:
    direction = str(payload.direction or "").strip().lower()
    author_type = str(payload.author_type or "").strip().lower()
    return direction == "inbound" and author_type in {"", "customer"}


def _next_autostage_pipeline_stage(session, deal: SalesDealRow) -> SalesPipelineStageRow | None:
    _, current_stage = _deal_pipeline_stage(session, deal)
    if current_stage is None or current_stage.is_terminal:
        return None
    return session.execute(
        select(SalesPipelineStageRow)
        .where(
            SalesPipelineStageRow.tenant_id == deal.tenant_id,
            SalesPipelineStageRow.pipeline_id == deal.pipeline_id,
            SalesPipelineStageRow.is_active == True,  # noqa: E712
            SalesPipelineStageRow.is_terminal == False,  # noqa: E712
            SalesPipelineStageRow.sort_order > current_stage.sort_order,
        )
        .order_by(SalesPipelineStageRow.sort_order.asc(), SalesPipelineStageRow.id.asc())
        .limit(1)
    ).scalar_one_or_none()


def _telegram_customer_message_count(session, deal: SalesDealRow) -> int:
    return int(
        session.execute(
            select(func.count())
            .select_from(SalesMessageRow)
            .where(
                SalesMessageRow.tenant_id == deal.tenant_id,
                SalesMessageRow.deal_id == deal.deal_id,
                SalesMessageRow.channel_provider == "telegram",
                SalesMessageRow.sender_type == "customer",
                SalesMessageRow.delivery_status == "received",
            )
        ).scalar_one()
        or 0
    )


def _maybe_advance_telegram_deal_stage(
    session,
    *,
    deal: SalesDealRow,
    lead: SalesLeadRow | None,
    payload: SalesTelegramSyncIn,
) -> None:
    if not _is_customer_inbound_telegram_message(payload):
        return
    interval = _telegram_autostage_message_interval()
    message_count = _telegram_customer_message_count(session, deal)
    if message_count == 0 or message_count % interval != 0:
        return
    next_stage = _next_autostage_pipeline_stage(session, deal)
    if next_stage is None:
        return
    _apply_stage(
        session,
        deal,
        next_stage.code,
        actor_type="system",
        actor_id="telegram-autostage",
        reason="telegram.autostage_every_3_messages",
        metadata={
            "telegram_message_count": message_count,
            "telegram_thread_id": payload.thread_id,
            "telegram_chat_id": payload.chat_id,
            "interval": interval,
            "mode": "deterministic",
        },
        force=True,
    )
    if lead is not None:
        if next_stage.code in {"warm_lead", "hot_lead", "enrichment_required"}:
            lead.status = next_stage.code
        if next_stage.code == "hot_lead":
            lead.lead_temperature = "hot"
        elif next_stage.code == "warm_lead":
            lead.lead_temperature = "warm"
        lead.updated_at = utc_now_iso()


def _resolve_stage_change_target_code(session, deal: SalesDealRow, payload: SalesDealStageChangeIn) -> str:
    target_stage_code = str(payload.target_stage_code or payload.stage_code or "").strip() or None
    target_stage_id = str(payload.target_stage_id or payload.stage_id or "").strip() or None
    if not target_stage_code and not target_stage_id:
        raise DealStateMachineError(
            status_code=400,
            error="stage_required",
            message="target_stage_code or target_stage_id is required",
            details={"deal_id": deal.deal_id},
        )
    if not target_stage_id:
        return target_stage_code or ""

    stage = session.execute(
        select(SalesPipelineStageRow).where(SalesPipelineStageRow.stage_id == target_stage_id)
    ).scalar_one_or_none()
    if stage is None:
        # Backward compatibility: old clients used stage_id for stable stage codes.
        return target_stage_code or target_stage_id
    if stage.tenant_id != deal.tenant_id:
        raise DealStateMachineError(
            status_code=403,
            error="cross_tenant_stage_forbidden",
            message="Stage belongs to another tenant",
            details={"stage_id": stage.stage_id, "deal_id": deal.deal_id},
        )
    if stage.pipeline_id != deal.pipeline_id:
        raise DealStateMachineError(
            status_code=400,
            error="invalid_stage_pipeline",
            message="Stage does not belong to deal pipeline",
            details={
                "stage_id": stage.stage_id,
                "stage_pipeline_id": stage.pipeline_id,
                "deal_pipeline_id": deal.pipeline_id,
            },
        )
    if target_stage_code and _normalize_stage_code(target_stage_code) != stage.code:
        raise DealStateMachineError(
            status_code=400,
            error="stage_target_mismatch",
            message="target_stage_id and target_stage_code point to different stages",
            details={"stage_id": stage.stage_id, "stage_code": stage.code, "target_stage_code": target_stage_code},
        )
    return stage.code


def _get_deal_for_state_change(session, deal_id: str, tenant_id: str) -> SalesDealRow:
    try:
        return _get_deal(session, deal_id, tenant_id)
    except HTTPException as exc:
        if exc.status_code == 404:
            raise DealStateMachineError(
                status_code=404,
                error="deal_not_found",
                message="Deal not found",
                details={"deal_id": deal_id},
            ) from exc
        raise


def _apply_payment_stage(session, deal: SalesDealRow, stage_code: str, *, reason: str, metadata: dict) -> None:
    _, current_stage = _deal_pipeline_stage(session, deal)
    if current_stage is not None and current_stage.code == stage_code:
        return
    try:
        _apply_stage(
            session,
            deal,
            stage_code,
            actor_type="system",
            actor_id=None,
            reason=reason,
            metadata=metadata,
        )
    except DealStateMachineError as exc:
        if exc.error != "invalid_stage_transition":
            raise
        _apply_stage(
            session,
            deal,
            stage_code,
            actor_type="system",
            actor_id=None,
            reason=reason,
            metadata={**metadata, "force_reason": "verified_payment_webhook"},
            force=True,
        )


def _payment_to_dict(row: SalesPaymentRow) -> dict[str, Any]:
    payload = _payment_to_out(row)
    if hasattr(payload, "model_dump"):
        return payload.model_dump()
    return payload.dict()


def _integration_settings(integration: TenantIntegrationRow | None) -> dict:
    return _json_dict(integration.settings_json if integration is not None else "{}")


def _payment_webhook_secret(integration: TenantIntegrationRow | None) -> str | None:
    settings = _integration_settings(integration)
    for key in ("webhook_secret", "payment_webhook_secret", "secret"):
        value = str(settings.get(key) or "").strip()
        if value:
            return value
    return None


def _payment_replay_window_seconds(integration: TenantIntegrationRow | None) -> int:
    settings = _integration_settings(integration)
    try:
        value = int(settings.get("webhook_replay_window_seconds") or settings.get("replay_window_seconds") or 600)
    except (TypeError, ValueError):
        value = 600
    return max(value, 60)


def _payment_allows_overpayment(integration: TenantIntegrationRow | None, metadata: dict | None) -> bool:
    settings = _integration_settings(integration)
    raw = (metadata or {}).get("allow_overpayment", settings.get("allow_overpayment", False))
    if isinstance(raw, bool):
        return raw
    return str(raw or "").strip().lower() in {"1", "true", "yes", "on"}


def _resolve_payment_integration(
    session,
    actor: dict,
    *,
    provider: str,
    provider_account_id: str | None,
    external_identifier: str | None = None,
) -> tuple[str, TenantIntegrationRow | None]:
    normalized_provider = str(provider or "").strip().lower() or "manual"
    normalized_account = str(provider_account_id or "").strip()
    normalized_external = str(external_identifier or "").strip()

    base_stmt = select(TenantIntegrationRow).where(
        func.lower(TenantIntegrationRow.provider_type) == "payment",
        func.lower(TenantIntegrationRow.provider_name) == normalized_provider,
        TenantIntegrationRow.is_active == True,  # noqa: E712
    )
    if normalized_account:
        row = session.execute(
            base_stmt.where(TenantIntegrationRow.provider_account_id == normalized_account).order_by(TenantIntegrationRow.id.desc())
        ).scalars().first()
        if row is not None:
            return row.tenant_id, row
    if normalized_external:
        row = session.execute(
            base_stmt.where(TenantIntegrationRow.external_identifier == normalized_external).order_by(TenantIntegrationRow.id.desc())
        ).scalars().first()
        if row is not None:
            return row.tenant_id, row

    actor_tenant_id = _optional_tenant_id(actor)
    if normalized_provider == "manual" and actor_tenant_id:
        return actor_tenant_id, None

    raise HTTPException(status_code=403, detail="Payment provider integration not found")


def _payment_webhook_event_duplicate_response(session, event: SalesPaymentWebhookEventRow) -> dict[str, Any]:
    payment = None
    if event.external_payment_id:
        payment = session.execute(
            select(SalesPaymentRow).where(
                SalesPaymentRow.tenant_id == event.tenant_id,
                SalesPaymentRow.payment_provider == event.payment_provider,
                SalesPaymentRow.external_payment_id == event.external_payment_id,
            )
        ).scalar_one_or_none()
    response = {
        "status": "duplicate",
        "event_status": event.event_status,
        "payment_id": payment.payment_id if payment is not None else None,
        "invoice_id": payment.invoice_id if payment is not None else None,
        "external_event_id": event.external_event_id,
        "external_payment_id": event.external_payment_id,
    }
    if event.normalized_payload_json:
        normalized = _json_dict(event.normalized_payload_json)
        result = normalized.get("result") if isinstance(normalized.get("result"), dict) else {}
        response["payment_id"] = response["payment_id"] or result.get("payment_id")
        response["invoice_id"] = response["invoice_id"] or result.get("invoice_id")
    return response


def _find_existing_webhook_event(
    session,
    *,
    tenant_id: str,
    provider: str,
    external_event_id: str | None,
) -> SalesPaymentWebhookEventRow | None:
    if not external_event_id:
        return None
    return session.execute(
        select(SalesPaymentWebhookEventRow).where(
            SalesPaymentWebhookEventRow.tenant_id == tenant_id,
            SalesPaymentWebhookEventRow.payment_provider == provider,
            SalesPaymentWebhookEventRow.external_event_id == external_event_id,
        )
    ).scalar_one_or_none()


def _mark_payment_webhook_event_failed(event: SalesPaymentWebhookEventRow, detail: Any) -> None:
    event.event_status = "failed"
    event.error_message = str(detail)
    event.updated_at = utc_now_iso()


def _payment_metadata(payload: dict) -> dict:
    metadata = payload.get("metadata") if isinstance(payload.get("metadata"), dict) else {}
    return dict(metadata)


def _payload_value(payload: dict, *keys: str) -> Any:
    sources: list[dict] = [payload]
    metadata = payload.get("metadata") if isinstance(payload.get("metadata"), dict) else {}
    sources.append(metadata)
    payment = payload.get("payment") if isinstance(payload.get("payment"), dict) else {}
    sources.append(payment)
    data_object = payload.get("data", {}).get("object") if isinstance(payload.get("data"), dict) else {}
    if isinstance(data_object, dict):
        sources.append(data_object)
    for source in sources:
        for key in keys:
            value = source.get(key)
            if value is not None and str(value).strip() != "":
                return value
    return None


def _normalized_payment_payload(payload: dict, headers: dict[str, str]) -> dict[str, Any]:
    provider = extract_payment_provider(payload, headers)
    provider_status = _payload_value(payload, "provider_status", "status", "payment_status")
    event_type = extract_event_type(payload, headers)
    normalized_status = normalize_payment_status(provider, str(provider_status or ""))
    normalized_event_type = normalize_payment_event_type(provider, str(provider_status or ""), event_type)
    return {
        "deal_id": _payload_value(payload, "deal_id"),
        "invoice_id": _payload_value(payload, "invoice_id"),
        "payment_provider": provider,
        "provider_account_id": extract_provider_account_id(payload, headers),
        "external_event_id": extract_external_event_id(payload, headers),
        "external_payment_id": extract_external_payment_id(payload, headers),
        "event_type": normalized_event_type,
        "provider_event_type": event_type,
        "provider_status": provider_status,
        "status": normalized_status,
        "amount": _payload_value(payload, "amount", "paid_amount"),
        "currency": str(_payload_value(payload, "currency") or "KZT").upper(),
        "paid_at": _payload_value(payload, "paid_at", "captured_at", "succeeded_at"),
        "payment_method": _payload_value(payload, "payment_method", "method"),
        "failure_reason": _payload_value(payload, "failure_reason", "error_message", "decline_reason"),
        "metadata": _payment_metadata(payload),
    }


def _coerce_payment_amount(value: Any) -> float:
    try:
        return float(value)
    except (TypeError, ValueError) as exc:
        raise HTTPException(status_code=400, detail="Invalid payment amount") from exc


def _paid_sum_for_invoice(session, *, tenant_id: str, invoice_id: str) -> float:
    value = session.execute(
        select(func.coalesce(func.sum(SalesPaymentRow.amount), 0.0)).where(
            SalesPaymentRow.tenant_id == tenant_id,
            SalesPaymentRow.invoice_id == invoice_id,
            SalesPaymentRow.status.in_(["success", "partial"]),
        )
    ).scalar_one()
    return float(value or 0.0)


def _cancel_paid_invoice_tasks(session, *, deal: SalesDealRow, invoice_id: str) -> None:
    now = utc_now_iso()
    rows = session.execute(
        select(SalesAutomationTaskRow).where(
            SalesAutomationTaskRow.tenant_id == deal.tenant_id,
            SalesAutomationTaskRow.deal_id == deal.deal_id,
            SalesAutomationTaskRow.task_type.in_({"send_invoice_reminder", "mark_invoice_overdue", "follow_up_customer"}),
            SalesAutomationTaskRow.status == "pending",
        )
    ).scalars().all()
    for row in rows:
        payload = _json_dict(row.payload_json)
        if row.task_type != "follow_up_customer" or payload.get("invoice_id") == invoice_id or payload.get("reason") == "invoice_payment":
            payload["canceled_reason"] = "invoice_paid"
            row.payload_json = json.dumps(payload, ensure_ascii=False)
            row.status = "canceled"
            row.updated_at = now


def _publish_payment_received(session, *, tenant_id: str, deal: SalesDealRow, payment: SalesPaymentRow) -> None:
    _publish_sales_event(
        session,
        tenant_id=tenant_id,
        event_type=sales_event_types.PAYMENT_RECEIVED,
        aggregate_type="payment",
        aggregate_id=payment.payment_id,
        actor_type="system",
        actor_id=None,
        payload={
            "deal_id": deal.deal_id,
            "invoice_id": payment.invoice_id,
            "payment_id": payment.payment_id,
            "payment_provider": payment.payment_provider,
            "external_payment_id": payment.external_payment_id,
            "amount": payment.amount,
            "currency": payment.currency,
            "status": payment.status,
        },
    )


def _apply_payment_update(
    session,
    *,
    tenant_id: str,
    deal: SalesDealRow,
    invoice: SalesInvoiceRow | None,
    payment_provider: str,
    external_payment_id: str | None,
    amount: Any,
    currency: str,
    status: str,
    paid_at: str | None,
    payment_method: str | None,
    failure_reason: str | None,
    metadata: dict,
    integration: TenantIntegrationRow | None = None,
    existing_payment: SalesPaymentRow | None = None,
) -> SalesPaymentRow:
    now = utc_now_iso()
    normalized_status = normalize_payment_status(payment_provider, status)
    if amount is None and normalized_status not in {"success", "partial"}:
        amount_value = 0.0
    else:
        amount_value = _coerce_payment_amount(amount)
    normalized_currency = str(currency or "KZT").strip().upper()
    if normalized_status in {"success", "partial"}:
        if amount_value <= 0:
            raise HTTPException(status_code=400, detail="Payment amount must be positive")
        if deal.status == "lost":
            raise HTTPException(status_code=400, detail="Cannot apply payment to lost deal")
        if invoice is not None:
            if invoice.status == "canceled":
                raise HTTPException(status_code=400, detail="Cannot apply payment to canceled invoice")
            if normalized_currency != str(invoice.currency or "").upper():
                raise HTTPException(status_code=400, detail="Payment currency does not match invoice currency")
            if amount_value > float(invoice.amount or 0) and not _payment_allows_overpayment(integration, metadata):
                raise HTTPException(status_code=400, detail="Payment amount exceeds invoice amount")

    row = existing_payment
    previous_payment_status = row.status if row is not None else None
    if row is None and external_payment_id:
        row = session.execute(
            select(SalesPaymentRow).where(
                SalesPaymentRow.tenant_id == tenant_id,
                SalesPaymentRow.payment_provider == payment_provider,
                SalesPaymentRow.external_payment_id == external_payment_id,
            )
        ).scalar_one_or_none()
        previous_payment_status = row.status if row is not None else None

    if row is None:
        row = SalesPaymentRow(
            payment_id=new_id("pay"),
            tenant_id=tenant_id,
            deal_id=deal.deal_id,
            invoice_id=invoice.invoice_id if invoice is not None else None,
            payment_provider=payment_provider,
            external_payment_id=external_payment_id,
            amount=amount_value,
            currency=normalized_currency,
            status=normalized_status,
            paid_at=paid_at or (now if normalized_status in {"success", "partial"} else None),
            payment_method=payment_method,
            failure_reason=failure_reason,
            metadata_json=json.dumps(metadata, ensure_ascii=False),
            created_at=now,
            updated_at=now,
        )
        session.add(row)
        session.flush()
    else:
        row.deal_id = deal.deal_id
        row.invoice_id = invoice.invoice_id if invoice is not None else row.invoice_id
        row.amount = amount_value
        row.currency = normalized_currency
        row.status = normalized_status
        row.paid_at = paid_at or row.paid_at or (now if normalized_status in {"success", "partial"} else None)
        row.payment_method = payment_method
        row.failure_reason = failure_reason
        row.metadata_json = json.dumps(metadata, ensure_ascii=False)
        row.updated_at = now
        session.flush()

    if normalized_status in {"success", "partial"} and previous_payment_status != normalized_status:
        _publish_payment_received(session, tenant_id=tenant_id, deal=deal, payment=row)

    if invoice is None:
        return row

    previous_invoice_status = invoice.status
    previous_deal_status = deal.status
    lead = _get_lead(session, deal.lead_id, tenant_id) if deal.lead_id else None
    counterparty = session.execute(
        select(SalesCounterpartyRow).where(SalesCounterpartyRow.deal_id == deal.deal_id, SalesCounterpartyRow.tenant_id == tenant_id)
    ).scalar_one_or_none()

    if normalized_status in {"success", "partial"}:
        paid_sum = _paid_sum_for_invoice(session, tenant_id=tenant_id, invoice_id=invoice.invoice_id)
        if paid_sum <= 0:
            pass
        elif paid_sum < float(invoice.amount or 0):
            invoice.status = "partially_paid"
            invoice.updated_at = now
            _apply_payment_stage(
                session,
                deal,
                "partially_paid",
                reason="payment.partial",
                metadata={
                    "invoice_id": invoice.invoice_id,
                    "payment_id": row.payment_id,
                    "external_payment_id": row.external_payment_id,
                    "payment_provider": row.payment_provider,
                    "paid_sum": paid_sum,
                },
            )
        else:
            invoice.status = "paid"
            invoice.paid_at = row.paid_at or now
            invoice.updated_at = now
            deal.final_amount = paid_sum
            payment_metadata = {
                "invoice_id": invoice.invoice_id,
                "payment_id": row.payment_id,
                "external_payment_id": row.external_payment_id,
                "payment_provider": row.payment_provider,
                "paid_sum": paid_sum,
            }
            if previous_deal_status != "won":
                _apply_payment_stage(session, deal, "paid", reason="payment.received", metadata=payment_metadata)
                _apply_payment_stage(session, deal, "won", reason="payment.received", metadata=payment_metadata)
            _ensure_crm_customer(session, lead=lead, deal=deal, counterparty=counterparty)
            if previous_invoice_status != "paid":
                _publish_sales_event(
                    session,
                    tenant_id=tenant_id,
                    event_type=sales_event_types.INVOICE_PAID,
                    aggregate_type="invoice",
                    aggregate_id=invoice.invoice_id,
                    actor_type="system",
                    actor_id=None,
                    payload={
                        "deal_id": deal.deal_id,
                        "invoice_id": invoice.invoice_id,
                        "paid_amount": paid_sum,
                        "currency": row.currency,
                        "paid_at": row.paid_at,
                    },
                )
            if previous_deal_status != "won":
                _publish_sales_event(
                    session,
                    tenant_id=tenant_id,
                    event_type=sales_event_types.DEAL_WON,
                    aggregate_type="deal",
                    aggregate_id=deal.deal_id,
                    actor_type="system",
                    actor_id=None,
                    payload={
                        "deal_id": deal.deal_id,
                        "invoice_id": invoice.invoice_id,
                        "payment_id": row.payment_id,
                        "final_amount": deal.final_amount,
                        "currency": deal.currency,
                        "won_reason": "payment_received",
                    },
                )
            _cancel_paid_invoice_tasks(session, deal=deal, invoice_id=invoice.invoice_id)
    elif normalized_status in {"failed", "canceled"}:
        invoice.updated_at = now
    return row


def _touch_deal_contact(deal: SalesDealRow) -> None:
    now = utc_now_iso()
    deal.last_contact_at = now
    deal.updated_at = now


def _start_communication(
    session,
    *,
    deal: SalesDealRow,
    lead: SalesLeadRow | None,
    payload: SalesCommunicationStartIn,
    actor_user: str | None = None,
    metadata: dict | None = None,
    skip_stage_transition: bool = False,
) -> SalesCommunicationSessionRow:
    now = utc_now_iso()
    metadata_payload = dict(metadata or {})
    skip_stage_transition = skip_stage_transition or bool(metadata_payload.pop("skip_stage_transition", False))
    channel_provider = str(metadata_payload.get("channel_provider") or "").strip()
    if not channel_provider:
        channel_provider = "voice" if payload.channel_type == "voice" else str(deal.current_channel if deal.current_channel != "voice" else (lead.preferred_channel if lead else "telegram") or "telegram")
        metadata_payload["channel_provider"] = channel_provider
    agent_type = payload.agent_type or ("voice_ai" if payload.channel_type == "voice" else "text_ai")
    communication = SalesCommunicationSessionRow(
        communication_id=new_id("com"),
        tenant_id=deal.tenant_id,
        deal_id=deal.deal_id,
        lead_id=lead.lead_id if lead else deal.lead_id,
        customer_id=deal.customer_id,
        channel_type=payload.channel_type,
        channel_provider=channel_provider,
        direction=payload.direction,
        agent_type=agent_type,
        started_at=now,
        ended_at=None,
        duration_sec=None,
        subject=payload.subject,
        status="active",
        summary=payload.summary,
        transcript_id=None,
        next_action_type=payload.next_action_type,
        next_action_at=payload.next_action_at,
        sentiment=None,
        result_code=None,
        metadata_json=json.dumps(metadata_payload, ensure_ascii=False),
        created_at=now,
        updated_at=now,
    )
    session.add(communication)
    deal.current_channel = "voice" if payload.channel_type == "voice" else (lead.preferred_channel if lead else deal.current_channel)
    _touch_deal_contact(deal)
    communication_actor_type = "system" if agent_type.endswith("_ai") else "human"
    _publish_sales_event(
        session,
        tenant_id=deal.tenant_id,
        event_type=sales_event_types.COMMUNICATION_STARTED,
        aggregate_type="communication",
        aggregate_id=communication.communication_id,
        actor_type=communication_actor_type,
        actor_id=actor_user,
        payload={
            "deal_id": deal.deal_id,
            "lead_id": communication.lead_id,
            "communication_session_id": communication.communication_id,
            "channel_type": communication.channel_type,
            "direction": communication.direction,
            "agent_type": communication.agent_type,
            "subject": communication.subject,
        },
    )
    if not skip_stage_transition:
        _apply_stage(
            session,
            deal,
            _infer_stage_for_channel(payload.channel_type),
            actor_type=communication_actor_type,
            actor_id=actor_user,
            reason=payload.subject or "communication.started",
        )
    if payload.next_action_type:
        _schedule_task(
            session,
            deal=deal,
            task_type=payload.next_action_type,
            run_at=payload.next_action_at,
            payload={"source": "communication.start", "communication_id": communication.communication_id},
        )
    return communication


def _resolve_deal_for_inbound(
    session,
    *,
    tenant_id: str,
    deal_id: str | None,
    lead_id: str | None,
    phone: str | None,
    subject: str | None,
    source_channel: str,
    preferred_channel: str,
) -> tuple[SalesLeadRow | None, SalesDealRow]:
    if deal_id:
        return None, _get_deal(session, deal_id, tenant_id)
    if lead_id:
        lead = _get_lead(session, lead_id, tenant_id)
        deal = session.execute(
            select(SalesDealRow)
            .where(SalesDealRow.lead_id == lead.lead_id, SalesDealRow.tenant_id == tenant_id)
            .order_by(SalesDealRow.id.desc())
        ).scalars().first()
        if deal is None:
            raise HTTPException(status_code=404, detail="Lead has no deal")
        return lead, deal

    normalized_phone = str(phone or "").strip()
    if normalized_phone:
        lead = session.execute(
            select(SalesLeadRow)
            .where(SalesLeadRow.phone == normalized_phone, SalesLeadRow.tenant_id == tenant_id)
            .order_by(SalesLeadRow.updated_at.desc(), SalesLeadRow.id.desc())
        ).scalars().first()
        if lead is not None:
            deal = session.execute(
                select(SalesDealRow)
                .where(
                    SalesDealRow.lead_id == lead.lead_id,
                    SalesDealRow.status == "active",
                    SalesDealRow.tenant_id == tenant_id,
                )
                .order_by(SalesDealRow.updated_at.desc(), SalesDealRow.id.desc())
            ).scalars().first()
            if deal is not None:
                return lead, deal

        crm_customer = _find_customer_by_phone(session, normalized_phone)
        now = utc_now_iso()
        pipeline = get_default_pipeline(session, tenant_id)
        stage = resolve_stage_by_code_or_id(
            session,
            tenant_id=tenant_id,
            pipeline_id=pipeline.pipeline_id,
            stage_id=DEFAULT_STAGE_CODE,
        )
        lead = SalesLeadRow(
            lead_id=new_id("sld"),
            tenant_id=tenant_id,
            source_type="inbound",
            source_channel=source_channel,
            source_campaign_id=None,
            full_name=crm_customer.display_name if crm_customer else "Новый лид",
            company_name=None,
            phone=normalized_phone,
            email=None,
            messenger_handles_json="{}",
            lead_temperature="warm",
            lead_score=55,
            customer_type=None,
            segment_type=None,
            initial_need_summary=subject,
            preferred_channel=preferred_channel,
            assigned_agent_type="voice_ai" if preferred_channel == "voice" else "text_ai",
            status="new_qualified_lead",
            crm_customer_id=crm_customer.customer_id if crm_customer else None,
            created_at=now,
            updated_at=now,
        )
        deal = SalesDealRow(
            deal_id=new_id("sde"),
            tenant_id=tenant_id,
            lead_id=lead.lead_id,
            customer_id=crm_customer.customer_id if crm_customer else None,
            pipeline_id=pipeline.pipeline_id,
            stage_id=stage.stage_id,
            scenario_type="quick_sale",
            priority=3,
            title=subject or f"Новый лид {normalized_phone}",
            need_summary=subject,
            product_context_json="{}",
            estimated_amount=None,
            final_amount=None,
            currency=DEFAULT_SALES_CURRENCY,
            payment_model=None,
            document_required=True,
            payment_required=True,
            assigned_human_user_id=None,
            assigned_ai_orchestrator_id=None,
            preferred_channel=preferred_channel,
            current_channel=preferred_channel,
            status="active",
            won_reason=None,
            lost_reason=None,
            close_reason=None,
            next_action_type=None,
            next_action_at=None,
            last_contact_at=None,
            closed_at=None,
            created_at=now,
            updated_at=now,
        )
        session.add(lead)
        session.add(deal)
        _record_stage_change(
            session,
            deal=deal,
            from_stage_id=None,
            to_stage_id=stage.stage_id,
            changed_by_type="system",
            changed_by_id=None,
            reason="lead.entered_crm",
        )
        _publish_lead_entered_event(session, lead=lead, deal=deal, stage=stage, actor_type="system", actor_id=None)
        return lead, deal

    raise HTTPException(status_code=400, detail="Cannot resolve inbound contact without deal_id, lead_id or phone")


def _bridge_voice_runtime(call: SalesCallRow, communication: SalesCommunicationSessionRow, deal: SalesDealRow) -> None:
    if not _voice_runtime_enabled():
        return
    metadata = _json_dict(communication.metadata_json)
    if metadata.get("voice_session_id"):
        return
    payload = {
        "call_id": call.external_call_id or call.call_id,
        "interaction_id": None,
        "customer_id": deal.customer_id,
        "queue_id": _default_handoff_queue_id(),
        "ai_session_id": None,
        "agent_profile": "voice_support",
        "language": "ru",
        "handoff_queue_id": _default_handoff_queue_id(),
        "greeting_text": f"Здравствуйте! Продолжим работу по сделке: {deal.title}.",
    }
    try:
        with httpx.Client(timeout=8.0) as client:
            response = client.post(
                f"{_voice_runtime_url()}/internal/voice-ai/sessions",
                json=payload,
                headers=_service_headers("svc:sales-service", "sales-service", "sales-service", deal.tenant_id),
            )
        response.raise_for_status()
        data = response.json()
        metadata["voice_session_id"] = data.get("voice_session_id")
        metadata["voice_runtime"] = "attached"
        communication.metadata_json = json.dumps(metadata, ensure_ascii=False)
        communication.updated_at = utc_now_iso()
    except Exception as exc:  # noqa: BLE001
        metadata["voice_runtime"] = "degraded"
        metadata["voice_runtime_error"] = str(exc)
        communication.metadata_json = json.dumps(metadata, ensure_ascii=False)
        communication.updated_at = utc_now_iso()


def _bridge_telegram_outbound(communication: SalesCommunicationSessionRow, message: SalesMessageRow, deal: SalesDealRow) -> None:
    if not _telegram_bridge_enabled():
        return
    metadata = _json_dict(communication.metadata_json)
    thread_id = str(metadata.get("telegram_thread_id") or "").strip()
    if not thread_id or message.channel_provider != "telegram":
        return
    payload = {"text": message.body}
    try:
        with httpx.Client(timeout=8.0) as client:
            response = client.post(
                f"{_telegram_adapter_url()}/integrations/telegram/threads/{thread_id}/messages",
                json=payload,
                headers=_service_headers("svc:sales-service", "sales-service", "sales-service", deal.tenant_id),
            )
        response.raise_for_status()
        message.delivery_status = "sent"
    except Exception as exc:  # noqa: BLE001
        message.delivery_status = "queued"
        metadata["telegram_bridge_error"] = str(exc)
        communication.metadata_json = json.dumps(metadata, ensure_ascii=False)
        communication.updated_at = utc_now_iso()


def _counterparty_status(payload: SalesCounterpartyUpsertIn) -> str:
    required_hits = sum(
        1
        for value in [
            payload.company_name or payload.full_name,
            payload.email_for_docs or payload.phone_for_docs,
            payload.bin_iin or payload.signer_name,
        ]
        if str(value or "").strip()
    )
    return "completed" if required_hits >= 2 else "requested"


def _invoice_number() -> str:
    return f"INV-{_now().strftime('%Y%m%d')}-{new_id('num').split('_', 1)[1].upper()}"


def _merge_communication_metadata(communication: SalesCommunicationSessionRow, values: dict[str, Any]) -> dict:
    metadata = _json_dict(communication.metadata_json)
    changed = False
    for key, value in values.items():
        if value is None:
            if key in metadata:
                metadata.pop(key, None)
                changed = True
            continue
        if isinstance(value, str):
            normalized = value.strip()
            if not normalized:
                if key in metadata:
                    metadata.pop(key, None)
                    changed = True
                continue
            value = normalized
        if metadata.get(key) != value:
            metadata[key] = value
            changed = True
    if changed:
        communication.metadata_json = json.dumps(metadata, ensure_ascii=False)
        communication.updated_at = utc_now_iso()
    channel_provider = str(metadata.get("channel_provider") or "").strip()
    if channel_provider and communication.channel_provider != channel_provider:
        communication.channel_provider = channel_provider
        communication.updated_at = utc_now_iso()
    return metadata


def _upsert_external_link(
    session,
    *,
    deal: SalesDealRow,
    channel_provider: str,
    communication_id: str | None = None,
    sales_call_id: str | None = None,
    external_thread_id: str | None = None,
    external_chat_id: str | None = None,
    external_call_id: str | None = None,
    voice_session_id: str | None = None,
    ai_session_id: str | None = None,
    interaction_id: str | None = None,
    customer_id: str | None = None,
    phone_number: str | None = None,
    external_status: str | None = None,
    metadata: dict[str, Any] | None = None,
) -> SalesExternalLinkRow:
    row = _find_external_link(
        session,
        tenant_id=deal.tenant_id,
        thread_id=external_thread_id,
        voice_session_id=voice_session_id,
        external_call_id=external_call_id,
        interaction_id=interaction_id,
    )
    now = utc_now_iso()
    if row is None:
        row = SalesExternalLinkRow(
            link_id=new_id("xlk"),
            tenant_id=deal.tenant_id,
            deal_id=deal.deal_id,
            communication_id=communication_id,
            sales_call_id=sales_call_id,
            channel_provider=channel_provider,
            external_thread_id=external_thread_id,
            external_chat_id=external_chat_id,
            external_call_id=external_call_id,
            voice_session_id=voice_session_id,
            ai_session_id=ai_session_id,
            interaction_id=interaction_id,
            customer_id=customer_id or deal.customer_id,
            phone_number=phone_number,
            external_status=external_status,
            link_metadata_json=json.dumps(metadata or {}, ensure_ascii=False),
            created_at=now,
            updated_at=now,
            last_sync_at=now,
        )
        session.add(row)
        return row

    row.deal_id = deal.deal_id
    row.communication_id = communication_id or row.communication_id
    row.sales_call_id = sales_call_id or row.sales_call_id
    row.channel_provider = channel_provider or row.channel_provider
    row.external_thread_id = str(external_thread_id or "").strip() or row.external_thread_id
    row.external_chat_id = str(external_chat_id or "").strip() or row.external_chat_id
    row.external_call_id = str(external_call_id or "").strip() or row.external_call_id
    row.voice_session_id = str(voice_session_id or "").strip() or row.voice_session_id
    row.ai_session_id = str(ai_session_id or "").strip() or row.ai_session_id
    row.interaction_id = str(interaction_id or "").strip() or row.interaction_id
    row.customer_id = str(customer_id or "").strip() or row.customer_id or deal.customer_id
    row.phone_number = str(phone_number or "").strip() or row.phone_number
    row.external_status = str(external_status or "").strip() or row.external_status
    merged_meta = _json_dict(row.link_metadata_json)
    merged_meta.update(metadata or {})
    row.link_metadata_json = json.dumps(merged_meta, ensure_ascii=False)
    row.updated_at = now
    row.last_sync_at = now
    return row


def _get_or_create_text_communication(
    session,
    *,
    deal: SalesDealRow,
    lead: SalesLeadRow | None,
    direction: str,
    subject: str | None,
    metadata: dict | None = None,
    thread_id: str | None = None,
) -> SalesCommunicationSessionRow:
    metadata_payload = dict(metadata or {})
    skip_stage_transition = bool(metadata_payload.pop("skip_stage_transition", False))
    if str(thread_id or "").strip():
        link = _find_external_link(session, tenant_id=deal.tenant_id, thread_id=thread_id)
        if link and link.communication_id:
            existing = session.execute(
                select(SalesCommunicationSessionRow).where(
                    SalesCommunicationSessionRow.communication_id == link.communication_id,
                    SalesCommunicationSessionRow.tenant_id == deal.tenant_id,
                )
            ).scalar_one_or_none()
            if existing is not None:
                _merge_communication_metadata(existing, metadata_payload)
                return existing
    existing = session.execute(
        select(SalesCommunicationSessionRow)
        .where(SalesCommunicationSessionRow.deal_id == deal.deal_id, SalesCommunicationSessionRow.tenant_id == deal.tenant_id)
        .where(SalesCommunicationSessionRow.channel_type == "text")
        .order_by(SalesCommunicationSessionRow.started_at.desc(), SalesCommunicationSessionRow.id.desc())
    ).scalars().first()
    if existing is not None:
        _merge_communication_metadata(existing, metadata_payload)
        return existing
    return _start_communication(
        session,
        deal=deal,
        lead=lead,
        payload=SalesCommunicationStartIn(
            channel_type="text",
            direction=direction,  # type: ignore[arg-type]
            agent_type="human" if direction == "outbound" else "text_ai",
            subject=subject,
        ),
        metadata=metadata_payload,
        skip_stage_transition=skip_stage_transition,
    )


def _get_or_create_voice_communication(
    session,
    *,
    deal: SalesDealRow,
    lead: SalesLeadRow | None,
    subject: str | None,
    metadata: dict | None = None,
    voice_session_id: str | None = None,
) -> SalesCommunicationSessionRow:
    if str(voice_session_id or "").strip():
        link = _find_external_link(session, tenant_id=deal.tenant_id, voice_session_id=voice_session_id)
        if link and link.communication_id:
            existing = session.execute(
                select(SalesCommunicationSessionRow).where(
                    SalesCommunicationSessionRow.communication_id == link.communication_id,
                    SalesCommunicationSessionRow.tenant_id == deal.tenant_id,
                )
            ).scalar_one_or_none()
            if existing is not None:
                _merge_communication_metadata(existing, metadata or {})
                return existing
    existing = session.execute(
        select(SalesCommunicationSessionRow)
        .where(SalesCommunicationSessionRow.deal_id == deal.deal_id, SalesCommunicationSessionRow.tenant_id == deal.tenant_id)
        .where(SalesCommunicationSessionRow.channel_type == "voice")
        .order_by(SalesCommunicationSessionRow.started_at.desc(), SalesCommunicationSessionRow.id.desc())
    ).scalars().first()
    if existing is not None:
        _merge_communication_metadata(existing, metadata or {})
        return existing
    return _start_communication(
        session,
        deal=deal,
        lead=lead,
        payload=SalesCommunicationStartIn(
            channel_type="voice",
            direction="inbound",
            agent_type="voice_ai",
            subject=subject,
        ),
        metadata=metadata,
    )


def _channel_type_for_value(value: str | None) -> str:
    return "voice" if str(value or "").strip().lower() == "voice" else "text"


def _stage_code_for_deal(session, deal: SalesDealRow) -> str | None:
    _, stage = _deal_pipeline_stage(session, deal)
    return stage.code if stage is not None else None


def _maybe_transition_deal_for_channel_switch(
    session,
    *,
    deal: SalesDealRow,
    to_channel: str,
    actor_type: str,
    actor_id: str | None,
    reason: str,
    metadata: dict,
) -> None:
    current_stage_code = _stage_code_for_deal(session, deal)
    if current_stage_code not in {"active_text_communication", "active_voice_communication"}:
        return
    target_stage_code = "active_voice_communication" if to_channel == "voice" else "active_text_communication"
    if current_stage_code == target_stage_code:
        return
    transition_deal_stage(
        session,
        tenant_id=deal.tenant_id,
        deal_id=deal.deal_id,
        target_stage_code=target_stage_code,
        actor_type=actor_type,
        actor_id=actor_id,
        reason=reason,
        metadata=metadata,
    )


def _create_switch_communication_session(
    session,
    *,
    deal: SalesDealRow,
    lead: SalesLeadRow | None,
    to_channel: str,
    reason: str,
    actor_user: str | None,
    metadata: dict,
) -> SalesCommunicationSessionRow:
    now = utc_now_iso()
    channel_provider = str(metadata.get("channel_provider") or ("voice" if to_channel == "voice" else "text")).strip()
    payload = dict(metadata)
    payload["channel_provider"] = channel_provider
    communication = SalesCommunicationSessionRow(
        communication_id=new_id("com"),
        tenant_id=deal.tenant_id,
        deal_id=deal.deal_id,
        lead_id=lead.lead_id if lead else deal.lead_id,
        customer_id=deal.customer_id,
        channel_type=to_channel,
        channel_provider=channel_provider,
        direction="outbound",
        agent_type="human",
        started_at=now,
        ended_at=None,
        duration_sec=None,
        subject=reason,
        status="active",
        summary=None,
        transcript_id=None,
        next_action_type=None,
        next_action_at=None,
        sentiment=None,
        result_code=None,
        metadata_json=json.dumps(payload, ensure_ascii=False),
        created_at=now,
        updated_at=now,
    )
    session.add(communication)
    _publish_sales_event(
        session,
        tenant_id=deal.tenant_id,
        event_type=sales_event_types.COMMUNICATION_STARTED,
        aggregate_type="communication",
        aggregate_id=communication.communication_id,
        actor_type="human",
        actor_id=actor_user,
        payload={
            "deal_id": deal.deal_id,
            "lead_id": communication.lead_id,
            "communication_session_id": communication.communication_id,
            "channel_type": communication.channel_type,
            "channel_provider": communication.channel_provider,
            "direction": communication.direction,
            "agent_type": communication.agent_type,
            "subject": communication.subject,
        },
    )
    return communication


def _complete_channel_switch_recommendation_task(
    session,
    *,
    tenant_id: str,
    deal_id: str,
    task_id: str | None,
    switch_id: str,
) -> None:
    normalized = str(task_id or "").strip()
    if not normalized:
        return
    task = session.execute(
        select(SalesAutomationTaskRow).where(
            SalesAutomationTaskRow.tenant_id == tenant_id,
            SalesAutomationTaskRow.deal_id == deal_id,
            SalesAutomationTaskRow.task_id == normalized,
            SalesAutomationTaskRow.task_type == "recommend_channel_switch",
        )
    ).scalar_one_or_none()
    if task is None:
        return
    payload = _json_dict(task.payload_json)
    payload["actual_switch_id"] = switch_id
    task.payload_json = json.dumps(payload, ensure_ascii=False)
    if task.status in {"pending", "running"}:
        task.status = "completed"
        task.completed_at = utc_now_iso()
    task.updated_at = utc_now_iso()


def _perform_channel_switch(
    session,
    *,
    deal: SalesDealRow,
    payload: SalesCommunicationSwitchChannelIn,
    actor_type: str,
    actor_id: str | None,
    communication: SalesCommunicationSessionRow | None = None,
) -> tuple[SalesChannelSwitchRow, SalesCommunicationSessionRow | None]:
    if payload.to_channel not in {"text", "voice"}:
        raise HTTPException(status_code=400, detail="Invalid target channel")
    if payload.recommended_by_task_id:
        existing = session.execute(
            select(SalesChannelSwitchRow)
            .where(
                SalesChannelSwitchRow.tenant_id == deal.tenant_id,
                SalesChannelSwitchRow.deal_id == deal.deal_id,
                SalesChannelSwitchRow.recommended_by_task_id == payload.recommended_by_task_id,
            )
            .order_by(SalesChannelSwitchRow.id.desc())
        ).scalars().first()
        if existing is not None:
            return existing, None

    now = utc_now_iso()
    from_channel = _channel_type_for_value(communication.channel_type if communication is not None else deal.current_channel)
    reason_text = payload.reason_text or payload.reason_for_channel_switch or payload.reason_code
    reason_code = str(payload.reason_code or "human_decision").strip() or "human_decision"
    metadata = dict(payload.metadata or {})
    metadata.update(
        {
            "reason_code": reason_code,
            "scheduled_at": payload.scheduled_at,
            "source_type": payload.source_type,
            "source_id": payload.source_id,
            "recommended_by_task_id": payload.recommended_by_task_id,
        }
    )
    new_communication = None
    if payload.create_session:
        lead = _get_lead(session, deal.lead_id, deal.tenant_id) if deal.lead_id else None
        new_communication = _create_switch_communication_session(
            session,
            deal=deal,
            lead=lead,
            to_channel=payload.to_channel,
            reason=reason_text,
            actor_user=actor_id,
            metadata=metadata,
        )

    row = SalesChannelSwitchRow(
        switch_id=new_id("swc"),
        tenant_id=deal.tenant_id,
        deal_id=deal.deal_id,
        communication_id=communication.communication_id if communication is not None else None,
        communication_session_id=communication.communication_id if communication is not None else None,
        previous_communication_session_id=communication.communication_id if communication is not None else None,
        new_communication_session_id=new_communication.communication_id if new_communication is not None else None,
        from_channel=from_channel,
        to_channel=payload.to_channel,
        reason_code=reason_code,
        reason_text=payload.reason_text,
        reason_for_channel_switch=reason_text,
        initiated_by_type=actor_type,
        initiated_by_id=actor_id,
        source_type=payload.source_type,
        source_id=payload.source_id,
        recommended_by_task_id=payload.recommended_by_task_id,
        switched_at=now,
        created_at=now,
    )
    session.add(row)
    deal.current_channel = payload.to_channel
    deal.updated_at = now
    _maybe_transition_deal_for_channel_switch(
        session,
        deal=deal,
        to_channel=payload.to_channel,
        actor_type=actor_type,
        actor_id=actor_id,
        reason=reason_text,
        metadata={"switch_id": row.switch_id, **metadata},
    )
    _complete_channel_switch_recommendation_task(
        session,
        tenant_id=deal.tenant_id,
        deal_id=deal.deal_id,
        task_id=payload.recommended_by_task_id,
        switch_id=row.switch_id,
    )
    _publish_sales_event(
        session,
        tenant_id=deal.tenant_id,
        event_type=sales_event_types.COMMUNICATION_CHANNEL_SWITCHED,
        aggregate_type="deal",
        aggregate_id=deal.deal_id,
        actor_type=actor_type,
        actor_id=actor_id,
        payload={
            "deal_id": deal.deal_id,
            "switch_id": row.switch_id,
            "from_channel": row.from_channel,
            "to_channel": row.to_channel,
            "reason_code": row.reason_code,
            "reason_text": row.reason_text,
            "communication_session_id": row.communication_session_id,
            "new_communication_session_id": row.new_communication_session_id,
            "recommended_by_task_id": row.recommended_by_task_id,
        },
    )
    return row, new_communication


def _build_workspace(session, deal: SalesDealRow) -> SalesWorkspaceOut:
    lead = _get_lead(session, deal.lead_id, deal.tenant_id) if deal.lead_id else None
    pipeline, stage = _deal_pipeline_stage(session, deal)
    communications = session.execute(
        select(SalesCommunicationSessionRow)
        .where(SalesCommunicationSessionRow.deal_id == deal.deal_id, SalesCommunicationSessionRow.tenant_id == deal.tenant_id)
        .order_by(SalesCommunicationSessionRow.started_at.desc(), SalesCommunicationSessionRow.id.desc())
    ).scalars().all()
    messages = session.execute(
        select(SalesMessageRow)
        .where(SalesMessageRow.deal_id == deal.deal_id, SalesMessageRow.tenant_id == deal.tenant_id)
        .order_by(SalesMessageRow.sent_at.desc(), SalesMessageRow.id.desc())
    ).scalars().all()
    calls = session.execute(
        select(SalesCallRow)
        .where(SalesCallRow.deal_id == deal.deal_id, SalesCallRow.tenant_id == deal.tenant_id)
        .order_by(SalesCallRow.started_at.desc(), SalesCallRow.id.desc())
    ).scalars().all()
    call_ids = [row.call_id for row in calls]
    transcripts = (
        session.execute(
            select(SalesTranscriptRow)
            .where(SalesTranscriptRow.tenant_id == deal.tenant_id, SalesTranscriptRow.call_id.in_(call_ids))
            .order_by(SalesTranscriptRow.updated_at.desc(), SalesTranscriptRow.id.desc())
        ).scalars().all()
        if call_ids
        else []
    )
    notes = session.execute(
        select(SalesNoteRow)
        .where(
            SalesNoteRow.deal_id == deal.deal_id,
            SalesNoteRow.tenant_id == deal.tenant_id,
            SalesNoteRow.archived_at.is_(None),
        )
        .order_by(SalesNoteRow.created_at.desc(), SalesNoteRow.id.desc())
    ).scalars().all()
    offers = session.execute(
        select(SalesOfferRow)
        .where(SalesOfferRow.deal_id == deal.deal_id, SalesOfferRow.tenant_id == deal.tenant_id)
        .order_by(SalesOfferRow.updated_at.desc(), SalesOfferRow.id.desc())
    ).scalars().all()
    conditions = session.execute(
        select(SalesConditionRow)
        .where(SalesConditionRow.deal_id == deal.deal_id, SalesConditionRow.tenant_id == deal.tenant_id)
        .order_by(SalesConditionRow.updated_at.desc(), SalesConditionRow.id.desc())
    ).scalars().all()
    counterparty = session.execute(
        select(SalesCounterpartyRow).where(SalesCounterpartyRow.deal_id == deal.deal_id, SalesCounterpartyRow.tenant_id == deal.tenant_id)
    ).scalar_one_or_none()
    documents = session.execute(
        select(SalesDocumentRow)
        .where(SalesDocumentRow.deal_id == deal.deal_id, SalesDocumentRow.tenant_id == deal.tenant_id)
        .order_by(SalesDocumentRow.updated_at.desc(), SalesDocumentRow.id.desc())
    ).scalars().all()
    invoices = session.execute(
        select(SalesInvoiceRow)
        .where(SalesInvoiceRow.deal_id == deal.deal_id, SalesInvoiceRow.tenant_id == deal.tenant_id)
        .order_by(SalesInvoiceRow.updated_at.desc(), SalesInvoiceRow.id.desc())
    ).scalars().all()
    payments = session.execute(
        select(SalesPaymentRow)
        .where(SalesPaymentRow.deal_id == deal.deal_id, SalesPaymentRow.tenant_id == deal.tenant_id)
        .order_by(SalesPaymentRow.updated_at.desc(), SalesPaymentRow.id.desc())
    ).scalars().all()
    escalations = session.execute(
        select(SalesEscalationRow)
        .where(SalesEscalationRow.deal_id == deal.deal_id, SalesEscalationRow.tenant_id == deal.tenant_id)
        .order_by(SalesEscalationRow.created_at.desc(), SalesEscalationRow.id.desc())
    ).scalars().all()
    tasks = session.execute(
        select(SalesAutomationTaskRow)
        .where(SalesAutomationTaskRow.deal_id == deal.deal_id, SalesAutomationTaskRow.tenant_id == deal.tenant_id)
        .order_by(SalesAutomationTaskRow.run_at.desc(), SalesAutomationTaskRow.id.desc())
    ).scalars().all()
    recommended_channel_switch = None
    for task in tasks:
        if task.task_type != "recommend_channel_switch":
            continue
        payload = _json_dict(task.payload_json)
        result = payload.get("automation_result") if isinstance(payload.get("automation_result"), dict) else None
        recommended_channel_switch = {
            "task_id": task.task_id,
            "status": task.status,
            **(result or payload),
        }
        break
    stage_history = session.execute(
        select(SalesStageHistoryRow)
        .where(SalesStageHistoryRow.deal_id == deal.deal_id, SalesStageHistoryRow.tenant_id == deal.tenant_id)
        .order_by(SalesStageHistoryRow.changed_at.desc(), SalesStageHistoryRow.id.desc())
    ).scalars().all()
    stage_refs = _stage_refs_by_id(
        session,
        deal.tenant_id,
        [stage.stage_id if stage else None]
        + [row.from_stage_id for row in stage_history]
        + [row.to_stage_id for row in stage_history],
    )
    channel_switches = session.execute(
        select(SalesChannelSwitchRow)
        .where(SalesChannelSwitchRow.deal_id == deal.deal_id, SalesChannelSwitchRow.tenant_id == deal.tenant_id)
        .order_by(SalesChannelSwitchRow.switched_at.desc(), SalesChannelSwitchRow.id.desc())
    ).scalars().all()

    timeline: list[SalesTimelineEventOut] = []
    for row in stage_history:
        to_stage = stage_refs.get(row.to_stage_id)
        timeline.append(
            SalesTimelineEventOut(
                ts=row.changed_at,
                kind="stage",
                title=f"Этап: {to_stage.name if to_stage else row.to_stage_id}",
                body=row.reason or "Изменение стадии сделки",
                meta={"from_stage_id": row.from_stage_id, "to_stage_id": row.to_stage_id},
            )
        )
    for row in communications:
        timeline.append(
            SalesTimelineEventOut(
                ts=row.started_at,
                kind="communication",
                title=f"Коммуникация: {'звонок' if row.channel_type == 'voice' else 'текст'}",
                body=row.summary or row.subject or "Открыта новая сессия коммуникации",
                meta={"communication_id": row.communication_id, "agent_type": row.agent_type},
            )
        )
    for row in offers:
        timeline.append(
            SalesTimelineEventOut(
                ts=row.updated_at,
                kind="offer",
                title=f"Оффер: {row.title}",
                body=f"Статус: {row.status}. Сумма: {row.total_amount:.2f} {row.currency}",
                meta={"offer_id": row.offer_id},
            )
        )
    for row in documents:
        timeline.append(
            SalesTimelineEventOut(
                ts=row.updated_at,
                kind="document",
                title=f"Документ: {row.document_type}",
                body=f"Статус: {row.status}",
                meta={"document_id": row.document_id},
            )
        )
    for row in invoices:
        timeline.append(
            SalesTimelineEventOut(
                ts=row.updated_at,
                kind="invoice",
                title=f"Счет: {row.invoice_number}",
                body=f"Статус: {row.status}. Сумма: {row.amount:.2f} {row.currency}",
                meta={"invoice_id": row.invoice_id},
            )
        )
    for row in payments:
        timeline.append(
            SalesTimelineEventOut(
                ts=row.updated_at,
                kind="payment",
                title=f"Оплата: {row.status}",
                body=f"Сумма: {row.amount:.2f} {row.currency}",
                meta={"payment_id": row.payment_id},
            )
        )
    for row in channel_switches:
        timeline.append(
            SalesTimelineEventOut(
                ts=row.switched_at,
                kind="channel_switch",
                title=f"Смена канала: {row.from_channel} -> {row.to_channel}",
                body=row.reason_for_channel_switch,
                meta={"switch_id": row.switch_id},
            )
        )
    for row in notes:
        timeline.append(
            SalesTimelineEventOut(
                ts=row.updated_at or row.created_at,
                kind="note",
                title=f"Заметка: {row.note_type}",
                body=row.content,
                meta={"note_id": row.note_id, "source_type": row.source_type, "source_id": row.source_id},
            )
        )
    for row in escalations:
        timeline.append(
            SalesTimelineEventOut(
                ts=row.updated_at or row.resolved_at or row.created_at,
                kind="escalation",
                title=f"Эскалация: {row.status}",
                body=row.resolution_summary or row.reason,
                meta={"escalation_id": row.escalation_id, "escalation_type": row.escalation_type},
            )
        )
    timeline.sort(key=lambda item: item.ts, reverse=True)
    active_escalation = next((row for row in escalations if row.status in {"open", "assigned", "in_progress"}), None)

    return SalesWorkspaceOut(
        lead=_lead_to_out(lead, deal) if lead else None,
        deal=_deal_to_out(deal, pipeline=pipeline, stage=stage),
        pipeline=_pipeline_to_out(pipeline) if pipeline else None,
        stage=_stage_to_ref(stage),
        communications=[_communication_to_out(row) for row in communications],
        messages=[_message_to_out(row) for row in messages],
        calls=[_call_to_out(row) for row in calls],
        transcripts=[_transcript_to_out(row) for row in transcripts],
        notes=[_note_to_out(row) for row in notes],
        offers=[_offer_to_out(row) for row in offers],
        conditions=[_condition_to_out(row) for row in conditions],
        counterparty=_counterparty_to_out(counterparty) if counterparty else None,
        documents=[_document_to_out(row) for row in documents],
        invoices=[_invoice_to_out(row) for row in invoices],
        payments=[_payment_to_out(row) for row in payments],
        escalations=[_escalation_to_out(row) for row in escalations],
        active_escalation=_escalation_to_out(active_escalation) if active_escalation else None,
        tasks=[_task_to_out(row) for row in tasks],
        stage_history=[_stage_history_to_out(row, stage_refs) for row in stage_history],
        channel_switches=[_channel_switch_to_out(row) for row in channel_switches],
        recommended_channel_switch=recommended_channel_switch,
        timeline=timeline[:40],
    )


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


@app.get("/api/v1/dashboard", response_model=SalesDashboardSummaryOut)
def sales_dashboard(actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST))) -> SalesDashboardSummaryOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        leads_total = session.execute(
            select(func.count()).select_from(SalesLeadRow).where(SalesLeadRow.tenant_id == tenant_id)
        ).scalar_one()
        deals_active = session.execute(
            select(func.count()).select_from(SalesDealRow).where(SalesDealRow.tenant_id == tenant_id, SalesDealRow.status == "active")
        ).scalar_one()
        deals_won = session.execute(
            select(func.count()).select_from(SalesDealRow).where(SalesDealRow.tenant_id == tenant_id, SalesDealRow.status == "won")
        ).scalar_one()
        deals_lost = session.execute(
            select(func.count()).select_from(SalesDealRow).where(SalesDealRow.tenant_id == tenant_id, SalesDealRow.status == "lost")
        ).scalar_one()
        overdue_invoices = session.execute(
            select(func.count()).select_from(SalesInvoiceRow).where(SalesInvoiceRow.tenant_id == tenant_id, SalesInvoiceRow.status == "overdue")
        ).scalar_one()
        payment_expected = session.execute(
            select(func.coalesce(func.sum(SalesInvoiceRow.amount), 0.0)).where(
                SalesInvoiceRow.tenant_id == tenant_id,
                SalesInvoiceRow.status.in_(["issued", "sent", "partially_paid", "overdue"])
            )
        ).scalar_one()
        payment_received = session.execute(
            select(func.coalesce(func.sum(SalesPaymentRow.amount), 0.0)).where(
                SalesPaymentRow.tenant_id == tenant_id,
                SalesPaymentRow.status.in_(["success", "partial"]),
            )
        ).scalar_one()
        voice_sessions = session.execute(
            select(func.count()).select_from(SalesCommunicationSessionRow).where(
                SalesCommunicationSessionRow.tenant_id == tenant_id,
                SalesCommunicationSessionRow.channel_type == "voice",
            )
        ).scalar_one()
        text_sessions = session.execute(
            select(func.count()).select_from(SalesCommunicationSessionRow).where(
                SalesCommunicationSessionRow.tenant_id == tenant_id,
                SalesCommunicationSessionRow.channel_type == "text",
            )
        ).scalar_one()
        channel_switches = session.execute(
            select(func.count()).select_from(SalesChannelSwitchRow).where(SalesChannelSwitchRow.tenant_id == tenant_id)
        ).scalar_one()
        human_escalations = session.execute(
            select(func.count()).select_from(SalesEscalationRow).where(SalesEscalationRow.tenant_id == tenant_id)
        ).scalar_one()

        stage_rows = session.execute(
            select(SalesDealRow.stage_id, func.count(), func.coalesce(func.sum(SalesDealRow.final_amount), func.sum(SalesDealRow.estimated_amount), 0.0))
            .where(SalesDealRow.tenant_id == tenant_id)
            .group_by(SalesDealRow.stage_id)
        ).all()
        stage_refs = _stage_refs_by_id(session, tenant_id, [stage_id for stage_id, _count, _amount in stage_rows])
        stage_counts = [
            SalesDashboardStageOut(
                stage_id=stage_id,
                label=stage_refs[stage_id].name if stage_id in stage_refs else stage_id,
                stage=_stage_to_ref(stage_refs.get(stage_id)),
                count=int(count or 0),
                amount=float(amount or 0.0),
            )
            for stage_id, count, amount in stage_rows
        ]
        hottest_rows = session.execute(
            select(SalesDealRow)
            .where(SalesDealRow.tenant_id == tenant_id)
            .order_by(SalesDealRow.priority.asc(), SalesDealRow.updated_at.desc(), SalesDealRow.id.desc())
            .limit(5)
        ).scalars().all()
        return SalesDashboardSummaryOut(
            leads_total=int(leads_total or 0),
            deals_active=int(deals_active or 0),
            deals_won=int(deals_won or 0),
            deals_lost=int(deals_lost or 0),
            overdue_invoices=int(overdue_invoices or 0),
            payment_expected=float(payment_expected or 0.0),
            payment_received=float(payment_received or 0.0),
            voice_sessions=int(voice_sessions or 0),
            text_sessions=int(text_sessions or 0),
            channel_switches=int(channel_switches or 0),
            human_escalations=int(human_escalations or 0),
            stage_counts=stage_counts,
            hottest_deals=[_deal_to_out_with_refs(session, row) for row in hottest_rows],
        )
    finally:
        session.close()


@app.get("/api/v1/pipelines", response_model=list[SalesPipelineOut])
def list_pipelines(actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST))) -> list[SalesPipelineOut]:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        ensure_default_pipeline(session, tenant_id)
        rows = session.execute(
            select(SalesPipelineRow)
            .where(SalesPipelineRow.tenant_id == tenant_id)
            .order_by(SalesPipelineRow.is_default.desc(), SalesPipelineRow.created_at.asc(), SalesPipelineRow.id.asc())
        ).scalars().all()
        session.commit()
        return [_pipeline_to_out(row) for row in rows]
    finally:
        session.close()


@app.get("/api/v1/pipelines/{pipeline_id}", response_model=SalesPipelineOut)
def get_pipeline(pipeline_id: str, actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST))) -> SalesPipelineOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        ensure_default_pipeline(session, tenant_id)
        return _pipeline_to_out(_get_pipeline(session, pipeline_id, tenant_id))
    finally:
        session.close()


@app.post("/api/v1/pipelines", response_model=SalesPipelineOut)
def create_pipeline(
    payload: SalesPipelineCreate,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesPipelineOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        ensure_default_pipeline(session, tenant_id)
        now = utc_now_iso()
        pipeline = SalesPipelineRow(
            pipeline_id=new_id("pip"),
            tenant_id=tenant_id,
            code=payload.code.strip(),
            name=payload.name,
            description=payload.description,
            is_default=False,
            is_active=payload.is_active,
            created_at=now,
            updated_at=now,
        )
        session.add(pipeline)
        if payload.is_default:
            _set_default_pipeline(session, pipeline)
        try:
            session.commit()
        except IntegrityError as exc:
            session.rollback()
            raise HTTPException(status_code=409, detail="Pipeline code already exists") from exc
        return _pipeline_to_out(pipeline)
    finally:
        session.close()


@app.patch("/api/v1/pipelines/{pipeline_id}", response_model=SalesPipelineOut)
def update_pipeline(
    pipeline_id: str,
    payload: SalesPipelineUpdate,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesPipelineOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        pipeline = _get_pipeline(session, pipeline_id, tenant_id)
        updates = payload.model_dump(exclude_unset=True)
        if "name" in updates:
            pipeline.name = updates["name"]
        if "description" in updates:
            pipeline.description = updates["description"]
        if updates.get("is_active") is False and pipeline.is_default:
            raise HTTPException(status_code=400, detail="Default pipeline cannot be deactivated")
        if "is_active" in updates:
            pipeline.is_active = bool(updates["is_active"])
        if updates.get("is_default") is True:
            if not pipeline.is_active:
                raise HTTPException(status_code=400, detail="Inactive pipeline cannot be default")
            _set_default_pipeline(session, pipeline)
        elif updates.get("is_default") is False and pipeline.is_default:
            raise HTTPException(status_code=400, detail="Use another pipeline set-default action")
        pipeline.updated_at = utc_now_iso()
        session.commit()
        return _pipeline_to_out(pipeline)
    finally:
        session.close()


@app.post("/api/v1/pipelines/{pipeline_id}/set-default", response_model=SalesPipelineOut)
def set_default_pipeline(
    pipeline_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesPipelineOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        pipeline = _get_pipeline(session, pipeline_id, tenant_id)
        if not pipeline.is_active:
            raise HTTPException(status_code=400, detail="Inactive pipeline cannot be default")
        _set_default_pipeline(session, pipeline)
        session.commit()
        return _pipeline_to_out(pipeline)
    finally:
        session.close()


@app.get("/api/v1/pipelines/{pipeline_id}/stages", response_model=list[SalesPipelineStageOut])
def list_pipeline_stages(
    pipeline_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> list[SalesPipelineStageOut]:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        pipeline = _get_pipeline(session, pipeline_id, tenant_id)
        if pipeline.code == DEFAULT_PIPELINE_CODE:
            ensure_default_stages(session, tenant_id, pipeline.pipeline_id)
        rows = session.execute(
            select(SalesPipelineStageRow)
            .where(SalesPipelineStageRow.tenant_id == tenant_id, SalesPipelineStageRow.pipeline_id == pipeline_id)
            .order_by(SalesPipelineStageRow.sort_order.asc(), SalesPipelineStageRow.id.asc())
        ).scalars().all()
        session.commit()
        return [_stage_to_out(row) for row in rows]
    finally:
        session.close()


@app.post("/api/v1/pipelines/{pipeline_id}/stages", response_model=SalesPipelineStageOut)
def create_pipeline_stage(
    pipeline_id: str,
    payload: SalesPipelineStageCreate,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesPipelineStageOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        _get_pipeline(session, pipeline_id, tenant_id)
        now = utc_now_iso()
        stage = SalesPipelineStageRow(
            stage_id=new_id("pst"),
            tenant_id=tenant_id,
            pipeline_id=pipeline_id,
            code=payload.code.strip(),
            name=payload.name,
            category=payload.category,
            sort_order=payload.sort_order,
            is_terminal=payload.is_terminal,
            is_system=payload.is_system,
            is_active=payload.is_active,
            created_at=now,
            updated_at=now,
        )
        session.add(stage)
        try:
            session.commit()
        except IntegrityError as exc:
            session.rollback()
            raise HTTPException(status_code=409, detail="Stage code already exists") from exc
        return _stage_to_out(stage)
    finally:
        session.close()


@app.patch("/api/v1/pipeline-stages/{stage_id}", response_model=SalesPipelineStageOut)
def update_pipeline_stage(
    stage_id: str,
    payload: SalesPipelineStageUpdate,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesPipelineStageOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        stage = _get_stage(session, stage_id, tenant_id)
        updates = payload.model_dump(exclude_unset=True)
        if updates.get("is_active") is False and stage.is_active:
            active_deals = session.execute(
                select(func.count())
                .select_from(SalesDealRow)
                .where(
                    SalesDealRow.tenant_id == tenant_id,
                    SalesDealRow.stage_id == stage.stage_id,
                    SalesDealRow.status == "active",
                )
            ).scalar_one()
            if int(active_deals or 0) > 0:
                raise HTTPException(status_code=400, detail="Stage has active deals")
        if "name" in updates:
            stage.name = updates["name"]
        if "category" in updates:
            stage.category = updates["category"]
        if "sort_order" in updates:
            stage.sort_order = int(updates["sort_order"])
        if "is_active" in updates:
            stage.is_active = bool(updates["is_active"])
        stage.updated_at = utc_now_iso()
        session.commit()
        return _stage_to_out(stage)
    finally:
        session.close()


@app.post("/api/v1/leads", response_model=SalesLeadOut)
def create_lead(
    payload: SalesLeadCreate,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesLeadOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        now = utc_now_iso()
        crm_customer = _find_customer_by_phone(session, payload.phone)
        pipeline = get_default_pipeline(session, tenant_id)
        initial_stage_code = payload.status if payload.status != "converted" else DEFAULT_STAGE_CODE
        try:
            stage = resolve_stage_by_code_or_id(
                session,
                tenant_id=tenant_id,
                pipeline_id=pipeline.pipeline_id,
                stage_id=initial_stage_code,
            )
        except HTTPException:
            stage = resolve_stage_by_code_or_id(
                session,
                tenant_id=tenant_id,
                pipeline_id=pipeline.pipeline_id,
                stage_id=DEFAULT_STAGE_CODE,
            )
        lead = SalesLeadRow(
            lead_id=new_id("sld"),
            tenant_id=tenant_id,
            source_type=payload.source_type,
            source_channel=payload.source_channel,
            source_campaign_id=payload.source_campaign_id,
            full_name=payload.full_name,
            company_name=payload.company_name,
            phone=payload.phone,
            email=payload.email,
            messenger_handles_json=json.dumps(payload.messenger_handles, ensure_ascii=False),
            lead_temperature=payload.lead_temperature,
            lead_score=payload.lead_score,
            customer_type=payload.customer_type,
            segment_type=payload.segment_type,
            initial_need_summary=payload.initial_need_summary,
            preferred_channel=payload.preferred_channel,
            assigned_agent_type=payload.assigned_agent_type,
            status=payload.status,
            crm_customer_id=crm_customer.customer_id if crm_customer else None,
            created_at=now,
            updated_at=now,
        )
        deal = SalesDealRow(
            deal_id=new_id("sde"),
            tenant_id=tenant_id,
            lead_id=lead.lead_id,
            customer_id=crm_customer.customer_id if crm_customer else None,
            pipeline_id=pipeline.pipeline_id,
            stage_id=stage.stage_id,
            scenario_type="quick_sale",
            priority=payload.priority,
            title=payload.title or f"{payload.full_name} • {payload.source_channel}",
            need_summary=payload.initial_need_summary,
            product_context_json="{}",
            estimated_amount=payload.estimated_amount,
            final_amount=None,
            currency=payload.currency or DEFAULT_SALES_CURRENCY,
            payment_model=None,
            document_required=True,
            payment_required=True,
            assigned_human_user_id=None,
            assigned_ai_orchestrator_id=None,
            preferred_channel=payload.preferred_channel,
            current_channel=payload.preferred_channel,
            status="active",
            won_reason=None,
            lost_reason=None,
            close_reason=None,
            next_action_type=None,
            next_action_at=None,
            last_contact_at=None,
            closed_at=None,
            created_at=now,
            updated_at=now,
        )
        session.add(lead)
        session.add(deal)
        _record_stage_change(
            session,
            deal=deal,
            from_stage_id=None,
            to_stage_id=stage.stage_id,
            changed_by_type="human",
            changed_by_id=actor["user"],
            reason="lead.entered_crm",
        )
        _publish_lead_entered_event(session, lead=lead, deal=deal, stage=stage, actor_type="human", actor_id=actor["user"])
        session.commit()
        return _lead_to_out(lead, deal)
    finally:
        session.close()


@app.get("/api/v1/leads", response_model=list[SalesLeadOut])
def list_leads(
    query: str | None = None,
    search: str | None = None,
    status: str | None = None,
    lead_temperature: str | None = None,
    source_channel: str | None = None,
    source_type: str | None = None,
    customer_type: str | None = None,
    preferred_channel: str | None = None,
    limit: int = Query(default=100, ge=1, le=200),
    offset: int = Query(default=0, ge=0),
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> list[SalesLeadOut]:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        stmt = (
            select(SalesLeadRow)
            .where(SalesLeadRow.tenant_id == tenant_id)
            .order_by(SalesLeadRow.updated_at.desc(), SalesLeadRow.id.desc())
        )
        if status:
            stmt = stmt.where(SalesLeadRow.status == status)
        if lead_temperature:
            stmt = stmt.where(SalesLeadRow.lead_temperature == lead_temperature)
        if source_channel:
            stmt = stmt.where(SalesLeadRow.source_channel == source_channel)
        if source_type:
            stmt = stmt.where(SalesLeadRow.source_type == source_type)
        if customer_type:
            stmt = stmt.where(SalesLeadRow.customer_type == customer_type)
        if preferred_channel:
            stmt = stmt.where(SalesLeadRow.preferred_channel == preferred_channel)
        search_term = (search or query or "").strip()
        if search_term:
            pattern = f"%{search_term.lower()}%"
            stmt = stmt.where(
                or_(
                    func.lower(func.coalesce(SalesLeadRow.full_name, "")).like(pattern),
                    func.lower(func.coalesce(SalesLeadRow.company_name, "")).like(pattern),
                    func.lower(func.coalesce(SalesLeadRow.phone, "")).like(pattern),
                    func.lower(func.coalesce(SalesLeadRow.email, "")).like(pattern),
                    func.lower(func.coalesce(SalesLeadRow.lead_id, "")).like(pattern),
                )
            )
        rows = session.execute(stmt.limit(limit).offset(offset)).scalars().all()
        deals_by_lead = _latest_deals_by_lead(session, [row.lead_id for row in rows], tenant_id)
        return [_lead_to_out(row, deals_by_lead.get(row.lead_id)) for row in rows]
    finally:
        session.close()


@app.get("/api/v1/leads/{lead_id}", response_model=SalesLeadOut)
def get_lead(
    lead_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> SalesLeadOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        lead = _get_lead(session, lead_id, tenant_id)
        return _lead_to_out(lead, _latest_deal_for_lead(session, lead.lead_id, tenant_id))
    finally:
        session.close()


@app.patch("/api/v1/leads/{lead_id}", response_model=SalesLeadOut)
def update_lead(
    lead_id: str,
    payload: SalesLeadUpdate,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesLeadOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        lead = _get_lead(session, lead_id, tenant_id)
        updates = payload.model_dump(exclude_unset=True)
        if "messenger_handles" in updates:
            lead.messenger_handles_json = json.dumps(updates.pop("messenger_handles") or {}, ensure_ascii=False)
        for key, value in updates.items():
            setattr(lead, key, value)
        lead.updated_at = utc_now_iso()
        session.commit()
        return _lead_to_out(lead, _latest_deal_for_lead(session, lead.lead_id, tenant_id))
    finally:
        session.close()


@app.post("/api/v1/leads/{lead_id}/enrich", response_model=SalesWorkspaceOut)
def enrich_lead(
    lead_id: str,
    payload: SalesLeadEnrichIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesWorkspaceOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        lead = _get_lead(session, lead_id, tenant_id)
        if payload.need_summary:
            lead.initial_need_summary = payload.need_summary
        if payload.customer_type:
            lead.customer_type = payload.customer_type
        if payload.preferred_channel:
            lead.preferred_channel = payload.preferred_channel
        lead.updated_at = utc_now_iso()
        deal = session.execute(
            select(SalesDealRow)
            .where(SalesDealRow.lead_id == lead.lead_id, SalesDealRow.tenant_id == tenant_id)
            .order_by(SalesDealRow.id.desc())
        ).scalars().first()
        if deal is None:
            raise HTTPException(status_code=404, detail="Deal not found for lead")
        if payload.need_summary:
            deal.need_summary = payload.need_summary
        if payload.product_context:
            deal.product_context_json = json.dumps(payload.product_context, ensure_ascii=False)
        if payload.preferred_channel:
            deal.preferred_channel = payload.preferred_channel
            deal.current_channel = payload.preferred_channel
        deal.updated_at = utc_now_iso()
        next_stage = _infer_stage_for_channel(_infer_text_channel(lead.preferred_channel))
        _apply_stage(
            session,
            deal,
            next_stage,
            actor_type="human",
            actor_id=actor["user"],
            reason="lead.enrichment_completed",
        )
        _publish_sales_event(
            session,
            tenant_id=tenant_id,
            event_type=sales_event_types.LEAD_ENRICHMENT_COMPLETED,
            aggregate_type="lead",
            aggregate_id=lead.lead_id,
            actor_type="human",
            actor_id=actor["user"],
            payload={
                "lead_id": lead.lead_id,
                "deal_id": deal.deal_id,
                "need_summary": lead.initial_need_summary,
                "customer_type": lead.customer_type,
                "preferred_channel": lead.preferred_channel,
            },
        )
        session.commit()
        return _build_workspace(session, deal)
    finally:
        session.close()


@app.post("/api/v1/leads/{lead_id}/convert-to-deal", response_model=SalesDealOut)
def convert_lead_to_deal(
    lead_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesDealOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        lead = _get_lead(session, lead_id, tenant_id)
        lead.status = "converted"
        lead.updated_at = utc_now_iso()
        deal = session.execute(
            select(SalesDealRow)
            .where(SalesDealRow.lead_id == lead.lead_id, SalesDealRow.tenant_id == tenant_id)
            .order_by(SalesDealRow.id.desc())
        ).scalars().first()
        if deal is None:
            raise HTTPException(status_code=404, detail="Deal not found for lead")
        session.commit()
        return _deal_to_out_with_refs(session, deal)
    finally:
        session.close()


@app.get("/api/v1/customers", response_model=list[SalesCustomerOut])
def list_customers(
    query: str | None = None,
    search: str | None = None,
    limit: int = Query(default=100, ge=1, le=200),
    offset: int = Query(default=0, ge=0),
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> list[SalesCustomerOut]:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        customer_ids = _sales_customer_ids_for_tenant(session, tenant_id)
        if not customer_ids:
            return []

        stmt = select(Customer).where(Customer.customer_id.in_(customer_ids))
        search_term = (search or query or "").strip()
        if search_term:
            pattern = f"%{search_term.lower()}%"
            stmt = stmt.where(
                or_(
                    func.lower(func.coalesce(Customer.customer_id, "")).like(pattern),
                    func.lower(func.coalesce(Customer.display_name, "")).like(pattern),
                    func.lower(func.coalesce(Customer.preferred_phone, "")).like(pattern),
                )
            )

        rows = session.execute(stmt.order_by(Customer.display_name.asc(), Customer.id.desc()).limit(limit).offset(offset)).scalars().all()
        counts = dict(
            session.execute(
                select(SalesDealRow.customer_id, func.count())
                .where(
                    SalesDealRow.tenant_id == tenant_id,
                    SalesDealRow.customer_id.in_([row.customer_id for row in rows] or [""]),
                )
                .group_by(SalesDealRow.customer_id)
            ).all()
        )
        return [_customer_to_out(row, int(counts.get(row.customer_id) or 0)) for row in rows]
    finally:
        session.close()


@app.get("/api/v1/customers/{customer_id}", response_model=SalesCustomerOut)
def get_customer(
    customer_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> SalesCustomerOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        customer = _get_sales_customer(session, customer_id, tenant_id)
        return _customer_to_out(customer, _customer_deal_count(session, customer.customer_id, tenant_id))
    finally:
        session.close()


@app.patch("/api/v1/customers/{customer_id}", response_model=SalesCustomerOut)
def update_customer(
    customer_id: str,
    payload: SalesCustomerUpdate,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesCustomerOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        customer = _get_sales_customer(session, customer_id, tenant_id)
        updates = payload.model_dump(exclude_unset=True)
        if "display_name" in updates and updates["display_name"] is not None:
            customer.display_name = str(updates["display_name"])
        if "phones" in updates and updates["phones"] is not None:
            customer.phones_json = json.dumps(updates["phones"], ensure_ascii=False)
        if "preferred_phone" in updates:
            customer.preferred_phone = updates["preferred_phone"]
        if "tags" in updates and updates["tags"] is not None:
            customer.tags_json = json.dumps(updates["tags"], ensure_ascii=False)
        session.commit()
        return _customer_to_out(customer, _customer_deal_count(session, customer.customer_id, tenant_id))
    finally:
        session.close()


@app.get("/api/v1/customers/{customer_id}/deals", response_model=list[SalesDealOut])
def list_customer_deals(
    customer_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> list[SalesDealOut]:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        _get_sales_customer(session, customer_id, tenant_id)
        rows = session.execute(
            select(SalesDealRow)
            .where(SalesDealRow.tenant_id == tenant_id, SalesDealRow.customer_id == customer_id)
            .order_by(SalesDealRow.updated_at.desc(), SalesDealRow.id.desc())
        ).scalars().all()
        return [_deal_to_out_with_refs(session, row) for row in rows]
    finally:
        session.close()


@app.get("/api/v1/customers/{customer_id}/communications", response_model=list[SalesCommunicationOut])
def list_customer_communications(
    customer_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> list[SalesCommunicationOut]:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        _get_sales_customer(session, customer_id, tenant_id)
        rows = session.execute(
            select(SalesCommunicationSessionRow)
            .where(SalesCommunicationSessionRow.tenant_id == tenant_id, SalesCommunicationSessionRow.customer_id == customer_id)
            .order_by(SalesCommunicationSessionRow.started_at.desc(), SalesCommunicationSessionRow.id.desc())
        ).scalars().all()
        return [_communication_to_out(row) for row in rows]
    finally:
        session.close()


@app.get("/api/v1/customers/{customer_id}/documents", response_model=list[SalesDocumentOut])
def list_customer_documents(
    customer_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> list[SalesDocumentOut]:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        _get_sales_customer(session, customer_id, tenant_id)
        rows = session.execute(
            select(SalesDocumentRow)
            .where(SalesDocumentRow.tenant_id == tenant_id, SalesDocumentRow.customer_id == customer_id)
            .order_by(SalesDocumentRow.updated_at.desc(), SalesDocumentRow.id.desc())
        ).scalars().all()
        return [_document_to_out(row) for row in rows]
    finally:
        session.close()


@app.get("/api/v1/customers/{customer_id}/invoices", response_model=list[SalesInvoiceOut])
def list_customer_invoices(
    customer_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> list[SalesInvoiceOut]:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        _get_sales_customer(session, customer_id, tenant_id)
        rows = session.execute(
            select(SalesInvoiceRow)
            .where(SalesInvoiceRow.tenant_id == tenant_id, SalesInvoiceRow.customer_id == customer_id)
            .order_by(SalesInvoiceRow.updated_at.desc(), SalesInvoiceRow.id.desc())
        ).scalars().all()
        return [_invoice_to_out(row) for row in rows]
    finally:
        session.close()


@app.get("/api/v1/customers/{customer_id}/payments", response_model=list[SalesPaymentOut])
def list_customer_payments(
    customer_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> list[SalesPaymentOut]:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        _get_sales_customer(session, customer_id, tenant_id)
        deal_ids = session.execute(
            select(SalesDealRow.deal_id).where(SalesDealRow.tenant_id == tenant_id, SalesDealRow.customer_id == customer_id)
        ).scalars().all()
        rows = session.execute(
            select(SalesPaymentRow)
            .where(SalesPaymentRow.tenant_id == tenant_id, SalesPaymentRow.deal_id.in_(deal_ids or [""]))
            .order_by(SalesPaymentRow.updated_at.desc(), SalesPaymentRow.id.desc())
        ).scalars().all()
        return [_payment_to_out(row) for row in rows]
    finally:
        session.close()


@app.post("/api/v1/deals", response_model=SalesDealOut)
def create_deal(
    payload: SalesDealCreate,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesDealOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        now = utc_now_iso()
        if payload.lead_id:
            _get_lead(session, payload.lead_id, tenant_id)
        pipeline = get_default_pipeline(session, tenant_id)
        stage_id_input = payload.stage_id
        if not stage_id_input and not payload.stage_code:
            stage_id_input = DEFAULT_STAGE_CODE
        stage = resolve_stage_by_code_or_id(
            session,
            tenant_id=tenant_id,
            pipeline_id=pipeline.pipeline_id,
            stage_id=stage_id_input,
            stage_code=payload.stage_code,
        )
        deal = SalesDealRow(
            deal_id=new_id("sde"),
            tenant_id=tenant_id,
            lead_id=payload.lead_id,
            customer_id=payload.customer_id,
            pipeline_id=pipeline.pipeline_id,
            stage_id=stage.stage_id,
            scenario_type=payload.scenario_type,
            priority=payload.priority,
            title=payload.title,
            need_summary=payload.need_summary,
            product_context_json=json.dumps(payload.product_context, ensure_ascii=False),
            estimated_amount=payload.estimated_amount,
            final_amount=payload.final_amount,
            currency=payload.currency or DEFAULT_SALES_CURRENCY,
            payment_model=payload.payment_model,
            document_required=payload.document_required,
            payment_required=payload.payment_required,
            assigned_human_user_id=None,
            assigned_ai_orchestrator_id=None,
            preferred_channel=payload.preferred_channel,
            current_channel=payload.current_channel,
            status="active",
            won_reason=None,
            lost_reason=None,
            close_reason=None,
            next_action_type=None,
            next_action_at=None,
            last_contact_at=None,
            closed_at=None,
            created_at=now,
            updated_at=now,
        )
        session.add(deal)
        _record_stage_change(
            session,
            deal=deal,
            from_stage_id=None,
            to_stage_id=stage.stage_id,
            changed_by_type="human",
            changed_by_id=actor["user"],
            reason="deal.created",
        )
        session.commit()
        return _deal_to_out(deal, pipeline=pipeline, stage=stage)
    finally:
        session.close()


@app.get("/api/v1/deals", response_model=list[SalesDealOut])
def list_deals(
    query: str | None = None,
    stage_id: str | None = None,
    status: str | None = None,
    limit: int = Query(default=100, ge=1, le=300),
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> list[SalesDealOut]:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        stmt = (
            select(SalesDealRow)
            .where(SalesDealRow.tenant_id == tenant_id)
            .order_by(SalesDealRow.updated_at.desc(), SalesDealRow.id.desc())
        )
        if stage_id:
            matching_stage_ids = _stage_filter_ids(session, tenant_id, stage_id)
            stmt = stmt.where(SalesDealRow.stage_id.in_(matching_stage_ids or [""]))
        if status:
            stmt = stmt.where(SalesDealRow.status == status)
        if query:
            normalized_query = query.strip().lower()
            pattern = f"%{normalized_query}%"
            matching_lead_ids = session.execute(
                select(SalesLeadRow.lead_id).where(
                    SalesLeadRow.tenant_id == tenant_id,
                    or_(
                        func.lower(func.coalesce(SalesLeadRow.full_name, "")).like(pattern),
                        func.lower(func.coalesce(SalesLeadRow.company_name, "")).like(pattern),
                        func.lower(func.coalesce(SalesLeadRow.phone, "")).like(pattern),
                        func.lower(func.coalesce(SalesLeadRow.email, "")).like(pattern),
                    )
                )
            ).scalars().all()
            matching_deal_ids = session.execute(
                select(SalesExternalLinkRow.deal_id).where(
                    SalesExternalLinkRow.tenant_id == tenant_id,
                    or_(
                        func.lower(func.coalesce(SalesExternalLinkRow.external_thread_id, "")).like(pattern),
                        func.lower(func.coalesce(SalesExternalLinkRow.external_chat_id, "")).like(pattern),
                        func.lower(func.coalesce(SalesExternalLinkRow.voice_session_id, "")).like(pattern),
                        func.lower(func.coalesce(SalesExternalLinkRow.external_call_id, "")).like(pattern),
                        func.lower(func.coalesce(SalesExternalLinkRow.interaction_id, "")).like(pattern),
                        func.lower(func.coalesce(SalesExternalLinkRow.phone_number, "")).like(pattern),
                        func.lower(func.coalesce(SalesExternalLinkRow.customer_id, "")).like(pattern),
                    )
                )
            ).scalars().all()
            stmt = stmt.where(
                or_(
                    func.lower(func.coalesce(SalesDealRow.title, "")).like(pattern),
                    func.lower(func.coalesce(SalesDealRow.need_summary, "")).like(pattern),
                    func.lower(func.coalesce(SalesDealRow.customer_id, "")).like(pattern),
                    SalesDealRow.lead_id.in_(matching_lead_ids or [""]),
                    SalesDealRow.deal_id.in_(matching_deal_ids or [""]),
                )
            )
        rows = session.execute(stmt.limit(limit)).scalars().all()
        return [_deal_to_out_with_refs(session, row) for row in rows]
    finally:
        session.close()


@app.get("/api/v1/deals/{deal_id}", response_model=SalesDealOut)
def get_deal(
    deal_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> SalesDealOut:
    session = get_session()
    try:
        return _deal_to_out_with_refs(session, _get_deal(session, deal_id, _tenant_id(actor)))
    finally:
        session.close()


@app.get("/api/v1/deals/{deal_id}/workspace", response_model=SalesWorkspaceOut)
def get_workspace(
    deal_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> SalesWorkspaceOut:
    session = get_session()
    try:
        return _build_workspace(session, _get_deal(session, deal_id, _tenant_id(actor)))
    finally:
        session.close()


@app.patch("/api/v1/deals/{deal_id}", response_model=SalesDealOut)
def update_deal(
    deal_id: str,
    payload: SalesDealUpdate,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesDealOut:
    session = get_session()
    try:
        deal = _get_deal(session, deal_id, _tenant_id(actor))
        updates = payload.model_dump(exclude_unset=True)
        if "product_context" in updates:
            deal.product_context_json = json.dumps(updates.pop("product_context") or {}, ensure_ascii=False)
        for key, value in updates.items():
            setattr(deal, key, value)
        deal.updated_at = utc_now_iso()
        session.commit()
        return _deal_to_out_with_refs(session, deal)
    finally:
        session.close()


@app.post("/api/v1/deals/{deal_id}/change-stage", response_model=SalesDealOut)
def change_stage(
    deal_id: str,
    payload: SalesDealStageChangeIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesDealOut:
    session = get_session()
    try:
        deal = _get_deal_for_state_change(session, deal_id, _tenant_id(actor))
        target_stage_code = _resolve_stage_change_target_code(session, deal, payload)
        actor_type = "admin" if payload.force and actor["role"] == Role.ADMIN.value else "human"
        transition_deal_stage(
            session,
            tenant_id=deal.tenant_id,
            deal_id=deal.deal_id,
            target_stage_code=target_stage_code,
            actor_type=actor_type,
            actor_id=actor["user"],
            reason=payload.reason,
            metadata=payload.metadata,
            force=payload.force,
        )
        session.commit()
        return _deal_to_out_with_refs(session, deal)
    finally:
        session.close()


@app.post("/api/v1/deals/{deal_id}/select-scenario", response_model=SalesDealOut)
def select_scenario(
    deal_id: str,
    payload: SalesDealScenarioIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesDealOut:
    session = get_session()
    try:
        deal = _get_deal(session, deal_id, _tenant_id(actor))
        previous_scenario_type = deal.scenario_type
        deal.scenario_type = payload.scenario_type
        deal.updated_at = utc_now_iso()
        _publish_sales_event(
            session,
            tenant_id=deal.tenant_id,
            event_type=sales_event_types.DEAL_SCENARIO_SELECTED,
            aggregate_type="deal",
            aggregate_id=deal.deal_id,
            actor_type="human",
            actor_id=actor["user"],
            payload={
                "deal_id": deal.deal_id,
                "scenario_type": deal.scenario_type,
                "previous_scenario_type": previous_scenario_type,
            },
        )
        session.commit()
        return _deal_to_out_with_refs(session, deal)
    finally:
        session.close()


@app.post("/api/v1/deals/{deal_id}/schedule-next-action", response_model=SalesDealOut)
def schedule_next_action(
    deal_id: str,
    payload: SalesDealNextActionIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesDealOut:
    session = get_session()
    try:
        deal = _get_deal(session, deal_id, _tenant_id(actor))
        deal.next_action_type = payload.next_action_type
        deal.next_action_at = payload.next_action_at
        deal.updated_at = utc_now_iso()
        _schedule_task(
            session,
            deal=deal,
            task_type=payload.next_action_type,
            run_at=payload.next_action_at,
            payload={**payload.payload, "scheduled_by": actor["user"]},
        )
        _apply_stage(
            session,
            deal,
            "follow_up_scheduled",
            actor_type="human",
            actor_id=actor["user"],
            reason="deal.next_action_scheduled",
        )
        _publish_sales_event(
            session,
            tenant_id=deal.tenant_id,
            event_type=sales_event_types.DEAL_NEXT_ACTION_SCHEDULED,
            aggregate_type="deal",
            aggregate_id=deal.deal_id,
            actor_type="human",
            actor_id=actor["user"],
            payload={
                "deal_id": deal.deal_id,
                "next_action_type": deal.next_action_type,
                "next_action_at": deal.next_action_at,
            },
        )
        session.commit()
        return _deal_to_out_with_refs(session, deal)
    finally:
        session.close()


@app.post("/api/v1/deals/{deal_id}/notes", response_model=SalesNoteOut)
def create_deal_note(
    deal_id: str,
    payload: SalesNoteCreateIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesNoteOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        deal = _get_deal(session, deal_id, tenant_id)
        now = utc_now_iso()
        row = SalesNoteRow(
            note_id=new_id("not"),
            tenant_id=tenant_id,
            deal_id=deal.deal_id,
            author_type="human",
            author_id=actor["user"],
            note_type=payload.note_type,
            content=payload.content,
            source_type=payload.source_type,
            source_id=payload.source_id,
            created_at=now,
            updated_at=now,
            archived_at=None,
        )
        session.add(row)
        _publish_sales_event(
            session,
            tenant_id=tenant_id,
            event_type=sales_event_types.DEAL_NOTE_CREATED,
            aggregate_type="deal",
            aggregate_id=deal.deal_id,
            actor_type="human",
            actor_id=actor["user"],
            payload={
                "deal_id": deal.deal_id,
                "note_id": row.note_id,
                "note_type": row.note_type,
                "source_type": row.source_type,
                "source_id": row.source_id,
            },
        )
        session.commit()
        return _note_to_out(row)
    finally:
        session.close()


@app.get("/api/v1/deals/{deal_id}/notes", response_model=list[SalesNoteOut])
def list_deal_notes(
    deal_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> list[SalesNoteOut]:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        _get_deal(session, deal_id, tenant_id)
        rows = session.execute(
            select(SalesNoteRow)
            .where(SalesNoteRow.deal_id == deal_id, SalesNoteRow.tenant_id == tenant_id, SalesNoteRow.archived_at.is_(None))
            .order_by(SalesNoteRow.created_at.desc(), SalesNoteRow.id.desc())
        ).scalars().all()
        return [_note_to_out(row) for row in rows]
    finally:
        session.close()


@app.patch("/api/v1/deals/{deal_id}/notes/{note_id}", response_model=SalesNoteOut)
def update_deal_note(
    deal_id: str,
    note_id: str,
    payload: SalesNoteUpdateIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesNoteOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        deal = _get_deal(session, deal_id, tenant_id)
        row = session.execute(
            select(SalesNoteRow).where(
                SalesNoteRow.note_id == note_id,
                SalesNoteRow.deal_id == deal.deal_id,
                SalesNoteRow.tenant_id == tenant_id,
                SalesNoteRow.archived_at.is_(None),
            )
        ).scalar_one_or_none()
        if row is None:
            raise HTTPException(status_code=404, detail="Note not found")
        updates = payload.model_dump(exclude_unset=True)
        for key, value in updates.items():
            setattr(row, key, value)
        row.updated_at = utc_now_iso()
        _publish_sales_event(
            session,
            tenant_id=tenant_id,
            event_type=sales_event_types.DEAL_NOTE_UPDATED,
            aggregate_type="deal",
            aggregate_id=deal.deal_id,
            actor_type="human",
            actor_id=actor["user"],
            payload={
                "deal_id": deal.deal_id,
                "note_id": row.note_id,
                "note_type": row.note_type,
                "source_type": row.source_type,
                "source_id": row.source_id,
            },
        )
        session.commit()
        return _note_to_out(row)
    finally:
        session.close()


@app.post("/api/v1/deals/{deal_id}/close", response_model=SalesDealOut)
def close_deal(
    deal_id: str,
    payload: SalesDealCloseIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesDealOut:
    session = get_session()
    try:
        deal = _get_deal(session, deal_id, _tenant_id(actor))
        deal.status = payload.status
        deal.closed_at = utc_now_iso()
        deal.close_reason = payload.reason
        if payload.status == "won":
            deal.won_reason = payload.reason
        elif payload.status == "lost":
            deal.lost_reason = payload.reason
        _apply_stage(
            session,
            deal,
            payload.status,
            actor_type="human",
            actor_id=actor["user"],
            reason=payload.reason or "deal.closed",
        )
        _publish_sales_event(
            session,
            tenant_id=deal.tenant_id,
            event_type=sales_event_types.DEAL_CLOSED,
            aggregate_type="deal",
            aggregate_id=deal.deal_id,
            actor_type="human",
            actor_id=actor["user"],
            payload={
                "deal_id": deal.deal_id,
                "status": deal.status,
                "reason": payload.reason,
                "closed_at": deal.closed_at,
            },
        )
        if payload.status == "won":
            _publish_sales_event(
                session,
                tenant_id=deal.tenant_id,
                event_type=sales_event_types.DEAL_WON,
                aggregate_type="deal",
                aggregate_id=deal.deal_id,
                actor_type="human",
                actor_id=actor["user"],
                payload={
                    "deal_id": deal.deal_id,
                    "final_amount": deal.final_amount,
                    "currency": deal.currency,
                    "won_reason": payload.reason or "deal.closed",
                },
            )
        elif payload.status == "lost":
            _publish_sales_event(
                session,
                tenant_id=deal.tenant_id,
                event_type=sales_event_types.DEAL_LOST,
                aggregate_type="deal",
                aggregate_id=deal.deal_id,
                actor_type="human",
                actor_id=actor["user"],
                payload={
                    "deal_id": deal.deal_id,
                    "lost_reason": payload.reason or "deal.closed",
                    "currency": deal.currency,
                },
            )
        session.commit()
        return _deal_to_out_with_refs(session, deal)
    finally:
        session.close()


@app.post("/api/v1/deals/{deal_id}/escalations", response_model=SalesEscalationOut)
@app.post("/api/v1/deals/{deal_id}/escalate", response_model=SalesEscalationOut)
def escalate_deal(
    deal_id: str,
    payload: SalesEscalationCreateIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesEscalationOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        deal = _get_deal(session, deal_id, tenant_id)
        existing = session.execute(
            select(SalesEscalationRow)
            .where(
                SalesEscalationRow.tenant_id == tenant_id,
                SalesEscalationRow.deal_id == deal.deal_id,
                SalesEscalationRow.escalation_type == payload.escalation_type,
                SalesEscalationRow.reason == payload.reason,
                SalesEscalationRow.status.in_(["open", "assigned", "in_progress"]),
            )
            .order_by(SalesEscalationRow.id.desc())
        ).scalars().first()
        if existing is not None:
            return _escalation_to_out(existing)
        now = utc_now_iso()
        row = SalesEscalationRow(
            escalation_id=new_id("esc"),
            tenant_id=deal.tenant_id,
            deal_id=deal.deal_id,
            escalation_type=payload.escalation_type,
            reason=payload.reason,
            severity=payload.severity,
            status="assigned" if payload.assigned_to_user_id else "open",
            assigned_to_user_id=payload.assigned_to_user_id,
            assigned_at=now if payload.assigned_to_user_id else None,
            started_at=None,
            created_at=now,
            resolved_at=None,
            canceled_at=None,
            resolution_code=None,
            resolution_summary=None,
            sla_due_at=payload.sla_due_at,
            source_channel=payload.source_channel or _channel_type_for_value(deal.current_channel),
            source_communication_session_id=payload.source_communication_session_id,
            source_automation_task_id=payload.source_automation_task_id,
            updated_at=now,
        )
        session.add(row)
        deal.assigned_human_user_id = payload.assigned_to_user_id or actor["user"]
        deal.scenario_type = "custom_human_escalation"
        try:
            transition_deal_stage(
                session,
                tenant_id=tenant_id,
                deal_id=deal.deal_id,
                target_stage_code="transferred_to_support",
                actor_type="human",
                actor_id=actor["user"],
                reason=payload.reason,
                metadata={"escalation_id": row.escalation_id},
            )
        except DealStateMachineError as exc:
            if exc.error != "invalid_stage_transition":
                raise
            transition_deal_stage(
                session,
                tenant_id=tenant_id,
                deal_id=deal.deal_id,
                target_stage_code="transferred_to_support",
                actor_type="admin" if actor["role"] == Role.ADMIN.value else "system",
                actor_id=actor["user"],
                reason=payload.reason,
                metadata={"escalation_id": row.escalation_id},
                force=True,
            )
        _publish_sales_event(
            session,
            tenant_id=tenant_id,
            event_type=sales_event_types.DEAL_ESCALATION_REQUESTED,
            aggregate_type="deal",
            aggregate_id=deal.deal_id,
            actor_type="human",
            actor_id=actor["user"],
            payload={
                "deal_id": deal.deal_id,
                "escalation_id": row.escalation_id,
                "escalation_type": row.escalation_type,
                "severity": row.severity,
                "assigned_to_user_id": row.assigned_to_user_id,
            },
        )
        _publish_sales_event(
            session,
            tenant_id=tenant_id,
            event_type=sales_event_types.ESCALATION_CREATED,
            aggregate_type="escalation",
            aggregate_id=row.escalation_id,
            actor_type="human",
            actor_id=actor["user"],
            payload={
                "deal_id": deal.deal_id,
                "escalation_id": row.escalation_id,
                "escalation_type": row.escalation_type,
                "status": row.status,
                "severity": row.severity,
            },
        )
        session.commit()
        return _escalation_to_out(row)
    finally:
        session.close()


@app.get("/api/v1/deals/{deal_id}/escalations", response_model=list[SalesEscalationOut])
def list_deal_escalations(
    deal_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> list[SalesEscalationOut]:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        _get_deal(session, deal_id, tenant_id)
        rows = session.execute(
            select(SalesEscalationRow)
            .where(SalesEscalationRow.deal_id == deal_id, SalesEscalationRow.tenant_id == tenant_id)
            .order_by(SalesEscalationRow.created_at.desc(), SalesEscalationRow.id.desc())
        ).scalars().all()
        return [_escalation_to_out(row) for row in rows]
    finally:
        session.close()


def _get_escalation(session, escalation_id: str, tenant_id: str) -> SalesEscalationRow:
    row = session.execute(
        select(SalesEscalationRow).where(SalesEscalationRow.escalation_id == escalation_id, SalesEscalationRow.tenant_id == tenant_id)
    ).scalar_one_or_none()
    if row is None:
        raise HTTPException(status_code=404, detail="Escalation not found")
    return row


def _publish_escalation_lifecycle_event(
    session,
    *,
    tenant_id: str,
    event_type: str,
    row: SalesEscalationRow,
    actor_user: str,
) -> None:
    _publish_sales_event(
        session,
        tenant_id=tenant_id,
        event_type=event_type,
        aggregate_type="escalation",
        aggregate_id=row.escalation_id,
        actor_type="human",
        actor_id=actor_user,
        payload={
            "deal_id": row.deal_id,
            "escalation_id": row.escalation_id,
            "escalation_type": row.escalation_type,
            "status": row.status,
            "assigned_to_user_id": row.assigned_to_user_id,
            "resolution_code": row.resolution_code,
        },
    )


@app.post("/api/v1/escalations/{escalation_id}/assign", response_model=SalesEscalationOut)
def assign_escalation(
    escalation_id: str,
    payload: SalesEscalationAssignIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesEscalationOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        row = _get_escalation(session, escalation_id, tenant_id)
        now = utc_now_iso()
        row.assigned_to_user_id = payload.assigned_to_user_id
        row.assigned_at = now
        if row.status == "open":
            row.status = "assigned"
        row.updated_at = now
        _publish_escalation_lifecycle_event(
            session,
            tenant_id=tenant_id,
            event_type=sales_event_types.ESCALATION_ASSIGNED,
            row=row,
            actor_user=actor["user"],
        )
        session.commit()
        return _escalation_to_out(row)
    finally:
        session.close()


@app.post("/api/v1/escalations/{escalation_id}/start", response_model=SalesEscalationOut)
def start_escalation(
    escalation_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesEscalationOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        row = _get_escalation(session, escalation_id, tenant_id)
        if row.status in {"resolved", "canceled"}:
            raise HTTPException(status_code=400, detail="Escalation is already closed")
        now = utc_now_iso()
        row.status = "in_progress"
        row.started_at = row.started_at or now
        row.updated_at = now
        _publish_escalation_lifecycle_event(
            session,
            tenant_id=tenant_id,
            event_type=sales_event_types.ESCALATION_STARTED,
            row=row,
            actor_user=actor["user"],
        )
        session.commit()
        return _escalation_to_out(row)
    finally:
        session.close()


@app.post("/api/v1/escalations/{escalation_id}/resolve", response_model=SalesEscalationOut)
def resolve_escalation(
    escalation_id: str,
    payload: SalesEscalationResolveIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesEscalationOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        row = _get_escalation(session, escalation_id, tenant_id)
        now = utc_now_iso()
        row.status = "resolved"
        row.resolved_at = now
        row.resolution_code = payload.resolution_code
        row.resolution_summary = payload.resolution_summary
        row.updated_at = now
        if payload.target_stage_code:
            transition_deal_stage(
                session,
                tenant_id=tenant_id,
                deal_id=row.deal_id,
                target_stage_code=payload.target_stage_code,
                actor_type="human",
                actor_id=actor["user"],
                reason=payload.resolution_summary or payload.resolution_code,
                metadata={"escalation_id": row.escalation_id, **payload.metadata},
            )
        _publish_escalation_lifecycle_event(
            session,
            tenant_id=tenant_id,
            event_type=sales_event_types.ESCALATION_RESOLVED,
            row=row,
            actor_user=actor["user"],
        )
        session.commit()
        return _escalation_to_out(row)
    finally:
        session.close()


@app.post("/api/v1/escalations/{escalation_id}/cancel", response_model=SalesEscalationOut)
def cancel_escalation(
    escalation_id: str,
    payload: SalesEscalationCancelIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesEscalationOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        row = _get_escalation(session, escalation_id, tenant_id)
        now = utc_now_iso()
        row.status = "canceled"
        row.canceled_at = now
        row.resolution_code = payload.resolution_code
        row.resolution_summary = payload.resolution_summary
        row.updated_at = now
        _publish_escalation_lifecycle_event(
            session,
            tenant_id=tenant_id,
            event_type=sales_event_types.ESCALATION_CANCELED,
            row=row,
            actor_user=actor["user"],
        )
        session.commit()
        return _escalation_to_out(row)
    finally:
        session.close()


@app.post("/api/v1/deals/{deal_id}/communications/text", response_model=SalesCommunicationOut)
def start_text_communication(
    deal_id: str,
    payload: SalesCommunicationStartIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesCommunicationOut:
    payload = payload.model_copy(update={"channel_type": "text"})
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        deal = _get_deal(session, deal_id, tenant_id)
        lead = _get_lead(session, deal.lead_id, tenant_id) if deal.lead_id else None
        row = _start_communication(session, deal=deal, lead=lead, payload=payload, actor_user=actor["user"])
        session.commit()
        return _communication_to_out(row)
    finally:
        session.close()


@app.post("/api/v1/deals/{deal_id}/communications/voice", response_model=SalesCommunicationOut)
def start_voice_communication(
    deal_id: str,
    payload: SalesCommunicationStartIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesCommunicationOut:
    payload = payload.model_copy(update={"channel_type": "voice"})
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        deal = _get_deal(session, deal_id, tenant_id)
        lead = _get_lead(session, deal.lead_id, tenant_id) if deal.lead_id else None
        row = _start_communication(session, deal=deal, lead=lead, payload=payload, actor_user=actor["user"])
        session.commit()
        return _communication_to_out(row)
    finally:
        session.close()


@app.get("/api/v1/deals/{deal_id}/communications", response_model=list[SalesCommunicationOut])
def list_communications(
    deal_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> list[SalesCommunicationOut]:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        _get_deal(session, deal_id, tenant_id)
        rows = session.execute(
            select(SalesCommunicationSessionRow)
            .where(SalesCommunicationSessionRow.deal_id == deal_id, SalesCommunicationSessionRow.tenant_id == tenant_id)
            .order_by(SalesCommunicationSessionRow.started_at.desc(), SalesCommunicationSessionRow.id.desc())
        ).scalars().all()
        return [_communication_to_out(row) for row in rows]
    finally:
        session.close()


@app.post("/api/v1/communications/{communication_id}/summary", response_model=SalesCommunicationOut)
def summarize_communication(
    communication_id: str,
    payload: SalesCommunicationSummaryIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesCommunicationOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        row = _get_communication(session, communication_id, tenant_id)
        row.summary = payload.summary
        row.result_code = payload.result_code
        row.sentiment = payload.sentiment
        row.next_action_type = payload.next_action_type
        row.next_action_at = payload.next_action_at
        row.updated_at = utc_now_iso()
        deal = _get_deal(session, row.deal_id, tenant_id)
        _touch_deal_contact(deal)
        if payload.next_action_type:
            deal.next_action_type = payload.next_action_type
            deal.next_action_at = payload.next_action_at
            _schedule_task(
                session,
                deal=deal,
                task_type=payload.next_action_type,
                run_at=payload.next_action_at,
                payload={"source": "communication.summary", "communication_id": row.communication_id},
            )
            _publish_sales_event(
                session,
                tenant_id=tenant_id,
                event_type=sales_event_types.DEAL_NEXT_ACTION_SCHEDULED,
                aggregate_type="deal",
                aggregate_id=deal.deal_id,
                actor_type="human",
                actor_id=actor["user"],
                payload={
                    "deal_id": deal.deal_id,
                    "next_action_type": payload.next_action_type,
                    "next_action_at": payload.next_action_at,
                    "communication_session_id": row.communication_id,
                },
            )
        _publish_sales_event(
            session,
            tenant_id=tenant_id,
            event_type=sales_event_types.COMMUNICATION_SUMMARY_CREATED,
            aggregate_type="communication",
            aggregate_id=row.communication_id,
            actor_type="human",
            actor_id=actor["user"],
            payload={
                "deal_id": deal.deal_id,
                "communication_session_id": row.communication_id,
                "summary": row.summary,
                "result_code": row.result_code,
                "sentiment": row.sentiment,
            },
        )
        session.commit()
        return _communication_to_out(row)
    finally:
        session.close()


@app.post("/api/v1/communications/{communication_id}/switch-channel", response_model=SalesChannelSwitchOut)
def switch_channel(
    communication_id: str,
    payload: SalesCommunicationSwitchChannelIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesChannelSwitchOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        communication = _get_communication(session, communication_id, tenant_id)
        deal = _get_deal(session, communication.deal_id, tenant_id)
        row, new_communication = _perform_channel_switch(
            session,
            deal=deal,
            payload=payload,
            actor_type="human",
            actor_id=actor["user"],
            communication=communication,
        )
        session.commit()
        return _channel_switch_to_out_with_session(row, new_communication)
    finally:
        session.close()


@app.post("/api/v1/deals/{deal_id}/switch-channel", response_model=SalesChannelSwitchOut)
def switch_deal_channel(
    deal_id: str,
    payload: SalesCommunicationSwitchChannelIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesChannelSwitchOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        deal = _get_deal(session, deal_id, tenant_id)
        communication = session.execute(
            select(SalesCommunicationSessionRow)
            .where(SalesCommunicationSessionRow.deal_id == deal.deal_id, SalesCommunicationSessionRow.tenant_id == tenant_id)
            .order_by(SalesCommunicationSessionRow.started_at.desc(), SalesCommunicationSessionRow.id.desc())
        ).scalars().first()
        row, new_communication = _perform_channel_switch(
            session,
            deal=deal,
            payload=payload,
            actor_type="human",
            actor_id=actor["user"],
            communication=communication,
        )
        session.commit()
        return _channel_switch_to_out_with_session(row, new_communication)
    finally:
        session.close()


@app.get("/api/v1/deals/{deal_id}/channel-switches", response_model=list[SalesChannelSwitchOut])
def list_deal_channel_switches(
    deal_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> list[SalesChannelSwitchOut]:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        _get_deal(session, deal_id, tenant_id)
        rows = session.execute(
            select(SalesChannelSwitchRow)
            .where(SalesChannelSwitchRow.deal_id == deal_id, SalesChannelSwitchRow.tenant_id == tenant_id)
            .order_by(SalesChannelSwitchRow.switched_at.desc(), SalesChannelSwitchRow.id.desc())
        ).scalars().all()
        return [_channel_switch_to_out(row) for row in rows]
    finally:
        session.close()


@app.post("/api/v1/communications/{communication_id}/bind-external", response_model=SalesCommunicationOut)
def bind_external_communication(
    communication_id: str,
    payload: SalesCommunicationBindExternalIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesCommunicationOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        communication = _get_communication(session, communication_id, tenant_id)
        deal = _get_deal(session, communication.deal_id, tenant_id)
        metadata_updates = dict(payload.metadata or {})
        if payload.channel_provider:
            metadata_updates["channel_provider"] = payload.channel_provider
        if payload.telegram_thread_id:
            metadata_updates["telegram_thread_id"] = payload.telegram_thread_id
        if payload.telegram_chat_id:
            metadata_updates["telegram_chat_id"] = payload.telegram_chat_id
        if payload.voice_session_id:
            metadata_updates["voice_session_id"] = payload.voice_session_id
        if payload.external_call_id:
            metadata_updates["external_call_id"] = payload.external_call_id
        _merge_communication_metadata(communication, metadata_updates)

        text_channels = {"telegram", "whatsapp", "webchat", "email"}
        channel_provider = str(payload.channel_provider or "").strip().lower()
        if payload.telegram_thread_id or channel_provider in text_channels:
            next_channel = channel_provider if channel_provider in text_channels else "telegram"
            deal.current_channel = next_channel
            deal.preferred_channel = next_channel
            if communication.channel_type != "voice":
                _apply_stage(
                    session,
                    deal,
                    "active_text_communication",
                    actor_type="human",
                    actor_id=actor["user"],
                    reason="external.channel_bound",
                )

        if payload.voice_session_id:
            deal.current_channel = "voice"
            deal.preferred_channel = "voice"
            _apply_stage(
                session,
                deal,
                "active_voice_communication",
                actor_type="human",
                actor_id=actor["user"],
                reason="voice.session_bound",
            )

        if payload.external_call_id:
            call = session.execute(
                select(SalesCallRow)
                .where(SalesCallRow.communication_id == communication.communication_id, SalesCallRow.tenant_id == tenant_id)
                .order_by(SalesCallRow.id.desc())
                .limit(1)
            ).scalar_one_or_none()
            if call is not None:
                call.external_call_id = payload.external_call_id.strip()
                call.updated_at = utc_now_iso()

        session.commit()
        return _communication_to_out(communication)
    finally:
        session.close()


@app.post("/internal/sales-sync/telegram", response_model=SalesWorkspaceOut)
def sync_telegram_thread(
    payload: SalesTelegramSyncIn,
    actor: dict = Depends(require_roles(Role.ADMIN)),
) -> SalesWorkspaceOut:
    session = get_session()
    try:
        tenant_id = _resolve_provider_tenant_id(
            session,
            actor,
            provider_type="message",
            provider_name="telegram",
            metadata=payload.metadata,
            external_identifier=payload.thread_id or payload.chat_id,
        )
        lead, deal = _resolve_or_create_sync_deal(
            session,
            tenant_id=tenant_id,
            source_channel="telegram",
            preferred_channel="telegram",
            customer_id=payload.customer_id,
            phone=payload.phone_number,
            display_name=payload.display_name,
            interaction_id=payload.interaction_id,
            thread_id=payload.thread_id,
            subject=payload.text or payload.display_name or "Telegram inquiry",
        )
        communication = _get_or_create_text_communication(
            session,
            deal=deal,
            lead=lead,
            direction=payload.direction,
            subject=(payload.text or payload.display_name or "Telegram conversation")[:120],
            metadata={
                "channel_provider": "telegram",
                "skip_stage_transition": True,
                "stage_progression_mode": "telegram_message_count",
                "telegram_thread_id": payload.thread_id,
                "telegram_chat_id": payload.chat_id,
                "interaction_id": payload.interaction_id,
                "queue_id": payload.queue_id,
                "ai_state": payload.ai_state,
                "ai_handoff_reason": payload.ai_handoff_reason,
                **(payload.metadata or {}),
            },
            thread_id=payload.thread_id,
        )
        event_ts = str(payload.happened_at or utc_now_iso()).strip() or utc_now_iso()
        dedupe_external_message_id = str(payload.external_message_id or payload.message_id or "").strip() or None
        if dedupe_external_message_id:
            message = session.execute(
                select(SalesMessageRow)
                .where(SalesMessageRow.tenant_id == tenant_id)
                .where(SalesMessageRow.channel_provider == "telegram")
                .where(SalesMessageRow.external_message_id == dedupe_external_message_id)
            ).scalar_one_or_none()
        else:
            message = None
        message_created = False
        if message is None and str(payload.text or "").strip():
            message = SalesMessageRow(
                message_id=new_id("msg"),
                tenant_id=deal.tenant_id,
                deal_id=deal.deal_id,
                communication_id=communication.communication_id,
                sender_type=payload.author_type or ("customer" if payload.direction == "inbound" else "human"),
                sender_id=payload.author_id,
                channel_provider="telegram",
                external_message_id=dedupe_external_message_id,
                body=payload.text.strip(),
                attachments_json="[]",
                delivery_status="sent" if payload.direction == "outbound" else "received",
                read_status="unread" if payload.direction == "inbound" else None,
                message_metadata_json=json.dumps(payload.metadata or {}, ensure_ascii=False),
                sent_at=event_ts,
                created_at=event_ts,
            )
            session.add(message)
            message_created = True

        _upsert_external_link(
            session,
            deal=deal,
            communication_id=communication.communication_id,
            channel_provider="telegram",
            external_thread_id=payload.thread_id,
            external_chat_id=payload.chat_id,
            interaction_id=payload.interaction_id,
            customer_id=payload.customer_id or deal.customer_id,
            phone_number=payload.phone_number,
            external_status=payload.status or payload.ai_state,
            metadata={
                "display_name": payload.display_name,
                "queue_id": payload.queue_id,
                "last_direction": payload.direction,
                "last_author_type": payload.author_type,
                "ai_state": payload.ai_state,
                "ai_handoff_reason": payload.ai_handoff_reason,
                **(payload.metadata or {}),
            },
        )
        deal.current_channel = "telegram"
        deal.preferred_channel = "telegram"
        deal.updated_at = utc_now_iso()
        if message_created:
            session.flush()
            _maybe_advance_telegram_deal_stage(session, deal=deal, lead=lead, payload=payload)
        session.commit()
        return _build_workspace(session, deal)
    finally:
        session.close()


@app.post("/internal/sales-sync/voice", response_model=SalesWorkspaceOut)
def sync_voice_session(
    payload: SalesVoiceSyncIn,
    actor: dict = Depends(require_roles(Role.ADMIN)),
) -> SalesWorkspaceOut:
    session = get_session()
    try:
        tenant_id = _resolve_provider_tenant_id(
            session,
            actor,
            provider_type="voice",
            provider_name="voice_runtime",
            metadata=payload.metadata,
            external_identifier=payload.voice_session_id or payload.call_id or payload.caller_number,
        )
        lead, deal = _resolve_or_create_sync_deal(
            session,
            tenant_id=tenant_id,
            source_channel="voice",
            preferred_channel="voice",
            phone=payload.caller_number,
            display_name=payload.caller_name,
            interaction_id=payload.interaction_id,
            voice_session_id=payload.voice_session_id,
            external_call_id=payload.call_id,
            subject=payload.summary or payload.caller_name or "Voice inquiry",
        )
        communication = _get_or_create_voice_communication(
            session,
            deal=deal,
            lead=lead,
            subject=(payload.summary or payload.caller_name or f"Voice call {payload.call_id}")[:120],
            metadata={
                "channel_provider": "voice",
                "voice_session_id": payload.voice_session_id,
                "ai_session_id": payload.ai_session_id,
                "interaction_id": payload.interaction_id,
                "queue_id": payload.queue_id,
                "queue_code": payload.queue_code,
                "ai_state": payload.ai_state,
                "handoff_reason": payload.handoff_reason,
                **(payload.metadata or {}),
            },
            voice_session_id=payload.voice_session_id,
        )
        call = None
        existing_link = _find_external_link(
            session,
            tenant_id=tenant_id,
            voice_session_id=payload.voice_session_id,
            external_call_id=payload.call_id,
            interaction_id=payload.interaction_id,
        )
        if existing_link and existing_link.sales_call_id:
            call = session.execute(
                select(SalesCallRow).where(SalesCallRow.call_id == existing_link.sales_call_id, SalesCallRow.tenant_id == tenant_id)
            ).scalar_one_or_none()
        if call is None:
            call = session.execute(
                select(SalesCallRow)
                .where(SalesCallRow.external_call_id == payload.call_id, SalesCallRow.tenant_id == tenant_id)
                .order_by(SalesCallRow.id.desc())
            ).scalars().first()
        started_at = str(payload.started_at or utc_now_iso()).strip() or utc_now_iso()
        ended_at = str(payload.ended_at or "").strip() or None
        if call is None:
            call = SalesCallRow(
                call_id=new_id("cal"),
                tenant_id=deal.tenant_id,
                deal_id=deal.deal_id,
                communication_id=communication.communication_id,
                phone_number=str(payload.caller_number or "").strip() or (lead.phone if lead else "") or "unknown",
                direction="inbound",
                provider="voice_runtime",
                external_call_id=payload.call_id,
                recording_url=None,
                transcript_status="ready" if str(payload.transcript_text or "").strip() else "pending",
                transcript_id=None,
                call_status=payload.call_status or payload.ai_state or "started",
                summary=payload.summary,
                started_at=started_at,
                ended_at=ended_at,
                duration_sec=None,
                created_at=started_at,
                updated_at=utc_now_iso(),
            )
            session.add(call)
        else:
            call.communication_id = communication.communication_id
            call.phone_number = str(payload.caller_number or "").strip() or call.phone_number
            call.provider = "voice_runtime"
            call.external_call_id = payload.call_id
            call.call_status = payload.call_status or payload.ai_state or call.call_status
            call.summary = payload.summary or call.summary
            call.started_at = started_at or call.started_at
            call.ended_at = ended_at or call.ended_at
            call.transcript_status = "ready" if str(payload.transcript_text or "").strip() else call.transcript_status
            call.updated_at = utc_now_iso()

        if str(payload.transcript_text or "").strip():
            transcript = None
            if call.transcript_id:
                transcript = session.execute(
                    select(SalesTranscriptRow).where(
                        SalesTranscriptRow.transcript_id == call.transcript_id,
                        SalesTranscriptRow.tenant_id == tenant_id,
                    )
                ).scalar_one_or_none()
            if transcript is None:
                transcript = SalesTranscriptRow(
                    transcript_id=new_id("trn"),
                    tenant_id=deal.tenant_id,
                    call_id=call.call_id,
                    language="ru",
                    transcript_text=payload.transcript_text.strip(),
                    diarization_json="{}",
                    extracted_entities_json=json.dumps(payload.metadata or {}, ensure_ascii=False),
                    created_at=utc_now_iso(),
                    updated_at=utc_now_iso(),
                )
                session.add(transcript)
                call.transcript_id = transcript.transcript_id
            else:
                transcript.transcript_text = payload.transcript_text.strip()
                transcript.extracted_entities_json = json.dumps(payload.metadata or {}, ensure_ascii=False)
                transcript.updated_at = utc_now_iso()

        _upsert_external_link(
            session,
            deal=deal,
            communication_id=communication.communication_id,
            sales_call_id=call.call_id,
            channel_provider="voice",
            external_call_id=payload.call_id,
            voice_session_id=payload.voice_session_id,
            ai_session_id=payload.ai_session_id,
            interaction_id=payload.interaction_id,
            customer_id=deal.customer_id,
            phone_number=payload.caller_number,
            external_status=payload.call_status or payload.ai_state or payload.telephony_status,
            metadata={
                "caller_name": payload.caller_name,
                "queue_id": payload.queue_id,
                "queue_code": payload.queue_code,
                "telephony_status": payload.telephony_status,
                "ai_state": payload.ai_state,
                "handoff_reason": payload.handoff_reason,
                **(payload.metadata or {}),
            },
        )
        deal.current_channel = "voice"
        deal.preferred_channel = "voice"
        deal.updated_at = utc_now_iso()
        session.commit()
        return _build_workspace(session, deal)
    finally:
        session.close()


@app.post("/api/v1/messages/inbound-webhook", response_model=SalesMessageOut)
def inbound_message(
    payload: SalesMessageWebhookIn,
    actor: dict = Depends(get_actor),
) -> SalesMessageOut:
    session = get_session()
    try:
        tenant_id = _resolve_provider_tenant_id(
            session,
            actor,
            provider_type="message",
            provider_name=payload.channel_provider,
            metadata=payload.metadata,
            external_identifier=payload.phone,
        )
        lead, deal = _resolve_deal_for_inbound(
            session,
            tenant_id=tenant_id,
            deal_id=payload.deal_id,
            lead_id=payload.lead_id,
            phone=payload.phone,
            subject=payload.body[:120],
            source_channel=payload.channel_provider,
            preferred_channel=payload.channel_provider if payload.channel_provider in {"telegram", "whatsapp", "webchat", "email"} else "telegram",
        )
        communication = _start_communication(
            session,
            deal=deal,
            lead=lead,
            payload=SalesCommunicationStartIn(
                channel_type="text",
                direction="inbound",
                agent_type="text_ai",
                subject=payload.body[:120],
                summary=None,
            ),
            metadata={**(payload.metadata or {}), "channel_provider": payload.channel_provider},
        )
        message = SalesMessageRow(
            message_id=new_id("msg"),
            tenant_id=deal.tenant_id,
            deal_id=deal.deal_id,
            communication_id=communication.communication_id,
            sender_type="customer",
            sender_id=payload.sender_id,
            channel_provider=payload.channel_provider,
            external_message_id=payload.external_message_id,
            body=payload.body,
            attachments_json=json.dumps(payload.attachments, ensure_ascii=False),
            delivery_status="received",
            read_status="unread",
            message_metadata_json=json.dumps(payload.metadata, ensure_ascii=False),
            sent_at=utc_now_iso(),
            created_at=utc_now_iso(),
        )
        session.add(message)
        _cancel_pending_automation_tasks(
            session,
            deal=deal,
            task_types={"follow_up_customer"},
            reason="customer_replied",
        )
        _publish_sales_event(
            session,
            tenant_id=deal.tenant_id,
            event_type=sales_event_types.MESSAGE_RECEIVED,
            aggregate_type="message",
            aggregate_id=message.message_id,
            actor_type="system",
            actor_id=None,
            payload={
                "deal_id": deal.deal_id,
                "communication_session_id": communication.communication_id,
                "message_id": message.message_id,
                "channel_type": communication.channel_type,
                "channel_provider": message.channel_provider,
                "external_message_id": message.external_message_id,
            },
        )
        session.commit()
        return _message_to_out(message)
    finally:
        session.close()


@app.post("/api/v1/messages/outbound", response_model=SalesMessageOut)
def outbound_message(
    payload: SalesMessageSendIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesMessageOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        deal = _get_deal(session, payload.deal_id, tenant_id)
        communication = None
        if payload.communication_id:
            communication = session.execute(
                select(SalesCommunicationSessionRow).where(
                    SalesCommunicationSessionRow.communication_id == payload.communication_id,
                    SalesCommunicationSessionRow.tenant_id == tenant_id,
                )
            ).scalar_one_or_none()
            if communication is not None and communication.deal_id != deal.deal_id:
                raise HTTPException(status_code=400, detail="Communication does not belong to deal")
        if communication is None:
            lead = _get_lead(session, deal.lead_id, tenant_id) if deal.lead_id else None
            communication = _start_communication(
                session,
                deal=deal,
                lead=lead,
                payload=SalesCommunicationStartIn(
                    channel_type="text",
                    direction="outbound",
                    agent_type=payload.sender_type if payload.sender_type == "human" else "text_ai",
                    subject=payload.body[:120],
                ),
                metadata={**(payload.metadata or {}), "channel_provider": payload.channel_provider},
            )
        else:
            _merge_communication_metadata(communication, {"channel_provider": payload.channel_provider})
        message = SalesMessageRow(
            message_id=new_id("msg"),
            tenant_id=deal.tenant_id,
            deal_id=deal.deal_id,
            communication_id=communication.communication_id,
            sender_type=payload.sender_type,
            sender_id=payload.sender_id,
            channel_provider=payload.channel_provider,
            external_message_id=None,
            body=payload.body,
            attachments_json=json.dumps(payload.attachments, ensure_ascii=False),
            delivery_status="queued",
            read_status=None,
            message_metadata_json=json.dumps(payload.metadata, ensure_ascii=False),
            sent_at=utc_now_iso(),
            created_at=utc_now_iso(),
        )
        session.add(message)
        _touch_deal_contact(deal)
        _bridge_telegram_outbound(communication, message, deal)
        _schedule_unique_task(
            session,
            deal=deal,
            task_type="follow_up_customer",
            run_at=_automation_run_at_after(days=1),
            payload={"message_id": message.message_id, "reason": "outbound_message.awaiting_reply"},
            match_payload={"message_id": message.message_id},
        )
        _publish_sales_event(
            session,
            tenant_id=deal.tenant_id,
            event_type=sales_event_types.MESSAGE_SENT,
            aggregate_type="message",
            aggregate_id=message.message_id,
            actor_type="human" if payload.sender_type == "human" else "system",
            actor_id=payload.sender_id or actor["user"],
            payload={
                "deal_id": deal.deal_id,
                "communication_session_id": communication.communication_id,
                "message_id": message.message_id,
                "channel_type": communication.channel_type,
                "channel_provider": message.channel_provider,
                "delivery_status": message.delivery_status,
            },
        )
        session.commit()
        return _message_to_out(message)
    finally:
        session.close()


@app.get("/api/v1/deals/{deal_id}/messages", response_model=list[SalesMessageOut])
def list_messages(
    deal_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> list[SalesMessageOut]:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        _get_deal(session, deal_id, tenant_id)
        rows = session.execute(
            select(SalesMessageRow)
            .where(SalesMessageRow.deal_id == deal_id, SalesMessageRow.tenant_id == tenant_id)
            .order_by(SalesMessageRow.sent_at.desc(), SalesMessageRow.id.desc())
        ).scalars().all()
        return [_message_to_out(row) for row in rows]
    finally:
        session.close()


@app.post("/api/v1/calls/inbound-webhook", response_model=SalesCallOut)
def inbound_call(
    payload: SalesCallWebhookIn,
    actor: dict = Depends(get_actor),
) -> SalesCallOut:
    session = get_session()
    try:
        tenant_id = _resolve_provider_tenant_id(
            session,
            actor,
            provider_type="voice",
            provider_name=payload.provider,
            metadata={"external_identifier": payload.external_call_id, "phone_number": payload.phone_number},
            external_identifier=payload.external_call_id or payload.phone_number,
        )
        lead, deal = _resolve_deal_for_inbound(
            session,
            tenant_id=tenant_id,
            deal_id=payload.deal_id,
            lead_id=payload.lead_id,
            phone=payload.phone_number,
            subject=payload.subject,
            source_channel="voice",
            preferred_channel="voice",
        )
        communication = _start_communication(
            session,
            deal=deal,
            lead=lead,
            payload=SalesCommunicationStartIn(
                channel_type="voice",
                direction="inbound",
                agent_type="voice_ai",
                subject=payload.subject or f"Входящий звонок {payload.phone_number}",
            ),
            metadata={"provider": payload.provider, "channel_provider": "voice"},
        )
        row = SalesCallRow(
            call_id=new_id("cal"),
            tenant_id=deal.tenant_id,
            deal_id=deal.deal_id,
            communication_id=communication.communication_id,
            phone_number=payload.phone_number,
            direction="inbound",
            provider=payload.provider,
            external_call_id=payload.external_call_id,
            recording_url=None,
            transcript_status="pending",
            transcript_id=None,
            call_status="started",
            summary=None,
            started_at=utc_now_iso(),
            ended_at=None,
            duration_sec=None,
            created_at=utc_now_iso(),
            updated_at=utc_now_iso(),
        )
        session.add(row)
        _bridge_voice_runtime(row, communication, deal)
        _publish_sales_event(
            session,
            tenant_id=deal.tenant_id,
            event_type=sales_event_types.CALL_RECEIVED,
            aggregate_type="call",
            aggregate_id=row.call_id,
            actor_type="system",
            actor_id=None,
            payload={
                "deal_id": deal.deal_id,
                "communication_session_id": communication.communication_id,
                "call_id": row.call_id,
                "phone_number": row.phone_number,
                "provider": row.provider,
                "external_call_id": row.external_call_id,
            },
        )
        session.commit()
        return _call_to_out(row)
    finally:
        session.close()


@app.post("/api/v1/calls/outbound", response_model=SalesCallOut)
def outbound_call(
    payload: SalesCallWebhookIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesCallOut:
    if not payload.deal_id:
        raise HTTPException(status_code=400, detail="deal_id is required for outbound calls")
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        deal = _get_deal(session, payload.deal_id, tenant_id)
        lead = _get_lead(session, deal.lead_id, tenant_id) if deal.lead_id else None
        communication = _start_communication(
            session,
            deal=deal,
            lead=lead,
            payload=SalesCommunicationStartIn(
                channel_type="voice",
                direction="outbound",
                agent_type="voice_ai",
                subject=payload.subject or f"Исходящий звонок {payload.phone_number}",
            ),
            metadata={"provider": payload.provider, "channel_provider": "voice"},
        )
        row = SalesCallRow(
            call_id=new_id("cal"),
            tenant_id=deal.tenant_id,
            deal_id=deal.deal_id,
            communication_id=communication.communication_id,
            phone_number=payload.phone_number,
            direction="outbound",
            provider=payload.provider,
            external_call_id=payload.external_call_id,
            recording_url=None,
            transcript_status="pending",
            transcript_id=None,
            call_status="started",
            summary=None,
            started_at=utc_now_iso(),
            ended_at=None,
            duration_sec=None,
            created_at=utc_now_iso(),
            updated_at=utc_now_iso(),
        )
        session.add(row)
        _bridge_voice_runtime(row, communication, deal)
        session.commit()
        return _call_to_out(row)
    finally:
        session.close()


@app.post("/api/v1/calls/{call_id}/complete", response_model=SalesCallOut)
def complete_call(
    call_id: str,
    payload: SalesCallCompleteIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesCallOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        row = session.execute(
            select(SalesCallRow).where(SalesCallRow.call_id == call_id, SalesCallRow.tenant_id == tenant_id)
        ).scalar_one_or_none()
        if row is None:
            raise HTTPException(status_code=404, detail="Call not found")
        row.summary = payload.summary
        row.recording_url = payload.recording_url
        row.call_status = payload.result_code or "completed"
        row.ended_at = utc_now_iso()
        if row.started_at:
            started = datetime.fromisoformat(row.started_at.replace("Z", "+00:00"))
            row.duration_sec = max(int((_now() - started).total_seconds()), 0)
        row.updated_at = utc_now_iso()
        communication = session.execute(
            select(SalesCommunicationSessionRow).where(
                SalesCommunicationSessionRow.communication_id == row.communication_id,
                SalesCommunicationSessionRow.tenant_id == tenant_id,
            )
        ).scalar_one_or_none()
        if communication is not None:
            communication.ended_at = row.ended_at
            communication.duration_sec = row.duration_sec
            communication.summary = payload.summary or communication.summary
            communication.result_code = payload.result_code
            communication.status = "completed"
            communication.updated_at = row.updated_at
        deal = _get_deal(session, row.deal_id, tenant_id)
        _touch_deal_contact(deal)
        _publish_sales_event(
            session,
            tenant_id=tenant_id,
            event_type=sales_event_types.CALL_COMPLETED,
            aggregate_type="call",
            aggregate_id=row.call_id,
            actor_type="human",
            actor_id=actor["user"],
            payload={
                "deal_id": deal.deal_id,
                "call_id": row.call_id,
                "communication_session_id": row.communication_id,
                "duration_sec": row.duration_sec,
                "result_code": payload.result_code,
                "transcript_id": row.transcript_id,
            },
        )
        if payload.summary:
            _publish_sales_event(
                session,
                tenant_id=tenant_id,
                event_type=sales_event_types.COMMUNICATION_SUMMARY_CREATED,
                aggregate_type="communication",
                aggregate_id=row.communication_id,
                actor_type="human",
                actor_id=actor["user"],
                payload={
                    "deal_id": deal.deal_id,
                    "call_id": row.call_id,
                    "communication_session_id": row.communication_id,
                    "duration_sec": row.duration_sec,
                    "transcript_id": row.transcript_id,
                    "summary": payload.summary,
                },
            )
        session.commit()
        return _call_to_out(row)
    finally:
        session.close()


@app.get("/api/v1/deals/{deal_id}/calls", response_model=list[SalesCallOut])
def list_calls(
    deal_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> list[SalesCallOut]:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        _get_deal(session, deal_id, tenant_id)
        rows = session.execute(
            select(SalesCallRow)
            .where(SalesCallRow.deal_id == deal_id, SalesCallRow.tenant_id == tenant_id)
            .order_by(SalesCallRow.started_at.desc(), SalesCallRow.id.desc())
        ).scalars().all()
        return [_call_to_out(row) for row in rows]
    finally:
        session.close()


@app.get("/api/v1/calls/{call_id}/transcript", response_model=SalesTranscriptOut)
def get_call_transcript(
    call_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> SalesTranscriptOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        call = session.execute(
            select(SalesCallRow).where(SalesCallRow.call_id == call_id, SalesCallRow.tenant_id == tenant_id)
        ).scalar_one_or_none()
        if call is None:
            raise HTTPException(status_code=404, detail="Call not found")
        row = session.execute(
            select(SalesTranscriptRow)
            .where(SalesTranscriptRow.call_id == call.call_id, SalesTranscriptRow.tenant_id == tenant_id)
            .order_by(SalesTranscriptRow.updated_at.desc(), SalesTranscriptRow.id.desc())
        ).scalars().first()
        if row is None:
            raise HTTPException(status_code=404, detail="Transcript not found")
        return _transcript_to_out(row)
    finally:
        session.close()


@app.post("/api/v1/calls/{call_id}/transcript", response_model=SalesTranscriptOut)
def attach_transcript(
    call_id: str,
    payload: SalesTranscriptIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesTranscriptOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        call = session.execute(
            select(SalesCallRow).where(SalesCallRow.call_id == call_id, SalesCallRow.tenant_id == tenant_id)
        ).scalar_one_or_none()
        if call is None:
            raise HTTPException(status_code=404, detail="Call not found")
        now = utc_now_iso()
        row = SalesTranscriptRow(
            transcript_id=new_id("trn"),
            tenant_id=tenant_id,
            call_id=call.call_id,
            language=payload.language,
            transcript_text=payload.transcript_text,
            diarization_json=json.dumps(payload.diarization, ensure_ascii=False),
            extracted_entities_json=json.dumps(payload.extracted_entities, ensure_ascii=False),
            created_at=now,
            updated_at=now,
        )
        session.add(row)
        call.transcript_id = row.transcript_id
        call.transcript_status = "ready"
        call.updated_at = now
        communication = session.execute(
            select(SalesCommunicationSessionRow).where(
                SalesCommunicationSessionRow.communication_id == call.communication_id,
                SalesCommunicationSessionRow.tenant_id == tenant_id,
            )
        ).scalar_one_or_none()
        if communication is not None:
            communication.transcript_id = row.transcript_id
            communication.updated_at = now
        session.commit()
        return _transcript_to_out(row)
    finally:
        session.close()


@app.post("/api/v1/deals/{deal_id}/offers", response_model=SalesOfferOut)
def create_offer(
    deal_id: str,
    payload: SalesOfferCreate,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesOfferOut:
    session = get_session()
    try:
        deal = _get_deal(session, deal_id, _tenant_id(actor))
        now = utc_now_iso()
        row = SalesOfferRow(
            offer_id=new_id("off"),
            tenant_id=deal.tenant_id,
            deal_id=deal.deal_id,
            offer_type=payload.offer_type,
            title=payload.title,
            description=payload.description,
            line_items_json=json.dumps(payload.line_items, ensure_ascii=False),
            pricing_json=json.dumps(payload.pricing, ensure_ascii=False),
            total_amount=payload.total_amount,
            currency=payload.currency,
            validity_until=payload.validity_until,
            status="draft",
            rendered_document_url=payload.rendered_document_url,
            created_by_type=actor["user"],
            created_at=now,
            updated_at=now,
        )
        session.add(row)
        _apply_stage(session, deal, "offer_preparing", actor_type="human", actor_id=actor["user"], reason="offer.created")
        _publish_sales_event(
            session,
            tenant_id=deal.tenant_id,
            event_type=sales_event_types.OFFER_CREATED,
            aggregate_type="offer",
            aggregate_id=row.offer_id,
            actor_type="human",
            actor_id=actor["user"],
            payload={
                "deal_id": deal.deal_id,
                "offer_id": row.offer_id,
                "offer_type": row.offer_type,
                "total_amount": row.total_amount,
                "currency": row.currency,
            },
        )
        session.commit()
        return _offer_to_out(row)
    finally:
        session.close()


@app.get("/api/v1/deals/{deal_id}/offers", response_model=list[SalesOfferOut])
def list_offers(
    deal_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> list[SalesOfferOut]:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        _get_deal(session, deal_id, tenant_id)
        rows = session.execute(
            select(SalesOfferRow)
            .where(SalesOfferRow.deal_id == deal_id, SalesOfferRow.tenant_id == tenant_id)
            .order_by(SalesOfferRow.updated_at.desc(), SalesOfferRow.id.desc())
        ).scalars().all()
        return [_offer_to_out(row) for row in rows]
    finally:
        session.close()


@app.get("/api/v1/offers/{offer_id}", response_model=SalesOfferOut)
def get_offer(
    offer_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> SalesOfferOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        row = session.execute(
            select(SalesOfferRow).where(SalesOfferRow.offer_id == offer_id, SalesOfferRow.tenant_id == tenant_id)
        ).scalar_one_or_none()
        if row is None:
            raise HTTPException(status_code=404, detail="Offer not found")
        return _offer_to_out(row)
    finally:
        session.close()


@app.post("/api/v1/offers/{offer_id}/send", response_model=SalesOfferOut)
def send_offer(
    offer_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesOfferOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        row = session.execute(
            select(SalesOfferRow).where(SalesOfferRow.offer_id == offer_id, SalesOfferRow.tenant_id == tenant_id)
        ).scalar_one_or_none()
        if row is None:
            raise HTTPException(status_code=404, detail="Offer not found")
        row.status = "sent"
        row.updated_at = utc_now_iso()
        deal = _get_deal(session, row.deal_id, tenant_id)
        _apply_stage(session, deal, "offer_sent", actor_type="system", actor_id=None, reason="offer.sent")
        _schedule_unique_task(
            session,
            deal=deal,
            task_type="follow_up_customer",
            run_at=_automation_run_at_after(days=1),
            payload={"offer_id": row.offer_id, "reason": "offer.sent"},
            match_payload={"offer_id": row.offer_id, "reason": "offer.sent"},
        )
        _publish_sales_event(
            session,
            tenant_id=tenant_id,
            event_type=sales_event_types.OFFER_SENT,
            aggregate_type="offer",
            aggregate_id=row.offer_id,
            actor_type="system",
            actor_id=None,
            payload={
                "deal_id": deal.deal_id,
                "offer_id": row.offer_id,
                "offer_type": row.offer_type,
                "total_amount": row.total_amount,
                "currency": row.currency,
            },
        )
        session.commit()
        return _offer_to_out(row)
    finally:
        session.close()


@app.post("/api/v1/offers/{offer_id}/accept", response_model=SalesOfferOut)
def accept_offer(
    offer_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesOfferOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        row = session.execute(
            select(SalesOfferRow).where(SalesOfferRow.offer_id == offer_id, SalesOfferRow.tenant_id == tenant_id)
        ).scalar_one_or_none()
        if row is None:
            raise HTTPException(status_code=404, detail="Offer not found")
        row.status = "accepted"
        row.updated_at = utc_now_iso()
        deal = _get_deal(session, row.deal_id, tenant_id)
        _apply_stage(session, deal, "conditions_negotiation", actor_type="system", actor_id=None, reason="offer.accepted")
        _publish_sales_event(
            session,
            tenant_id=tenant_id,
            event_type=sales_event_types.OFFER_ACCEPTED,
            aggregate_type="offer",
            aggregate_id=row.offer_id,
            actor_type="system",
            actor_id=None,
            payload={"deal_id": deal.deal_id, "offer_id": row.offer_id, "offer_type": row.offer_type},
        )
        session.commit()
        return _offer_to_out(row)
    finally:
        session.close()


@app.post("/api/v1/offers/{offer_id}/reject", response_model=SalesOfferOut)
def reject_offer(
    offer_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesOfferOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        row = session.execute(
            select(SalesOfferRow).where(SalesOfferRow.offer_id == offer_id, SalesOfferRow.tenant_id == tenant_id)
        ).scalar_one_or_none()
        if row is None:
            raise HTTPException(status_code=404, detail="Offer not found")
        row.status = "rejected"
        row.updated_at = utc_now_iso()
        deal = _get_deal(session, row.deal_id, tenant_id)
        _apply_stage(session, deal, "need_clarification", actor_type="system", actor_id=None, reason="offer.rejected")
        _publish_sales_event(
            session,
            tenant_id=tenant_id,
            event_type=sales_event_types.OFFER_REJECTED,
            aggregate_type="offer",
            aggregate_id=row.offer_id,
            actor_type="system",
            actor_id=None,
            payload={"deal_id": deal.deal_id, "offer_id": row.offer_id, "offer_type": row.offer_type},
        )
        session.commit()
        return _offer_to_out(row)
    finally:
        session.close()


@app.get("/api/v1/deals/{deal_id}/conditions", response_model=SalesConditionOut | None)
def get_deal_conditions(
    deal_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> SalesConditionOut | None:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        _get_deal(session, deal_id, tenant_id)
        row = session.execute(
            select(SalesConditionRow)
            .where(SalesConditionRow.deal_id == deal_id, SalesConditionRow.tenant_id == tenant_id)
            .order_by(SalesConditionRow.updated_at.desc(), SalesConditionRow.id.desc())
        ).scalars().first()
        return _condition_to_out(row) if row else None
    finally:
        session.close()


@app.post("/api/v1/deals/{deal_id}/conditions", response_model=SalesConditionOut)
@app.patch("/api/v1/deals/{deal_id}/conditions", response_model=SalesConditionOut)
@app.put("/api/v1/deals/{deal_id}/conditions", response_model=SalesConditionOut)
def upsert_conditions(
    deal_id: str,
    payload: SalesConditionUpsertIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesConditionOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        deal = _get_deal(session, deal_id, tenant_id)
        row = session.execute(
            select(SalesConditionRow).where(SalesConditionRow.deal_id == deal.deal_id, SalesConditionRow.tenant_id == tenant_id)
        ).scalar_one_or_none()
        now = utc_now_iso()
        if row is None:
            row = SalesConditionRow(
                condition_id=new_id("con"),
                tenant_id=deal.tenant_id,
                deal_id=deal.deal_id,
                product_name=payload.product_name,
                service_name=payload.service_name,
                quantity=payload.quantity,
                unit=payload.unit,
                delivery_mode=payload.delivery_mode,
                execution_date=payload.execution_date,
                start_date=payload.start_date,
                end_date=payload.end_date,
                payment_terms=payload.payment_terms,
                custom_terms_json=json.dumps(payload.custom_terms, ensure_ascii=False),
                agreed_price=payload.agreed_price,
                currency=payload.currency,
                confirmed_at=now,
                created_at=now,
                updated_at=now,
            )
            session.add(row)
        else:
            row.product_name = payload.product_name
            row.service_name = payload.service_name
            row.quantity = payload.quantity
            row.unit = payload.unit
            row.delivery_mode = payload.delivery_mode
            row.execution_date = payload.execution_date
            row.start_date = payload.start_date
            row.end_date = payload.end_date
            row.payment_terms = payload.payment_terms
            row.custom_terms_json = json.dumps(payload.custom_terms, ensure_ascii=False)
            row.agreed_price = payload.agreed_price
            row.currency = payload.currency
            row.confirmed_at = now
            row.updated_at = now
        if payload.agreed_price is not None:
            deal.final_amount = payload.agreed_price
        _apply_stage_if_needed(session, deal, "conditions_negotiation", actor_type="human", actor_id=actor["user"], reason="deal.conditions_confirmed")
        _publish_sales_event(
            session,
            tenant_id=tenant_id,
            event_type=sales_event_types.DEAL_CONDITIONS_CONFIRMED,
            aggregate_type="deal",
            aggregate_id=deal.deal_id,
            actor_type="human",
            actor_id=actor["user"],
            payload={
                "deal_id": deal.deal_id,
                "condition_id": row.condition_id,
                "agreed_price": row.agreed_price,
                "currency": row.currency,
                "payment_terms": row.payment_terms,
            },
        )
        session.commit()
        return _condition_to_out(row)
    finally:
        session.close()


@app.post("/api/v1/deals/{deal_id}/conditions/confirm", response_model=SalesConditionOut)
def confirm_deal_conditions(
    deal_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesConditionOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        deal = _get_deal(session, deal_id, tenant_id)
        row = session.execute(
            select(SalesConditionRow)
            .where(SalesConditionRow.deal_id == deal.deal_id, SalesConditionRow.tenant_id == tenant_id)
            .order_by(SalesConditionRow.updated_at.desc(), SalesConditionRow.id.desc())
        ).scalars().first()
        if row is None:
            raise HTTPException(status_code=404, detail="Conditions not found")
        now = utc_now_iso()
        row.confirmed_at = row.confirmed_at or now
        row.updated_at = now
        if row.agreed_price is not None:
            deal.final_amount = row.agreed_price
        _apply_stage_if_needed(session, deal, "conditions_negotiation", actor_type="human", actor_id=actor["user"], reason="deal.conditions_confirmed")
        _publish_sales_event(
            session,
            tenant_id=tenant_id,
            event_type=sales_event_types.DEAL_CONDITIONS_CONFIRMED,
            aggregate_type="deal",
            aggregate_id=deal.deal_id,
            actor_type="human",
            actor_id=actor["user"],
            payload={
                "deal_id": deal.deal_id,
                "condition_id": row.condition_id,
                "agreed_price": row.agreed_price,
                "currency": row.currency,
                "payment_terms": row.payment_terms,
            },
        )
        session.commit()
        return _condition_to_out(row)
    finally:
        session.close()


@app.post("/api/v1/deals/{deal_id}/counterparty", response_model=SalesCounterpartyOut)
def create_counterparty(
    deal_id: str,
    payload: SalesCounterpartyUpsertIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesCounterpartyOut:
    return patch_counterparty(deal_id, payload, actor)


@app.patch("/api/v1/deals/{deal_id}/counterparty", response_model=SalesCounterpartyOut)
def patch_counterparty(
    deal_id: str,
    payload: SalesCounterpartyUpsertIn,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesCounterpartyOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        deal = _get_deal(session, deal_id, tenant_id)
        row = session.execute(
            select(SalesCounterpartyRow).where(SalesCounterpartyRow.deal_id == deal.deal_id, SalesCounterpartyRow.tenant_id == tenant_id)
        ).scalar_one_or_none()
        now = utc_now_iso()
        status = _counterparty_status(payload)
        if row is None:
            row = SalesCounterpartyRow(
                counterparty_id=new_id("ctp"),
                tenant_id=deal.tenant_id,
                deal_id=deal.deal_id,
                customer_id=deal.customer_id,
                company_name=payload.company_name,
                full_name=payload.full_name,
                bin_iin=payload.bin_iin,
                address=payload.address,
                bank_details_json=json.dumps(payload.bank_details, ensure_ascii=False),
                signer_name=payload.signer_name,
                signer_role=payload.signer_role,
                signer_basis=payload.signer_basis,
                email_for_docs=payload.email_for_docs,
                phone_for_docs=payload.phone_for_docs,
                completeness_status=status,
                created_at=now,
                updated_at=now,
            )
            session.add(row)
        else:
            row.company_name = payload.company_name
            row.full_name = payload.full_name
            row.bin_iin = payload.bin_iin
            row.address = payload.address
            row.bank_details_json = json.dumps(payload.bank_details, ensure_ascii=False)
            row.signer_name = payload.signer_name
            row.signer_role = payload.signer_role
            row.signer_basis = payload.signer_basis
            row.email_for_docs = payload.email_for_docs
            row.phone_for_docs = payload.phone_for_docs
            row.completeness_status = status
            row.updated_at = now
        row.customer_id = _ensure_crm_customer(
            session,
            lead=_get_lead(session, deal.lead_id, tenant_id) if deal.lead_id else None,
            deal=deal,
            counterparty=row,
        )
        _apply_stage_if_needed(
            session,
            deal,
            "counterparty_data_received" if status == "completed" else "counterparty_data_requested",
            actor_type="human",
            actor_id=actor["user"],
            reason="counterparty.completed" if status == "completed" else "counterparty.requested",
        )
        if status == "completed":
            _publish_sales_event(
                session,
                tenant_id=tenant_id,
                event_type=sales_event_types.COUNTERPARTY_COMPLETED,
                aggregate_type="counterparty",
                aggregate_id=row.counterparty_id,
                actor_type="human",
                actor_id=actor["user"],
                payload={
                    "deal_id": deal.deal_id,
                    "counterparty_id": row.counterparty_id,
                    "customer_id": row.customer_id,
                    "completeness_status": row.completeness_status,
                },
            )
        session.commit()
        return _counterparty_to_out(row)
    finally:
        session.close()


@app.get("/api/v1/deals/{deal_id}/counterparty", response_model=SalesCounterpartyOut)
def get_counterparty(
    deal_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> SalesCounterpartyOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        _get_deal(session, deal_id, tenant_id)
        row = session.execute(
            select(SalesCounterpartyRow).where(SalesCounterpartyRow.deal_id == deal_id, SalesCounterpartyRow.tenant_id == tenant_id)
        ).scalar_one_or_none()
        if row is None:
            raise HTTPException(status_code=404, detail="Counterparty not found")
        return _counterparty_to_out(row)
    finally:
        session.close()


@app.get("/api/v1/deals/{deal_id}/documents", response_model=list[SalesDocumentOut])
def list_documents(
    deal_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> list[SalesDocumentOut]:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        _get_deal(session, deal_id, tenant_id)
        rows = session.execute(
            select(SalesDocumentRow)
            .where(SalesDocumentRow.deal_id == deal_id, SalesDocumentRow.tenant_id == tenant_id)
            .order_by(SalesDocumentRow.updated_at.desc(), SalesDocumentRow.id.desc())
        ).scalars().all()
        return [_document_to_out(row) for row in rows]
    finally:
        session.close()


@app.post("/api/v1/deals/{deal_id}/documents", response_model=SalesDocumentOut)
def create_document(
    deal_id: str,
    payload: SalesDocumentCreate,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesDocumentOut:
    session = get_session()
    try:
        deal = _get_deal(session, deal_id, _tenant_id(actor))
        now = utc_now_iso()
        row = SalesDocumentRow(
            document_id=new_id("doc"),
            tenant_id=deal.tenant_id,
            deal_id=deal.deal_id,
            customer_id=deal.customer_id,
            document_type=payload.document_type,
            template_id=payload.template_id,
            version=payload.version,
            status="draft",
            file_url=payload.file_url,
            rendered_payload_json=json.dumps(payload.rendered_payload, ensure_ascii=False),
            external_sign_provider_id=None,
            signed_at=None,
            created_at=now,
            updated_at=now,
        )
        session.add(row)
        _apply_stage(session, deal, "document_preparing", actor_type="human", actor_id=actor["user"], reason="document.created")
        _publish_sales_event(
            session,
            tenant_id=deal.tenant_id,
            event_type=sales_event_types.DOCUMENT_CREATED,
            aggregate_type="document",
            aggregate_id=row.document_id,
            actor_type="human",
            actor_id=actor["user"],
            payload={
                "deal_id": deal.deal_id,
                "document_id": row.document_id,
                "document_type": row.document_type,
                "status": row.status,
            },
        )
        session.commit()
        return _document_to_out(row)
    finally:
        session.close()


@app.get("/api/v1/documents/{document_id}", response_model=SalesDocumentOut)
def get_document(
    document_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> SalesDocumentOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        row = session.execute(
            select(SalesDocumentRow).where(SalesDocumentRow.document_id == document_id, SalesDocumentRow.tenant_id == tenant_id)
        ).scalar_one_or_none()
        if row is None:
            raise HTTPException(status_code=404, detail="Document not found")
        return _document_to_out(row)
    finally:
        session.close()


@app.post("/api/v1/documents/{document_id}/send", response_model=SalesDocumentOut)
def send_document(
    document_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesDocumentOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        row = session.execute(
            select(SalesDocumentRow).where(SalesDocumentRow.document_id == document_id, SalesDocumentRow.tenant_id == tenant_id)
        ).scalar_one_or_none()
        if row is None:
            raise HTTPException(status_code=404, detail="Document not found")
        row.status = "sent"
        row.updated_at = utc_now_iso()
        deal = _get_deal(session, row.deal_id, tenant_id)
        _apply_stage(session, deal, "document_sent", actor_type="system", actor_id=None, reason="document.sent")
        _schedule_unique_task(
            session,
            deal=deal,
            task_type="follow_up_customer",
            run_at=_automation_run_at_after(days=1),
            payload={"document_id": row.document_id, "reason": "document.sent"},
            match_payload={"document_id": row.document_id, "reason": "document.sent"},
        )
        _publish_sales_event(
            session,
            tenant_id=tenant_id,
            event_type=sales_event_types.DOCUMENT_SENT,
            aggregate_type="document",
            aggregate_id=row.document_id,
            actor_type="system",
            actor_id=None,
            payload={
                "deal_id": deal.deal_id,
                "document_id": row.document_id,
                "document_type": row.document_type,
                "status": row.status,
            },
        )
        session.commit()
        return _document_to_out(row)
    finally:
        session.close()


@app.post("/api/v1/documents/{document_id}/confirm", response_model=SalesDocumentOut)
def confirm_document(
    document_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesDocumentOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        row = session.execute(
            select(SalesDocumentRow).where(SalesDocumentRow.document_id == document_id, SalesDocumentRow.tenant_id == tenant_id)
        ).scalar_one_or_none()
        if row is None:
            raise HTTPException(status_code=404, detail="Document not found")
        row.status = "confirmed"
        row.updated_at = utc_now_iso()
        deal = _get_deal(session, row.deal_id, tenant_id)
        _apply_stage(session, deal, "document_confirmed", actor_type="system", actor_id=None, reason="document.confirmed")
        _publish_sales_event(
            session,
            tenant_id=tenant_id,
            event_type=sales_event_types.DOCUMENT_CONFIRMED,
            aggregate_type="document",
            aggregate_id=row.document_id,
            actor_type="system",
            actor_id=None,
            payload={
                "deal_id": deal.deal_id,
                "document_id": row.document_id,
                "document_type": row.document_type,
                "status": row.status,
            },
        )
        session.commit()
        return _document_to_out(row)
    finally:
        session.close()


@app.post("/api/v1/documents/{document_id}/sign-status-webhook", response_model=SalesDocumentOut)
def sign_document(
    document_id: str,
    x_provider_account_id: str | None = Header(default=None, alias="X-Provider-Account-ID"),
    x_provider_name: str | None = Header(default=None, alias="X-Provider-Name"),
    actor: dict = Depends(get_actor),
) -> SalesDocumentOut:
    session = get_session()
    try:
        tenant_id = _resolve_provider_tenant_id(
            session,
            actor,
            provider_type="document_sign",
            provider_name=x_provider_name or "generic",
            provider_account_id=x_provider_account_id,
        )
        row = session.execute(
            select(SalesDocumentRow).where(SalesDocumentRow.document_id == document_id, SalesDocumentRow.tenant_id == tenant_id)
        ).scalar_one_or_none()
        if row is None:
            raise HTTPException(status_code=404, detail="Document not found")
        row.status = "signed"
        row.signed_at = utc_now_iso()
        row.updated_at = row.signed_at
        _publish_sales_event(
            session,
            tenant_id=tenant_id,
            event_type=sales_event_types.DOCUMENT_SIGNED,
            aggregate_type="document",
            aggregate_id=row.document_id,
            actor_type="system",
            actor_id=None,
            payload={
                "deal_id": row.deal_id,
                "document_id": row.document_id,
                "document_type": row.document_type,
                "status": row.status,
                "signed_at": row.signed_at,
            },
        )
        session.commit()
        return _document_to_out(row)
    finally:
        session.close()


@app.get("/api/v1/deals/{deal_id}/invoices", response_model=list[SalesInvoiceOut])
def list_invoices(
    deal_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> list[SalesInvoiceOut]:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        _get_deal(session, deal_id, tenant_id)
        rows = session.execute(
            select(SalesInvoiceRow)
            .where(SalesInvoiceRow.deal_id == deal_id, SalesInvoiceRow.tenant_id == tenant_id)
            .order_by(SalesInvoiceRow.updated_at.desc(), SalesInvoiceRow.id.desc())
        ).scalars().all()
        return [_invoice_to_out(row) for row in rows]
    finally:
        session.close()


@app.post("/api/v1/deals/{deal_id}/invoices", response_model=SalesInvoiceOut)
def create_invoice(
    deal_id: str,
    payload: SalesInvoiceCreate,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesInvoiceOut:
    session = get_session()
    try:
        deal = _get_deal(session, deal_id, _tenant_id(actor))
        now = utc_now_iso()
        row = SalesInvoiceRow(
            invoice_id=new_id("inv"),
            tenant_id=deal.tenant_id,
            deal_id=deal.deal_id,
            customer_id=deal.customer_id,
            invoice_number=_invoice_number(),
            basis_document_id=payload.basis_document_id,
            amount=payload.amount,
            currency=payload.currency,
            due_date=payload.due_date,
            status="draft",
            payment_link=payload.payment_link,
            line_items_json=json.dumps(payload.line_items, ensure_ascii=False),
            metadata_json=json.dumps(payload.metadata, ensure_ascii=False),
            issued_at=None,
            paid_at=None,
            created_at=now,
            updated_at=now,
        )
        session.add(row)
        _apply_stage(session, deal, "invoice_preparing", actor_type="human", actor_id=actor["user"], reason="invoice.created")
        _publish_sales_event(
            session,
            tenant_id=deal.tenant_id,
            event_type=sales_event_types.INVOICE_CREATED,
            aggregate_type="invoice",
            aggregate_id=row.invoice_id,
            actor_type="human",
            actor_id=actor["user"],
            payload={
                "deal_id": deal.deal_id,
                "invoice_id": row.invoice_id,
                "invoice_number": row.invoice_number,
                "amount": row.amount,
                "currency": row.currency,
                "due_date": row.due_date,
            },
        )
        session.commit()
        return _invoice_to_out(row)
    finally:
        session.close()


@app.get("/api/v1/invoices/{invoice_id}", response_model=SalesInvoiceOut)
def get_invoice(
    invoice_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> SalesInvoiceOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        row = session.execute(
            select(SalesInvoiceRow).where(SalesInvoiceRow.invoice_id == invoice_id, SalesInvoiceRow.tenant_id == tenant_id)
        ).scalar_one_or_none()
        if row is None:
            raise HTTPException(status_code=404, detail="Invoice not found")
        return _invoice_to_out(row)
    finally:
        session.close()


@app.post("/api/v1/invoices/{invoice_id}/send", response_model=SalesInvoiceOut)
def send_invoice(
    invoice_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesInvoiceOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        row = session.execute(
            select(SalesInvoiceRow).where(SalesInvoiceRow.invoice_id == invoice_id, SalesInvoiceRow.tenant_id == tenant_id)
        ).scalar_one_or_none()
        if row is None:
            raise HTTPException(status_code=404, detail="Invoice not found")
        row.status = "sent"
        row.issued_at = utc_now_iso()
        row.updated_at = row.issued_at
        deal = _get_deal(session, row.deal_id, tenant_id)
        deal.next_action_type = "payment_follow_up"
        deal.next_action_at = row.due_date
        _schedule_unique_task(
            session,
            deal=deal,
            task_type="send_invoice_reminder",
            run_at=_invoice_reminder_run_at(row.due_date),
            payload={"invoice_id": row.invoice_id, "invoice_number": row.invoice_number, "reason": "invoice.sent"},
            match_payload={"invoice_id": row.invoice_id, "reason": "invoice.sent"},
        )
        _schedule_unique_task(
            session,
            deal=deal,
            task_type="mark_invoice_overdue",
            run_at=_invoice_overdue_run_at(row.due_date),
            payload={"invoice_id": row.invoice_id, "invoice_number": row.invoice_number, "reason": "invoice.due_date"},
            match_payload={"invoice_id": row.invoice_id, "reason": "invoice.due_date"},
        )
        _apply_stage(session, deal, "invoice_sent", actor_type="system", actor_id=None, reason="invoice.sent")
        _publish_sales_event(
            session,
            tenant_id=tenant_id,
            event_type=sales_event_types.INVOICE_SENT,
            aggregate_type="invoice",
            aggregate_id=row.invoice_id,
            actor_type="system",
            actor_id=None,
            payload={
                "deal_id": deal.deal_id,
                "invoice_id": row.invoice_id,
                "invoice_number": row.invoice_number,
                "amount": row.amount,
                "currency": row.currency,
                "due_date": row.due_date,
            },
        )
        session.commit()
        return _invoice_to_out(row)
    finally:
        session.close()


@app.post("/api/v1/invoices/{invoice_id}/mark-overdue", response_model=SalesInvoiceOut)
def mark_invoice_overdue(
    invoice_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesInvoiceOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        row = session.execute(
            select(SalesInvoiceRow).where(SalesInvoiceRow.invoice_id == invoice_id, SalesInvoiceRow.tenant_id == tenant_id)
        ).scalar_one_or_none()
        if row is None:
            raise HTTPException(status_code=404, detail="Invoice not found")
        row.status = "overdue"
        row.updated_at = utc_now_iso()
        deal = _get_deal(session, row.deal_id, tenant_id)
        _apply_stage(session, deal, "payment_overdue", actor_type="system", actor_id=None, reason="invoice.overdue")
        _schedule_unique_task(
            session,
            deal=deal,
            task_type="send_invoice_reminder",
            run_at=utc_now_iso(),
            payload={"invoice_id": row.invoice_id, "invoice_number": row.invoice_number, "reason": "invoice.overdue"},
            match_payload={"invoice_id": row.invoice_id, "reason": "invoice.overdue"},
        )
        _publish_sales_event(
            session,
            tenant_id=tenant_id,
            event_type=sales_event_types.INVOICE_OVERDUE,
            aggregate_type="invoice",
            aggregate_id=row.invoice_id,
            actor_type="system",
            actor_id=None,
            payload={
                "deal_id": deal.deal_id,
                "invoice_id": row.invoice_id,
                "invoice_number": row.invoice_number,
                "amount": row.amount,
                "currency": row.currency,
                "due_date": row.due_date,
            },
        )
        session.commit()
        return _invoice_to_out(row)
    finally:
        session.close()


@app.post("/api/v1/payments/webhook")
async def payment_webhook(
    request: Request,
    actor: dict = Depends(get_actor),
):
    raw_body = await request.body()
    headers = normalize_headers(request.headers)
    try:
        raw_payload = safe_json_loads(raw_body)
    except (json.JSONDecodeError, UnicodeDecodeError, ValueError) as exc:
        raise HTTPException(status_code=400, detail="Invalid payment webhook payload") from exc

    normalized_payload = _normalized_payment_payload(raw_payload, headers)
    provider = normalized_payload["payment_provider"]
    provider_account_id = normalized_payload.get("provider_account_id")
    external_event_id = normalized_payload.get("external_event_id")
    external_payment_id = normalized_payload.get("external_payment_id")
    event_type = normalized_payload["event_type"]
    session = get_session()
    try:
        tenant_id, integration = _resolve_payment_integration(
            session,
            actor,
            provider=provider,
            provider_account_id=provider_account_id,
            external_identifier=normalized_payload["metadata"].get("payment_link_id")
            or normalized_payload["metadata"].get("invoice_external_reference"),
        )
        now = utc_now_iso()
        existing_event = _find_existing_webhook_event(
            session,
            tenant_id=tenant_id,
            provider=provider,
            external_event_id=external_event_id,
        )
        if existing_event is not None:
            try:
                signature_required = provider != "manual" or bool(_payment_webhook_secret(integration))
                verify_payment_webhook_signature(
                    raw_body=raw_body,
                    headers=headers,
                    secret=_payment_webhook_secret(integration),
                    replay_window_seconds=_payment_replay_window_seconds(integration),
                    required=signature_required,
                )
            except PaymentWebhookVerificationError as exc:
                raise HTTPException(status_code=exc.status_code, detail=exc.message) from exc
            return _payment_webhook_event_duplicate_response(session, existing_event)

        event = SalesPaymentWebhookEventRow(
            tenant_id=tenant_id,
            payment_provider=provider,
            provider_account_id=provider_account_id,
            external_event_id=external_event_id,
            external_payment_id=external_payment_id,
            event_type=event_type,
            event_status="received",
            signature_status="unchecked",
            raw_payload_hash=raw_payload_hash(raw_body),
            normalized_payload_json=json.dumps(normalized_payload, ensure_ascii=False),
            headers_json=json.dumps(headers, ensure_ascii=False),
            received_at=now,
            processed_at=None,
            error_message=None,
            created_at=now,
            updated_at=now,
        )
        session.add(event)
        session.flush()

        try:
            signature_required = provider != "manual" or bool(_payment_webhook_secret(integration))
            event.signature_status = verify_payment_webhook_signature(
                raw_body=raw_body,
                headers=headers,
                secret=_payment_webhook_secret(integration),
                replay_window_seconds=_payment_replay_window_seconds(integration),
                required=signature_required,
            )
        except PaymentWebhookVerificationError as exc:
            event.event_status = "rejected"
            event.signature_status = exc.signature_status
            event.error_message = exc.message
            event.updated_at = utc_now_iso()
            session.commit()
            raise HTTPException(status_code=exc.status_code, detail=exc.message) from exc

        event.event_status = "verified"
        event.updated_at = utc_now_iso()

        deal_id = str(normalized_payload.get("deal_id") or "").strip()
        invoice_id = str(normalized_payload.get("invoice_id") or "").strip()
        if not deal_id and invoice_id:
            invoice_for_deal = session.execute(
                select(SalesInvoiceRow).where(SalesInvoiceRow.invoice_id == invoice_id, SalesInvoiceRow.tenant_id == tenant_id)
            ).scalar_one_or_none()
            if invoice_for_deal is not None:
                deal_id = invoice_for_deal.deal_id
        if not deal_id:
            _mark_payment_webhook_event_failed(event, "Payment webhook must include deal_id or invoice_id")
            session.commit()
            raise HTTPException(status_code=400, detail="Payment webhook must include deal_id or invoice_id")

        try:
            deal = _get_deal(session, deal_id, tenant_id)
        except HTTPException as exc:
            _mark_payment_webhook_event_failed(event, exc.detail)
            session.commit()
            raise
        invoice = None
        if invoice_id:
            invoice = session.execute(
                select(SalesInvoiceRow).where(SalesInvoiceRow.invoice_id == invoice_id, SalesInvoiceRow.tenant_id == tenant_id)
            ).scalar_one_or_none()
            if invoice is None:
                _mark_payment_webhook_event_failed(event, "Invoice not found")
                session.commit()
                raise HTTPException(status_code=404, detail="Invoice not found")
            if invoice.deal_id != deal.deal_id:
                _mark_payment_webhook_event_failed(event, "Invoice does not belong to deal")
                session.commit()
                raise HTTPException(status_code=400, detail="Invoice does not belong to deal")

        try:
            payment_metadata = dict(normalized_payload["metadata"])
            payment_metadata.update(
                {
                    "external_event_id": external_event_id,
                    "event_type": event_type,
                    "webhook_event_row_id": event.id,
                    "raw_payload_hash": event.raw_payload_hash,
                }
            )
            row = _apply_payment_update(
                session,
                tenant_id=tenant_id,
                deal=deal,
                invoice=invoice,
                payment_provider=provider,
                external_payment_id=external_payment_id,
                amount=normalized_payload.get("amount"),
                currency=normalized_payload.get("currency") or "KZT",
                status=normalized_payload.get("status") or "pending",
                paid_at=normalized_payload.get("paid_at"),
                payment_method=normalized_payload.get("payment_method"),
                failure_reason=normalized_payload.get("failure_reason"),
                metadata=payment_metadata,
                integration=integration,
            )
        except HTTPException as exc:
            _mark_payment_webhook_event_failed(event, exc.detail)
            session.commit()
            raise

        result = _payment_to_dict(row)
        normalized_payload["result"] = {
            "payment_id": row.payment_id,
            "invoice_id": row.invoice_id,
            "deal_id": row.deal_id,
            "status": row.status,
        }
        event.normalized_payload_json = json.dumps(normalized_payload, ensure_ascii=False)
        event.event_status = "processed"
        event.processed_at = utc_now_iso()
        event.updated_at = event.processed_at
        session.commit()
        return result
    except IntegrityError:
        session.rollback()
        resolved_tenant_id = tenant_id if "tenant_id" in locals() else ""
        existing_event = _find_existing_webhook_event(
            session,
            tenant_id=resolved_tenant_id,
            provider=provider,
            external_event_id=external_event_id,
        )
        if existing_event is not None:
            return _payment_webhook_event_duplicate_response(session, existing_event)
        raise
    finally:
        session.close()


@app.get("/api/v1/deals/{deal_id}/automation-tasks", response_model=list[SalesAutomationTaskOut])
def list_automation_tasks(
    deal_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> list[SalesAutomationTaskOut]:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        _get_deal(session, deal_id, tenant_id)
        rows = session.execute(
            select(SalesAutomationTaskRow)
            .where(SalesAutomationTaskRow.deal_id == deal_id, SalesAutomationTaskRow.tenant_id == tenant_id)
            .order_by(SalesAutomationTaskRow.run_at.desc(), SalesAutomationTaskRow.id.desc())
        ).scalars().all()
        return [_task_to_out(row) for row in rows]
    finally:
        session.close()


@app.get("/api/v1/automation-tasks", response_model=list[SalesAutomationTaskOut])
def list_tenant_automation_tasks(
    status: str | None = Query(default=None),
    task_type: str | None = Query(default=None),
    run_at: str | None = Query(default=None),
    deal_search: str | None = Query(default=None),
    failed_only: bool = Query(default=False),
    pending_only: bool = Query(default=False),
    limit: int = Query(default=100, ge=1, le=500),
    offset: int = Query(default=0, ge=0),
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> list[SalesAutomationTaskOut]:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        conditions = [SalesAutomationTaskRow.tenant_id == tenant_id]

        if status:
            conditions.append(SalesAutomationTaskRow.status == status)
        if task_type:
            conditions.append(SalesAutomationTaskRow.task_type == task_type)
        if failed_only:
            conditions.append(
                or_(
                    SalesAutomationTaskRow.status == "failed",
                    SalesAutomationTaskRow.failed_at.is_not(None),
                    SalesAutomationTaskRow.last_error.is_not(None),
                )
            )
        elif pending_only:
            conditions.append(SalesAutomationTaskRow.status == "pending")

        now = datetime.now(tz=timezone.utc).replace(microsecond=0)
        if run_at == "today":
            day_start = now.replace(hour=0, minute=0, second=0)
            day_end = day_start + timedelta(days=1)
            conditions.append(SalesAutomationTaskRow.run_at >= day_start.isoformat())
            conditions.append(SalesAutomationTaskRow.run_at < day_end.isoformat())
        elif run_at == "overdue":
            conditions.append(SalesAutomationTaskRow.status == "pending")
            conditions.append(SalesAutomationTaskRow.run_at < now.isoformat())
        elif run_at == "future":
            conditions.append(SalesAutomationTaskRow.run_at >= now.isoformat())

        if deal_search:
            term = f"%{deal_search.strip().lower()}%"
            if term != "%%":
                conditions.append(
                    or_(
                        func.lower(SalesAutomationTaskRow.deal_id).like(term),
                        func.lower(SalesDealRow.title).like(term),
                        func.lower(SalesDealRow.customer_id).like(term),
                    )
                )

        rows = session.execute(
            select(SalesAutomationTaskRow, SalesDealRow)
            .join(
                SalesDealRow,
                (SalesDealRow.deal_id == SalesAutomationTaskRow.deal_id)
                & (SalesDealRow.tenant_id == SalesAutomationTaskRow.tenant_id),
                isouter=True,
            )
            .where(*conditions)
            .order_by(SalesAutomationTaskRow.run_at.asc(), SalesAutomationTaskRow.id.desc())
            .limit(limit)
            .offset(offset)
        ).all()
        return [_task_to_out(task, deal) for task, deal in rows]
    finally:
        session.close()


@app.post("/api/v1/automation-tasks/{task_id}/cancel", response_model=SalesAutomationTaskOut)
def cancel_automation_task(
    task_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesAutomationTaskOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        row = session.execute(
            select(SalesAutomationTaskRow).where(SalesAutomationTaskRow.task_id == task_id, SalesAutomationTaskRow.tenant_id == tenant_id)
        ).scalar_one_or_none()
        if row is None:
            raise HTTPException(status_code=404, detail="Automation task not found")
        if row.status != "pending":
            raise HTTPException(status_code=400, detail={"code": "automation_task_not_pending", "message": "Only pending automation tasks can be canceled"})
        payload = _json_dict(row.payload_json)
        payload["canceled_by"] = actor["user"]
        payload["canceled_reason"] = "manual"
        row.payload_json = json.dumps(payload, ensure_ascii=False)
        row.status = "canceled"
        row.locked_at = None
        row.locked_by = None
        row.updated_at = utc_now_iso()
        session.commit()
        return _task_to_out(row)
    finally:
        session.close()


@app.post("/api/v1/automation-tasks/{task_id}/run-now", response_model=SalesAutomationTaskOut)
def run_automation_task_now(
    task_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesAutomationTaskOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        row = session.execute(
            select(SalesAutomationTaskRow).where(SalesAutomationTaskRow.task_id == task_id, SalesAutomationTaskRow.tenant_id == tenant_id)
        ).scalar_one_or_none()
        if row is None:
            raise HTTPException(status_code=404, detail="Automation task not found")
        if row.status not in {"pending", "failed", "canceled"}:
            raise HTTPException(status_code=400, detail={"code": "automation_task_not_pending", "message": "Automation task cannot be queued"})
        row.status = "pending"
        row.run_at = utc_now_iso()
        row.locked_at = None
        row.locked_by = None
        row.failed_at = None
        row.completed_at = None
        row.last_error = None
        row.updated_at = row.run_at
        session.commit()
        return _task_to_out(row)
    finally:
        session.close()


@app.get("/api/v1/deals/{deal_id}/payments", response_model=list[SalesPaymentOut])
def list_payments(
    deal_id: str,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR, Role.ANALYST)),
) -> list[SalesPaymentOut]:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        _get_deal(session, deal_id, tenant_id)
        rows = session.execute(
            select(SalesPaymentRow)
            .where(SalesPaymentRow.deal_id == deal_id, SalesPaymentRow.tenant_id == tenant_id)
            .order_by(SalesPaymentRow.updated_at.desc(), SalesPaymentRow.id.desc())
        ).scalars().all()
        return [_payment_to_out(row) for row in rows]
    finally:
        session.close()


@app.post("/api/v1/payments/{payment_id}/reconcile", response_model=SalesPaymentOut)
def reconcile_payment(
    payment_id: str,
    payload: SalesPaymentReconcileIn | None = None,
    actor: dict = Depends(require_roles(Role.ADMIN, Role.SUPERVISOR, Role.OPERATOR)),
) -> SalesPaymentOut:
    session = get_session()
    try:
        tenant_id = _tenant_id(actor)
        row = session.execute(
            select(SalesPaymentRow).where(SalesPaymentRow.payment_id == payment_id, SalesPaymentRow.tenant_id == tenant_id)
        ).scalar_one_or_none()
        if row is None:
            raise HTTPException(status_code=404, detail="Payment not found")
        deal = _get_deal(session, row.deal_id, tenant_id)
        invoice = None
        if row.invoice_id:
            invoice = session.execute(
                select(SalesInvoiceRow).where(SalesInvoiceRow.invoice_id == row.invoice_id, SalesInvoiceRow.tenant_id == tenant_id)
            ).scalar_one_or_none()
            if invoice is None:
                raise HTTPException(status_code=404, detail="Invoice not found")

        provider = str(row.payment_provider or "manual").strip().lower() or "manual"
        integration = session.execute(
            select(TenantIntegrationRow)
            .where(
                TenantIntegrationRow.tenant_id == tenant_id,
                func.lower(TenantIntegrationRow.provider_type) == "payment",
                func.lower(TenantIntegrationRow.provider_name) == provider,
                TenantIntegrationRow.is_active == True,  # noqa: E712
            )
            .order_by(TenantIntegrationRow.id.desc())
        ).scalars().first()

        metadata = _json_dict(row.metadata_json)
        failure_reason = payload.failure_reason if payload is not None else row.failure_reason
        if payload is not None:
            status = payload.status
            metadata.update(payload.metadata or {})
        else:
            adapter = get_payment_provider_adapter(provider)
            if adapter is None:
                if provider == "manual":
                    status = row.status
                else:
                    raise HTTPException(status_code=400, detail="Payment provider adapter is not configured")
            elif not row.external_payment_id:
                raise HTTPException(status_code=400, detail="Payment external_payment_id is required for reconcile")
            else:
                provider_result = adapter.get_status(
                    row.external_payment_id,
                    context={
                        "tenant_id": tenant_id,
                        "payment_id": row.payment_id,
                        "payment_provider": provider,
                        "integration_id": integration.integration_id if integration is not None else None,
                    },
                )
                if isinstance(provider_result, dict):
                    status = normalize_payment_status(provider, str(provider_result.get("status") or provider_result.get("provider_status") or ""))
                    failure_reason = str(provider_result.get("failure_reason") or failure_reason or "") or None
                    metadata.update(provider_result.get("metadata") if isinstance(provider_result.get("metadata"), dict) else {})
                    if provider_result.get("amount") is not None:
                        row.amount = _coerce_payment_amount(provider_result.get("amount"))
                    if provider_result.get("currency"):
                        row.currency = str(provider_result.get("currency")).upper()
                else:
                    status = normalize_payment_status(provider, str(provider_result))

        updated = _apply_payment_update(
            session,
            tenant_id=tenant_id,
            deal=deal,
            invoice=invoice,
            payment_provider=provider,
            external_payment_id=row.external_payment_id,
            amount=row.amount,
            currency=row.currency,
            status=status,
            paid_at=row.paid_at,
            payment_method=row.payment_method,
            failure_reason=failure_reason,
            metadata=metadata,
            integration=integration,
            existing_payment=row,
        )
        session.commit()
        return _payment_to_out(updated)
    finally:
        session.close()
