feat(customer): 网站客户身份签发与长期负责人绑定

可信客户身份由站点后端签发，长期负责人与本次接待人分开保存。

backend/app/api/crm.py

@@ -0,0 +1,308 @@
+"""
+Copyright 2024-2026 ReplyPier
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+import base64
+import hashlib
+import hmac
+import secrets
+import time
+from typing import List
+from urllib.parse import urlencode
+from uuid import UUID
+
+from fastapi import APIRouter, Depends, HTTPException, Request
+from fastapi.responses import RedirectResponse
+from sqlalchemy.orm import Session
+
+from app.core.auth import get_current_organization, require_permissions
+from app.core.config import settings
+from app.core.logger import get_logger
+from app.core.oauth_state import pop_state, store_state
+from app.crm.base import CrmAuthError, CrmTransientError, basic_auth_header
+from app.crm.registry import get_adapter
+from app.database import get_db
+from app.models.crm import CrmConnectionStatus, CrmProvider
+from app.models.organization import Organization
+from app.models.schemas.crm import CrmConnectionOut, CrmTestResult
+from app.models.user import User
+from app.repositories.crm import CrmConnectionRepository, CrmSyncJobRepository
+from app.services.crm_tokens import get_valid_tokens, tokens_from_connection
+from app.services.feature_gate import check_feature_access
+
+router = APIRouter()
+logger = get_logger(__name__)
+
+UPGRADE_MESSAGE = "CRM sync is not available in your current plan."
+# HubSpot signs webhooks with a timestamp; reject anything older than this.
+WEBHOOK_TIMESTAMP_MAX_AGE_SECONDS = 300
+
+def _provider_credentials(provider: str) -> tuple:
+    return {
+        CrmProvider.HUBSPOT.value: (settings.HUBSPOT_CLIENT_ID, settings.HUBSPOT_CLIENT_SECRET),
+        CrmProvider.PIPEDRIVE.value: (settings.PIPEDRIVE_CLIENT_ID, settings.PIPEDRIVE_CLIENT_SECRET),
+    }.get(provider, ("", ""))
+
+
+def _require_provider(provider: str) -> None:
+    if get_adapter(provider) is None:
+        raise HTTPException(status_code=404, detail="Unknown CRM provider")
+
+
+def _store_state(state: str, org_id: str, provider: str) -> None:
+    store_state("crm", state, f"{org_id}:{provider}")
+
+
+def _pop_state(state: str) -> tuple:
+    value = pop_state("crm", state)
+    if not value or ":" not in value:
+        return None, None
+    org_id, provider = value.split(":", 1)
+    return org_id, provider
+
+
+def _redirect_uri(provider: str) -> str:
+    return f"{settings.BACKEND_URL.rstrip('/')}{settings.API_V1_STR}/crm/{provider}/callback"
+
+
+def _settings_redirect(provider: str, status: str, reason: str = None) -> RedirectResponse:
+    params = {"status": status, "integration": provider}
+    if reason:
+        params["reason"] = reason
+    return RedirectResponse(f"{settings.FRONTEND_URL}/settings/integrations?{urlencode(params)}")
+
+
+def _get_org_connection_or_404(db: Session, organization: Organization, provider: str):
+    connection = CrmConnectionRepository(db).get_by_org_provider(organization.id, provider)
+    if connection is None:
+        raise HTTPException(status_code=404, detail="CRM is not connected")
+    return connection
+
+
+@router.get("/connections", response_model=List[CrmConnectionOut])
+async def list_connections(
+    current_user: User = Depends(require_permissions("manage_organization")),
+    organization: Organization = Depends(get_current_organization),
+    db: Session = Depends(get_db),
+):
+    """Connection status per provider for the integrations page, with a
+    failed-push count so a quietly broken connection is visible."""
+    check_feature_access(db, organization.id, "crm_sync", UPGRADE_MESSAGE)
+    job_repo = CrmSyncJobRepository(db)
+    out = []
+    for connection in CrmConnectionRepository(db).list_by_org(organization.id):
+        item = CrmConnectionOut.model_validate(connection)
+        item.recent_failures = job_repo.count_recent_failures(
+            organization.id, connection.provider)
+        out.append(item)
+    return out
+
+
+@router.get("/{provider}/install")
+async def install_crm(
+    provider: str,
+    current_user: User = Depends(require_permissions("manage_organization")),
+    organization: Organization = Depends(get_current_organization),
+    db: Session = Depends(get_db),
+):
+    """Start the OAuth install: redirect the browser to the CRM's consent page
+    with a CSRF state bound to (organization, provider)."""
+    _require_provider(provider)
+    check_feature_access(db, organization.id, "crm_sync", UPGRADE_MESSAGE)
+    client_id, client_secret = _provider_credentials(provider)
+    if not client_id or not client_secret:
+        raise HTTPException(status_code=400,
+                            detail=f"{provider} app credentials are not configured")
+    state = secrets.token_urlsafe(24)
+    _store_state(state, str(organization.id), provider)
+    adapter = get_adapter(provider)
+    return RedirectResponse(adapter.get_authorize_url(state, _redirect_uri(provider)))
+
+
+@router.get("/{provider}/callback")
+async def crm_oauth_callback(
+    provider: str,
+    code: str = None,
+    state: str = None,
+    error: str = None,
+    db: Session = Depends(get_db),
+):
+    """Complete the OAuth install: exchange the code, upsert the connection,
+    and bounce back to the settings page."""
+    _require_provider(provider)
+    if error:
+        return _settings_redirect(
+            provider, "failure", "cancelled" if error == "access_denied" else error)
+    org_id, state_provider = _pop_state(state or "")
+    if org_id is None or state_provider != provider or not code:
+        return _settings_redirect(provider, "failure", "invalid_state")
+
+    adapter = get_adapter(provider)
+    try:
+        tokens = await adapter.exchange_code(code, _redirect_uri(provider))
+    except (CrmAuthError, CrmTransientError) as e:
+        logger.error(f"{provider} OAuth exchange failed: {e}")
+        return _settings_redirect(provider, "failure", "oauth_failed")
+
+    repo = CrmConnectionRepository(db)
+    existing = repo.get_by_external_id(provider, tokens.external_account_id)
+    if existing is not None and str(existing.organization_id) != org_id:
+        return _settings_redirect(provider, "failure", "account_connected_elsewhere")
+
+    repo.create_or_update(
+        organization_id=UUID(org_id),
+        provider=provider,
+        external_account_id=tokens.external_account_id,
+        credentials={
+            "access_token": tokens.access_token,
+            "refresh_token": tokens.refresh_token,
+            "api_domain": tokens.api_domain,
+        },
+        display_name=tokens.display_name,
+        access_token_expires_at=tokens.expires_at,
+        refresh_token_expires_at=tokens.refresh_token_expires_at,
+    )
+    # A reconnect is the fix for a dead connection — requeue anything that
+    # terminally failed while it was down so those leads reach the CRM.
+    CrmSyncJobRepository(db).reopen_failed(UUID(org_id), provider)
+    logger.info(f"Connected {provider} ({tokens.display_name}) for org {org_id}")
+    return _settings_redirect(provider, "success")
+
+
+@router.post("/{provider}/test", response_model=CrmTestResult)
+async def test_crm_connection(
+    provider: str,
+    current_user: User = Depends(require_permissions("manage_organization")),
+    organization: Organization = Depends(get_current_organization),
+    db: Session = Depends(get_db),
+):
+    """Read-only connectivity check — never writes records into the CRM."""
+    _require_provider(provider)
+    check_feature_access(db, organization.id, "crm_sync", UPGRADE_MESSAGE)
+    connection = _get_org_connection_or_404(db, organization, provider)
+    try:
+        tokens = await get_valid_tokens(db, connection)
+    except CrmAuthError:
+        return CrmTestResult(ok=False, error="Connection expired — please reconnect")
+    except CrmTransientError as e:
+        return CrmTestResult(ok=False, error=str(e))
+    return CrmTestResult(**(await get_adapter(provider).test_connection(tokens)))
+
+
+@router.delete("/{provider}")
+async def disconnect_crm(
+    provider: str,
+    current_user: User = Depends(require_permissions("manage_organization")),
+    organization: Organization = Depends(get_current_organization),
+    db: Session = Depends(get_db),
+):
+    """Disconnect: best-effort token revocation, drop the row, cancel queued
+    pushes. Agents' crm_sync_target stays set — it's simply inert until a
+    reconnect, and the UI warns about the mismatch."""
+    _require_provider(provider)
+    connection = _get_org_connection_or_404(db, organization, provider)
+    repo = CrmConnectionRepository(db)
+    try:
+        await get_adapter(provider).revoke(tokens_from_connection(repo, connection))
+    except Exception as e:
+        logger.warning(f"{provider} revoke on disconnect failed (ignored): {e}")
+    CrmSyncJobRepository(db).skip_pending_for_connection(
+        organization.id, provider, "disconnected")
+    repo.delete(connection)
+    return {"status": "disconnected"}
+
+
+# ---- uninstall webhooks (marketplace requirement) --------------------------
+
+
+@router.post("/hubspot/uninstall")
+async def hubspot_uninstall_webhook(request: Request, db: Session = Depends(get_db)):
+    """HubSpot app-uninstall events, authenticated by the v3 signature
+    (HMAC-SHA256 over method+url+body+timestamp with the client secret)."""
+    body = await request.body()
+    if not _valid_hubspot_signature(request, body):
+        raise HTTPException(status_code=404)
+    try:
+        events = (await request.json()) or []
+    except Exception:
+        events = []
+    for event in events if isinstance(events, list) else [events]:
+        portal_id = str(event.get("portalId", "") or event.get("hub_id", ""))
+        event_kind = f"{event.get('subscriptionType', '')}{event.get('eventType', '')}".lower()
+        if portal_id and "uninstall" in event_kind:
+            _revoke_connection(db, CrmProvider.HUBSPOT.value, portal_id)
+    return {"ok": True}
+
+
+@router.post("/pipedrive/uninstall")
+async def pipedrive_uninstall_webhook(request: Request, db: Session = Depends(get_db)):
+    """Pipedrive app_uninstalled callback, authenticated with HTTP Basic —
+    the Developer Hub webhook is configured with our client_id:client_secret."""
+    if not _valid_pipedrive_basic_auth(request.headers.get("Authorization", "")):
+        raise HTTPException(status_code=404)
+    try:
+        payload = await request.json()
+    except Exception:
+        payload = {}
+    company_id = str(payload.get("company_id", ""))
+    if company_id:
+        _revoke_connection(db, CrmProvider.PIPEDRIVE.value, company_id)
+    return {"ok": True}
+
+
+def _revoke_connection(db: Session, provider: str, external_account_id: str) -> None:
+    repo = CrmConnectionRepository(db)
+    connection = repo.get_by_external_id(provider, external_account_id)
+    if connection is None:
+        return
+    repo.set_status(connection, CrmConnectionStatus.REVOKED.value,
+                    last_error="App uninstalled from the CRM")
+    CrmSyncJobRepository(db).skip_pending_for_connection(
+        connection.organization_id, provider, "app uninstalled")
+    logger.info(f"{provider} app uninstalled for org {connection.organization_id}")
+
+
+def _valid_hubspot_signature(request: Request, body: bytes) -> bool:
+    """HubSpot v3: base64(HMAC-SHA256(secret, METHOD + full_url + body + ts)),
+    with a freshness window against replay."""
+    signature = request.headers.get("X-HubSpot-Signature-v3", "")
+    timestamp = request.headers.get("X-HubSpot-Request-Timestamp", "")
+    if not signature or not timestamp or not settings.HUBSPOT_CLIENT_SECRET:
+        return False
+    try:
+        age_ms = time.time() * 1000 - int(timestamp)
+    except ValueError:
+        return False
+    if age_ms > WEBHOOK_TIMESTAMP_MAX_AGE_SECONDS * 1000:
+        return False
+    # HubSpot signs the public https URL; behind the reverse proxy the
+    # reconstructed URL may say http — accept either scheme, nothing else.
+    url = str(request.url)
+    candidate_urls = {url, url.replace("http://", "https://", 1)}
+    for candidate in candidate_urls:
+        message = f"{request.method}{candidate}".encode() + body + timestamp.encode()
+        expected = base64.b64encode(
+            hmac.new(settings.HUBSPOT_CLIENT_SECRET.encode(), message, hashlib.sha256).digest()
+        ).decode()
+        if hmac.compare_digest(expected, signature):
+            return True
+    return False
+
+
+def _valid_pipedrive_basic_auth(header: str) -> bool:
+    client_id, client_secret = _provider_credentials(CrmProvider.PIPEDRIVE.value)
+    if not client_id or not client_secret or not header.startswith("Basic "):
+        return False
+    return hmac.compare_digest(header[6:], basic_auth_header(client_id, client_secret))

backend/app/api/lead_capture.py

@@ -0,0 +1,113 @@
+"""
+Copyright 2024-2026 ReplyPier
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+from fastapi import APIRouter, Depends, HTTPException, status
+from sqlalchemy.orm import Session
+from uuid import UUID
+
+from app.core.logger import get_logger
+from app.database import get_db
+from app.core.auth import require_any_permission, require_permissions
+from app.models.user import User
+from app.models.agent import Agent
+from app.repositories.lead_capture import LeadCaptureConfigRepository
+from app.models.schemas.lead_capture import (
+    LeadCaptureConfigUpdate, LeadCaptureConfigResponse,
+)
+
+# Enterprise gating — Lead Capture is a Pro-plan feature where the enterprise
+# module is installed; OSS/community deployments are unrestricted (same shape as
+# the MCP/Workflow/Advanced gates, which are all Pro+ only).
+try:
+    from app.enterprise.repositories.plan import PlanRepository
+    from app.enterprise.services.feature_access import require_accessible_subscription
+    HAS_ENTERPRISE = True
+except ImportError:
+    HAS_ENTERPRISE = False
+
+router = APIRouter()
+logger = get_logger(__name__)
+
+
+def check_lead_capture_access(current_user: User, db: Session) -> None:
+    """Gate Lead Capture behind the Pro plan when the enterprise module is present.
+    Requires an accessible subscription whose plan has the 'lead_capture' feature
+    (Pro/Enterprise only — Free and Base do not). OSS deployments are unrestricted."""
+    if not HAS_ENTERPRISE:
+        return
+    # Raises 403 when the org has no accessible plan.
+    subscription = require_accessible_subscription(db, current_user.organization_id)
+    if not PlanRepository(db).check_feature_availability(str(subscription.plan_id), 'lead_capture'):
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="Lead Capture is not available in your current plan. Please upgrade to Pro to access this feature.",
+        )
+
+
+def _get_owned_agent(agent_id: UUID, current_user: User, db: Session) -> Agent:
+    agent = db.query(Agent).filter(Agent.id == agent_id).first()
+    # Return 404 for both missing and cross-org agents (don't reveal existence).
+    if not agent or agent.organization_id != current_user.organization_id:
+        raise HTTPException(status_code=404, detail="Agent not found")
+    return agent
+
+
+@router.get("/{agent_id}/lead-capture", response_model=LeadCaptureConfigResponse)
+async def get_lead_capture_config(
+    agent_id: UUID,
+    current_user: User = Depends(
+        require_any_permission("view_agents", "manage_agents")),
+    db: Session = Depends(get_db),
+):
+    """Fetch the agent's lead-capture config, lazily creating a default (OFF) row."""
+    try:
+        _get_owned_agent(agent_id, current_user, db)
+        config = LeadCaptureConfigRepository(db).get_or_create(agent_id)
+        return config
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Error fetching lead capture config: {str(e)}")
+        raise HTTPException(status_code=500, detail="Failed to fetch lead capture config")
+
+
+@router.put("/{agent_id}/lead-capture", response_model=LeadCaptureConfigResponse)
+async def update_lead_capture_config(
+    agent_id: UUID,
+    data: LeadCaptureConfigUpdate,
+    current_user: User = Depends(require_permissions("manage_agents")),
+    db: Session = Depends(get_db),
+):
+    """Full-replace the agent's lead-capture config (whole tab saved at once)."""
+    try:
+        _get_owned_agent(agent_id, current_user, db)
+        check_lead_capture_access(current_user, db)
+        # Assignment target (stored-only in phase 1) must belong to this org — reject
+        # cross-org user ids so the routing FK can never dangle when routing ships.
+        if data.assignment_target_user_id is not None:
+            target = db.query(User).filter(
+                User.id == data.assignment_target_user_id,
+                User.organization_id == current_user.organization_id,
+            ).first()
+            if not target:
+                raise HTTPException(status_code=400, detail="Assignment target user not found in this organization")
+        config = LeadCaptureConfigRepository(db).update(agent_id, data)
+        return config
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Error updating lead capture config: {str(e)}")
+        raise HTTPException(status_code=500, detail="Failed to update lead capture config")

backend/app/crm/__init__.py

@@ -0,0 +1,27 @@
+"""
+Copyright 2024-2026 ReplyPier
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+from app.crm.base import CrmAdapter, OAuthTokens, LeadPayload, CrmPushResult
+from app.crm.registry import get_adapter, register_adapter
+
+__all__ = [
+    "CrmAdapter",
+    "OAuthTokens",
+    "LeadPayload",
+    "CrmPushResult",
+    "get_adapter",
+    "register_adapter",
+]

backend/app/crm/base.py

@@ -0,0 +1,163 @@
+"""
+Copyright 2024-2026 ReplyPier
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+import base64
+import re
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import ClassVar, Optional
+from uuid import UUID
+
+import httpx
+
+from app.core.logger import get_logger
+
+logger = get_logger(__name__)
+
+REQUEST_TIMEOUT_SECONDS = 20
+
+
+def basic_auth_header(client_id: str, client_secret: str) -> str:
+    """base64(client_id:client_secret) for HTTP Basic — Pipedrive OAuth + webhook."""
+    return base64.b64encode(f"{client_id}:{client_secret}".encode()).decode()
+
+
+def summarize_provider_error(text: Optional[str]) -> str:
+    """A short, single-line, tag-stripped snippet of a provider error body for a
+    user-facing message. The full body is logged separately — never echo an
+    untrusted third-party string verbatim into an admin surface."""
+    stripped = re.sub(r"<[^>]+>", " ", text or "")
+    return " ".join(stripped.split())[:200]
+
+_http_client: Optional[httpx.AsyncClient] = None
+
+
+def get_http_client() -> httpx.AsyncClient:
+    """Shared AsyncClient for all CRM adapters (tests swap in a MockTransport)."""
+    global _http_client
+    if _http_client is None or _http_client.is_closed:
+        _http_client = httpx.AsyncClient(timeout=REQUEST_TIMEOUT_SECONDS)
+    return _http_client
+
+
+class CrmAuthError(Exception):
+    """The provider definitively rejected our grant (invalid_grant/revoked) —
+    the connection needs a human reconnect, retrying cannot help."""
+
+
+class CrmTransientError(Exception):
+    """Temporary provider failure (5xx/network/429) — safe to retry later."""
+
+
+@dataclass
+class OAuthTokens:
+    """Decrypted per-tenant OAuth state, passed into every adapter call.
+
+    `api_domain` is the per-tenant API host where the provider has one
+    (Pipedrive api_domain; later Salesforce instance_url / Zoho DC host).
+    """
+    access_token: str
+    refresh_token: str
+    expires_at: Optional[datetime] = None
+    api_domain: Optional[str] = None
+    refresh_token_expires_at: Optional[datetime] = None
+    external_account_id: str = ""
+    display_name: Optional[str] = None
+    raw: dict = field(default_factory=dict)
+
+
+@dataclass
+class LeadPayload:
+    """Provider-agnostic view of one captured lead (built by crm.mapping)."""
+    lead_response_id: UUID
+    email: str
+    name: Optional[str] = None
+    company: Optional[str] = None
+    phone: Optional[str] = None
+    summary: Optional[str] = None
+    # Human label -> value, from the agent's configured custom fields.
+    custom_fields: dict = field(default_factory=dict)
+    source_url: Optional[str] = None
+
+
+@dataclass
+class CrmPushResult:
+    ok: bool
+    action: Optional[str] = None          # 'created' | 'updated'
+    contact_id: Optional[str] = None      # HubSpot contact / Pipedrive person
+    secondary_id: Optional[str] = None    # Pipedrive lead id
+    record_url: Optional[str] = None      # deep link for the UI
+    error: Optional[str] = None
+    retryable: bool = False
+    retry_after_seconds: Optional[int] = None
+    auth_failed: bool = False
+
+
+def classify_error_response(response: httpx.Response) -> CrmPushResult:
+    """Map a failed provider response onto retry semantics.
+
+    401 → dead token (refresh-and-retry, else reconnect); 429 → back off,
+    honoring Retry-After; 5xx → transient; other 4xx → permanent (payload/
+    scope problems a retry can't fix).
+    """
+    logger.warning(f"CRM provider HTTP {response.status_code}: {response.text[:1000]}")
+    error = f"HTTP {response.status_code}: {summarize_provider_error(response.text)}"
+    if response.status_code == 401:
+        return CrmPushResult(ok=False, error=error, auth_failed=True)
+    if response.status_code == 429:
+        retry_after = response.headers.get("Retry-After")
+        seconds = int(retry_after) if retry_after and retry_after.isdigit() else None
+        return CrmPushResult(ok=False, error=error, retryable=True,
+                             retry_after_seconds=seconds)
+    if response.status_code >= 500:
+        return CrmPushResult(ok=False, error=error, retryable=True)
+    return CrmPushResult(ok=False, error=error)
+
+
+class CrmAdapter(ABC):
+    """One stateless singleton per CRM. Tokens are passed into every call —
+    the adapter holds no per-tenant state (mirrors ChannelAdapter)."""
+
+    provider: ClassVar[str]
+
+    @abstractmethod
+    def get_authorize_url(self, state: str, redirect_uri: str) -> str:
+        """Provider consent-screen URL for the OAuth install redirect."""
+
+    @abstractmethod
+    async def exchange_code(self, code: str, redirect_uri: str) -> OAuthTokens:
+        """Swap the callback code for tokens + account identity.
+        Raises CrmAuthError on a rejected code, CrmTransientError on 5xx."""
+
+    @abstractmethod
+    async def refresh_tokens(self, tokens: OAuthTokens) -> OAuthTokens:
+        """Mint a fresh access token. Raises CrmAuthError on invalid_grant
+        (connection dead), CrmTransientError on temporary failure."""
+
+    @abstractmethod
+    async def push_lead(self, tokens: OAuthTokens, payload: LeadPayload) -> CrmPushResult:
+        """Create-or-update the lead in the CRM, deduping by email.
+        Never raises — failures come back classified in CrmPushResult."""
+
+    @abstractmethod
+    async def revoke(self, tokens: OAuthTokens) -> None:
+        """Best-effort token revocation on disconnect. Never raises."""
+
+    @abstractmethod
+    async def test_connection(self, tokens: OAuthTokens) -> dict:
+        """Cheap read-only authenticated call.
+        Returns {"ok": bool, "account_name": str | None, "error": str | None}."""

backend/app/crm/hubspot.py

@@ -0,0 +1,210 @@
+"""
+Copyright 2024-2026 ReplyPier
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+from datetime import datetime, timedelta, timezone
+from urllib.parse import urlencode
+
+import httpx
+
+from app.core.config import settings
+from app.core.logger import get_logger
+from app.crm.base import (
+    CrmAdapter, CrmAuthError, CrmPushResult, CrmTransientError,
+    LeadPayload, OAuthTokens, classify_error_response, get_http_client,
+    summarize_provider_error,
+)
+from app.crm.mapping import build_note_body, split_name
+from app.crm.registry import register_adapter
+from app.models.crm import CrmProvider
+from app.core.external_urls import external_base
+
+logger = get_logger(__name__)
+
+AUTHORIZE_URL = "https://app.hubspot.com/oauth/authorize"
+API_BASE = external_base("HUBSPOT_API_BASE", "https://api.hubapi.com")
+TOKEN_URL = f"{API_BASE}/oauth/v1/token"
+# Request only what the push needs — the marketplace review checks scope usage.
+SCOPES = "crm.objects.contacts.read crm.objects.contacts.write"
+# HubSpot-defined association type: note -> contact.
+NOTE_TO_CONTACT_ASSOCIATION_TYPE_ID = 202
+
+
+class HubSpotAdapter(CrmAdapter):
+    """Pushes a lead as a Contact (email upsert) plus a context Note.
+
+    HubSpot's separate "Leads" object is deliberately not used: it requires a
+    paid-seat owner and an existing contact, which doesn't fit lead capture.
+    """
+
+    provider = CrmProvider.HUBSPOT.value
+
+    def get_authorize_url(self, state: str, redirect_uri: str) -> str:
+        return f"{AUTHORIZE_URL}?" + urlencode({
+            "client_id": settings.HUBSPOT_CLIENT_ID,
+            "redirect_uri": redirect_uri,
+            "scope": SCOPES,
+            "state": state,
+        })
+
+    async def exchange_code(self, code: str, redirect_uri: str) -> OAuthTokens:
+        tokens = await self._token_request({
+            "grant_type": "authorization_code",
+            "client_id": settings.HUBSPOT_CLIENT_ID,
+            "client_secret": settings.HUBSPOT_CLIENT_SECRET,
+            "redirect_uri": redirect_uri,
+            "code": code,
+        })
+        # Token introspection carries the portal identity (hub_id/domain).
+        response = await get_http_client().get(
+            f"{API_BASE}/oauth/v1/access-tokens/{tokens.access_token}")
+        if response.status_code == 200:
+            info = response.json()
+            tokens.external_account_id = str(info.get("hub_id", ""))
+            tokens.display_name = info.get("hub_domain")
+        return tokens
+
+    async def refresh_tokens(self, tokens: OAuthTokens) -> OAuthTokens:
+        refreshed = await self._token_request({
+            "grant_type": "refresh_token",
+            "client_id": settings.HUBSPOT_CLIENT_ID,
+            "client_secret": settings.HUBSPOT_CLIENT_SECRET,
+            "refresh_token": tokens.refresh_token,
+        })
+        refreshed.external_account_id = tokens.external_account_id
+        refreshed.display_name = tokens.display_name
+        return refreshed
+
+    async def push_lead(self, tokens: OAuthTokens, payload: LeadPayload) -> CrmPushResult:
+        properties = self._contact_properties(payload)
+        try:
+            response = await get_http_client().post(
+                f"{API_BASE}/crm/v3/objects/contacts/batch/upsert",
+                headers=self._auth_header(tokens),
+                json={"inputs": [{
+                    "id": payload.email,
+                    "idProperty": "email",
+                    "properties": properties,
+                }]},
+            )
+        except httpx.HTTPError as e:
+            return CrmPushResult(ok=False, error=f"network error: {e}", retryable=True)
+
+        if response.status_code not in (200, 201):
+            return classify_error_response(response)
+
+        results = response.json().get("results") or [{}]
+        contact_id = str(results[0].get("id", ""))
+        action = "created" if results[0].get("new") else "updated"
+
+        await self._attach_note(tokens, contact_id, payload)
+
+        record_url = None
+        if tokens.external_account_id and contact_id:
+            record_url = (f"https://app.hubspot.com/contacts/"
+                          f"{tokens.external_account_id}/record/0-1/{contact_id}")
+        return CrmPushResult(ok=True, action=action, contact_id=contact_id,
+                             record_url=record_url)
+
+    async def revoke(self, tokens: OAuthTokens) -> None:
+        try:
+            await get_http_client().delete(
+                f"{API_BASE}/oauth/v1/refresh-tokens/{tokens.refresh_token}")
+        except Exception as e:
+            logger.warning(f"HubSpot token revocation failed (ignored): {e}")
+
+    async def test_connection(self, tokens: OAuthTokens) -> dict:
+        try:
+            response = await get_http_client().get(
+                f"{API_BASE}/oauth/v1/access-tokens/{tokens.access_token}")
+        except httpx.HTTPError as e:
+            return {"ok": False, "account_name": None, "error": f"network error: {e}"}
+        if response.status_code != 200:
+            return {"ok": False, "account_name": None,
+                    "error": f"HTTP {response.status_code}"}
+        return {"ok": True, "account_name": response.json().get("hub_domain"),
+                "error": None}
+
+    async def _token_request(self, data: dict) -> OAuthTokens:
+        try:
+            response = await get_http_client().post(
+                TOKEN_URL, data=data,
+                headers={"Content-Type": "application/x-www-form-urlencoded"},
+            )
+        except httpx.HTTPError as e:
+            raise CrmTransientError(f"HubSpot token endpoint unreachable: {e}")
+        if response.status_code >= 500:
+            raise CrmTransientError(f"HubSpot token endpoint HTTP {response.status_code}")
+        if response.status_code != 200:
+            # 400 invalid_grant / bad client config — retrying cannot help.
+            logger.warning(f"HubSpot token request rejected: {response.text[:500]}")
+            raise CrmAuthError(f"HubSpot rejected the token request: {summarize_provider_error(response.text)}")
+        body = response.json()
+        return OAuthTokens(
+            access_token=body["access_token"],
+            refresh_token=body.get("refresh_token", data.get("refresh_token", "")),
+            expires_at=datetime.now(timezone.utc) + timedelta(seconds=body.get("expires_in", 1800)),
+            raw=body,
+        )
+
+    def _contact_properties(self, payload: LeadPayload) -> dict:
+        first, last = split_name(payload.name)
+        properties = {
+            "email": payload.email,
+            "firstname": first,
+            "lastname": last,
+            "phone": payload.phone,
+            "company": payload.company,
+            # Backward lifecycle moves are ignored by HubSpot, so this is
+            # safe on updates too.
+            "lifecyclestage": "lead",
+        }
+        return {k: v for k, v in properties.items() if v}
+
+    async def _attach_note(self, tokens: OAuthTokens, contact_id: str,
+                           payload: LeadPayload) -> None:
+        """Context note (summary/custom answers/source). Best-effort: the
+        contact is the push — a failed note never fails the job."""
+        if not contact_id:
+            return
+        try:
+            response = await get_http_client().post(
+                f"{API_BASE}/crm/v3/objects/notes",
+                headers=self._auth_header(tokens),
+                json={
+                    "properties": {
+                        "hs_timestamp": int(datetime.now(timezone.utc).timestamp() * 1000),
+                        "hs_note_body": build_note_body(payload),
+                    },
+                    "associations": [{
+                        "to": {"id": contact_id},
+                        "types": [{
+                            "associationCategory": "HUBSPOT_DEFINED",
+                            "associationTypeId": NOTE_TO_CONTACT_ASSOCIATION_TYPE_ID,
+                        }],
+                    }],
+                },
+            )
+            if response.status_code not in (200, 201):
+                logger.warning(f"HubSpot note attach failed: HTTP {response.status_code}")
+        except httpx.HTTPError as e:
+            logger.warning(f"HubSpot note attach failed: {e}")
+
+    @staticmethod
+    def _auth_header(tokens: OAuthTokens) -> dict:
+        return {"Authorization": f"Bearer {tokens.access_token}"}
+
+
+register_adapter(HubSpotAdapter())

backend/app/crm/mapping.py

@@ -0,0 +1,143 @@
+"""
+Copyright 2024-2026 ReplyPier
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+import html
+from typing import Optional, Tuple
+
+from app.crm.base import LeadPayload
+from app.models.customer import Customer
+from app.models.lead_capture import LeadCaptureConfig, LeadCaptureResponse
+
+# Standard field keys the agent collects; everything else is a custom field.
+_STANDARD_KEYS = {"email", "name", "company", "phone"}
+
+
+def _payload_from_values(
+    values: dict,
+    customer: Optional[Customer],
+    config: Optional[LeadCaptureConfig],
+    summary: Optional[str],
+    ref_id,
+) -> LeadPayload:
+    """Fold captured field values into the provider-agnostic push payload.
+
+    `values` holds what the agent extracted conversationally (email/name/
+    company/phone + custom_N); the Customer row backfills anything blank. Custom
+    answers are labeled from the agent's field config. Company has no Customer
+    column, so it comes from the captured values only.
+    """
+    values = values or {}
+    email = (values.get("email") or (customer.email if customer else "") or "").strip().lower()
+
+    labels = _custom_field_labels(config)
+    custom_fields = {
+        labels.get(key, key): str(value)
+        for key, value in values.items()
+        if key not in _STANDARD_KEYS and value not in (None, "")
+    }
+
+    lead_source = (customer.lead_source if customer else None) or {}
+
+    return LeadPayload(
+        lead_response_id=ref_id,
+        email=email,
+        name=_first_truthy(values.get("name"), customer.full_name if customer else None),
+        company=values.get("company") or None,
+        phone=_first_truthy(values.get("phone"), customer.phone if customer else None),
+        summary=summary,
+        custom_fields=custom_fields,
+        source_url=lead_source.get("page_url"),
+    )
+
+
+def build_lead_payload(
+    response: LeadCaptureResponse,
+    config: Optional[LeadCaptureConfig],
+    customer: Optional[Customer],
+) -> LeadPayload:
+    """Auto path: fold one captured lead response into the push payload."""
+    return _payload_from_values(
+        response.field_values, customer, config, response.summary, response.id)
+
+
+def build_customer_payload(
+    customer: Customer,
+    attributes: Optional[dict] = None,
+    summary: Optional[str] = None,
+    config: Optional[LeadCaptureConfig] = None,
+) -> LeadPayload:
+    """Manual "Sync now": build from the person's captured attributes (the same
+    merged field_values the People drawer shows) so name/company/phone/custom
+    answers all sync — not just the sparse Customer columns. Integrator-supplied
+    meta_data is folded in as extra custom fields."""
+    payload = _payload_from_values(attributes, customer, config, summary, customer.id)
+    meta = customer.meta_data if isinstance(customer.meta_data, dict) else {}
+    for key, value in meta.items():
+        if value not in (None, "") and str(key) not in payload.custom_fields:
+            payload.custom_fields[str(key)] = str(value)
+    return payload
+
+
+def split_name(name: Optional[str]) -> Tuple[str, str]:
+    """Best-effort (first, last) split for CRMs with separate name fields."""
+    if not name or not name.strip():
+        return "", ""
+    parts = name.strip().split()
+    if len(parts) == 1:
+        return parts[0], ""
+    return " ".join(parts[:-1]), parts[-1]
+
+
+def build_note_body(payload: LeadPayload) -> str:
+    """The context note attached to the CRM record: AI qualification summary,
+    custom answers, and where the lead came from. Shared across providers.
+
+    Returns HTML — both HubSpot (hs_note_body) and Pipedrive (note content)
+    render notes as HTML, so plain newlines would collapse and run the summary
+    into the header. Labelled lines keep the AI summary unmistakable; dynamic
+    values are escaped."""
+    lines = ["<b>Lead captured by ReplyPier</b>"]
+    if payload.summary:
+        lines.append(f"<b>AI summary:</b> {html.escape(payload.summary)}")
+    for label, value in (payload.custom_fields or {}).items():
+        lines.append(f"<b>{html.escape(str(label))}:</b> {html.escape(str(value))}")
+    if payload.source_url:
+        url = html.escape(payload.source_url)
+        # Only render an anchor for http(s); a javascript:/data: page_url would
+        # otherwise become a clickable link in the CRM note.
+        if payload.source_url.lower().startswith(("http://", "https://")):
+            lines.append(f'Captured on: <a href="{url}">{url}</a>')
+        else:
+            lines.append(f"Captured on: {url}")
+    return "<br>".join(lines)
+
+
+def _custom_field_labels(config: Optional[LeadCaptureConfig]) -> dict:
+    """key -> human label for the agent's configured custom fields."""
+    if config is None or not config.fields:
+        return {}
+    return {
+        f["key"]: f.get("label") or f["key"]
+        for f in config.fields
+        if isinstance(f, dict) and f.get("key")
+    }
+
+
+def _first_truthy(*values: Optional[str]) -> Optional[str]:
+    for value in values:
+        if value and str(value).strip():
+            return str(value).strip()
+    return None

backend/app/crm/pipedrive.py

@@ -0,0 +1,324 @@
+"""
+Copyright 2024-2026 ReplyPier
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+from datetime import datetime, timedelta, timezone
+from typing import Optional
+from urllib.parse import urlencode, urlparse
+
+import httpx
+
+from app.core.config import settings
+from app.core.external_urls import external_base
+from app.core.logger import get_logger
+from app.crm.base import (
+    CrmAdapter, CrmAuthError, CrmPushResult, CrmTransientError,
+    LeadPayload, OAuthTokens, basic_auth_header, classify_error_response,
+    get_http_client, summarize_provider_error,
+)
+from app.crm.mapping import build_note_body
+from app.crm.registry import register_adapter
+from app.models.crm import CrmProvider
+
+logger = get_logger(__name__)
+
+OAUTH_BASE = external_base("PIPEDRIVE_OAUTH_BASE", "https://oauth.pipedrive.com")
+AUTHORIZE_URL = f"{OAUTH_BASE}/oauth/authorize"
+TOKEN_URL = f"{OAUTH_BASE}/oauth/token"
+REVOKE_URL = f"{OAUTH_BASE}/oauth/revoke"
+# The API host is per company (the token response's api_domain), so there is no
+# fixed base to swap. When set, this is the one extra api_domain we accept.
+API_BASE = external_base("PIPEDRIVE_API_BASE", "")
+# Refresh tokens die 60 days after their LAST use (sliding window) — every
+# refresh restamps this; the worker's proactive sweep keeps idle tenants alive.
+REFRESH_TOKEN_LIFETIME_DAYS = 60
+
+
+def _validated_api_domain(api_domain) -> Optional[str]:
+    """The per-tenant API host comes from Pipedrive's token response (over TLS),
+    but every subsequent call uses it as the base URL — so defensively require
+    an https pipedrive.com host and reject anything else, rather than letting a
+    malformed value point our requests at an arbitrary host."""
+    if not api_domain:
+        return None
+    if API_BASE and str(api_domain).rstrip("/") == API_BASE:
+        return API_BASE
+    try:
+        host = urlparse(api_domain).hostname or ""
+    except Exception:
+        return None
+    if api_domain.startswith("https://") and (host == "pipedrive.com" or host.endswith(".pipedrive.com")):
+        return api_domain
+    logger.warning(f"Pipedrive returned an unexpected api_domain, ignoring: {api_domain!r}")
+    return None
+
+
+class PipedriveAdapter(CrmAdapter):
+    """Pushes a lead as Person (v2, email-deduped) + Lead (v1) + context Note.
+
+    All API calls go to the tenant's `api_domain` from the token response.
+    A push is ~4 sequential calls, well under the 10 req/2s search cap; the
+    worker additionally serializes pushes per organization.
+    """
+
+    provider = CrmProvider.PIPEDRIVE.value
+
+    def get_authorize_url(self, state: str, redirect_uri: str) -> str:
+        # Scopes are declared on the app in Developer Hub, not in the URL.
+        return f"{AUTHORIZE_URL}?" + urlencode({
+            "client_id": settings.PIPEDRIVE_CLIENT_ID,
+            "redirect_uri": redirect_uri,
+            "state": state,
+        })
+
+    async def exchange_code(self, code: str, redirect_uri: str) -> OAuthTokens:
+        tokens = await self._token_request({
+            "grant_type": "authorization_code",
+            "code": code,
+            "redirect_uri": redirect_uri,
+        })
+        # Company identity comes from users/me on the tenant's api_domain.
+        # Skip when the domain was missing/rejected — there's no valid base to call.
+        if not tokens.api_domain:
+            return tokens
+        try:
+            response = await get_http_client().get(
+                f"{tokens.api_domain}/api/v1/users/me",
+                headers=self._auth_header(tokens))
+            if response.status_code == 200:
+                data = response.json().get("data") or {}
+                tokens.external_account_id = str(data.get("company_id", ""))
+                tokens.display_name = data.get("company_name")
+        except httpx.HTTPError as e:
+            logger.warning(f"Pipedrive users/me lookup failed (ignored): {e}")
+        return tokens
+
+    async def refresh_tokens(self, tokens: OAuthTokens) -> OAuthTokens:
+        refreshed = await self._token_request({
+            "grant_type": "refresh_token",
+            "refresh_token": tokens.refresh_token,
+        }, fallback_refresh_token=tokens.refresh_token)
+        refreshed.api_domain = refreshed.api_domain or tokens.api_domain
+        refreshed.external_account_id = tokens.external_account_id
+        refreshed.display_name = tokens.display_name
+        return refreshed
+
+    async def push_lead(self, tokens: OAuthTokens, payload: LeadPayload) -> CrmPushResult:
+        try:
+            org_id = await self._upsert_organization(tokens, payload.company)
+            person_id, action, person_error = await self._upsert_person(tokens, payload, org_id)
+            if person_error is not None:
+                return person_error
+            lead_id, lead_error = await self._ensure_open_lead(tokens, person_id, payload, org_id)
+            if lead_error is not None:
+                return lead_error
+            if lead_id:
+                await self._attach_note(tokens, lead_id, payload)
+        except httpx.HTTPError as e:
+            return CrmPushResult(ok=False, error=f"network error: {e}", retryable=True)
+
+        return CrmPushResult(
+            ok=True, action=action,
+            contact_id=str(person_id), secondary_id=str(lead_id) if lead_id else None,
+            record_url=f"{tokens.api_domain}/person/{person_id}" if tokens.api_domain else None,
+        )
+
+    async def revoke(self, tokens: OAuthTokens) -> None:
+        try:
+            await get_http_client().post(
+                REVOKE_URL,
+                headers={"Authorization": f"Basic {self._basic_secret()}"},
+                data={"token": tokens.refresh_token,
+                      "token_type_hint": "refresh_token"},
+            )
+        except Exception as e:
+            logger.warning(f"Pipedrive token revocation failed (ignored): {e}")
+
+    async def test_connection(self, tokens: OAuthTokens) -> dict:
+        try:
+            response = await get_http_client().get(
+                f"{tokens.api_domain}/api/v1/users/me",
+                headers=self._auth_header(tokens))
+        except httpx.HTTPError as e:
+            return {"ok": False, "account_name": None, "error": f"network error: {e}"}
+        if response.status_code != 200:
+            return {"ok": False, "account_name": None,
+                    "error": f"HTTP {response.status_code}"}
+        data = response.json().get("data") or {}
+        return {"ok": True, "account_name": data.get("company_name"), "error": None}
+
+    # ---- push steps -------------------------------------------------------
+
+    async def _upsert_organization(self, tokens: OAuthTokens, company: str | None):
+        """Find-or-create the organization by name so the person is linked to a
+        company. Best-effort: a failure returns None and the person still syncs,
+        just unlinked. Uses the contacts scope (persons + orgs), so no extra
+        permission is needed."""
+        if not company or not company.strip():
+            return None
+        company = company.strip()
+        try:
+            search = await get_http_client().get(
+                f"{tokens.api_domain}/api/v2/organizations/search?" + urlencode({
+                    "term": company, "fields": "name", "exact_match": "true"}),
+                headers=self._auth_header(tokens))
+            if search.status_code == 200:
+                items = ((search.json().get("data") or {}).get("items")) or []
+                if items:
+                    return (items[0].get("item") or {}).get("id")
+            create = await get_http_client().post(
+                f"{tokens.api_domain}/api/v2/organizations",
+                headers=self._auth_header(tokens), json={"name": company})
+            if create.status_code in (200, 201):
+                return (create.json().get("data") or {}).get("id")
+            logger.warning(f"Pipedrive organization upsert failed: HTTP {create.status_code}")
+        except httpx.HTTPError as e:
+            logger.warning(f"Pipedrive organization upsert failed: {e}")
+        return None
+
+    async def _upsert_person(self, tokens: OAuthTokens, payload: LeadPayload, org_id=None):
+        """Search by email; update blanks on a hit, create on a miss.
+        Returns (person_id, action, error_result)."""
+        response = await get_http_client().get(
+            f"{tokens.api_domain}/api/v2/persons/search?" + urlencode({
+                "term": payload.email, "fields": "email", "exact_match": "true"}),
+            headers=self._auth_header(tokens))
+        if response.status_code != 200:
+            return None, None, classify_error_response(response)
+
+        items = ((response.json().get("data") or {}).get("items")) or []
+        if items:
+            item = items[0].get("item") or {}
+            person_id = item.get("id")
+            update = self._blank_filling_update(item, payload, org_id)
+            if update:
+                patch = await get_http_client().patch(
+                    f"{tokens.api_domain}/api/v2/persons/{person_id}",
+                    headers=self._auth_header(tokens), json=update)
+                if patch.status_code not in (200, 201):
+                    logger.warning(
+                        f"Pipedrive person update failed: HTTP {patch.status_code}")
+            return person_id, "updated", None
+
+        create = await get_http_client().post(
+            f"{tokens.api_domain}/api/v2/persons",
+            headers=self._auth_header(tokens),
+            json=self._person_body(payload, org_id))
+        if create.status_code not in (200, 201):
+            return None, None, classify_error_response(create)
+        person_id = (create.json().get("data") or {}).get("id")
+        return person_id, "created", None
+
+    async def _ensure_open_lead(self, tokens: OAuthTokens, person_id,
+                                payload: LeadPayload, org_id=None):
+        """Create a Lead unless the person already has an open one — repeat
+        captures must not spam the leads inbox. Returns (lead_id, error_result).
+
+        A Lead carries its own organization_id (separate from the person's
+        org), so set it too or the lead shows no company."""
+        listing = await get_http_client().get(
+            f"{tokens.api_domain}/api/v1/leads?" + urlencode({
+                "person_id": person_id, "archived_status": "not_archived"}),
+            headers=self._auth_header(tokens))
+        if listing.status_code != 200:
+            # Guessing here risks a duplicate lead — classify and let the
+            # (idempotent) job retry instead.
+            return None, classify_error_response(listing)
+        if listing.json().get("data") or []:
+            return None, None  # open lead exists — nothing to create
+
+        body = {"title": f"{payload.name or payload.email} — ReplyPier lead",
+                "person_id": person_id}
+        if org_id:
+            body["organization_id"] = org_id
+        create = await get_http_client().post(
+            f"{tokens.api_domain}/api/v1/leads",
+            headers=self._auth_header(tokens), json=body)
+        if create.status_code not in (200, 201):
+            return None, classify_error_response(create)
+        return (create.json().get("data") or {}).get("id"), None
+
+    async def _attach_note(self, tokens: OAuthTokens, lead_id,
+                           payload: LeadPayload) -> None:
+        """Context note on the lead. Best-effort — never fails the job."""
+        try:
+            response = await get_http_client().post(
+                f"{tokens.api_domain}/api/v1/notes",
+                headers=self._auth_header(tokens),
+                json={"lead_id": lead_id,
+                      "content": build_note_body(payload)})
+            if response.status_code not in (200, 201):
+                logger.warning(f"Pipedrive note attach failed: HTTP {response.status_code}")
+        except httpx.HTTPError as e:
+            logger.warning(f"Pipedrive note attach failed: {e}")
+
+    # ---- helpers ----------------------------------------------------------
+
+    async def _token_request(self, data: dict,
+                             fallback_refresh_token: str = "") -> OAuthTokens:
+        try:
+            response = await get_http_client().post(
+                TOKEN_URL, data=data,
+                headers={"Authorization": f"Basic {self._basic_secret()}",
+                         "Content-Type": "application/x-www-form-urlencoded"},
+            )
+        except httpx.HTTPError as e:
+            raise CrmTransientError(f"Pipedrive token endpoint unreachable: {e}")
+        if response.status_code >= 500:
+            raise CrmTransientError(f"Pipedrive token endpoint HTTP {response.status_code}")
+        if response.status_code != 200:
+            logger.warning(f"Pipedrive token request rejected: {response.text[:500]}")
+            raise CrmAuthError(f"Pipedrive rejected the token request: {summarize_provider_error(response.text)}")
+        body = response.json()
+        now = datetime.now(timezone.utc)
+        return OAuthTokens(
+            access_token=body["access_token"],
+            refresh_token=body.get("refresh_token") or fallback_refresh_token,
+            expires_at=now + timedelta(seconds=body.get("expires_in", 3600)),
+            api_domain=_validated_api_domain(body.get("api_domain")),
+            # Sliding window restarts on every successful token use.
+            refresh_token_expires_at=now + timedelta(days=REFRESH_TOKEN_LIFETIME_DAYS),
+            raw=body,
+        )
+
+    def _person_body(self, payload: LeadPayload, org_id=None) -> dict:
+        body = {"name": payload.name or payload.email,
+                "emails": [{"value": payload.email, "primary": True}]}
+        if payload.phone:
+            body["phones"] = [{"value": payload.phone, "primary": True}]
+        if org_id:
+            body["org_id"] = org_id
+        return body
+
+    def _blank_filling_update(self, item: dict, payload: LeadPayload, org_id=None) -> dict:
+        """Only fill fields the existing person is missing — never clobber."""
+        update = {}
+        if payload.name and not item.get("name"):
+            update["name"] = payload.name
+        if payload.phone and not item.get("phones"):
+            update["phones"] = [{"value": payload.phone, "primary": True}]
+        if org_id and not item.get("org_id"):
+            update["org_id"] = org_id
+        return update
+
+    def _auth_header(self, tokens: OAuthTokens) -> dict:
+        return {"Authorization": f"Bearer {tokens.access_token}"}
+
+    @staticmethod
+    def _basic_secret() -> str:
+        return basic_auth_header(settings.PIPEDRIVE_CLIENT_ID, settings.PIPEDRIVE_CLIENT_SECRET)
+
+
+register_adapter(PipedriveAdapter())

backend/app/crm/registry.py

@@ -0,0 +1,47 @@
+"""
+Copyright 2024-2026 ReplyPier
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+from typing import Dict, Optional
+
+from app.crm.base import CrmAdapter
+
+_adapters: Dict[str, CrmAdapter] = {}
+
+
+def register_adapter(adapter: CrmAdapter) -> CrmAdapter:
+    """Register a stateless adapter singleton for its provider."""
+    _adapters[adapter.provider] = adapter
+    return adapter
+
+
+def get_adapter(provider: str) -> Optional[CrmAdapter]:
+    """Adapter for a CRM provider, or None for unknown providers."""
+    _ensure_loaded()
+    return _adapters.get(provider)
+
+
+_loaded = False
+
+
+def _ensure_loaded() -> None:
+    """Import adapter modules on first use (each registers itself).
+    Lazy so importing app.crm never pulls every adapter's deps."""
+    global _loaded
+    if _loaded:
+        return
+    _loaded = True
+    from app.crm import hubspot  # noqa: F401
+    from app.crm import pipedrive  # noqa: F401

backend/app/services/app_first_routing.py

@@ -0,0 +1,98 @@
+"""
+Copyright 2024-2026 ReplyPier
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+Letting an installed extension answer a channel message before the AI does
+(see app.channels.app_hooks), and announcing every inbound message to it.
+"""
+
+import asyncio
+from datetime import datetime, timezone
+from typing import Optional
+
+from sqlalchemy.orm import Session
+
+from app.channels.app_hooks import (
+    ChannelEvent, EventKind, HandledBy, InboundContext, emit_channel_event,
+    inbound_event_id, inbound_handler,
+)
+from app.channels.base import InboundMessage
+from app.core.logger import get_logger
+from app.models.channels import ChannelAccount
+from app.models.session_to_agent import SessionStatus
+
+logger = get_logger(__name__)
+
+# How long the customer's message waits on the extension before the AI (or
+# inbox) takes it. The extension answers synchronously, so this is added to
+# every reply the AI then gives — kept short.
+APP_HANDLER_TIMEOUT_SECONDS = 4.0
+
+
+async def app_claims(db: Session, account: ChannelAccount, session_record, customer_id: str,
+                     inbound: InboundMessage, received_at: datetime) -> bool:
+    """Offer the message to the extension; True when it has dealt with it.
+
+    Only a conversation nobody holds is offered — once a human has it, or a
+    transfer is pending, the human answers. A timeout or a failing extension
+    counts as "not handled", so the customer is never left waiting on it.
+    """
+    handler = inbound_handler()
+    if handler is None or not _open_and_unheld(session_record):
+        return False
+    # Nothing may sit in an open transaction while an outside service thinks
+    db.commit()
+    context = InboundContext(
+        event_id=inbound_event_id(account.id, inbound.external_message_id),
+        organization_id=account.organization_id,
+        channel_account_id=account.id,
+        channel_type=account.channel_type,
+        session_id=str(session_record.session_id),
+        customer_id=customer_id,
+        inbound=inbound,
+        received_at=received_at,
+    )
+    try:
+        # Exactly True: anything else (a stray truthy string, a forgotten
+        # await) must not silence the AI
+        return await asyncio.wait_for(handler(context), APP_HANDLER_TIMEOUT_SECONDS) is True
+    except asyncio.TimeoutError:
+        logger.warning(f"Extension did not answer {inbound.external_message_id} within "
+                       f"{APP_HANDLER_TIMEOUT_SECONDS}s; continuing without it")
+    except Exception as e:
+        logger.error(f"Extension failed on {inbound.external_message_id}; continuing without it: {e}")
+    return False
+
+
+async def announce_inbound(account: ChannelAccount, session_record, customer_id: str,
+                           inbound: InboundMessage, handled_by: HandledBy,
+                           received_at: Optional[datetime] = None) -> None:
+    """Tell listeners about an inbound message and who is answering it. Sent
+    once per message on every path, before any AI reply, under an id derived
+    from the message so a redelivered webhook can't announce it twice."""
+    await emit_channel_event(ChannelEvent(
+        kind=EventKind.INBOUND,
+        event_id=inbound_event_id(account.id, inbound.external_message_id),
+        organization_id=account.organization_id,
+        channel_account_id=account.id,
+        occurred_at=received_at or datetime.now(timezone.utc),
+        session_id=str(session_record.session_id),
+        customer_id=customer_id,
+        inbound=inbound,
+        handled_by=handled_by,
+    ))
+
+
+def _open_and_unheld(session_record) -> bool:
+    return session_record.user_id is None and session_record.status == SessionStatus.OPEN

backend/app/services/contact_capture.py

@@ -0,0 +1,88 @@
+"""
+Copyright 2024-2026 ReplyPier
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+from typing import Optional, Dict, Any
+from app.repositories.customer import CustomerRepository
+
+# Where a handoff email goes when the customer row can't take it — another
+# customer in the org already owns that address (the (email, organization_id)
+# unique constraint), or it is a channel identity key. meta_data renders as a
+# label/value list in the inbox customer panel, so an agent still sees the
+# address and can follow up. Dropping it is how a visitor asks to be contacted
+# and then never is.
+CONTACT_EMAIL_META_KEY = "contact_email_provided"
+
+
+def retain_unstored_email(customer_repo, customer_id, submitted_email: str, update_result: dict) -> bool:
+    """Keep a submitted handoff email that ``update_contact`` could not apply.
+
+    Returns True when it was retained on meta_data. No-op when the address did
+    land on the customer, when it is already their stored address, or when
+    nothing was submitted — so a normal capture writes no metadata.
+    """
+    if not submitted_email or update_result.get('email_updated'):
+        return False
+    stored = (update_result.get('email') or '').lower()
+    if submitted_email.lower() == stored:
+        return False
+    return customer_repo.update_meta_data(
+        customer_id, {CONTACT_EMAIL_META_KEY: submitted_email}
+    ) is not None
+
+
+def build_handoff_contact_form(
+    customer,
+    collect_email: bool,
+    collect_name: bool,
+) -> Optional[Dict[str, Any]]:
+    """Return ``form_data`` (``form_type='contact'``) for the handoff contact prompt, or
+    ``None`` when there is nothing to collect (toggles off, or the customer already has the
+    info). The widget renders this with its existing inline form UI.
+    """
+    if customer is None:
+        return None
+
+    needs_email = bool(collect_email) and CustomerRepository.is_placeholder_email(getattr(customer, 'email', None))
+    full_name = (getattr(customer, 'full_name', None) or '').strip()
+    needs_name = bool(collect_name) and not full_name
+
+    if not needs_email and not needs_name:
+        return None
+
+    fields = []
+    if needs_email:
+        fields.append({
+            'name': 'email',
+            'type': 'email',
+            'label': 'Email',
+            'placeholder': 'you@example.com',
+            'required': True,
+        })
+    if needs_name:
+        fields.append({
+            'name': 'name',
+            'type': 'text',
+            'label': 'Name',
+            'placeholder': 'Your name',
+            'required': False,
+        })
+
+    return {
+        'form_type': 'contact',
+        'title': 'Before we connect you to a teammate',
+        'description': 'Share your details so we can follow up.',
+        'fields': fields,
+    }

backend/app/services/crm_sync.py

@@ -0,0 +1,241 @@
+"""
+Copyright 2024-2026 ReplyPier
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+from typing import Optional
+
+from sqlalchemy.orm import Session
+
+from app.core.logger import get_logger
+from app.crm.base import CrmAuthError, CrmPushResult, CrmTransientError
+from app.crm.mapping import build_customer_payload, build_lead_payload
+from app.crm.registry import get_adapter
+from app.models.crm import CrmConnectionStatus, CrmSyncJob
+from app.models.customer import Customer
+from app.models.lead_capture import CrmSyncTarget, LeadCaptureConfig, LeadCaptureResponse
+from app.repositories.crm import (
+    CrmConnectionRepository, CrmSyncJobRepository, CrmCustomerSyncRepository,
+)
+from app.services.crm_tokens import get_valid_tokens
+from app.services.feature_gate import feature_allowed
+
+logger = get_logger(__name__)
+
+
+def enqueue_crm_sync(db: Session, config: LeadCaptureConfig,
+                     response: LeadCaptureResponse) -> Optional[CrmSyncJob]:
+    """Queue a captured lead for CRM push, or quietly decide not to.
+
+    All the "should this sync at all" logic lives here so the capture path
+    stays dumb: no target configured, plan lacks the feature (fails closed),
+    or no active connection → no job. Idempotent per (lead, provider)."""
+    provider = str(config.crm_sync_target or CrmSyncTarget.NONE.value)
+    if provider == CrmSyncTarget.NONE.value:
+        return None
+    if get_adapter(provider) is None:
+        logger.warning(f"Lead {response.id}: unknown CRM target '{provider}', not syncing")
+        return None
+    if not feature_allowed(db, response.organization_id, "crm_sync"):
+        logger.info(f"Lead {response.id}: plan lacks crm_sync, not syncing")
+        return None
+    connection = CrmConnectionRepository(db).get_active(response.organization_id, provider)
+    if connection is None:
+        logger.info(f"Lead {response.id}: no active {provider} connection, not syncing")
+        return None
+    job = CrmSyncJobRepository(db).enqueue(
+        organization_id=response.organization_id,
+        lead_response_id=response.id,
+        provider=provider,
+        agent_id=response.agent_id,
+    )
+    if job is not None:
+        logger.info(f"Queued {provider} sync for lead {response.id}")
+    return job
+
+
+async def process_job(db: Session, job: CrmSyncJob) -> CrmSyncJob:
+    """Run one claimed sync job to a terminal or retry state.
+
+    Safe to re-run: the adapters' push sequences converge (HubSpot upsert,
+    Pipedrive search-first + open-lead check)."""
+    jobs = CrmSyncJobRepository(db)
+
+    # Re-checked at run time: the org may have downgraded since enqueue.
+    if not feature_allowed(db, job.organization_id, "crm_sync"):
+        return jobs.skip(job, "plan no longer includes crm_sync")
+
+    connections = CrmConnectionRepository(db)
+    connection = connections.get_active(job.organization_id, job.provider)
+    if connection is None:
+        # Terminal: retrying cannot help until a human reconnects, and the
+        # reconnect path enqueues nothing retroactively — the UI surfaces it.
+        return jobs.fail(job, "no_connection")
+
+    response = db.get(LeadCaptureResponse, job.lead_response_id)
+    if response is None:
+        return jobs.skip(job, "lead response deleted")
+    payload = build_lead_payload(
+        response,
+        _config_for_agent(db, response.agent_id),
+        db.get(Customer, response.customer_id) if response.customer_id else None,
+    )
+    if not payload.email:
+        return jobs.skip(job, "lead has no email")
+
+    adapter = get_adapter(job.provider)
+    try:
+        tokens = await get_valid_tokens(db, connection)
+    except CrmAuthError as e:
+        return jobs.fail(job, f"auth: {e}", connection_id=connection.id)
+    except CrmTransientError as e:
+        return jobs.schedule_retry(job, f"token refresh failed: {e}")
+
+    result = await adapter.push_lead(tokens, payload)
+
+    if result.auth_failed:
+        # The stored expiry looked fine but the provider said no — force one
+        # refresh and retry before declaring the connection dead.
+        try:
+            tokens = await get_valid_tokens(db, connection, force_refresh=True)
+        except (CrmAuthError, CrmTransientError) as e:
+            return jobs.fail(job, f"auth: {e}", connection_id=connection.id)
+        result = await adapter.push_lead(tokens, payload)
+        if result.auth_failed:
+            connections.set_status(connection, CrmConnectionStatus.EXPIRED.value,
+                                   last_error=result.error)
+            return jobs.fail(job, f"auth: {result.error}", connection_id=connection.id)
+
+    if result.ok:
+        logger.info(f"Pushed lead {response.id} to {job.provider} "
+                    f"({result.action} {result.contact_id})")
+        # Record the per-person link so the People drawer can show it.
+        if response.customer_id:
+            _record_customer_sync(db, job.organization_id, response.customer_id,
+                                  job.provider, result)
+        return jobs.complete(job, {
+            "action": result.action,
+            "contact_id": result.contact_id,
+            "secondary_id": result.secondary_id,
+            "record_url": result.record_url,
+        }, connection_id=connection.id)
+    if result.retryable:
+        return jobs.schedule_retry(job, result.error or "retryable failure",
+                                   retry_after_seconds=result.retry_after_seconds)
+    return jobs.fail(job, result.error or "push failed", connection_id=connection.id)
+
+
+class CrmManualSyncError(Exception):
+    """A manual "Sync now" could not be completed (surfaced to the user)."""
+
+
+async def sync_customer_to_crm(db: Session, customer: Customer) -> list:
+    """Manual "Sync now" from the People drawer: push this person to every
+    active CRM connection in their org and record each link. Synchronous so
+    the UI gets an immediate result. Returns the list of recorded links.
+
+    Raises CrmManualSyncError for user-facing failures (no email, no
+    connection, plan gate, or a definitive provider rejection)."""
+    from app.repositories.customer import CustomerRepository
+    # Anonymous visitors carry a placeholder stand-in (…@noemail.com / .channel)
+    # that can't receive mail — never push it to a CRM.
+    if CustomerRepository.is_placeholder_email(customer.email):
+        raise CrmManualSyncError("Add an email to this person before syncing to a CRM.")
+    if not feature_allowed(db, customer.organization_id, "crm_sync"):
+        raise CrmManualSyncError("CRM sync is not available in your current plan.")
+
+    connections = CrmConnectionRepository(db).list_by_org(customer.organization_id)
+    active = [c for c in connections if c.status == CrmConnectionStatus.ACTIVE.value]
+    if not active:
+        raise CrmManualSyncError("Connect a CRM in Settings → Integrations first.")
+
+    attributes, summary, config = _lead_capture_context(db, customer)
+    payload = build_customer_payload(customer, attributes=attributes, summary=summary, config=config)
+    records, errors = [], []
+    for connection in active:
+        adapter = get_adapter(connection.provider)
+        if adapter is None:
+            continue
+        try:
+            tokens = await get_valid_tokens(db, connection)
+            result = await adapter.push_lead(tokens, payload)
+            if result.auth_failed:
+                tokens = await get_valid_tokens(db, connection, force_refresh=True)
+                result = await adapter.push_lead(tokens, payload)
+        except CrmAuthError:
+            errors.append(f"{connection.provider}: reconnect required")
+            continue
+        except CrmTransientError:
+            errors.append(f"{connection.provider}: temporary error, try again")
+            continue
+        if result.ok:
+            records.append(_record_customer_sync(
+                db, customer.organization_id, customer.id, connection.provider, result))
+        elif result.auth_failed:
+            connections_repo = CrmConnectionRepository(db)
+            connections_repo.set_status(connection, CrmConnectionStatus.EXPIRED.value,
+                                        last_error=result.error)
+            errors.append(f"{connection.provider}: reconnect required")
+        else:
+            errors.append(f"{connection.provider}: {result.error or 'push failed'}")
+
+    if not records and errors:
+        raise CrmManualSyncError("; ".join(errors))
+    return records
+
+
+def _record_customer_sync(db: Session, organization_id, customer_id, provider: str,
+                          result: CrmPushResult):
+    return CrmCustomerSyncRepository(db).record(
+        organization_id=organization_id,
+        customer_id=customer_id,
+        provider=provider,
+        contact_id=result.contact_id,
+        secondary_id=result.secondary_id,
+        record_url=result.record_url,
+    )
+
+
+def _lead_capture_context(db: Session, customer: Customer):
+    """Merge everything the person's lead captures produced — attributes
+    (latest value wins per key), the most recent AI summary, and the latest
+    response's agent config for custom-field labels — mirroring what the People
+    drawer shows, so a manual push sends the full picture, not just email.
+    Returns (attributes, summary, config)."""
+    responses = (
+        db.query(LeadCaptureResponse)
+        .filter(LeadCaptureResponse.customer_id == customer.id)
+        .order_by(LeadCaptureResponse.created_at)
+        .all()
+    )
+    attributes: dict = {}
+    summary = None
+    latest_agent_id = None
+    for r in responses:
+        for key, value in (r.field_values or {}).items():
+            if value not in (None, ""):
+                attributes[key] = value
+        if r.summary:
+            summary = r.summary
+        if r.agent_id:
+            latest_agent_id = r.agent_id
+    config = _config_for_agent(db, latest_agent_id) if latest_agent_id else None
+    return attributes, summary, config
+
+
+def _config_for_agent(db: Session, agent_id) -> Optional[LeadCaptureConfig]:
+    if agent_id is None:
+        return None
+    return db.query(LeadCaptureConfig).filter(
+        LeadCaptureConfig.agent_id == agent_id).first()

backend/app/services/crm_tokens.py

@@ -0,0 +1,122 @@
+"""
+Copyright 2024-2026 ReplyPier
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+from datetime import datetime, timedelta, timezone
+from typing import Optional
+
+from sqlalchemy.orm import Session
+
+from app.core.logger import get_logger
+from app.crm.base import CrmAuthError, OAuthTokens
+from app.crm.registry import get_adapter
+from app.models.crm import CrmConnection, CrmConnectionStatus
+from app.repositories.crm import CrmConnectionRepository
+
+logger = get_logger(__name__)
+
+# Refresh when the access token has less runway than this — covers clock skew
+# and the request's own latency.
+EXPIRY_MARGIN_SECONDS = 120
+
+
+def tokens_from_connection(repo: CrmConnectionRepository,
+                           connection: CrmConnection) -> OAuthTokens:
+    """Decrypt a connection into the adapter-facing token dataclass."""
+    credentials = repo.get_credentials(connection)
+    return OAuthTokens(
+        access_token=credentials.get("access_token", ""),
+        refresh_token=credentials.get("refresh_token", ""),
+        expires_at=_aware(connection.access_token_expires_at),
+        api_domain=credentials.get("api_domain"),
+        refresh_token_expires_at=_aware(connection.refresh_token_expires_at),
+        external_account_id=connection.external_account_id or "",
+        display_name=connection.display_name,
+    )
+
+
+def persist_tokens(repo: CrmConnectionRepository, connection: CrmConnection,
+                   tokens: OAuthTokens) -> None:
+    """Re-encrypt the blob and mirror expiries onto queryable columns."""
+    repo.save_credentials(
+        connection,
+        {
+            "access_token": tokens.access_token,
+            "refresh_token": tokens.refresh_token,
+            "api_domain": tokens.api_domain,
+        },
+        access_token_expires_at=tokens.expires_at,
+        refresh_token_expires_at=tokens.refresh_token_expires_at,
+    )
+
+
+async def get_valid_tokens(db: Session, connection: CrmConnection,
+                           force_refresh: bool = False) -> OAuthTokens:
+    """Decrypted tokens with a live access token, refreshing when needed.
+
+    Raises CrmAuthError after marking the connection expired when the provider
+    definitively rejects our grant (the UI then shows "Reconnect");
+    CrmTransientError propagates untouched so callers can retry.
+    """
+    repo = CrmConnectionRepository(db)
+    tokens = tokens_from_connection(repo, connection)
+    if not force_refresh and _has_runway(tokens.expires_at):
+        return tokens
+
+    # A refresh is needed. Serialize per connection: Pipedrive rotates the
+    # refresh token on every use, so two concurrent refreshers would send the
+    # same (now-consumed) token — the loser gets invalid_grant and would
+    # falsely mark the connection expired. Lock the row and re-read the current
+    # tokens under the lock (populate_existing bypasses the identity-map cache),
+    # so we refresh at most once and always with the latest token. FOR UPDATE is
+    # a no-op on SQLite (tests), which is fine.
+    locked = (
+        db.query(CrmConnection)
+        .filter(CrmConnection.id == connection.id)
+        .populate_existing()
+        .with_for_update()
+        .first()
+    ) or connection
+    tokens = tokens_from_connection(repo, locked)
+    if not force_refresh and _has_runway(tokens.expires_at):
+        db.commit()  # another worker refreshed while we waited — release the lock
+        return tokens
+
+    adapter = get_adapter(locked.provider)
+    if adapter is None:
+        raise CrmAuthError(f"No adapter for provider '{locked.provider}'")
+    try:
+        refreshed = await adapter.refresh_tokens(tokens)
+    except CrmAuthError as e:
+        logger.warning(f"CRM refresh rejected for {locked.provider} "
+                       f"connection {locked.id}: {e}")
+        repo.set_status(locked, CrmConnectionStatus.EXPIRED.value, last_error=str(e))
+        raise
+    persist_tokens(repo, locked, refreshed)
+    return refreshed
+
+
+def _has_runway(expires_at: Optional[datetime]) -> bool:
+    if expires_at is None:
+        return False
+    margin = datetime.now(timezone.utc) + timedelta(seconds=EXPIRY_MARGIN_SECONDS)
+    return _aware(expires_at) > margin
+
+
+def _aware(dt: Optional[datetime]) -> Optional[datetime]:
+    """Normalize DB datetimes: PG returns tz-aware, SQLite (tests) naive-UTC."""
+    if dt is None or dt.tzinfo is not None:
+        return dt
+    return dt.replace(tzinfo=timezone.utc)

backend/app/services/lead_capture.py

@@ -0,0 +1,323 @@
+"""
+Copyright 2024-2026 ReplyPier
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+import re
+from typing import Optional, Dict, Any
+from uuid import UUID
+from datetime import datetime, timezone
+from sqlalchemy.orm import Session
+
+from app.models.lead_capture import LeadCaptureConfig, LeadCaptureResponse
+from app.repositories.customer import CustomerRepository
+from app.services.widget_identity import can_act_as
+from app.core.logger import get_logger
+
+logger = get_logger(__name__)
+
+_EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
+
+
+def has_captured_lead(db: Session, customer_id, agent_id) -> bool:
+    """True once a lead has been recorded for this customer + agent (capture once)."""
+    return db.query(LeadCaptureResponse.id).filter(
+        LeadCaptureResponse.customer_id == customer_id,
+        LeadCaptureResponse.agent_id == agent_id,
+    ).first() is not None
+
+
+def _valid_email(email: Optional[str]) -> bool:
+    return bool(email and _EMAIL_RE.match(email.strip()))
+
+
+def record_lead_capture(
+    db: Session,
+    config: LeadCaptureConfig,
+    organization_id: UUID,
+    agent_id,
+    customer_id,
+    session_id,
+    lead_data: Optional[Dict[str, Any]],
+    summary: Optional[str],
+    consent: bool,
+    page_url: Optional[str] = None,
+    channel: str = "widget",
+    *,
+    scope: Optional[str],
+) -> Optional[LeadCaptureResponse]:
+    """Persist an AI-captured lead (fields the agent extracted conversationally +
+    an AI qualification summary). Returns None — recording nothing — when there is
+    no valid email, or consent is required but not given (GDPR).
+
+    Never records when lead capture is disabled for the agent, or when the customer
+    is integration-authenticated (identified via generate-token) — those are the
+    business's existing customers, not leads.
+
+    If the captured email already belongs to ANOTHER customer in the org, the
+    anonymous visitor is MERGED into that existing customer (history/sessions
+    reassigned, visitor row marked merged) so People shows one entry. The returned
+    response's customer_id is the final (possibly merged-into) customer.
+
+    `scope` is the conversation-token scope of the visitor (VERIFIED_SCOPE for
+    channels, whose identity the platform vouches for). A widget visitor may be
+    bound to a customer by an email they only typed, so a lead is only recorded
+    on a customer that visitor created (widget_identity.can_act_as), and only
+    merged into one it may also act as. Otherwise nothing is recorded.
+
+    A recorded lead is queued for CRM push (post-commit, best-effort) when the
+    agent has a crm_sync_target configured. Slack/assignment routing remains
+    stored-only.
+    """
+    from app.models.customer import Customer
+
+    # Never capture when lead capture is off for this agent (authoritative guard,
+    # independent of any caller-side check).
+    if not (config and config.enabled):
+        return None
+
+    # Never capture integration-authenticated people (identified via generate-token) —
+    # they're already the business's customers, not leads, regardless of the agent's
+    # toggle. Uses the explicit flag, not meta_data (see Customer.is_authenticated).
+    customer = db.query(Customer).filter(Customer.id == customer_id).first()
+    if customer is not None and customer.is_authenticated:
+        return None
+
+    lead_data = lead_data or {}
+    # Keep only configured field keys (defensive: the model shouldn't inject others).
+    field_keys = {f.get("key") for f in (config.fields or []) if f.get("key")}
+    if field_keys:
+        field_values = {k: v for k, v in lead_data.items() if k in field_keys and v}
+    else:
+        field_values = {k: v for k, v in lead_data.items() if v}
+
+    email = (str(field_values.get("email") or "")).strip()
+    if not _valid_email(email):
+        return None  # need at least a valid email to have a usable lead
+    if config.require_consent and not consent:
+        return None  # GDPR: never record without explicit consent
+
+    name = (str(field_values.get("name") or "")).strip() or None
+
+    # If this email already belongs to a different customer in the org, merge the
+    # current (anonymous) visitor into that customer and record the lead there.
+    existing = (
+        db.query(Customer)
+        .filter(
+            Customer.organization_id == organization_id,
+            Customer.email == email,
+            Customer.id != customer_id,
+            Customer.merged_into_customer_id.is_(None),
+        )
+        .first()
+    )
+    # A visitor bound to this customer by an email they only typed records
+    # nothing against them: no lead, no contact details, no CRM push - and the
+    # real customer is still asked for their details next time.
+    if not can_act_as(db, customer, scope):
+        logger.info(f"Lead capture: not recording for customer {customer_id}; the visitor does not own it")
+        return None
+
+    # Merging moves everything of one customer onto another, so the visitor must
+    # be able to act as both. For a widget visitor that rules out merging into
+    # anyone who already existed - an email they typed proves nothing - and the
+    # lead stays on their own row.
+    if existing is not None and not can_act_as(db, existing, scope):
+        logger.info(f"Lead capture: not merging customer {customer_id}; the visitor does not own both")
+        existing = None
+
+    # Merge (reassigning history across rows), the new response row, the contact
+    # update, and the promotion must all commit together — roll back on any failure
+    # so a mid-merge error can never leave a half-merged customer.
+    try:
+        target_customer_id = customer_id
+        if existing is not None:
+            _merge_customer(db, source_id=customer_id, target=existing)
+            target_customer_id = existing.id
+
+        response = LeadCaptureResponse(
+            organization_id=organization_id,
+            agent_id=agent_id,
+            customer_id=target_customer_id,
+            session_id=session_id,
+            field_values=field_values or None,
+            summary=(summary or None),
+            consent=bool(consent),
+            qualified=True,
+        )
+        db.add(response)
+
+        # Update the customer's contact details so the People page / inbox show a real
+        # email/name (reuses the handoff-capture logic + unique-constraint handling).
+        # Phone included: an AI-captured number makes a widget lead WhatsApp-addressable.
+        # Deliberately strict: a captured number without a '+' country code is a
+        # national number we cannot resolve to one person worldwide, so it is
+        # dropped (logged) rather than guessed at — a wrong-country identity key
+        # is worse than an unset one.
+        CustomerRepository(db).update_contact(
+            target_customer_id, email=email or None, full_name=name,
+            phone=(field_values or {}).get('phone'))
+
+        # Promote Visitor -> Lead; never downgrade an existing Customer.
+        _promote_to_lead(db, target_customer_id, session_id, page_url=page_url, channel=channel)
+
+        db.commit()
+    except Exception:
+        db.rollback()
+        raise
+    db.refresh(response)
+
+    # Post-commit so a job can never reference a rolled-back lead; wrapped so
+    # CRM problems can never break the conversation flow.
+    try:
+        from app.services.crm_sync import enqueue_crm_sync
+        enqueue_crm_sync(db, config, response)
+    except Exception as e:
+        logger.error(f"CRM sync enqueue failed (lead still recorded): {e}")
+
+    return response
+
+
+def _merge_customer(db: Session, source_id, target) -> None:
+    """Merge the (anonymous) source customer into the identified target: reassign
+    conversation history / sessions / lead responses, then mark the source row as
+    merged so it disappears from People and stale device tokens resolve to target."""
+    from app.models.customer import Customer
+    from app.models.chat_history import ChatHistory
+    from app.models.session_to_agent import SessionToAgent
+    from app.models.rating import Rating
+
+    source = db.query(Customer).filter(Customer.id == source_id).first()
+    if not source or source.id == target.id:
+        return
+
+    db.query(ChatHistory).filter(ChatHistory.customer_id == source_id).update(
+        {ChatHistory.customer_id: target.id}, synchronize_session=False)
+    db.query(SessionToAgent).filter(SessionToAgent.customer_id == source_id).update(
+        {SessionToAgent.customer_id: target.id}, synchronize_session=False)
+    db.query(LeadCaptureResponse).filter(LeadCaptureResponse.customer_id == source_id).update(
+        {LeadCaptureResponse.customer_id: target.id}, synchronize_session=False)
+    db.query(Rating).filter(Rating.customer_id == source_id).update(
+        {Rating.customer_id: target.id}, synchronize_session=False)
+
+    # Carry over a name the target lacks; keep the row for token resolution.
+    if not (target.full_name or "").strip() and (source.full_name or "").strip():
+        target.full_name = source.full_name
+    # Identifiers union on merge: the phone follows the surviving person. The
+    # source is cleared and FLUSHED before the target is set — the unique index
+    # on (organization_id, phone) is checked per-UPDATE, and SQLAlchemy orders
+    # same-mapper UPDATEs by primary key (random UUIDs), so without the flush
+    # the target could gain the phone while the source still holds it in the
+    # database, failing the whole merge intermittently.
+    #
+    # The clear is UNCONDITIONAL, even when the target already has a number and
+    # there is nowhere to move it to. A merged row is invisible in People and
+    # resolves to the target everywhere a human looks — but phone lookup is the
+    # FIRST thing get_or_create_customer tries, so a tombstone left holding a
+    # live number would silently capture that person's next WhatsApp message
+    # onto a row nobody can see or edit.
+    if source.phone:
+        moved_phone = source.phone
+        source.phone = None
+        db.flush()
+        if not target.phone:
+            target.phone = moved_phone
+        elif target.phone != moved_phone:
+            # Two real numbers, one column: the survivor keeps theirs. Logged
+            # because it is the one thing this merge cannot preserve — a
+            # customer_identifiers table is what would (see the plan's
+            # evolution path), not a second column.
+            logger.info(
+                f"Merge {source.id} -> {target.id}: dropped second phone "
+                f"{moved_phone} (target keeps {target.phone})"
+            )
+    source.merged_into_customer_id = target.id
+    source.is_active = False
+
+
+def _promote_to_lead(db: Session, customer_id, session_id, page_url: Optional[str] = None,
+                     channel: str = "widget") -> None:
+    from app.models.customer import Customer, LeadStage
+    customer = db.query(Customer).filter(Customer.id == customer_id).first()
+    if not customer:
+        return
+    if customer.lead_stage == LeadStage.VISITOR:
+        customer.lead_stage = LeadStage.LEAD
+        customer.lead_qualified_at = datetime.now(timezone.utc)
+    # Always keep the capture context fresh (also for repeat captures on a Lead).
+    source = dict(customer.lead_source or {})
+    source.setdefault("channel", channel)
+    source["captured_at"] = datetime.now(timezone.utc).isoformat()
+    source["session_id"] = str(session_id) if session_id else None
+    if page_url:
+        source["page_url"] = page_url[:500]
+    customer.lead_source = source
+
+
+def record_lead_from_response(
+    db: Session,
+    response,
+    organization_id,
+    agent_id,
+    customer_id,
+    session_id,
+    page_url: Optional[str] = None,
+    channel: str = "widget",
+    *,
+    scope: Optional[str],
+) -> Optional[LeadCaptureResponse]:
+    """Persist a lead from a ChatResponse whose request_lead_capture is set.
+
+    Shared by the widget socket handler and external channel processors so
+    every channel records leads identically: loads the agent's lead-capture
+    config, assembles lead_data from the explicit scalar fields (reliable
+    under strict structured outputs) plus any free-form lead_data dict, and
+    delegates validation/merge/promotion to record_lead_capture. Returns the
+    LeadCaptureResponse (whose customer_id may differ from the input when the
+    visitor was merged into an existing customer), or None if nothing was
+    recorded. Never raises.
+    """
+    try:
+        if not getattr(response, 'request_lead_capture', False):
+            return None
+        from app.repositories.agent import AgentRepository
+        agent = AgentRepository(db).get_agent(agent_id)
+        config = getattr(agent, 'lead_capture_config', None) if agent else None
+        if not (config and config.enabled) or has_captured_lead(db, customer_id, agent_id):
+            return None
+
+        lead_data = dict(getattr(response, 'lead_data', None) or {})
+        for key, attr in (('email', 'lead_email'), ('name', 'lead_name'),
+                          ('company', 'lead_company'), ('phone', 'lead_phone')):
+            val = getattr(response, attr, None)
+            if val and not lead_data.get(key):
+                lead_data[key] = val
+
+        return record_lead_capture(
+            db, config,
+            organization_id=organization_id,
+            agent_id=agent_id,
+            customer_id=customer_id,
+            session_id=session_id,
+            lead_data=lead_data,
+            summary=getattr(response, 'lead_summary', None),
+            consent=getattr(response, 'lead_consent', False),
+            page_url=page_url,
+            channel=channel,
+            scope=scope,
+        )
+    except Exception as e:
+        # Lead capture must never break the conversation flow.
+        logger.error(f"Lead-capture record error: {e}")
+        return None