Implement sales CRM workflow foundation

services/sales_service/automation_worker.py

@@ -0,0 +1,695 @@
+from __future__ import annotations
+
+import json
+import socket
+from datetime import datetime, timedelta, timezone
+from typing import Any, Callable
+
+from sqlalchemy import select
+
+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.shared.core import new_id, utc_now_iso
+from services.shared.db import get_session
+from services.shared.sales_sql_models import (
+    SalesAutomationTaskRow,
+    SalesChannelSwitchRow,
+    SalesCommunicationSessionRow,
+    SalesDealRow,
+    SalesEscalationRow,
+    SalesInvoiceRow,
+    SalesPipelineStageRow,
+)
+
+
+RETRY_DELAYS_SECONDS = {
+    1: 60,
+    2: 300,
+    3: 900,
+}
+ACTIVE_TASK_STATUSES = {"pending", "running"}
+TERMINAL_TASK_STATUSES = {"completed", "failed", "canceled"}
+
+
+def _now_dt() -> datetime:
+    return datetime.now(timezone.utc).replace(microsecond=0)
+
+
+def _iso(dt: datetime) -> str:
+    return dt.astimezone(timezone.utc).replace(microsecond=0).isoformat()
+
+
+def _payload(row: SalesAutomationTaskRow) -> dict[str, Any]:
+    try:
+        parsed = json.loads(row.payload_json or "{}")
+    except json.JSONDecodeError:
+        return {}
+    return parsed if isinstance(parsed, dict) else {}
+
+
+def _set_payload(row: SalesAutomationTaskRow, payload: dict[str, Any]) -> None:
+    row.payload_json = json.dumps(payload, ensure_ascii=False)
+
+
+def _task_error_message(exc: Exception) -> str:
+    message = str(exc).strip()
+    return message[:1000] if message else exc.__class__.__name__
+
+
+class SalesAutomationWorker:
+    def __init__(self, *, worker_id: str | None = None, batch_size: int = 10) -> None:
+        self.worker_id = worker_id or f"sales-worker-{socket.gethostname()}-{new_id('wrk')}"
+        self.batch_size = batch_size
+        self.handlers: dict[str, Callable[[Any, SalesAutomationTaskRow], None]] = {
+            "follow_up_customer": self._handle_follow_up_customer,
+            "mark_invoice_overdue": self._handle_mark_invoice_overdue,
+            "send_invoice_reminder": self._handle_send_invoice_reminder,
+            "recommend_channel_switch": self._handle_recommend_channel_switch,
+            "close_stale_communication": self._handle_close_stale_communication,
+            "escalate_to_human": self._handle_escalate_to_human,
+            "post_sale_transfer": self._handle_post_sale_transfer,
+        }
+
+    def claim_pending_tasks(self, session, *, limit: int | None = None, now: str | None = None) -> list[SalesAutomationTaskRow]:
+        claim_limit = limit or self.batch_size
+        now_text = now or utc_now_iso()
+        stmt = (
+            select(SalesAutomationTaskRow)
+            .where(
+                SalesAutomationTaskRow.status == "pending",
+                SalesAutomationTaskRow.run_at <= now_text,
+            )
+            .order_by(SalesAutomationTaskRow.run_at.asc(), SalesAutomationTaskRow.id.asc())
+            .limit(claim_limit)
+        )
+        if session.bind is not None and session.bind.dialect.name == "postgresql":
+            stmt = stmt.with_for_update(skip_locked=True)
+        rows = session.execute(stmt).scalars().all()
+        for row in rows:
+            row.status = "running"
+            row.locked_at = now_text
+            row.locked_by = self.worker_id
+            row.last_error = None
+            row.updated_at = now_text
+        session.flush()
+        return list(rows)
+
+    def run_once(self, *, limit: int | None = None) -> int:
+        session = get_session()
+        try:
+            tasks = self.claim_pending_tasks(session, limit=limit)
+            task_ids = [task.task_id for task in tasks]
+            session.commit()
+        finally:
+            session.close()
+
+        for task_id in task_ids:
+            self._execute_claimed_task(task_id)
+        return len(task_ids)
+
+    def _execute_claimed_task(self, task_id: str) -> None:
+        session = get_session()
+        try:
+            task = session.execute(
+                select(SalesAutomationTaskRow).where(
+                    SalesAutomationTaskRow.task_id == task_id,
+                    SalesAutomationTaskRow.status == "running",
+                    SalesAutomationTaskRow.locked_by == self.worker_id,
+                )
+            ).scalar_one_or_none()
+            if task is None:
+                session.rollback()
+                return
+            try:
+                self._execute_task(session, task)
+            except Exception as exc:  # noqa: BLE001
+                self._mark_failed_or_retry(task, exc)
+            else:
+                self._mark_completed(task)
+            session.commit()
+        finally:
+            session.close()
+
+    def _execute_task(self, session, task: SalesAutomationTaskRow) -> None:
+        handler = self.handlers.get(task.task_type)
+        if handler is None:
+            raise RuntimeError(f"Unsupported automation task type: {task.task_type}")
+        handler(session, task)
+
+    def _mark_completed(self, task: SalesAutomationTaskRow) -> None:
+        now = utc_now_iso()
+        task.status = "completed"
+        task.completed_at = now
+        task.failed_at = None
+        task.locked_at = None
+        task.locked_by = None
+        task.last_error = None
+        task.updated_at = now
+
+    def _mark_failed_or_retry(self, task: SalesAutomationTaskRow, exc: Exception) -> None:
+        now_dt = _now_dt()
+        task.retry_count += 1
+        task.last_error = _task_error_message(exc)
+        task.locked_at = None
+        task.locked_by = None
+        task.updated_at = _iso(now_dt)
+        max_retries = int(task.max_retries or 3)
+        if task.retry_count < max_retries:
+            task.status = "pending"
+            delay_seconds = RETRY_DELAYS_SECONDS.get(task.retry_count, 900)
+            task.run_at = _iso(now_dt + timedelta(seconds=delay_seconds))
+            return
+        task.status = "failed"
+        task.failed_at = _iso(now_dt)
+
+    def _get_deal(self, session, task: SalesAutomationTaskRow) -> SalesDealRow:
+        row = session.execute(
+            select(SalesDealRow).where(
+                SalesDealRow.tenant_id == task.tenant_id,
+                SalesDealRow.deal_id == task.deal_id,
+            )
+        ).scalar_one_or_none()
+        if row is None:
+            raise RuntimeError(f"Deal not found: {task.deal_id}")
+        return row
+
+    def _get_invoice(self, session, task: SalesAutomationTaskRow, invoice_id: str) -> SalesInvoiceRow:
+        row = session.execute(
+            select(SalesInvoiceRow).where(
+                SalesInvoiceRow.tenant_id == task.tenant_id,
+                SalesInvoiceRow.invoice_id == invoice_id,
+            )
+        ).scalar_one_or_none()
+        if row is None:
+            raise RuntimeError(f"Invoice not found: {invoice_id}")
+        if row.deal_id != task.deal_id:
+            raise RuntimeError("Invoice does not belong to task deal")
+        return row
+
+    def _stage_code(self, session, deal: SalesDealRow) -> str | None:
+        stage = session.execute(
+            select(SalesPipelineStageRow).where(
+                SalesPipelineStageRow.tenant_id == deal.tenant_id,
+                SalesPipelineStageRow.stage_id == deal.stage_id,
+            )
+        ).scalar_one_or_none()
+        return stage.code if stage is not None else None
+
+    def _transition_stage(
+        self,
+        session,
+        *,
+        task: SalesAutomationTaskRow,
+        target_stage_code: str,
+        reason: str,
+        force_on_invalid: bool = True,
+    ) -> None:
+        metadata = {"automation_task_id": task.task_id, "task_type": task.task_type}
+        try:
+            transition_deal_stage(
+                session,
+                tenant_id=task.tenant_id,
+                deal_id=task.deal_id,
+                target_stage_code=target_stage_code,
+                actor_type="system",
+                reason=reason,
+                metadata=metadata,
+            )
+        except DealStateMachineError as exc:
+            if not force_on_invalid or exc.error != "invalid_stage_transition":
+                raise
+            transition_deal_stage(
+                session,
+                tenant_id=task.tenant_id,
+                deal_id=task.deal_id,
+                target_stage_code=target_stage_code,
+                actor_type="system",
+                reason=reason,
+                metadata={**metadata, "force_reason": reason},
+                force=True,
+            )
+
+    def _publish(
+        self,
+        session,
+        *,
+        task: SalesAutomationTaskRow,
+        event_type: str,
+        aggregate_type: str,
+        aggregate_id: str,
+        payload: dict[str, Any],
+    ) -> None:
+        SalesEventPublisher.publish_sales_event(
+            session,
+            tenant_id=task.tenant_id,
+            event_type=event_type,
+            aggregate_type=aggregate_type,
+            aggregate_id=aggregate_id,
+            actor_type="system",
+            actor_id=self.worker_id,
+            payload={
+                "deal_id": task.deal_id,
+                "automation_task_id": task.task_id,
+                "task_type": task.task_type,
+                **payload,
+            },
+        )
+
+    def _find_related_task(
+        self,
+        session,
+        *,
+        tenant_id: str,
+        deal_id: str,
+        task_type: str,
+        match_payload: dict[str, Any],
+    ) -> SalesAutomationTaskRow | None:
+        rows = session.execute(
+            select(SalesAutomationTaskRow)
+            .where(
+                SalesAutomationTaskRow.tenant_id == tenant_id,
+                SalesAutomationTaskRow.deal_id == deal_id,
+                SalesAutomationTaskRow.task_type == task_type,
+                SalesAutomationTaskRow.status.in_(["pending", "running", "completed"]),
+            )
+            .order_by(SalesAutomationTaskRow.id.desc())
+        ).scalars().all()
+        for row in rows:
+            payload = _payload(row)
+            if all(payload.get(key) == value for key, value in match_payload.items()):
+                return row
+        return None
+
+    def _create_task_once(
+        self,
+        session,
+        *,
+        tenant_id: str,
+        deal_id: str,
+        task_type: str,
+        run_at: str | None = None,
+        payload: dict[str, Any] | None = None,
+        match_payload: dict[str, Any] | None = None,
+    ) -> SalesAutomationTaskRow:
+        payload_data = payload or {}
+        existing = self._find_related_task(
+            session,
+            tenant_id=tenant_id,
+            deal_id=deal_id,
+            task_type=task_type,
+            match_payload=match_payload or payload_data,
+        )
+        if existing is not None:
+            return existing
+        now = utc_now_iso()
+        row = SalesAutomationTaskRow(
+            task_id=new_id("tsk"),
+            tenant_id=tenant_id,
+            deal_id=deal_id,
+            task_type=task_type,
+            payload_json=json.dumps(payload_data, 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(row)
+        return row
+
+    def _handle_follow_up_customer(self, session, task: SalesAutomationTaskRow) -> None:
+        deal = self._get_deal(session, task)
+        data = _payload(task)
+        if deal.status != "active":
+            data["automation_result"] = {"status": "skipped", "reason": f"deal_status_{deal.status}"}
+            _set_payload(task, data)
+            return
+
+        now = utc_now_iso()
+        deal.next_action_type = str(data.get("next_action_type") or "follow_up_customer")
+        deal.next_action_at = now
+        deal.updated_at = now
+        current_stage = self._stage_code(session, deal)
+        preferred_channel = str(data.get("preferred_channel") or deal.current_channel or deal.preferred_channel or "text")
+        if current_stage == "follow_up_scheduled":
+            target = "active_voice_communication" if preferred_channel == "voice" else "active_text_communication"
+            self._transition_stage(
+                session,
+                task=task,
+                target_stage_code=target,
+                reason="automation.follow_up_customer",
+            )
+        data["automation_result"] = {
+            "status": "follow_up_ready",
+            "next_action_type": deal.next_action_type,
+            "next_action_at": deal.next_action_at,
+        }
+        _set_payload(task, data)
+        self._publish(
+            session,
+            task=task,
+            event_type=sales_event_types.DEAL_FOLLOW_UP_READY,
+            aggregate_type="deal",
+            aggregate_id=deal.deal_id,
+            payload={"next_action_type": deal.next_action_type, "next_action_at": deal.next_action_at},
+        )
+
+    def _handle_mark_invoice_overdue(self, session, task: SalesAutomationTaskRow) -> None:
+        data = _payload(task)
+        invoice_id = str(data.get("invoice_id") or "").strip()
+        if not invoice_id:
+            raise RuntimeError("invoice_id is required")
+        invoice = self._get_invoice(session, task, invoice_id)
+        deal = self._get_deal(session, task)
+        if invoice.status in {"paid", "canceled"}:
+            data["automation_result"] = {"status": "skipped", "invoice_status": invoice.status}
+            _set_payload(task, data)
+            return
+
+        previous_status = invoice.status
+        now = utc_now_iso()
+        invoice.status = "overdue"
+        invoice.updated_at = now
+        if deal.status == "active":
+            self._transition_stage(
+                session,
+                task=task,
+                target_stage_code="payment_overdue",
+                reason="automation.mark_invoice_overdue",
+            )
+        if previous_status != "overdue":
+            self._publish(
+                session,
+                task=task,
+                event_type=sales_event_types.INVOICE_OVERDUE,
+                aggregate_type="invoice",
+                aggregate_id=invoice.invoice_id,
+                payload={
+                    "invoice_id": invoice.invoice_id,
+                    "invoice_number": invoice.invoice_number,
+                    "amount": invoice.amount,
+                    "currency": invoice.currency,
+                    "due_date": invoice.due_date,
+                },
+            )
+        reminder = self._create_task_once(
+            session,
+            tenant_id=task.tenant_id,
+            deal_id=task.deal_id,
+            task_type="send_invoice_reminder",
+            payload={"invoice_id": invoice.invoice_id, "source_task_id": task.task_id, "reason": "invoice.overdue"},
+            match_payload={"invoice_id": invoice.invoice_id, "reason": "invoice.overdue"},
+        )
+        data["automation_result"] = {"status": "invoice_overdue", "reminder_task_id": reminder.task_id}
+        _set_payload(task, data)
+
+    def _handle_send_invoice_reminder(self, session, task: SalesAutomationTaskRow) -> None:
+        data = _payload(task)
+        invoice_id = str(data.get("invoice_id") or "").strip()
+        if not invoice_id:
+            raise RuntimeError("invoice_id is required")
+        invoice = self._get_invoice(session, task, invoice_id)
+        deal = self._get_deal(session, task)
+        if invoice.status in {"paid", "canceled"} or deal.status != "active":
+            data["automation_result"] = {
+                "status": "skipped",
+                "invoice_status": invoice.status,
+                "deal_status": deal.status,
+            }
+            _set_payload(task, data)
+            return
+
+        now = utc_now_iso()
+        deal.next_action_type = "invoice_reminder"
+        deal.next_action_at = now
+        deal.updated_at = now
+        follow_up = self._create_task_once(
+            session,
+            tenant_id=task.tenant_id,
+            deal_id=task.deal_id,
+            task_type="follow_up_customer",
+            payload={"invoice_id": invoice.invoice_id, "source_task_id": task.task_id, "reason": "invoice.reminder"},
+            match_payload={"invoice_id": invoice.invoice_id, "reason": "invoice.reminder"},
+        )
+        data["automation_result"] = {
+            "status": "invoice_reminder_ready",
+            "follow_up_task_id": follow_up.task_id,
+        }
+        _set_payload(task, data)
+        self._publish(
+            session,
+            task=task,
+            event_type=sales_event_types.INVOICE_REMINDER_SENT,
+            aggregate_type="invoice",
+            aggregate_id=invoice.invoice_id,
+            payload={
+                "invoice_id": invoice.invoice_id,
+                "invoice_number": invoice.invoice_number,
+                "invoice_status": invoice.status,
+                "follow_up_task_id": follow_up.task_id,
+            },
+        )
+
+    def _handle_recommend_channel_switch(self, session, task: SalesAutomationTaskRow) -> None:
+        deal = self._get_deal(session, task)
+        data = _payload(task)
+        recommendation = {
+            "recommended_from_channel": data.get("recommended_from_channel") or deal.current_channel,
+            "recommended_to_channel": data.get("recommended_to_channel") or data.get("to_channel") or "voice",
+            "reason": data.get("reason") or "automation_recommendation",
+        }
+        deal.next_action_type = "channel_switch_recommended"
+        deal.next_action_at = utc_now_iso()
+        deal.updated_at = deal.next_action_at
+        data["automation_result"] = recommendation
+        if data.get("auto_apply"):
+            existing_switch = session.execute(
+                select(SalesChannelSwitchRow)
+                .where(
+                    SalesChannelSwitchRow.tenant_id == task.tenant_id,
+                    SalesChannelSwitchRow.deal_id == task.deal_id,
+                    SalesChannelSwitchRow.recommended_by_task_id == task.task_id,
+                )
+                .order_by(SalesChannelSwitchRow.id.desc())
+            ).scalars().first()
+            if existing_switch is None:
+                now = utc_now_iso()
+                to_channel = "voice" if recommendation["recommended_to_channel"] == "voice" else "text"
+                from_channel = "voice" if recommendation["recommended_from_channel"] == "voice" else "text"
+                existing_switch = SalesChannelSwitchRow(
+                    switch_id=new_id("swc"),
+                    tenant_id=task.tenant_id,
+                    deal_id=task.deal_id,
+                    communication_id=None,
+                    communication_session_id=None,
+                    previous_communication_session_id=None,
+                    new_communication_session_id=None,
+                    from_channel=from_channel,
+                    to_channel=to_channel,
+                    reason_code=str(data.get("reason_code") or "system_recommendation"),
+                    reason_text=recommendation["reason"],
+                    reason_for_channel_switch=recommendation["reason"],
+                    initiated_by_type="system",
+                    initiated_by_id=None,
+                    source_type="automation_task",
+                    source_id=task.task_id,
+                    recommended_by_task_id=task.task_id,
+                    switched_at=now,
+                    created_at=now,
+                )
+                session.add(existing_switch)
+                deal.current_channel = to_channel
+                deal.updated_at = now
+                current_stage = self._stage_code(session, deal)
+                target_stage = "active_voice_communication" if to_channel == "voice" else "active_text_communication"
+                if current_stage in {"active_text_communication", "active_voice_communication"} and current_stage != target_stage:
+                    transition_deal_stage(
+                        session,
+                        tenant_id=task.tenant_id,
+                        deal_id=task.deal_id,
+                        target_stage_code=target_stage,
+                        actor_type="system",
+                        reason="automation.recommend_channel_switch",
+                        metadata={"automation_task_id": task.task_id, "switch_id": existing_switch.switch_id},
+                    )
+                self._publish(
+                    session,
+                    task=task,
+                    event_type=sales_event_types.COMMUNICATION_CHANNEL_SWITCHED,
+                    aggregate_type="deal",
+                    aggregate_id=deal.deal_id,
+                    payload={
+                        "switch_id": existing_switch.switch_id,
+                        "from_channel": existing_switch.from_channel,
+                        "to_channel": existing_switch.to_channel,
+                        "reason_code": existing_switch.reason_code,
+                        "recommended_by_task_id": task.task_id,
+                    },
+                )
+            data["actual_switch_id"] = existing_switch.switch_id
+        _set_payload(task, data)
+        self._publish(
+            session,
+            task=task,
+            event_type=sales_event_types.DEAL_CHANNEL_SWITCH_RECOMMENDED,
+            aggregate_type="deal",
+            aggregate_id=deal.deal_id,
+            payload=recommendation,
+        )
+
+    def _handle_close_stale_communication(self, session, task: SalesAutomationTaskRow) -> None:
+        data = _payload(task)
+        communication_id = str(data.get("communication_id") or "").strip()
+        if not communication_id:
+            raise RuntimeError("communication_id is required")
+        communication = session.execute(
+            select(SalesCommunicationSessionRow).where(
+                SalesCommunicationSessionRow.tenant_id == task.tenant_id,
+                SalesCommunicationSessionRow.deal_id == task.deal_id,
+                SalesCommunicationSessionRow.communication_id == communication_id,
+            )
+        ).scalar_one_or_none()
+        if communication is None:
+            raise RuntimeError(f"Communication not found: {communication_id}")
+        if communication.status != "completed":
+            now = utc_now_iso()
+            communication.status = "completed"
+            communication.ended_at = communication.ended_at or now
+            communication.result_code = communication.result_code or str(data.get("result_code") or "stale_closed")
+            communication.updated_at = now
+        follow_up = self._create_task_once(
+            session,
+            tenant_id=task.tenant_id,
+            deal_id=task.deal_id,
+            task_type="follow_up_customer",
+            payload={"communication_id": communication.communication_id, "source_task_id": task.task_id, "reason": "stale_communication"},
+            match_payload={"communication_id": communication.communication_id, "reason": "stale_communication"},
+        )
+        data["automation_result"] = {"status": "communication_closed", "follow_up_task_id": follow_up.task_id}
+        _set_payload(task, data)
+
+    def _handle_escalate_to_human(self, session, task: SalesAutomationTaskRow) -> None:
+        deal = self._get_deal(session, task)
+        data = _payload(task)
+        escalation_type = str(data.get("escalation_type") or "automation_escalation").strip()
+        reason = str(data.get("reason") or "automation.escalate_to_human").strip()
+        assigned_to = str(data.get("assigned_to_user_id") or deal.assigned_human_user_id or "").strip() or None
+        existing = session.execute(
+            select(SalesEscalationRow)
+            .where(
+                SalesEscalationRow.tenant_id == task.tenant_id,
+                SalesEscalationRow.deal_id == task.deal_id,
+                SalesEscalationRow.escalation_type == escalation_type,
+                SalesEscalationRow.status.in_(["open", "assigned", "in_progress"]),
+            )
+            .order_by(SalesEscalationRow.id.desc())
+        ).scalars().first()
+        now = utc_now_iso()
+        if existing is None:
+            existing = SalesEscalationRow(
+                escalation_id=new_id("esc"),
+                tenant_id=task.tenant_id,
+                deal_id=task.deal_id,
+                escalation_type=escalation_type,
+                reason=reason,
+                severity=str(data.get("severity") or "medium"),
+                status="assigned" if assigned_to else "open",
+                assigned_to_user_id=assigned_to,
+                assigned_at=now if assigned_to else None,
+                started_at=None,
+                created_at=now,
+                resolved_at=None,
+                canceled_at=None,
+                resolution_code=None,
+                resolution_summary=None,
+                sla_due_at=data.get("sla_due_at"),
+                source_channel=deal.current_channel,
+                source_communication_session_id=data.get("communication_id"),
+                source_automation_task_id=task.task_id,
+                updated_at=now,
+            )
+            session.add(existing)
+            self._publish(
+                session,
+                task=task,
+                event_type=sales_event_types.ESCALATION_CREATED,
+                aggregate_type="escalation",
+                aggregate_id=existing.escalation_id,
+                payload={
+                    "escalation_id": existing.escalation_id,
+                    "escalation_type": existing.escalation_type,
+                    "status": existing.status,
+                    "severity": existing.severity,
+                },
+            )
+        else:
+            existing.updated_at = now
+        deal.assigned_human_user_id = assigned_to
+        deal.scenario_type = "custom_human_escalation"
+        deal.updated_at = now
+        if self._stage_code(session, deal) != "transferred_to_support":
+            transition_deal_stage(
+                session,
+                tenant_id=task.tenant_id,
+                deal_id=task.deal_id,
+                target_stage_code="transferred_to_support",
+                actor_type="system",
+                reason="automation.escalate_to_human",
+                metadata={"automation_task_id": task.task_id, "escalation_id": existing.escalation_id},
+                force=True,
+            )
+        data["automation_result"] = {"status": "escalated", "escalation_id": existing.escalation_id}
+        _set_payload(task, data)
+        self._publish(
+            session,
+            task=task,
+            event_type=sales_event_types.DEAL_ESCALATION_REQUESTED,
+            aggregate_type="deal",
+            aggregate_id=deal.deal_id,
+            payload={
+                "escalation_id": existing.escalation_id,
+                "escalation_type": existing.escalation_type,
+                "severity": existing.severity,
+                "assigned_to_user_id": existing.assigned_to_user_id,
+            },
+        )
+
+    def _handle_post_sale_transfer(self, session, task: SalesAutomationTaskRow) -> None:
+        deal = self._get_deal(session, task)
+        current_stage = self._stage_code(session, deal)
+        data = _payload(task)
+        if current_stage == "transferred_to_execution":
+            data["automation_result"] = {"status": "skipped", "reason": "already_transferred"}
+            _set_payload(task, data)
+            return
+        if deal.status != "won":
+            data["automation_result"] = {"status": "skipped", "reason": f"deal_status_{deal.status}"}
+            _set_payload(task, data)
+            return
+        transition_deal_stage(
+            session,
+            tenant_id=task.tenant_id,
+            deal_id=task.deal_id,
+            target_stage_code="transferred_to_execution",
+            actor_type="system",
+            reason="automation.post_sale_transfer",
+            metadata={"automation_task_id": task.task_id},
+            force=True,
+        )
+        data["automation_result"] = {"status": "transferred_post_sale"}
+        _set_payload(task, data)
+        self._publish(
+            session,
+            task=task,
+            event_type=sales_event_types.DEAL_TRANSFERRED_POST_SALE,
+            aggregate_type="deal",
+            aggregate_id=deal.deal_id,
+            payload={"transfer_status": "stub_created"},
+        )

services/sales_service/deal_state_machine.py

@@ -0,0 +1,648 @@
+from __future__ import annotations
+
+from typing import Any
+
+from sqlalchemy import select
+
+from services.sales_service import sales_events as sales_event_types
+from services.sales_service.event_publisher import SalesEventPublisher
+from services.shared.core import new_id, utc_now_iso
+from services.shared.sales_sql_models import (
+    SalesConditionRow,
+    SalesCounterpartyRow,
+    SalesDealRow,
+    SalesDocumentRow,
+    SalesInvoiceRow,
+    SalesOfferRow,
+    SalesPaymentRow,
+    SalesPipelineStageRow,
+    SalesStageHistoryRow,
+)
+
+
+ALLOWED_TRANSITIONS: dict[str, list[str]] = {
+    "new_qualified_lead": [
+        "warm_lead",
+        "hot_lead",
+        "enrichment_required",
+        "active_text_communication",
+        "active_voice_communication",
+        "lost",
+    ],
+    "warm_lead": [
+        "active_text_communication",
+        "active_voice_communication",
+        "enrichment_required",
+        "lost",
+    ],
+    "hot_lead": [
+        "active_text_communication",
+        "active_voice_communication",
+        "offer_selection",
+        "lost",
+    ],
+    "enrichment_required": [
+        "active_text_communication",
+        "active_voice_communication",
+        "need_clarification",
+        "lost",
+    ],
+    "active_text_communication": [
+        "waiting_customer_reply",
+        "need_clarification",
+        "need_confirmed",
+        "active_voice_communication",
+        "follow_up_scheduled",
+        "transferred_to_support",
+        "lost",
+    ],
+    "active_voice_communication": [
+        "waiting_customer_reply",
+        "need_clarification",
+        "need_confirmed",
+        "active_text_communication",
+        "follow_up_scheduled",
+        "transferred_to_support",
+        "lost",
+    ],
+    "waiting_customer_reply": [
+        "active_text_communication",
+        "active_voice_communication",
+        "follow_up_scheduled",
+        "lost",
+    ],
+    "need_clarification": [
+        "active_text_communication",
+        "active_voice_communication",
+        "need_confirmed",
+        "transferred_to_support",
+        "lost",
+    ],
+    "need_confirmed": [
+        "offer_selection",
+        "offer_preparing",
+        "transferred_to_support",
+        "lost",
+    ],
+    "offer_selection": [
+        "offer_preparing",
+        "conditions_negotiation",
+        "lost",
+    ],
+    "offer_preparing": [
+        "offer_sent",
+        "conditions_negotiation",
+        "lost",
+    ],
+    "offer_sent": [
+        "conditions_negotiation",
+        "counterparty_data_requested",
+        "follow_up_scheduled",
+        "lost",
+    ],
+    "conditions_negotiation": [
+        "counterparty_data_requested",
+        "counterparty_data_received",
+        "document_preparing",
+        "invoice_preparing",
+        "transferred_to_support",
+        "lost",
+    ],
+    "counterparty_data_requested": [
+        "counterparty_data_received",
+        "follow_up_scheduled",
+        "lost",
+    ],
+    "counterparty_data_received": [
+        "document_preparing",
+        "invoice_preparing",
+        "lost",
+    ],
+    "document_preparing": [
+        "document_sent",
+        "lost",
+    ],
+    "document_sent": [
+        "document_under_review",
+        "document_confirmed",
+        "follow_up_scheduled",
+        "lost",
+    ],
+    "document_under_review": [
+        "document_confirmed",
+        "transferred_to_support",
+        "lost",
+    ],
+    "document_confirmed": [
+        "invoice_preparing",
+        "invoice_sent",
+        "lost",
+    ],
+    "invoice_preparing": [
+        "invoice_sent",
+        "lost",
+    ],
+    "invoice_sent": [
+        "payment_expected",
+        "partially_paid",
+        "paid",
+        "payment_overdue",
+        "lost",
+    ],
+    "payment_expected": [
+        "partially_paid",
+        "paid",
+        "payment_overdue",
+        "lost",
+    ],
+    "partially_paid": [
+        "paid",
+        "payment_overdue",
+        "lost",
+    ],
+    "payment_overdue": [
+        "payment_expected",
+        "partially_paid",
+        "paid",
+        "lost",
+    ],
+    "paid": [
+        "won",
+        "transferred_to_execution",
+        "transferred_to_support",
+    ],
+    "follow_up_scheduled": [
+        "active_text_communication",
+        "active_voice_communication",
+        "waiting_customer_reply",
+        "lost",
+    ],
+    "postponed": [
+        "follow_up_scheduled",
+        "active_text_communication",
+        "active_voice_communication",
+        "lost",
+    ],
+    "transferred_to_support": [
+        "active_text_communication",
+        "active_voice_communication",
+        "conditions_negotiation",
+        "lost",
+        "won",
+    ],
+    "won": [],
+    "lost": [],
+    "transferred_to_execution": [],
+}
+
+LEGACY_STAGE_ALIASES = {
+    "new": "new_qualified_lead",
+    "active_text": "active_text_communication",
+    "invoice": "invoice_sent",
+    "paid": "paid",
+    "won": "won",
+}
+TERMINAL_STAGE_CODES = {"won", "lost", "transferred_to_execution"}
+SCENARIOS_REQUIRING_OFFER = {"quotation_based_sale", "booking_based_sale", "subscription_sale"}
+
+
+class DealStateMachineError(Exception):
+    def __init__(
+        self,
+        *,
+        status_code: int,
+        error: str,
+        message: str,
+        details: dict[str, Any] | None = None,
+    ) -> None:
+        super().__init__(message)
+        self.status_code = status_code
+        self.error = error
+        self.message = message
+        self.details = details or {}
+
+    def response(self) -> dict[str, Any]:
+        return {"error": self.error, "message": self.message, "details": self.details}
+
+
+def _raise(status_code: int, error: str, message: str, details: dict[str, Any] | None = None) -> None:
+    raise DealStateMachineError(status_code=status_code, error=error, message=message, details=details)
+
+
+def _normalize_stage_code(value: str | None) -> str:
+    code = str(value or "").strip()
+    return LEGACY_STAGE_ALIASES.get(code, code)
+
+
+def _get_deal(session, *, tenant_id: str, deal_id: str) -> SalesDealRow:
+    row = session.execute(
+        select(SalesDealRow).where(
+            SalesDealRow.tenant_id == tenant_id,
+            SalesDealRow.deal_id == deal_id,
+        )
+    ).scalar_one_or_none()
+    if row is None:
+        _raise(404, "deal_not_found", "Deal not found", {"deal_id": deal_id})
+    return row
+
+
+def _get_current_stage(session, deal: SalesDealRow) -> SalesPipelineStageRow:
+    row = session.execute(
+        select(SalesPipelineStageRow).where(
+            SalesPipelineStageRow.tenant_id == deal.tenant_id,
+            SalesPipelineStageRow.stage_id == deal.stage_id,
+        )
+    ).scalar_one_or_none()
+    if row is None:
+        _raise(
+            404,
+            "stage_not_found",
+            "Current deal stage not found",
+            {"deal_id": deal.deal_id, "stage_id": deal.stage_id},
+        )
+    if row.pipeline_id != deal.pipeline_id:
+        _raise(
+            400,
+            "invalid_stage_pipeline",
+            "Current deal stage does not belong to deal pipeline",
+            {
+                "deal_id": deal.deal_id,
+                "stage_id": row.stage_id,
+                "stage_pipeline_id": row.pipeline_id,
+                "deal_pipeline_id": deal.pipeline_id,
+            },
+        )
+    return row
+
+
+def _get_target_stage(session, deal: SalesDealRow, target_stage_code: str) -> SalesPipelineStageRow:
+    code = _normalize_stage_code(target_stage_code)
+    row = session.execute(
+        select(SalesPipelineStageRow).where(
+            SalesPipelineStageRow.tenant_id == deal.tenant_id,
+            SalesPipelineStageRow.pipeline_id == deal.pipeline_id,
+            SalesPipelineStageRow.code == code,
+        )
+    ).scalar_one_or_none()
+    if row is None:
+        _raise(
+            404,
+            "stage_not_found",
+            "Target stage not found",
+            {"deal_id": deal.deal_id, "target_stage_code": code},
+        )
+    if not row.is_active:
+        _raise(
+            400,
+            "stage_inactive",
+            "Target stage is inactive",
+            {"stage_id": row.stage_id, "stage_code": row.code},
+        )
+    return row
+
+
+def _first_offer(session, deal: SalesDealRow, statuses: set[str] | None = None) -> SalesOfferRow | None:
+    stmt = select(SalesOfferRow).where(
+        SalesOfferRow.tenant_id == deal.tenant_id,
+        SalesOfferRow.deal_id == deal.deal_id,
+    )
+    if statuses is not None:
+        stmt = stmt.where(SalesOfferRow.status.in_(statuses))
+    return session.execute(stmt.order_by(SalesOfferRow.updated_at.desc(), SalesOfferRow.id.desc())).scalars().first()
+
+
+def _counterparty(session, deal: SalesDealRow) -> SalesCounterpartyRow | None:
+    return session.execute(
+        select(SalesCounterpartyRow).where(
+            SalesCounterpartyRow.tenant_id == deal.tenant_id,
+            SalesCounterpartyRow.deal_id == deal.deal_id,
+        )
+    ).scalar_one_or_none()
+
+
+def _first_document(session, deal: SalesDealRow, statuses: set[str] | None = None) -> SalesDocumentRow | None:
+    stmt = select(SalesDocumentRow).where(
+        SalesDocumentRow.tenant_id == deal.tenant_id,
+        SalesDocumentRow.deal_id == deal.deal_id,
+    )
+    if statuses is not None:
+        stmt = stmt.where(SalesDocumentRow.status.in_(statuses))
+    return session.execute(stmt.order_by(SalesDocumentRow.updated_at.desc(), SalesDocumentRow.id.desc())).scalars().first()
+
+
+def _first_invoice(session, deal: SalesDealRow, statuses: set[str] | None = None) -> SalesInvoiceRow | None:
+    stmt = select(SalesInvoiceRow).where(
+        SalesInvoiceRow.tenant_id == deal.tenant_id,
+        SalesInvoiceRow.deal_id == deal.deal_id,
+    )
+    if statuses is not None:
+        stmt = stmt.where(SalesInvoiceRow.status.in_(statuses))
+    return session.execute(stmt.order_by(SalesInvoiceRow.updated_at.desc(), SalesInvoiceRow.id.desc())).scalars().first()
+
+
+def _successful_payment(session, deal: SalesDealRow) -> SalesPaymentRow | None:
+    return session.execute(
+        select(SalesPaymentRow)
+        .where(
+            SalesPaymentRow.tenant_id == deal.tenant_id,
+            SalesPaymentRow.deal_id == deal.deal_id,
+            SalesPaymentRow.status == "success",
+        )
+        .order_by(SalesPaymentRow.updated_at.desc(), SalesPaymentRow.id.desc())
+    ).scalars().first()
+
+
+def _confirmed_conditions(session, deal: SalesDealRow) -> SalesConditionRow | None:
+    return session.execute(
+        select(SalesConditionRow)
+        .where(
+            SalesConditionRow.tenant_id == deal.tenant_id,
+            SalesConditionRow.deal_id == deal.deal_id,
+            SalesConditionRow.confirmed_at.is_not(None),
+        )
+        .order_by(SalesConditionRow.updated_at.desc(), SalesConditionRow.id.desc())
+    ).scalars().first()
+
+
+def _has_paid_invoice_or_payment(session, deal: SalesDealRow) -> bool:
+    return _first_invoice(session, deal, {"paid"}) is not None or _successful_payment(session, deal) is not None
+
+
+def _scenario_allows_transition(deal: SalesDealRow, from_code: str, to_code: str) -> bool:
+    if deal.scenario_type == "quick_sale" and from_code == "need_confirmed":
+        return to_code in {"invoice_preparing", "invoice_sent"}
+    if deal.scenario_type == "custom_human_escalation" and to_code == "transferred_to_support":
+        return True
+    return False
+
+
+def _validate_allowed_transition(
+    deal: SalesDealRow,
+    from_stage: SalesPipelineStageRow,
+    to_stage: SalesPipelineStageRow,
+) -> None:
+    if from_stage.stage_id == to_stage.stage_id:
+        return
+    allowed = ALLOWED_TRANSITIONS.get(from_stage.code, [])
+    if to_stage.code in allowed or _scenario_allows_transition(deal, from_stage.code, to_stage.code):
+        return
+    _raise(
+        400,
+        "invalid_stage_transition",
+        f"Cannot transition deal from {from_stage.code} to {to_stage.code}",
+        {
+            "deal_id": deal.deal_id,
+            "from_stage_code": from_stage.code,
+            "to_stage_code": to_stage.code,
+        },
+    )
+
+
+def _validate_preconditions(
+    session,
+    *,
+    deal: SalesDealRow,
+    to_stage: SalesPipelineStageRow,
+    reason: str | None,
+    metadata: dict[str, Any],
+) -> None:
+    target = to_stage.code
+    if target == "need_confirmed" and not str(deal.need_summary or "").strip():
+        _raise(
+            400,
+            "missing_required_need_summary",
+            "need_summary is required before moving deal to need_confirmed",
+            {"deal_id": deal.deal_id, "to_stage_code": target},
+        )
+
+    if target == "invoice_preparing" and deal.scenario_type in SCENARIOS_REQUIRING_OFFER:
+        if _first_offer(session, deal) is None:
+            _raise(
+                400,
+                "offer_required_before_invoice",
+                "Offer is required before invoice preparation for this scenario",
+                {"deal_id": deal.deal_id, "scenario_type": deal.scenario_type},
+            )
+
+    if target == "offer_sent" and _first_offer(session, deal, {"draft", "ready", "sent"}) is None:
+        _raise(
+            400,
+            "offer_required_before_offer_sent",
+            "Offer is required before moving deal to offer_sent",
+            {"deal_id": deal.deal_id, "to_stage_code": target},
+        )
+
+    if target == "counterparty_data_received":
+        counterparty = _counterparty(session, deal)
+        status = str(counterparty.completeness_status if counterparty else "").strip().lower()
+        if status not in {"complete", "completed", "partial_acceptable"}:
+            _raise(
+                400,
+                "counterparty_required_before_document",
+                "Complete or acceptable counterparty data is required",
+                {"deal_id": deal.deal_id, "to_stage_code": target},
+            )
+
+    if target == "document_sent" and _first_document(session, deal, {"rendered", "sent"}) is None:
+        _raise(
+            400,
+            "document_required_before_document_sent",
+            "Rendered or sent document is required before moving deal to document_sent",
+            {"deal_id": deal.deal_id, "to_stage_code": target},
+        )
+
+    if target == "invoice_sent":
+        if _first_invoice(session, deal, {"issued", "sent"}) is None:
+            _raise(
+                400,
+                "invoice_required_before_invoice_sent",
+                "Issued or sent invoice is required before moving deal to invoice_sent",
+                {"deal_id": deal.deal_id, "to_stage_code": target},
+            )
+        if deal.document_required and _first_document(session, deal, {"confirmed", "signed"}) is None:
+            _raise(
+                400,
+                "document_confirmed_required_before_invoice_sent",
+                "Confirmed document is required before sending invoice",
+                {"deal_id": deal.deal_id, "to_stage_code": target},
+            )
+
+    if target == "paid" and not _has_paid_invoice_or_payment(session, deal):
+        _raise(
+            400,
+            "payment_required_before_paid",
+            "Successful payment or paid invoice is required before moving deal to paid",
+            {"deal_id": deal.deal_id, "to_stage_code": target},
+        )
+
+    if target == "won":
+        if deal.payment_required and not _has_paid_invoice_or_payment(session, deal):
+            _raise(
+                400,
+                "payment_required_before_won",
+                "Successful payment or paid invoice is required before moving deal to won",
+                {"deal_id": deal.deal_id, "to_stage_code": target},
+            )
+        if not deal.payment_required and _confirmed_conditions(session, deal) is None:
+            _raise(
+                400,
+                "conditions_required_before_won",
+                "Confirmed conditions are required before moving deal to won without payment",
+                {"deal_id": deal.deal_id, "to_stage_code": target},
+            )
+
+    if target == "lost":
+        close_reason = (
+            str(metadata.get("lost_reason") or "").strip()
+            or str(metadata.get("close_reason") or "").strip()
+            or str(reason or "").strip()
+            or str(deal.lost_reason or "").strip()
+            or str(deal.close_reason or "").strip()
+        )
+        if not close_reason:
+            _raise(
+                400,
+                "lost_reason_required",
+                "lost_reason or close_reason is required before moving deal to lost",
+                {"deal_id": deal.deal_id, "to_stage_code": target},
+            )
+
+
+def _record_stage_change(
+    session,
+    *,
+    deal: SalesDealRow,
+    from_stage_id: str | None,
+    to_stage_id: str,
+    actor_type: str,
+    actor_id: str | None,
+    reason: str | None,
+    changed_at: str,
+) -> 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=actor_type,
+            changed_by_id=actor_id,
+            reason=reason,
+            changed_at=changed_at,
+        )
+    )
+
+
+def _apply_terminal_state(deal: SalesDealRow, *, to_stage: SalesPipelineStageRow, reason: str | None, metadata: dict[str, Any], now: str) -> None:
+    if to_stage.code == "won":
+        deal.status = "won"
+        deal.closed_at = deal.closed_at or now
+        deal.won_reason = str(metadata.get("won_reason") or reason or deal.won_reason or "deal.won").strip()
+        return
+
+    if to_stage.code == "lost":
+        close_reason = (
+            str(metadata.get("lost_reason") or "").strip()
+            or str(metadata.get("close_reason") or "").strip()
+            or str(reason or "").strip()
+            or str(deal.lost_reason or "").strip()
+            or str(deal.close_reason or "").strip()
+        )
+        deal.status = "lost"
+        deal.closed_at = deal.closed_at or now
+        deal.lost_reason = close_reason
+        deal.close_reason = deal.close_reason or close_reason
+        return
+
+    if to_stage.code == "transferred_to_execution":
+        deal.status = "closed"
+        deal.closed_at = deal.closed_at or now
+        deal.close_reason = str(metadata.get("close_reason") or reason or deal.close_reason or "transferred_to_execution").strip()
+        return
+
+    if deal.status in {"won", "lost", "closed"}:
+        deal.status = "active"
+        deal.closed_at = None
+
+
+def transition_deal_stage(
+    session,
+    *,
+    tenant_id: str,
+    deal_id: str,
+    target_stage_code: str,
+    actor_type: str,
+    actor_id: str | None = None,
+    reason: str | None = None,
+    metadata: dict | None = None,
+    force: bool = False,
+) -> SalesPipelineStageRow:
+    reason_text = str(reason or "").strip() or None
+    event_metadata = dict(metadata or {})
+    if force:
+        if actor_type not in {"system", "admin"}:
+            _raise(
+                403,
+                "force_transition_forbidden",
+                "force transition is allowed only for system or admin actors",
+                {"actor_type": actor_type},
+            )
+        if not reason_text:
+            _raise(
+                400,
+                "force_transition_reason_required",
+                "reason is required for force transition",
+                {"deal_id": deal_id, "target_stage_code": target_stage_code},
+            )
+        event_metadata["force"] = True
+
+    deal = _get_deal(session, tenant_id=tenant_id, deal_id=deal_id)
+    current_stage = _get_current_stage(session, deal)
+    target_stage = _get_target_stage(session, deal, target_stage_code)
+
+    if not force:
+        _validate_allowed_transition(deal, current_stage, target_stage)
+        _validate_preconditions(session, deal=deal, to_stage=target_stage, reason=reason_text, metadata=event_metadata)
+
+    now = utc_now_iso()
+    if current_stage.stage_id == target_stage.stage_id:
+        deal.updated_at = now
+        _apply_terminal_state(deal, to_stage=target_stage, reason=reason_text, metadata=event_metadata, now=now)
+        return target_stage
+
+    previous_stage_id = deal.stage_id
+    deal.stage_id = target_stage.stage_id
+    _apply_terminal_state(deal, to_stage=target_stage, reason=reason_text, metadata=event_metadata, now=now)
+    deal.updated_at = now
+
+    _record_stage_change(
+        session,
+        deal=deal,
+        from_stage_id=previous_stage_id,
+        to_stage_id=target_stage.stage_id,
+        actor_type=actor_type,
+        actor_id=actor_id,
+        reason=reason_text,
+        changed_at=now,
+    )
+    SalesEventPublisher.publish_sales_event(
+        session,
+        tenant_id=deal.tenant_id,
+        event_type=sales_event_types.DEAL_STAGE_CHANGED,
+        aggregate_type="deal",
+        aggregate_id=deal.deal_id,
+        actor_type=actor_type,
+        actor_id=actor_id,
+        payload={
+            "deal_id": deal.deal_id,
+            "pipeline_id": deal.pipeline_id,
+            "from_stage_id": previous_stage_id,
+            "from_stage_code": current_stage.code,
+            "to_stage_id": target_stage.stage_id,
+            "to_stage_code": target_stage.code,
+            "reason": reason_text,
+            "metadata": event_metadata,
+        },
+    )
+    return target_stage

services/sales_service/payment_webhooks.py

@@ -0,0 +1,226 @@
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+from datetime import datetime, timezone
+from typing import Any, Protocol
+
+
+class PaymentWebhookVerificationError(Exception):
+    def __init__(self, signature_status: str, message: str, *, status_code: int = 400) -> None:
+        super().__init__(message)
+        self.signature_status = signature_status
+        self.message = message
+        self.status_code = status_code
+
+
+class PaymentProviderAdapter(Protocol):
+    def get_status(self, external_payment_id: str, *, context: dict[str, Any] | None = None) -> dict[str, Any] | str:
+        ...
+
+
+_PROVIDER_ADAPTERS: dict[str, PaymentProviderAdapter] = {}
+
+
+def register_payment_provider_adapter(provider: str, adapter: PaymentProviderAdapter) -> None:
+    normalized = normalize_provider(provider)
+    if normalized:
+        _PROVIDER_ADAPTERS[normalized] = adapter
+
+
+def get_payment_provider_adapter(provider: str | None) -> PaymentProviderAdapter | None:
+    return _PROVIDER_ADAPTERS.get(normalize_provider(provider))
+
+
+def normalize_provider(provider: str | None) -> str:
+    return str(provider or "").strip().lower() or "manual"
+
+
+def raw_payload_hash(raw_body: bytes) -> str:
+    return hashlib.sha256(raw_body).hexdigest()
+
+
+def safe_json_loads(raw_body: bytes) -> dict[str, Any]:
+    if not raw_body:
+        return {}
+    parsed = json.loads(raw_body.decode("utf-8"))
+    if not isinstance(parsed, dict):
+        raise ValueError("Webhook payload must be a JSON object")
+    return parsed
+
+
+def normalize_headers(headers: Any) -> dict[str, str]:
+    return {str(key).lower(): str(value) for key, value in dict(headers).items()}
+
+
+def header_value(headers: dict[str, str], *names: str) -> str | None:
+    for name in names:
+        value = headers.get(name.lower())
+        normalized = str(value or "").strip()
+        if normalized:
+            return normalized
+    return None
+
+
+def metadata_value(payload: dict[str, Any] | None, *keys: str) -> str | None:
+    data = payload or {}
+    metadata = data.get("metadata") if isinstance(data.get("metadata"), dict) else {}
+    candidates = [data, metadata]
+    payment = data.get("payment") if isinstance(data.get("payment"), dict) else {}
+    candidates.append(payment)
+    nested_metadata = payment.get("metadata") if isinstance(payment.get("metadata"), dict) else {}
+    candidates.append(nested_metadata)
+    data_object = data.get("data", {}).get("object") if isinstance(data.get("data"), dict) else {}
+    if isinstance(data_object, dict):
+        candidates.append(data_object)
+
+    for source in candidates:
+        for key in keys:
+            value = source.get(key)
+            normalized = str(value or "").strip()
+            if normalized:
+                return normalized
+    return None
+
+
+def extract_payment_provider(payload: dict[str, Any], headers: dict[str, str]) -> str:
+    return normalize_provider(
+        header_value(headers, "x-payment-provider", "x-provider")
+        or metadata_value(payload, "payment_provider", "provider")
+    )
+
+
+def extract_provider_account_id(payload: dict[str, Any], headers: dict[str, str]) -> str | None:
+    return header_value(
+        headers,
+        "x-provider-account-id",
+        "x-merchant-id",
+        "x-terminal-id",
+        "x-integration-id",
+    ) or metadata_value(
+        payload,
+        "provider_account_id",
+        "payment_provider_account_id",
+        "merchant_id",
+        "terminal_id",
+        "account_id",
+        "integration_id",
+    )
+
+
+def extract_external_event_id(payload: dict[str, Any], headers: dict[str, str]) -> str | None:
+    return header_value(headers, "x-webhook-event-id", "x-event-id") or metadata_value(
+        payload,
+        "external_event_id",
+        "webhook_event_id",
+        "event_id",
+    )
+
+
+def extract_external_payment_id(payload: dict[str, Any], headers: dict[str, str]) -> str | None:
+    return header_value(headers, "x-payment-id", "x-external-payment-id") or metadata_value(
+        payload,
+        "external_payment_id",
+        "external_payment_ref",
+        "transaction_id",
+        "provider_payment_id",
+        "payment_id",
+        "id",
+    )
+
+
+def extract_event_type(payload: dict[str, Any], headers: dict[str, str]) -> str:
+    return (
+        header_value(headers, "x-webhook-event-type", "x-event-type")
+        or metadata_value(payload, "event_type", "type")
+        or normalize_payment_event_type(None, metadata_value(payload, "status", "provider_status"))
+    )
+
+
+def normalize_payment_status(provider: str | None, provider_status: str | None) -> str:
+    normalized = str(provider_status or "").strip().lower()
+    normalized = normalized.replace("-", "_").replace(" ", "_")
+    if normalized in {"succeeded", "success", "paid", "captured", "settled", "completed"}:
+        return "success"
+    if normalized in {"authorized", "authorised", "processing", "pending", "created", "new"}:
+        return "pending"
+    if normalized in {"failed", "declined", "error", "rejected"}:
+        return "failed"
+    if normalized in {"canceled", "cancelled", "voided", "expired"}:
+        return "canceled"
+    if normalized in {"partial", "partially_paid", "partial_paid"}:
+        return "partial"
+    return normalized if normalized in {"pending", "success", "failed", "canceled", "partial"} else "pending"
+
+
+def normalize_payment_event_type(provider: str | None, provider_status: str | None, event_type: str | None = None) -> str:
+    raw_event_type = str(event_type or "").strip().lower()
+    if raw_event_type.startswith("payment."):
+        if raw_event_type in {"payment.succeeded", "payment.captured", "payment.paid"}:
+            return "payment.received"
+        if raw_event_type in {"payment.cancelled", "payment.canceled"}:
+            return "payment.canceled"
+        return raw_event_type
+
+    status = normalize_payment_status(provider, provider_status)
+    if status in {"success", "partial"}:
+        return "payment.received"
+    if status == "failed":
+        return "payment.failed"
+    if status == "canceled":
+        return "payment.canceled"
+    return "payment.pending"
+
+
+def _signature_candidates(signature_header: str) -> set[str]:
+    raw = str(signature_header or "").strip()
+    if not raw:
+        return set()
+    candidates = {raw}
+    if raw.startswith("sha256="):
+        candidates.add(raw.split("=", 1)[1])
+    for part in raw.split(","):
+        key, sep, value = part.partition("=")
+        if sep and key.strip() in {"v1", "sha256"} and value.strip():
+            candidates.add(value.strip())
+    return candidates
+
+
+def verify_payment_webhook_signature(
+    *,
+    raw_body: bytes,
+    headers: dict[str, str],
+    secret: str | None,
+    replay_window_seconds: int = 600,
+    now: datetime | None = None,
+    required: bool = True,
+) -> str:
+    normalized_secret = str(secret or "").strip()
+    if not required and not normalized_secret:
+        return "not_required"
+    if not normalized_secret:
+        raise PaymentWebhookVerificationError("missing_secret", "Payment webhook secret is not configured", status_code=403)
+
+    timestamp = header_value(headers, "x-webhook-timestamp", "x-provider-timestamp", "x-timestamp")
+    if timestamp:
+        try:
+            event_ts = int(float(timestamp))
+        except ValueError as exc:
+            raise PaymentWebhookVerificationError("invalid_timestamp", "Invalid webhook timestamp") from exc
+        now_ts = int((now or datetime.now(timezone.utc)).timestamp())
+        if abs(now_ts - event_ts) > max(int(replay_window_seconds), 1):
+            raise PaymentWebhookVerificationError("expired", "Webhook timestamp is outside the replay window")
+
+    signature_header = header_value(headers, "x-webhook-signature", "x-signature", "stripe-signature")
+    if not signature_header:
+        raise PaymentWebhookVerificationError("missing", "Payment webhook signature is missing", status_code=403)
+
+    body_to_sign = raw_body
+    if timestamp:
+        body_to_sign = f"{timestamp}.".encode("utf-8") + raw_body
+    expected = hmac.new(normalized_secret.encode("utf-8"), body_to_sign, hashlib.sha256).hexdigest()
+    for candidate in _signature_candidates(signature_header):
+        if hmac.compare_digest(candidate, expected):
+            return "valid"
+    raise PaymentWebhookVerificationError("invalid", "Payment webhook signature is invalid", status_code=403)

services/sales_service/sales_events.py

@@ -9,8 +9,13 @@ DEAL_NEXT_ACTION_SCHEDULED = "deal.next_action_scheduled"
 DEAL_CLOSED = "deal.closed"
 DEAL_LOST = "deal.lost"
 DEAL_WON = "deal.won"
+DEAL_ESCALATION_REQUESTED = "deal.escalation_requested"
+DEAL_TRANSFERRED_POST_SALE = "deal.transferred_post_sale"
+DEAL_CHANNEL_SWITCH_RECOMMENDED = "deal.channel_switch_recommended"
+DEAL_FOLLOW_UP_READY = "deal.follow_up_ready"
 
 COMMUNICATION_STARTED = "communication.started"
+COMMUNICATION_CHANNEL_SWITCHED = "communication.channel_switched"
 COMMUNICATION_SUMMARY_CREATED = "communication.summary_created"
 MESSAGE_RECEIVED = "message.received"
 MESSAGE_SENT = "message.sent"
@@ -22,6 +27,8 @@ OFFER_SENT = "offer.sent"
 OFFER_ACCEPTED = "offer.accepted"
 OFFER_REJECTED = "offer.rejected"
 DEAL_CONDITIONS_CONFIRMED = "deal.conditions_confirmed"
+DEAL_NOTE_CREATED = "deal.note.created"
+DEAL_NOTE_UPDATED = "deal.note.updated"
 
 COUNTERPARTY_COMPLETED = "counterparty.completed"
 DOCUMENT_CREATED = "document.created"
@@ -32,9 +39,16 @@ DOCUMENT_SIGNED = "document.signed"
 INVOICE_CREATED = "invoice.created"
 INVOICE_SENT = "invoice.sent"
 INVOICE_OVERDUE = "invoice.overdue"
+INVOICE_REMINDER_SENT = "invoice.reminder_sent"
 PAYMENT_RECEIVED = "payment.received"
 INVOICE_PAID = "invoice.paid"
 
+ESCALATION_CREATED = "escalation.created"
+ESCALATION_ASSIGNED = "escalation.assigned"
+ESCALATION_STARTED = "escalation.started"
+ESCALATION_RESOLVED = "escalation.resolved"
+ESCALATION_CANCELED = "escalation.canceled"
+
 
 SALES_EVENT_TYPES = {
     LEAD_ENTERED_CRM,
@@ -45,7 +59,12 @@ SALES_EVENT_TYPES = {
     DEAL_CLOSED,
     DEAL_LOST,
     DEAL_WON,
+    DEAL_ESCALATION_REQUESTED,
+    DEAL_TRANSFERRED_POST_SALE,
+    DEAL_CHANNEL_SWITCH_RECOMMENDED,
+    DEAL_FOLLOW_UP_READY,
     COMMUNICATION_STARTED,
+    COMMUNICATION_CHANNEL_SWITCHED,
     COMMUNICATION_SUMMARY_CREATED,
     MESSAGE_RECEIVED,
     MESSAGE_SENT,
@@ -56,6 +75,8 @@ SALES_EVENT_TYPES = {
     OFFER_ACCEPTED,
     OFFER_REJECTED,
     DEAL_CONDITIONS_CONFIRMED,
+    DEAL_NOTE_CREATED,
+    DEAL_NOTE_UPDATED,
     COUNTERPARTY_COMPLETED,
     DOCUMENT_CREATED,
     DOCUMENT_SENT,
@@ -64,6 +85,12 @@ SALES_EVENT_TYPES = {
     INVOICE_CREATED,
     INVOICE_SENT,
     INVOICE_OVERDUE,
+    INVOICE_REMINDER_SENT,
     PAYMENT_RECEIVED,
     INVOICE_PAID,
+    ESCALATION_CREATED,
+    ESCALATION_ASSIGNED,
+    ESCALATION_STARTED,
+    ESCALATION_RESOLVED,
+    ESCALATION_CANCELED,
 }

services/shared/sales_models.py

@@ -38,9 +38,10 @@ SalesDocumentStatus = Literal["draft", "sent", "under_review", "confirmed", "sig
 SalesInvoiceStatus = Literal["draft", "issued", "sent", "partially_paid", "paid", "overdue", "canceled"]
 SalesPaymentStatus = Literal["pending", "success", "failed", "canceled", "partial"]
 SalesEscalationSeverity = Literal["low", "medium", "high", "critical"]
-SalesEscalationStatus = Literal["open", "in_progress", "resolved"]
+SalesEscalationStatus = Literal["open", "assigned", "in_progress", "resolved", "canceled"]
 SalesAutomationStatus = Literal["pending", "running", "completed", "failed", "canceled"]
 SalesPipelineStageCategory = Literal["entry", "communication", "commercial", "paperwork", "finance", "closing"]
+SalesSwitchChannel = Literal["text", "voice"]
 
 
 class SalesStageRefOut(BaseModel):
@@ -222,7 +223,11 @@ class SalesDealUpdate(BaseModel):
 class SalesDealStageChangeIn(BaseModel):
     stage_id: str | None = Field(default=None, min_length=2)
     stage_code: str | None = Field(default=None, min_length=2)
+    target_stage_id: str | None = Field(default=None, min_length=2)
+    target_stage_code: str | None = Field(default=None, min_length=2)
     reason: str | None = None
+    metadata: dict = Field(default_factory=dict)
+    force: bool = False
 
 
 class SalesDealScenarioIn(BaseModel):
@@ -295,8 +300,16 @@ class SalesCommunicationSummaryIn(BaseModel):
 
 
 class SalesCommunicationSwitchChannelIn(BaseModel):
-    to_channel: SalesPreferredChannel
-    reason_for_channel_switch: str = Field(min_length=3)
+    to_channel: SalesSwitchChannel
+    reason_code: str = "human_decision"
+    reason_text: str | None = None
+    reason_for_channel_switch: str | None = None
+    create_session: bool = False
+    scheduled_at: str | None = None
+    metadata: dict = Field(default_factory=dict)
+    source_type: str | None = None
+    source_id: str | None = None
+    recommended_by_task_id: str | None = None
 
 
 class SalesCommunicationBindExternalIn(BaseModel):
@@ -315,6 +328,7 @@ class SalesCommunicationOut(BaseModel):
     lead_id: str | None = None
     customer_id: str | None = None
     channel_type: SalesChannelType
+    channel_provider: str | None = None
     direction: SalesDirection
     agent_type: SalesAgentType
     started_at: str
@@ -580,7 +594,11 @@ class SalesPaymentWebhookIn(BaseModel):
     deal_id: str
     invoice_id: str | None = None
     payment_provider: str = "manual"
+    provider_account_id: str | None = None
+    external_event_id: str | None = None
     external_payment_id: str | None = None
+    event_type: str | None = None
+    provider_status: str | None = None
     amount: float = Field(ge=0)
     currency: str = "KZT"
     status: SalesPaymentStatus = "success"
@@ -618,6 +636,10 @@ class SalesEscalationCreateIn(BaseModel):
     reason: str = Field(min_length=3)
     severity: SalesEscalationSeverity = "medium"
     assigned_to_user_id: str | None = None
+    source_channel: str | None = None
+    source_communication_session_id: str | None = None
+    source_automation_task_id: str | None = None
+    sla_due_at: str | None = None
 
 
 class SalesEscalationOut(BaseModel):
@@ -628,8 +650,35 @@ class SalesEscalationOut(BaseModel):
     severity: SalesEscalationSeverity
     status: SalesEscalationStatus
     assigned_to_user_id: str | None = None
+    assigned_at: str | None = None
+    started_at: str | None = None
     created_at: str
     resolved_at: str | None = None
+    canceled_at: str | None = None
+    resolution_code: str | None = None
+    resolution_summary: str | None = None
+    sla_due_at: str | None = None
+    source_channel: str | None = None
+    source_communication_session_id: str | None = None
+    source_automation_task_id: str | None = None
+    updated_at: str | None = None
+
+
+class SalesEscalationAssignIn(BaseModel):
+    assigned_to_user_id: str = Field(min_length=1)
+
+
+class SalesEscalationResolveIn(BaseModel):
+    resolution_code: str = Field(min_length=2)
+    resolution_summary: str | None = None
+    target_stage_code: str | None = None
+    metadata: dict = Field(default_factory=dict)
+
+
+class SalesEscalationCancelIn(BaseModel):
+    resolution_code: str = "canceled"
+    resolution_summary: str | None = None
+    metadata: dict = Field(default_factory=dict)
 
 
 class SalesAutomationTaskOut(BaseModel):
@@ -640,6 +689,11 @@ class SalesAutomationTaskOut(BaseModel):
     run_at: str
     status: SalesAutomationStatus
     retry_count: int = 0
+    max_retries: int = 3
+    locked_at: str | None = None
+    locked_by: str | None = None
+    completed_at: str | None = None
+    failed_at: str | None = None
     last_error: str | None = None
     created_at: str
     updated_at: str
@@ -662,10 +716,50 @@ class SalesChannelSwitchOut(BaseModel):
     switch_id: str
     deal_id: str
     communication_id: str | None = None
+    communication_session_id: str | None = None
+    previous_communication_session_id: str | None = None
+    new_communication_session_id: str | None = None
     from_channel: str
     to_channel: str
+    reason_code: str
+    reason_text: str | None = None
     reason_for_channel_switch: str
+    initiated_by_type: str
+    initiated_by_id: str | None = None
+    source_type: str | None = None
+    source_id: str | None = None
+    recommended_by_task_id: str | None = None
     switched_at: str
+    created_at: str
+    new_communication: SalesCommunicationOut | None = None
+
+
+class SalesNoteCreateIn(BaseModel):
+    note_type: str = "general"
+    content: str = Field(min_length=1)
+    source_type: str | None = None
+    source_id: str | None = None
+
+
+class SalesNoteUpdateIn(BaseModel):
+    note_type: str | None = None
+    content: str | None = Field(default=None, min_length=1)
+    source_type: str | None = None
+    source_id: str | None = None
+
+
+class SalesNoteOut(BaseModel):
+    note_id: str
+    deal_id: str
+    author_type: str
+    author_id: str | None = None
+    note_type: str
+    content: str
+    source_type: str | None = None
+    source_id: str | None = None
+    created_at: str
+    updated_at: str | None = None
+    archived_at: str | None = None
 
 
 class SalesTimelineEventOut(BaseModel):
@@ -684,6 +778,8 @@ class SalesWorkspaceOut(BaseModel):
     communications: list[SalesCommunicationOut] = Field(default_factory=list)
     messages: list[SalesMessageOut] = Field(default_factory=list)
     calls: list[SalesCallOut] = Field(default_factory=list)
+    transcripts: list[SalesTranscriptOut] = Field(default_factory=list)
+    notes: list[SalesNoteOut] = Field(default_factory=list)
     offers: list[SalesOfferOut] = Field(default_factory=list)
     conditions: list[SalesConditionOut] = Field(default_factory=list)
     counterparty: SalesCounterpartyOut | None = None
@@ -691,9 +787,11 @@ class SalesWorkspaceOut(BaseModel):
     invoices: list[SalesInvoiceOut] = Field(default_factory=list)
     payments: list[SalesPaymentOut] = Field(default_factory=list)
     escalations: list[SalesEscalationOut] = Field(default_factory=list)
+    active_escalation: SalesEscalationOut | None = None
     tasks: list[SalesAutomationTaskOut] = Field(default_factory=list)
     stage_history: list[SalesStageHistoryOut] = Field(default_factory=list)
     channel_switches: list[SalesChannelSwitchOut] = Field(default_factory=list)
+    recommended_channel_switch: dict | None = None
     timeline: list[SalesTimelineEventOut] = Field(default_factory=list)
 
 

services/shared/sales_sql_models.py

@@ -178,6 +178,7 @@ class SalesCommunicationSessionRow(Base):
     lead_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
     customer_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
     channel_type: Mapped[str] = mapped_column(String(16), index=True)
+    channel_provider: Mapped[str | None] = mapped_column(String(32), nullable=True, index=True)
     direction: Mapped[str] = mapped_column(String(16), index=True)
     agent_type: Mapped[str] = mapped_column(String(32), index=True)
     started_at: Mapped[str] = mapped_column(String(64), index=True)
@@ -275,7 +276,11 @@ class SalesNoteRow(Base):
     author_id: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True)
     note_type: Mapped[str] = mapped_column(String(64), index=True)
     content: Mapped[str] = mapped_column(Text)
+    source_type: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    source_id: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True)
     created_at: Mapped[str] = mapped_column(String(64), index=True)
+    updated_at: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    archived_at: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
 
 
 class SalesOfferRow(Base):
@@ -403,6 +408,15 @@ class SalesPaymentRow(Base):
     __tablename__ = "sales_payments"
     __table_args__ = (
         Index("idx_sales_payments_deal_status", "deal_id", "status"),
+        Index(
+            "idx_sales_payments_tenant_external_payment_unique",
+            "tenant_id",
+            "payment_provider",
+            "external_payment_id",
+            unique=True,
+            sqlite_where=text("external_payment_id IS NOT NULL"),
+            postgresql_where=text("external_payment_id IS NOT NULL"),
+        ),
     )
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
@@ -423,6 +437,41 @@ class SalesPaymentRow(Base):
     updated_at: Mapped[str] = mapped_column(String(64), index=True)
 
 
+class SalesPaymentWebhookEventRow(Base):
+    __tablename__ = "sales_payment_webhook_events"
+    __table_args__ = (
+        Index("idx_sales_payment_webhook_events_received", "tenant_id", "received_at"),
+        Index("idx_sales_payment_webhook_events_payment", "tenant_id", "payment_provider", "external_payment_id"),
+        Index(
+            "idx_sales_payment_webhook_events_external_event_unique",
+            "tenant_id",
+            "payment_provider",
+            "external_event_id",
+            unique=True,
+            sqlite_where=text("external_event_id IS NOT NULL"),
+            postgresql_where=text("external_event_id IS NOT NULL"),
+        ),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    tenant_id: Mapped[str] = mapped_column(String(64), index=True)
+    payment_provider: Mapped[str] = mapped_column(String(64), index=True)
+    provider_account_id: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True)
+    external_event_id: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True)
+    external_payment_id: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True)
+    event_type: Mapped[str] = mapped_column(String(128), index=True)
+    event_status: Mapped[str] = mapped_column(String(32), index=True, default="received")
+    signature_status: Mapped[str] = mapped_column(String(32), index=True, default="unchecked")
+    raw_payload_hash: Mapped[str] = mapped_column(String(64), index=True)
+    normalized_payload_json: Mapped[str] = mapped_column(Text, default="{}")
+    headers_json: Mapped[str] = mapped_column(Text, default="{}")
+    received_at: Mapped[str] = mapped_column(String(64), index=True)
+    processed_at: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    error_message: Mapped[str | None] = mapped_column(Text, nullable=True)
+    created_at: Mapped[str] = mapped_column(String(64), index=True)
+    updated_at: Mapped[str] = mapped_column(String(64), index=True)
+
+
 class SalesStageHistoryRow(Base):
     __tablename__ = "sales_stage_history"
     __table_args__ = (
@@ -456,14 +505,25 @@ class SalesEscalationRow(Base):
     severity: Mapped[str] = mapped_column(String(16), index=True)
     status: Mapped[str] = mapped_column(String(32), index=True, default="open")
     assigned_to_user_id: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True)
+    assigned_at: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    started_at: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
     created_at: Mapped[str] = mapped_column(String(64), index=True)
     resolved_at: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    canceled_at: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    resolution_code: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    resolution_summary: Mapped[str | None] = mapped_column(Text, nullable=True)
+    sla_due_at: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    source_channel: Mapped[str | None] = mapped_column(String(32), nullable=True, index=True)
+    source_communication_session_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    source_automation_task_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    updated_at: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
 
 
 class SalesAutomationTaskRow(Base):
     __tablename__ = "sales_automation_tasks"
     __table_args__ = (
         Index("idx_sales_tasks_run_status", "run_at", "status"),
+        Index("idx_sales_tasks_lock", "status", "locked_at", "locked_by"),
     )
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
@@ -475,6 +535,11 @@ class SalesAutomationTaskRow(Base):
     run_at: Mapped[str] = mapped_column(String(64), index=True)
     status: Mapped[str] = mapped_column(String(32), index=True, default="pending")
     retry_count: Mapped[int] = mapped_column(Integer, default=0)
+    max_retries: Mapped[int] = mapped_column(Integer, default=3)
+    locked_at: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    locked_by: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True)
+    completed_at: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    failed_at: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
     last_error: Mapped[str | None] = mapped_column(Text, nullable=True)
     created_at: Mapped[str] = mapped_column(String(64), index=True)
     updated_at: Mapped[str] = mapped_column(String(64), index=True)
@@ -491,10 +556,21 @@ class SalesChannelSwitchRow(Base):
     tenant_id: Mapped[str] = mapped_column(String(64), index=True)
     deal_id: Mapped[str] = mapped_column(String(64), index=True)
     communication_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    communication_session_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    previous_communication_session_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    new_communication_session_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
     from_channel: Mapped[str] = mapped_column(String(32), index=True)
     to_channel: Mapped[str] = mapped_column(String(32), index=True)
+    reason_code: Mapped[str] = mapped_column(String(64), default="human_decision", index=True)
+    reason_text: Mapped[str | None] = mapped_column(Text, nullable=True)
     reason_for_channel_switch: Mapped[str] = mapped_column(Text)
+    initiated_by_type: Mapped[str] = mapped_column(String(32), default="human", index=True)
+    initiated_by_id: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True)
+    source_type: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
+    source_id: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True)
+    recommended_by_task_id: Mapped[str | None] = mapped_column(String(64), nullable=True, index=True)
     switched_at: Mapped[str] = mapped_column(String(64), index=True)
+    created_at: Mapped[str] = mapped_column(String(64), index=True)
 
 
 class SalesExternalLinkRow(Base):

services/shared/sql_init.py

@@ -601,6 +601,145 @@ def _apply_runtime_schema_compatibility() -> None:
                     )
                 )
 
+        if "sales_automation_tasks" in table_names:
+            columns = _table_columns(inspector, "sales_automation_tasks")
+            _add_column_if_missing(conn, columns, "sales_automation_tasks", "max_retries", "INTEGER DEFAULT 3")
+            _add_column_if_missing(conn, columns, "sales_automation_tasks", "locked_at", "VARCHAR(64)")
+            _add_column_if_missing(conn, columns, "sales_automation_tasks", "locked_by", "VARCHAR(128)")
+            _add_column_if_missing(conn, columns, "sales_automation_tasks", "completed_at", "VARCHAR(64)")
+            _add_column_if_missing(conn, columns, "sales_automation_tasks", "failed_at", "VARCHAR(64)")
+            conn.execute(
+                text(
+                    """
+                    UPDATE sales_automation_tasks
+                    SET max_retries = 3
+                    WHERE max_retries IS NULL
+                    """
+                )
+            )
+            indexes = _table_indexes(inspector, "sales_automation_tasks")
+            if "idx_sales_tasks_lock" not in indexes:
+                conn.execute(
+                    text(
+                        "CREATE INDEX IF NOT EXISTS idx_sales_tasks_lock "
+                        "ON sales_automation_tasks(status, locked_at, locked_by)"
+                    )
+                )
+
+        if "sales_communication_sessions" in table_names:
+            columns = _table_columns(inspector, "sales_communication_sessions")
+            _add_column_if_missing(conn, columns, "sales_communication_sessions", "channel_provider", "VARCHAR(32)")
+            indexes = _table_indexes(inspector, "sales_communication_sessions")
+            if "idx_sales_comm_channel_provider" not in indexes:
+                conn.execute(
+                    text(
+                        "CREATE INDEX IF NOT EXISTS idx_sales_comm_channel_provider "
+                        "ON sales_communication_sessions(channel_provider)"
+                    )
+                )
+
+        if "sales_notes" in table_names:
+            columns = _table_columns(inspector, "sales_notes")
+            _add_column_if_missing(conn, columns, "sales_notes", "source_type", "VARCHAR(64)")
+            _add_column_if_missing(conn, columns, "sales_notes", "source_id", "VARCHAR(128)")
+            _add_column_if_missing(conn, columns, "sales_notes", "updated_at", "VARCHAR(64)")
+            _add_column_if_missing(conn, columns, "sales_notes", "archived_at", "VARCHAR(64)")
+            conn.execute(
+                text(
+                    """
+                    UPDATE sales_notes
+                    SET updated_at = created_at
+                    WHERE updated_at IS NULL OR TRIM(updated_at) = ''
+                    """
+                )
+            )
+
+        if "sales_escalations" in table_names:
+            columns = _table_columns(inspector, "sales_escalations")
+            _add_column_if_missing(conn, columns, "sales_escalations", "assigned_at", "VARCHAR(64)")
+            _add_column_if_missing(conn, columns, "sales_escalations", "started_at", "VARCHAR(64)")
+            _add_column_if_missing(conn, columns, "sales_escalations", "canceled_at", "VARCHAR(64)")
+            _add_column_if_missing(conn, columns, "sales_escalations", "resolution_code", "VARCHAR(64)")
+            _add_column_if_missing(conn, columns, "sales_escalations", "resolution_summary", "TEXT")
+            _add_column_if_missing(conn, columns, "sales_escalations", "sla_due_at", "VARCHAR(64)")
+            _add_column_if_missing(conn, columns, "sales_escalations", "source_channel", "VARCHAR(32)")
+            _add_column_if_missing(conn, columns, "sales_escalations", "source_communication_session_id", "VARCHAR(64)")
+            _add_column_if_missing(conn, columns, "sales_escalations", "source_automation_task_id", "VARCHAR(64)")
+            _add_column_if_missing(conn, columns, "sales_escalations", "updated_at", "VARCHAR(64)")
+            conn.execute(
+                text(
+                    """
+                    UPDATE sales_escalations
+                    SET updated_at = COALESCE(updated_at, resolved_at, created_at)
+                    WHERE updated_at IS NULL OR TRIM(updated_at) = ''
+                    """
+                )
+            )
+
+        if "sales_channel_switches" in table_names:
+            columns = _table_columns(inspector, "sales_channel_switches")
+            _add_column_if_missing(conn, columns, "sales_channel_switches", "communication_session_id", "VARCHAR(64)")
+            _add_column_if_missing(conn, columns, "sales_channel_switches", "previous_communication_session_id", "VARCHAR(64)")
+            _add_column_if_missing(conn, columns, "sales_channel_switches", "new_communication_session_id", "VARCHAR(64)")
+            _add_column_if_missing(conn, columns, "sales_channel_switches", "reason_code", "VARCHAR(64) DEFAULT 'human_decision'")
+            _add_column_if_missing(conn, columns, "sales_channel_switches", "reason_text", "TEXT")
+            _add_column_if_missing(conn, columns, "sales_channel_switches", "initiated_by_type", "VARCHAR(32) DEFAULT 'human'")
+            _add_column_if_missing(conn, columns, "sales_channel_switches", "initiated_by_id", "VARCHAR(128)")
+            _add_column_if_missing(conn, columns, "sales_channel_switches", "source_type", "VARCHAR(64)")
+            _add_column_if_missing(conn, columns, "sales_channel_switches", "source_id", "VARCHAR(128)")
+            _add_column_if_missing(conn, columns, "sales_channel_switches", "recommended_by_task_id", "VARCHAR(64)")
+            _add_column_if_missing(conn, columns, "sales_channel_switches", "created_at", "VARCHAR(64)")
+            conn.execute(
+                text(
+                    """
+                    UPDATE sales_channel_switches
+                    SET communication_session_id = COALESCE(communication_session_id, communication_id),
+                        previous_communication_session_id = COALESCE(previous_communication_session_id, communication_id),
+                        reason_code = COALESCE(reason_code, 'human_decision'),
+                        reason_text = COALESCE(reason_text, reason_for_channel_switch),
+                        initiated_by_type = COALESCE(initiated_by_type, 'human'),
+                        created_at = COALESCE(created_at, switched_at)
+                    WHERE created_at IS NULL OR TRIM(created_at) = ''
+                    """
+                )
+            )
+
+        if "sales_payment_webhook_events" in table_names:
+            indexes = _table_indexes(inspector, "sales_payment_webhook_events")
+            if "idx_sales_payment_webhook_events_received" not in indexes:
+                conn.execute(
+                    text(
+                        "CREATE INDEX IF NOT EXISTS idx_sales_payment_webhook_events_received "
+                        "ON sales_payment_webhook_events(tenant_id, received_at)"
+                    )
+                )
+            if "idx_sales_payment_webhook_events_payment" not in indexes:
+                conn.execute(
+                    text(
+                        "CREATE INDEX IF NOT EXISTS idx_sales_payment_webhook_events_payment "
+                        "ON sales_payment_webhook_events(tenant_id, payment_provider, external_payment_id)"
+                    )
+                )
+            if "idx_sales_payment_webhook_events_external_event_unique" not in indexes:
+                conn.execute(
+                    text(
+                        "CREATE UNIQUE INDEX IF NOT EXISTS idx_sales_payment_webhook_events_external_event_unique "
+                        "ON sales_payment_webhook_events(tenant_id, payment_provider, external_event_id) "
+                        "WHERE external_event_id IS NOT NULL"
+                    )
+                )
+
+        if "sales_payments" in table_names:
+            indexes = _table_indexes(inspector, "sales_payments")
+            if "idx_sales_payments_tenant_external_payment_unique" not in indexes:
+                conn.execute(
+                    text(
+                        "CREATE UNIQUE INDEX IF NOT EXISTS idx_sales_payments_tenant_external_payment_unique "
+                        "ON sales_payments(tenant_id, payment_provider, external_payment_id) "
+                        "WHERE external_payment_id IS NOT NULL"
+                    )
+                )
+
 
 def init_sql_schema() -> None:
     global _BASE_SCHEMA_INITIALIZED