feat: Ipad app production readiness, Colony orchestration, Social posting (#44)

#38 Ipad app production readiness, Colony orchestration, Social posting

Co-authored-by: Sayan Datta <sayan@Sayans-MacBook-Air.local>
Reviewed-on: sagnik/Project_Velocity#44

backend/api/routes_admin_surface.py

@@ -36,6 +36,7 @@ from backend.auth.dependencies import get_current_user
 logger = logging.getLogger("velocity.admin_surface")
 
 router = APIRouter()
+dashboard_router = APIRouter()
 
 # ── RBAC guard ────────────────────────────────────────────────────────────────
 
@@ -61,6 +62,128 @@ def _pool(request: Request):
     return pool
 
 
+class OfflineReplayAuditRecord(BaseModel):
+    id: str
+    kind: str
+    operation: str
+    targetId: str | None = None
+    queuedAt: str
+    attemptCount: int
+    lastAttemptAt: str | None = None
+    lastError: str | None = None
+
+
+class OfflineReplayAuditRequest(BaseModel):
+    records: list[OfflineReplayAuditRecord] = Field(default_factory=list)
+    pendingCount: int
+
+
+@dashboard_router.get("/metrics", summary="Canonical dashboard metrics for WebOS and iPad parity")
+async def get_dashboard_metrics(
+    request: Request,
+    user=Depends(get_current_user),
+):
+    pool = _pool(request)
+    tenant_id = user.tenant_id or "tenant_velocity"
+    async with pool.acquire() as conn:
+        row = await conn.fetchrow(
+            """
+            SELECT
+                (SELECT COUNT(*) FROM crm_leads WHERE tenant_id = $1)::int AS lead_count,
+                (
+                    SELECT COUNT(DISTINCT p.person_id)::int
+                    FROM crm_people p
+                    LEFT JOIN LATERAL (
+                        SELECT current_value
+                        FROM intel_qd_scores q
+                        WHERE q.person_id = p.person_id
+                        ORDER BY q.computed_at DESC
+                        LIMIT 1
+                    ) q ON TRUE
+                    WHERE p.tenant_id = $1
+                      AND (
+                        COALESCE(p.buyer_type, '') ILIKE '%whale%'
+                        OR COALESCE(q.current_value, 0) >= 0.90
+                      )
+                ) AS whale_lead_count,
+                (SELECT COUNT(*) FROM inventory_properties WHERE tenant_id = $1 AND status <> 'archived')::int AS property_count,
+                (
+                    SELECT COUNT(*)
+                    FROM user_calendar_events
+                    WHERE tenant_id = $1
+                      AND owner_user_id = $2
+                      AND status NOT IN ('cancelled', 'done')
+                      AND start_at >= date_trunc('day', NOW())
+                      AND start_at < date_trunc('day', NOW()) + INTERVAL '1 day'
+                )::int AS today_calendar_count,
+                (
+                    SELECT COUNT(*)
+                    FROM intel_reminders
+                    WHERE COALESCE(tenant_id, $1) = $1
+                      AND status IN ('pending', 'open', 'scheduled', 'snoozed', 'confirmed')
+                )::int AS pending_task_count,
+                (
+                    SELECT COUNT(*)
+                    FROM intel_reminders
+                    WHERE COALESCE(tenant_id, $1) = $1
+                      AND status IN ('pending', 'open', 'scheduled', 'snoozed', 'confirmed')
+                      AND priority IN ('urgent', 'high')
+                )::int AS urgent_task_count,
+                (SELECT COUNT(*) FROM insight_recommendations WHERE tenant_id = $1 AND status = 'pending')::int AS pending_insights,
+                (SELECT COUNT(*) FROM edge_transcription_jobs WHERE tenant_id = $1 AND status = 'pending')::int AS pending_transcriptions
+            """,
+            tenant_id,
+            user.user_id,
+        )
+    return {
+        "status": "ok",
+        "data": {
+            "leadCount": row["lead_count"],
+            "whaleLeadCount": row["whale_lead_count"],
+            "propertyCount": row["property_count"],
+            "todayCalendarCount": row["today_calendar_count"],
+            "pendingTaskCount": row["pending_task_count"],
+            "urgentTaskCount": row["urgent_task_count"],
+            "pendingInsights": row["pending_insights"],
+            "pendingTranscriptions": row["pending_transcriptions"],
+        },
+        "meta": {"generatedAt": datetime.now(timezone.utc).isoformat()},
+    }
+
+
+@dashboard_router.post("/offline-replay/audit", summary="Publish native offline replay queue observability")
+async def publish_offline_replay_audit(
+    body: OfflineReplayAuditRequest,
+    request: Request,
+    user=Depends(get_current_user),
+):
+    pool = _pool(request)
+    async with pool.acquire() as conn:
+        await conn.execute(
+            """
+            CREATE TABLE IF NOT EXISTS mobile_offline_replay_audits (
+                audit_id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
+                tenant_id TEXT NOT NULL,
+                user_id TEXT NOT NULL,
+                pending_count INT NOT NULL,
+                records JSONB NOT NULL,
+                created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+            )
+            """
+        )
+        await conn.execute(
+            """
+            INSERT INTO mobile_offline_replay_audits (tenant_id, user_id, pending_count, records)
+            VALUES ($1, $2, $3, $4::jsonb)
+            """,
+            user.tenant_id,
+            user.user_id,
+            body.pendingCount,
+            json.dumps([record.model_dump() for record in body.records]),
+        )
+    return {"status": "ok", "pendingCount": body.pendingCount}
+
+
 # ── Pydantic Models ───────────────────────────────────────────────────────────
 
 VALID_ACTION_TYPES = {

backend/api/routes_catalyst.py

@@ -10,6 +10,8 @@ Routes:
   POST /api/catalyst/auth/meta             — OAuth token acquisition
 """
 
+from __future__ import annotations
+
 import os
 import uuid
 import hashlib
@@ -17,9 +19,11 @@ import logging
 from typing import Any
 from datetime import datetime
 
-from fastapi import APIRouter, HTTPException, Query, Request, status
+import httpx
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
 from pydantic import BaseModel, Field
 
+from backend.auth.dependencies import UserPrincipal, get_current_user
 from backend.services.ad_network_service import (
     AdInsight,
     BidStrategyUpdate,
@@ -27,6 +31,17 @@ from backend.services.ad_network_service import (
     Platform,
     ad_network_service,
 )
+from backend.services.social_posting import (
+    PostRequest,
+    PostStatus,
+    SocialPlatform,
+    SocialPostingConfigurationError,
+    SocialPostingError,
+    get_post,
+    list_posts,
+    publish_content,
+    publish_due_scheduled,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -91,6 +106,13 @@ def _ok(data: Any, meta: dict | None = None) -> dict:
     return {"status": "ok", "data": data, "meta": meta or {}}
 
 
+def _get_db_pool(request: Request) -> Any:
+    pool = getattr(request.app.state, "db_pool", None)
+    if pool is None:
+        raise HTTPException(status_code=503, detail="Database unavailable.")
+    return pool
+
+
 def _sha256_hash(value: str) -> str:
     """SHA-256 hash an email for Meta's hashed audience upload."""
     return hashlib.sha256(value.strip().lower().encode()).hexdigest()
@@ -510,3 +532,91 @@ async def meta_oauth(payload: MetaAuthRequest) -> dict:
         "token_type":   token_data.get("token_type", "bearer"),
         "expires_in":   token_data.get("expires_in"),
     })
+
+
+# ── 6. Social publishing ─────────────────────────────────────────────────────
+
+@router.post("/publish", status_code=status.HTTP_201_CREATED, summary="Publish or schedule content to social channels")
+async def api_publish_content(
+    payload: PostRequest,
+    request: Request,
+    user: UserPrincipal = Depends(get_current_user),
+) -> dict:
+    try:
+        result = await publish_content(
+            pool=_get_db_pool(request),
+            tenant_id=user.tenant_id,
+            actor_id=user.user_id,
+            payload=payload,
+        )
+    except SocialPostingConfigurationError as exc:
+        raise HTTPException(status_code=503, detail=str(exc)) from exc
+    except ValueError as exc:
+        raise HTTPException(status_code=422, detail=f"Invalid schedule_time: {exc}") from exc
+    except (SocialPostingError, httpx.HTTPError) as exc:
+        raise HTTPException(status_code=502, detail=str(exc)) from exc
+    return _ok(result)
+
+
+@router.get("/posts", summary="List tenant-scoped social posts")
+async def api_list_social_posts(
+    request: Request,
+    platform: SocialPlatform | None = Query(default=None),
+    post_status: PostStatus | None = Query(default=None, alias="status"),
+    limit: int = Query(default=50, ge=1, le=200),
+    user: UserPrincipal = Depends(get_current_user),
+) -> dict:
+    posts = await list_posts(
+        pool=_get_db_pool(request),
+        tenant_id=user.tenant_id,
+        platform=platform,
+        status=post_status,
+        limit=limit,
+    )
+    return _ok(posts, meta={"count": len(posts)})
+
+
+@router.get("/posts/{post_id}", summary="Get a tenant-scoped social post")
+async def api_get_social_post(
+    post_id: str,
+    request: Request,
+    user: UserPrincipal = Depends(get_current_user),
+) -> dict:
+    post = await get_post(pool=_get_db_pool(request), tenant_id=user.tenant_id, post_id=post_id)
+    if post is None:
+        raise HTTPException(status_code=404, detail=f"Social post '{post_id}' not found.")
+    return _ok(post)
+
+
+@router.get("/scheduled", summary="List scheduled social posts for the authenticated tenant")
+async def api_scheduled_posts(
+    request: Request,
+    limit: int = Query(default=50, ge=1, le=200),
+    user: UserPrincipal = Depends(get_current_user),
+) -> dict:
+    posts = await list_posts(
+        pool=_get_db_pool(request),
+        tenant_id=user.tenant_id,
+        status=PostStatus.SCHEDULED,
+        limit=limit,
+    )
+    return _ok(posts, meta={"count": len(posts)})
+
+
+@router.post("/scheduled/publish-due", summary="Publish due scheduled social posts for the authenticated tenant")
+async def api_publish_due_scheduled(
+    request: Request,
+    limit: int = Query(default=20, ge=1, le=100),
+    user: UserPrincipal = Depends(get_current_user),
+) -> dict:
+    try:
+        result = await publish_due_scheduled(
+            pool=_get_db_pool(request),
+            tenant_id=user.tenant_id,
+            limit=limit,
+        )
+    except SocialPostingConfigurationError as exc:
+        raise HTTPException(status_code=503, detail=str(exc)) from exc
+    except (SocialPostingError, httpx.HTTPError) as exc:
+        raise HTTPException(status_code=502, detail=str(exc)) from exc
+    return _ok(result)

backend/api/routes_colony.py

@@ -0,0 +1,251 @@
+from __future__ import annotations
+
+import uuid
+from typing import Any, Literal
+
+import httpx
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
+from pydantic import BaseModel, Field
+
+from backend.auth.dependencies import UserPrincipal, get_current_user
+from backend.services.colony_gateway import ColonyConfigurationError, ColonyGateway, ColonyGatewayError
+from backend.services.colony_repository import ColonyRepository
+
+router = APIRouter()
+
+MissionType = Literal["oracle_advisory", "crm_lead_intelligence", "catalyst_strategy_brief"]
+RiskLevel = Literal["low", "medium", "high"]
+SensitivityClass = Literal["public", "internal", "confidential"]
+
+
+class MissionCreateRequest(BaseModel):
+    mission_type: MissionType
+    user_goal: str = Field(..., min_length=1, max_length=2000)
+    normalized_goal: str | None = Field(default=None, max_length=2000)
+    origin_surface: str = Field(default="api", min_length=1, max_length=128)
+    actor_role: str | None = Field(default=None, max_length=128)
+    risk_level: RiskLevel = "low"
+    sensitivity_class: SensitivityClass = "internal"
+    time_budget_ms: int = Field(default=30000, gt=0, le=300000)
+    token_budget: int = Field(default=4096, gt=0, le=200000)
+    context_refs: dict[str, Any] = Field(default_factory=dict)
+    requested_outputs: list[str] = Field(default_factory=list)
+    payload: dict[str, Any] = Field(default_factory=dict)
+
+
+class ApprovalRequest(BaseModel):
+    reason: str | None = Field(default=None, max_length=2000)
+
+
+def _get_repo(request: Request) -> ColonyRepository:
+    pool = getattr(request.app.state, "db_pool", None)
+    if pool is None:
+        raise HTTPException(status_code=503, detail="Database unavailable.")
+    return ColonyRepository(pool)
+
+
+def _serialize_mission(row: dict[str, Any], dispatch: dict[str, Any] | None = None) -> dict[str, Any]:
+    data = {
+        "mission_id": str(row["mission_id"]),
+        "tenant_id": row["tenant_id"],
+        "mission_type": row["mission_type"],
+        "origin_surface": row["origin_surface"],
+        "actor_id": row["actor_id"],
+        "actor_role": row["actor_role"],
+        "risk_level": row["risk_level"],
+        "sensitivity_class": row["sensitivity_class"],
+        "status": row["status"],
+        "review_status": row["review_status"],
+        "time_budget_ms": row["time_budget_ms"],
+        "token_budget": row["token_budget"],
+        "user_goal": row["user_goal"],
+        "normalized_goal": row["normalized_goal"],
+        "context_refs": row["context_refs"] or {},
+        "requested_outputs": row["requested_outputs"] or [],
+        "payload": row["payload"] or {},
+        "created_at": row["created_at"].isoformat() if row.get("created_at") else None,
+        "updated_at": row["updated_at"].isoformat() if row.get("updated_at") else None,
+        "completed_at": row["completed_at"].isoformat() if row.get("completed_at") else None,
+    }
+    if dispatch is not None:
+        data["dispatch"] = dispatch
+    return data
+
+
+@router.post("/missions", status_code=status.HTTP_201_CREATED, summary="Create and dispatch a colony mission")
+async def create_mission(
+    body: MissionCreateRequest,
+    request: Request,
+    user: UserPrincipal = Depends(get_current_user),
+) -> dict[str, Any]:
+    repo = _get_repo(request)
+    try:
+        gateway = ColonyGateway()
+    except ColonyConfigurationError as exc:
+        raise HTTPException(status_code=503, detail=str(exc)) from exc
+
+    mission_id = str(uuid.uuid4())
+    mission = {
+        "mission_id": mission_id,
+        "mission_type": body.mission_type,
+        "origin_surface": body.origin_surface,
+        "tenant_id": user.tenant_id,
+        "actor_id": user.user_id,
+        "actor_role": body.actor_role or user.role,
+        "risk_level": body.risk_level,
+        "sensitivity_class": body.sensitivity_class,
+        "time_budget_ms": body.time_budget_ms,
+        "token_budget": body.token_budget,
+        "user_goal": body.user_goal,
+        "normalized_goal": body.normalized_goal or body.user_goal,
+        "context_refs": body.context_refs,
+        "requested_outputs": body.requested_outputs,
+        "payload": body.payload,
+    }
+    row = await repo.create_mission(mission)
+    await repo.log_event(
+        mission_id=mission_id,
+        tenant_id=user.tenant_id,
+        event_type="mission_created",
+        actor=user.user_id,
+        detail={"mission_type": body.mission_type},
+    )
+
+    try:
+        dispatch = await gateway.dispatch_mission(mission)
+    except ColonyGatewayError as exc:
+        failed = await repo.update_status(mission_id, user.tenant_id, "dispatch_failed")
+        await repo.log_event(
+            mission_id=mission_id,
+            tenant_id=user.tenant_id,
+            event_type="mission_dispatch_failed",
+            actor=user.user_id,
+            detail={"error": str(exc)},
+        )
+        raise HTTPException(
+            status_code=502,
+            detail={
+                "message": str(exc),
+                "mission": _serialize_mission(failed or row),
+            },
+        ) from exc
+
+    queued = await repo.update_status(mission_id, user.tenant_id, "queued")
+    await repo.log_event(
+        mission_id=mission_id,
+        tenant_id=user.tenant_id,
+        event_type="mission_dispatched",
+        actor=user.user_id,
+        detail={"dispatch": dispatch},
+    )
+    return {"status": "ok", "data": _serialize_mission(queued or row, dispatch=dispatch)}
+
+
+@router.get("/missions", summary="List colony missions for the authenticated tenant")
+async def list_missions(
+    request: Request,
+    limit: int = Query(default=50, ge=1, le=200),
+    offset: int = Query(default=0, ge=0),
+    user: UserPrincipal = Depends(get_current_user),
+) -> dict[str, Any]:
+    rows = await _get_repo(request).list_missions(user.tenant_id, limit=limit, offset=offset)
+    return {"status": "ok", "data": [_serialize_mission(row) for row in rows], "meta": {"count": len(rows)}}
+
+
+@router.get("/missions/{mission_id}", summary="Get a colony mission")
+async def get_mission(
+    mission_id: str,
+    request: Request,
+    user: UserPrincipal = Depends(get_current_user),
+) -> dict[str, Any]:
+    row = await _get_repo(request).get_mission(mission_id, user.tenant_id)
+    if row is None:
+        raise HTTPException(status_code=404, detail=f"Mission '{mission_id}' not found.")
+    return {"status": "ok", "data": _serialize_mission(row)}
+
+
+@router.get("/missions/{mission_id}/artifacts", summary="Get mission tasks, results, and writeback proposals")
+async def get_artifacts(
+    mission_id: str,
+    request: Request,
+    user: UserPrincipal = Depends(get_current_user),
+) -> dict[str, Any]:
+    artifacts = await _get_repo(request).artifacts(mission_id, user.tenant_id)
+    if not artifacts:
+        raise HTTPException(status_code=404, detail=f"Mission '{mission_id}' not found.")
+    return {"status": "ok", "data": artifacts}
+
+
+@router.post("/missions/{mission_id}/approve", summary="Approve all pending writeback proposals for a mission")
+async def approve_writebacks(
+    mission_id: str,
+    body: ApprovalRequest,
+    request: Request,
+    user: UserPrincipal = Depends(get_current_user),
+) -> dict[str, Any]:
+    del body
+    repo = _get_repo(request)
+    if await repo.get_mission(mission_id, user.tenant_id) is None:
+        raise HTTPException(status_code=404, detail=f"Mission '{mission_id}' not found.")
+    count = await repo.approve_pending_writebacks(mission_id, user.tenant_id, user.user_id)
+    if count == 0:
+        raise HTTPException(status_code=404, detail="No pending writeback proposals.")
+    await repo.log_event(
+        mission_id=mission_id,
+        tenant_id=user.tenant_id,
+        event_type="writeback_approved",
+        actor=user.user_id,
+        detail={"approved": count},
+    )
+    return {"status": "ok", "data": {"approved": count}}
+
+
+@router.post("/missions/{mission_id}/reject", summary="Reject all pending writeback proposals for a mission")
+async def reject_writebacks(
+    mission_id: str,
+    body: ApprovalRequest,
+    request: Request,
+    user: UserPrincipal = Depends(get_current_user),
+) -> dict[str, Any]:
+    repo = _get_repo(request)
+    if await repo.get_mission(mission_id, user.tenant_id) is None:
+        raise HTTPException(status_code=404, detail=f"Mission '{mission_id}' not found.")
+    count = await repo.reject_pending_writebacks(
+        mission_id,
+        user.tenant_id,
+        user.user_id,
+        body.reason or "Rejected by operator.",
+    )
+    if count == 0:
+        raise HTTPException(status_code=404, detail="No pending writeback proposals.")
+    await repo.log_event(
+        mission_id=mission_id,
+        tenant_id=user.tenant_id,
+        event_type="writeback_rejected",
+        actor=user.user_id,
+        detail={"rejected": count, "reason": body.reason},
+    )
+    return {"status": "ok", "data": {"rejected": count}}
+
+
+@router.get("/health", summary="Check colony root persistence and orchestrator connectivity")
+async def colony_health(
+    request: Request,
+    user: UserPrincipal = Depends(get_current_user),
+) -> dict[str, Any]:
+    repo = _get_repo(request)
+    try:
+        gateway = ColonyGateway()
+        service = await gateway.health()
+    except (ColonyConfigurationError, ColonyGatewayError, httpx.HTTPError) as exc:  # type: ignore[name-defined]
+        raise HTTPException(status_code=503, detail=str(exc)) from exc
+    rows = await repo.list_missions(user.tenant_id, limit=1, offset=0)
+    return {
+        "status": "ok",
+        "data": {
+            "tenant_id": user.tenant_id,
+            "root_db": "connected",
+            "orchestrator": service,
+            "has_missions": bool(rows),
+        },
+    }

backend/api/routes_comms.py

@@ -18,7 +18,7 @@ from pydantic import BaseModel
 
 from backend.auth.dependencies import UserPrincipal, get_current_user
 from backend.services.comms_evolution_provider import EvolutionProvider
-from backend.services.comms_ingest import ingest_inbound_message
+from backend.services.comms_ingest import TranscriptionError, ingest_inbound_message, transcribe_recording as run_transcription
 from backend.services.comms_provider import MockProvider
 from backend.services.comms_waha_provider import WahaProvider
 
@@ -46,6 +46,8 @@ class NoteBody(BaseModel):
 class TaskBody(BaseModel):
     title: str
     dueAt: str | None = None
+    notes: str | None = None
+    priority: str = "normal"
 
 
 class SettingsPatch(BaseModel):
@@ -158,6 +160,37 @@ def _record_value(row: Any, key: str, default: Any = None) -> Any:
         return default
 
 
+def _optional_datetime(value: str | None) -> datetime | None:
+    if not value or not value.strip():
+        return None
+    normalized = value.strip().replace("Z", "+00:00")
+    try:
+        return datetime.fromisoformat(normalized)
+    except ValueError as exc:
+        raise HTTPException(status_code=422, detail="dueAt must be an ISO-8601 timestamp.") from exc
+
+
+async def _thread_context(conn, thread_id: str, tenant_id: str):
+    thread = await conn.fetchrow("SELECT * FROM comms_threads WHERE thread_id = $1::uuid", thread_id)
+    if not thread:
+        raise HTTPException(status_code=404, detail="Thread not found")
+    lead_id = None
+    if thread["person_id"]:
+        lead_id = await conn.fetchval(
+            """
+            SELECT lead_id
+            FROM crm_leads
+            WHERE person_id = $1::uuid
+              AND tenant_id = $2
+            ORDER BY updated_at DESC NULLS LAST, created_at DESC NULLS LAST
+            LIMIT 1
+            """,
+            thread["person_id"],
+            tenant_id,
+        )
+    return thread, lead_id
+
+
 async def _ensure_schema(pool) -> None:
     global _SCHEMA_READY
     if _SCHEMA_READY:
@@ -182,6 +215,19 @@ async def _ensure_schema(pool) -> None:
                 created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
                 updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
             );
+            ALTER TABLE comms_threads ADD COLUMN IF NOT EXISTS provider TEXT NOT NULL DEFAULT 'mock';
+            ALTER TABLE comms_threads ADD COLUMN IF NOT EXISTS external_thread_id TEXT;
+            ALTER TABLE comms_threads ADD COLUMN IF NOT EXISTS person_id UUID NULL REFERENCES crm_people(person_id) ON DELETE SET NULL;
+            ALTER TABLE comms_threads ADD COLUMN IF NOT EXISTS phone_e164 TEXT NOT NULL DEFAULT '';
+            ALTER TABLE comms_threads ADD COLUMN IF NOT EXISTS display_name TEXT;
+            ALTER TABLE comms_threads ADD COLUMN IF NOT EXISTS channel TEXT NOT NULL DEFAULT 'whatsapp';
+            ALTER TABLE comms_threads ADD COLUMN IF NOT EXISTS status TEXT NOT NULL DEFAULT 'open';
+            ALTER TABLE comms_threads ADD COLUMN IF NOT EXISTS assigned_user_id UUID NULL;
+            ALTER TABLE comms_threads ADD COLUMN IF NOT EXISTS last_message_at TIMESTAMPTZ;
+            ALTER TABLE comms_threads ADD COLUMN IF NOT EXISTS unread_count INT NOT NULL DEFAULT 0;
+            ALTER TABLE comms_threads ADD COLUMN IF NOT EXISTS metadata_json JSONB NOT NULL DEFAULT '{}'::jsonb;
+            ALTER TABLE comms_threads ADD COLUMN IF NOT EXISTS created_at TIMESTAMPTZ NOT NULL DEFAULT NOW();
+            ALTER TABLE comms_threads ADD COLUMN IF NOT EXISTS updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW();
             CREATE INDEX IF NOT EXISTS idx_comms_threads_phone_provider ON comms_threads(provider, phone_e164);
             CREATE INDEX IF NOT EXISTS idx_comms_threads_person ON comms_threads(person_id) WHERE person_id IS NOT NULL;
             CREATE INDEX IF NOT EXISTS idx_comms_threads_status ON comms_threads(status);
@@ -203,6 +249,19 @@ async def _ensure_schema(pool) -> None:
                 raw_payload JSONB NOT NULL DEFAULT '{}'::jsonb,
                 created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
             );
+            ALTER TABLE comms_messages ADD COLUMN IF NOT EXISTS provider TEXT NOT NULL DEFAULT 'mock';
+            ALTER TABLE comms_messages ADD COLUMN IF NOT EXISTS external_message_id TEXT;
+            ALTER TABLE comms_messages ADD COLUMN IF NOT EXISTS direction TEXT NOT NULL DEFAULT 'system';
+            ALTER TABLE comms_messages ADD COLUMN IF NOT EXISTS message_type TEXT NOT NULL DEFAULT 'text';
+            ALTER TABLE comms_messages ADD COLUMN IF NOT EXISTS body TEXT NOT NULL DEFAULT '';
+            ALTER TABLE comms_messages ADD COLUMN IF NOT EXISTS media_url TEXT;
+            ALTER TABLE comms_messages ADD COLUMN IF NOT EXISTS media_mime_type TEXT;
+            ALTER TABLE comms_messages ADD COLUMN IF NOT EXISTS delivery_status TEXT NOT NULL DEFAULT 'pending';
+            ALTER TABLE comms_messages ADD COLUMN IF NOT EXISTS sent_at TIMESTAMPTZ;
+            ALTER TABLE comms_messages ADD COLUMN IF NOT EXISTS delivered_at TIMESTAMPTZ;
+            ALTER TABLE comms_messages ADD COLUMN IF NOT EXISTS read_at TIMESTAMPTZ;
+            ALTER TABLE comms_messages ADD COLUMN IF NOT EXISTS raw_payload JSONB NOT NULL DEFAULT '{}'::jsonb;
+            ALTER TABLE comms_messages ADD COLUMN IF NOT EXISTS created_at TIMESTAMPTZ NOT NULL DEFAULT NOW();
             CREATE INDEX IF NOT EXISTS idx_comms_messages_thread ON comms_messages(thread_id, created_at DESC);
             CREATE INDEX IF NOT EXISTS idx_comms_messages_external ON comms_messages(external_message_id) WHERE external_message_id IS NOT NULL;
             CREATE TABLE IF NOT EXISTS comms_call_logs (
@@ -223,6 +282,21 @@ async def _ensure_schema(pool) -> None:
                 raw_payload JSONB NOT NULL DEFAULT '{}'::jsonb,
                 created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
             );
+            ALTER TABLE comms_call_logs ADD COLUMN IF NOT EXISTS thread_id UUID NULL REFERENCES comms_threads(thread_id) ON DELETE SET NULL;
+            ALTER TABLE comms_call_logs ADD COLUMN IF NOT EXISTS person_id UUID NULL REFERENCES crm_people(person_id) ON DELETE SET NULL;
+            ALTER TABLE comms_call_logs ADD COLUMN IF NOT EXISTS provider TEXT NOT NULL DEFAULT 'mock';
+            ALTER TABLE comms_call_logs ADD COLUMN IF NOT EXISTS external_call_id TEXT;
+            ALTER TABLE comms_call_logs ADD COLUMN IF NOT EXISTS phone_e164 TEXT NOT NULL DEFAULT '';
+            ALTER TABLE comms_call_logs ADD COLUMN IF NOT EXISTS direction TEXT NOT NULL DEFAULT 'inbound';
+            ALTER TABLE comms_call_logs ADD COLUMN IF NOT EXISTS status TEXT NOT NULL DEFAULT 'completed';
+            ALTER TABLE comms_call_logs ADD COLUMN IF NOT EXISTS started_at TIMESTAMPTZ NOT NULL DEFAULT NOW();
+            ALTER TABLE comms_call_logs ADD COLUMN IF NOT EXISTS ended_at TIMESTAMPTZ;
+            ALTER TABLE comms_call_logs ADD COLUMN IF NOT EXISTS duration_seconds INT;
+            ALTER TABLE comms_call_logs ADD COLUMN IF NOT EXISTS recording_url TEXT;
+            ALTER TABLE comms_call_logs ADD COLUMN IF NOT EXISTS transcript_id UUID;
+            ALTER TABLE comms_call_logs ADD COLUMN IF NOT EXISTS transcript_text TEXT;
+            ALTER TABLE comms_call_logs ADD COLUMN IF NOT EXISTS raw_payload JSONB NOT NULL DEFAULT '{}'::jsonb;
+            ALTER TABLE comms_call_logs ADD COLUMN IF NOT EXISTS created_at TIMESTAMPTZ NOT NULL DEFAULT NOW();
             CREATE INDEX IF NOT EXISTS idx_comms_call_logs_phone ON comms_call_logs(phone_e164);
             CREATE INDEX IF NOT EXISTS idx_comms_call_logs_thread ON comms_call_logs(thread_id) WHERE thread_id IS NOT NULL;
             CREATE TABLE IF NOT EXISTS comms_settings (
@@ -422,6 +496,54 @@ async def list_messages(
     return {"messages": messages, "thread": await get_thread(thread_id, request)}
 
 
+@router.get("/threads/{thread_id}/calls")
+async def list_thread_calls(
+    thread_id: str,
+    request: Request,
+    limit: int = 25,
+    offset: int = 0,
+    _: UserPrincipal = Depends(get_current_user),
+):
+    pool = await _pool(request)
+    limit = max(1, min(limit, 100))
+    offset = max(0, offset)
+    async with pool.acquire() as conn:
+        rows = await conn.fetch(
+            """
+            SELECT *
+            FROM comms_call_logs
+            WHERE thread_id = $1::uuid
+            ORDER BY started_at DESC, created_at DESC
+            LIMIT $2 OFFSET $3
+            """,
+            thread_id,
+            limit,
+            offset,
+        )
+    calls = [
+        {
+            "callId": str(row["call_id"]),
+            "threadId": str(row["thread_id"]) if row["thread_id"] else None,
+            "personId": str(row["person_id"]) if row["person_id"] else None,
+            "provider": row["provider"],
+            "externalCallId": row["external_call_id"],
+            "phoneE164": row["phone_e164"],
+            "direction": row["direction"],
+            "status": row["status"],
+            "startedAt": row["started_at"].isoformat(),
+            "endedAt": row["ended_at"].isoformat() if row["ended_at"] else None,
+            "durationSeconds": row["duration_seconds"],
+            "recordingUrl": row["recording_url"],
+            "transcriptId": str(row["transcript_id"]) if row["transcript_id"] else None,
+            "transcriptText": row["transcript_text"],
+            "rawPayload": _json_obj(row["raw_payload"]),
+            "createdAt": row["created_at"].isoformat(),
+        }
+        for row in rows
+    ]
+    return {"calls": calls, "thread": await get_thread(thread_id, request)}
+
+
 @router.post("/threads/{thread_id}/messages")
 async def send_message(
     thread_id: str,
@@ -465,11 +587,15 @@ async def link_person(
     thread_id: str,
     body: LinkPersonBody,
     request: Request,
-    _: UserPrincipal = Depends(get_current_user),
+    user: UserPrincipal = Depends(get_current_user),
 ):
     pool = await _pool(request)
     async with pool.acquire() as conn:
-        exists = await conn.fetchval("SELECT EXISTS (SELECT 1 FROM crm_people WHERE person_id = $1::uuid)", body.personId)
+        exists = await conn.fetchval(
+            "SELECT EXISTS (SELECT 1 FROM crm_people WHERE person_id = $1::uuid AND tenant_id = $2)",
+            body.personId,
+            user.tenant_id,
+        )
         if not exists:
             raise HTTPException(status_code=404, detail="CRM person not found")
         updated = await conn.execute(
@@ -481,36 +607,127 @@ async def link_person(
 
 
 @router.post("/threads/{thread_id}/notes")
-async def add_note(thread_id: str, body: NoteBody, request: Request, _: UserPrincipal = Depends(get_current_user)):
+async def add_note(thread_id: str, body: NoteBody, request: Request, user: UserPrincipal = Depends(get_current_user)):
     pool = await _pool(request)
     async with pool.acquire() as conn:
-        msg_id = await conn.fetchval(
-            """
-            INSERT INTO comms_messages (thread_id, provider, direction, message_type, body, delivery_status)
-            VALUES ($1::uuid, 'system', 'system', 'text', $2, 'delivered')
-            RETURNING message_id
-            """,
-            thread_id,
-            f"Note: {body.content}",
-        )
-    return {"messageId": str(msg_id)}
+        async with conn.transaction():
+            thread, lead_id = await _thread_context(conn, thread_id, user.tenant_id)
+            msg_id = await conn.fetchval(
+                """
+                INSERT INTO comms_messages (thread_id, provider, direction, message_type, body, delivery_status)
+                VALUES ($1::uuid, 'system', 'system', 'text', $2, 'delivered')
+                RETURNING message_id
+                """,
+                thread_id,
+                f"Note: {body.content}",
+            )
+            interaction_id = None
+            canonical_message_id = None
+            if thread["person_id"]:
+                interaction_id = await conn.fetchval(
+                    """
+                    INSERT INTO intel_interactions (
+                        interaction_id, tenant_id, person_id, lead_id, channel,
+                        interaction_type, happened_at, summary, source_ref, metadata_json
+                    ) VALUES (
+                        gen_random_uuid(), COALESCE($1, 'tenant_velocity'), $2::uuid, $3::uuid, 'whatsapp',
+                        'operator_note', NOW(), $4, $5, $6::jsonb
+                    )
+                    RETURNING interaction_id
+                    """,
+                    user.tenant_id,
+                    thread["person_id"],
+                    lead_id,
+                    body.content,
+                    f"comms:{thread_id}",
+                    json.dumps({"source": "comms_thread_note", "thread_id": thread_id, "message_id": str(msg_id)}),
+                )
+                canonical_message_id = await conn.fetchval(
+                    """
+                    INSERT INTO intel_messages (
+                        message_id, interaction_id, thread_id, sender_role, sender_name,
+                        message_text, delivered_at, metadata_json
+                    ) VALUES (
+                        gen_random_uuid(), $1::uuid, $2::uuid, 'operator', 'iPad operator',
+                        $3, NOW(), $4::jsonb
+                    )
+                    RETURNING message_id
+                    """,
+                    interaction_id,
+                    thread_id,
+                    body.content,
+                    json.dumps({"source": "comms_thread_note"}),
+                )
+    return {
+        "messageId": str(msg_id),
+        "canonicalInteractionId": str(interaction_id) if interaction_id else None,
+        "canonicalMessageId": str(canonical_message_id) if canonical_message_id else None,
+    }
 
 
 @router.post("/threads/{thread_id}/tasks")
-async def add_task(thread_id: str, body: TaskBody, request: Request, _: UserPrincipal = Depends(get_current_user)):
+async def add_task(thread_id: str, body: TaskBody, request: Request, user: UserPrincipal = Depends(get_current_user)):
     pool = await _pool(request)
     text = f"Task: {body.title}" + (f" (Due: {body.dueAt})" if body.dueAt else "")
     async with pool.acquire() as conn:
-        msg_id = await conn.fetchval(
-            """
-            INSERT INTO comms_messages (thread_id, provider, direction, message_type, body, delivery_status)
-            VALUES ($1::uuid, 'system', 'system', 'text', $2, 'delivered')
-            RETURNING message_id
-            """,
-            thread_id,
-            text,
-        )
-    return {"messageId": str(msg_id)}
+        async with conn.transaction():
+            thread, lead_id = await _thread_context(conn, thread_id, user.tenant_id)
+            msg_id = await conn.fetchval(
+                """
+                INSERT INTO comms_messages (thread_id, provider, direction, message_type, body, delivery_status)
+                VALUES ($1::uuid, 'system', 'system', 'text', $2, 'delivered')
+                RETURNING message_id
+                """,
+                thread_id,
+                text,
+            )
+            reminder_id = None
+            interaction_id = None
+            if thread["person_id"]:
+                interaction_id = await conn.fetchval(
+                    """
+                    INSERT INTO intel_interactions (
+                        interaction_id, tenant_id, person_id, lead_id, channel,
+                        interaction_type, happened_at, summary, source_ref, metadata_json
+                    ) VALUES (
+                        gen_random_uuid(), COALESCE($1, 'tenant_velocity'), $2::uuid, $3::uuid, 'whatsapp',
+                        'next_best_action', NOW(), $4, $5, $6::jsonb
+                    )
+                    RETURNING interaction_id
+                    """,
+                    user.tenant_id,
+                    thread["person_id"],
+                    lead_id,
+                    body.title,
+                    f"comms:{thread_id}",
+                    json.dumps({"source": "comms_thread_task", "thread_id": thread_id, "message_id": str(msg_id)}),
+                )
+                reminder_id = await conn.fetchval(
+                    """
+                    INSERT INTO intel_reminders (
+                        reminder_id, tenant_id, person_id, lead_id, interaction_id,
+                        reminder_type, title, notes, due_at, status, priority,
+                        created_by_type, created_at
+                    ) VALUES (
+                        gen_random_uuid(), COALESCE($1, 'tenant_velocity'), $2::uuid, $3::uuid, $4::uuid,
+                        'follow_up', $5, $6, $7, 'pending', $8, 'human', NOW()
+                    )
+                    RETURNING reminder_id
+                    """,
+                    user.tenant_id,
+                    thread["person_id"],
+                    lead_id,
+                    interaction_id,
+                    body.title,
+                    body.notes,
+                    _optional_datetime(body.dueAt),
+                    body.priority,
+                )
+    return {
+        "messageId": str(msg_id),
+        "canonicalInteractionId": str(interaction_id) if interaction_id else None,
+        "canonicalReminderId": str(reminder_id) if reminder_id else None,
+    }
 
 
 @router.post("/webhooks/{provider}")
@@ -572,17 +789,53 @@ async def test_provider(request: Request, _: UserPrincipal = Depends(get_current
 @router.post("/recordings/transcribe")
 async def transcribe_recording(body: TranscribeBody, request: Request, _: UserPrincipal = Depends(get_current_user)):
     pool = await _pool(request)
+    config = await _load_config(pool)
+    configured_provider = str(config.get("transcription_provider") or "").strip().lower()
+    env_provider = os.getenv("COMMS_TRANSCRIPTION_PROVIDER", "none").strip().lower()
+    provider = env_provider if configured_provider in {"", "none", "disabled"} else configured_provider
+    recording_url = body.recordingUrl
+    if body.callId and not recording_url:
+        async with pool.acquire() as conn:
+            recording_url = await conn.fetchval(
+                "SELECT recording_url FROM comms_call_logs WHERE call_id = $1::uuid",
+                body.callId,
+            )
+    if not recording_url:
+        raise HTTPException(status_code=422, detail="recordingUrl is required when callId has no stored recording_url.")
+
+    try:
+        result = await run_transcription(recording_url, provider=provider)
+    except TranscriptionError as exc:
+        if body.callId:
+            async with pool.acquire() as conn:
+                await conn.execute(
+                    "UPDATE comms_call_logs SET transcript_text = $1 WHERE call_id = $2::uuid",
+                    f"Transcription failed: {exc}",
+                    body.callId,
+                )
+        raise HTTPException(status_code=503, detail=str(exc)) from exc
+
     if body.callId:
         async with pool.acquire() as conn:
             await conn.execute(
-                "UPDATE comms_call_logs SET transcript_text = $1 WHERE call_id = $2::uuid",
-                "Transcription pending. Configure COMMS_TRANSCRIPTION_PROVIDER to enable processing.",
+                """
+                UPDATE comms_call_logs
+                SET transcript_text = $1,
+                    raw_payload = COALESCE(raw_payload, '{}'::jsonb) || $2::jsonb
+                WHERE call_id = $3::uuid
+                """,
+                result["text"],
+                json.dumps({"transcription": {"provider": result["provider"], "language": result["language"]}}),
                 body.callId,
             )
     return {
         "success": True,
-        "status": "pending",
-        "message": "Transcription intake recorded. A real transcription worker/provider is still required.",
+        "status": "completed",
+        "message": "Transcription completed.",
         "callId": body.callId,
         "recordingUrl": body.recordingUrl,
+        "provider": result["provider"],
+        "language": result["language"],
+        "text": result["text"],
+        "segments": result["segments"],
     }

backend/api/routes_crm_imports.py

@@ -77,6 +77,56 @@ CANONICAL_OPPORTUNITY_STAGES = (
     "closed_won",
     "closed_lost",
 )
+IMPORT_DUPLICATE_POLICIES = ("create_new", "update_existing", "skip_duplicate")
+CANONICAL_URGENCY_VALUES = ("low", "medium", "high", "critical")
+CANONICAL_TASK_PRIORITIES = ("low", "normal", "high", "urgent")
+CANONICAL_BUYER_TYPES = (
+    "end_user",
+    "hni_end_user",
+    "nri_investor",
+    "family_office",
+    "founder_buyer",
+    "broker_referral",
+    "investor",
+)
+DREAM_WEAVER_ROOM_TYPES = (
+    ("bedroom", "Bedroom", "bed.double"),
+    ("living_room", "Living Room", "sofa"),
+    ("bathroom", "Bathroom", "drop"),
+    ("kitchen", "Kitchen", "refrigerator"),
+    ("dining_room", "Dining Room", "fork.knife"),
+    ("home_office", "Office", "desktopcomputer"),
+    ("hallway", "Hallway", "door.left.hand.open"),
+    ("balcony", "Balcony", "sun.max"),
+)
+
+
+def _label_for_vocab(value: str) -> str:
+    return value.replace("_", " ").title()
+
+
+def _vocab_options(values: tuple[str, ...], descriptions: dict[str, str] | None = None) -> list[dict[str, str]]:
+    descriptions = descriptions or {}
+    return [
+        {
+            "value": value,
+            "label": _label_for_vocab(value),
+            "description": descriptions.get(value, ""),
+        }
+        for value in values
+    ]
+
+
+def _room_type_options() -> list[dict[str, str]]:
+    return [
+        {
+            "value": value,
+            "label": label,
+            "description": "",
+            "icon": icon,
+        }
+        for value, label, icon in DREAM_WEAVER_ROOM_TYPES
+    ]
 
 
 def _now() -> str:
@@ -176,12 +226,53 @@ def _opportunity_payload(row) -> dict[str, Any]:
     }
 
 
+@router.get("/crm/vocabularies", tags=["CRM Vocabularies"])
+async def get_crm_vocabularies(
+    request: Request,
+    user: UserPrincipal = Depends(get_current_user),
+) -> dict[str, Any]:
+    """
+    Canonical business vocabularies for native clients.
+
+    The iPad must not own CRM funnel semantics, duplicate merge policy values,
+    task priorities, buyer personas, or Dream Weaver room vocabularies. This
+    endpoint gives authenticated clients a single backend-owned contract.
+    """
+    await _get_pool(request)
+    return {
+        "status": "ok",
+        "data": {
+            "lead_statuses": _vocab_options(CANONICAL_LEAD_STAGES),
+            "urgencies": _vocab_options(CANONICAL_URGENCY_VALUES),
+            "buyer_types": _vocab_options(CANONICAL_BUYER_TYPES),
+            "task_priorities": _vocab_options(CANONICAL_TASK_PRIORITIES),
+            "lead_stages": _vocab_options(CANONICAL_LEAD_STAGES),
+            "opportunity_stages": _vocab_options(CANONICAL_OPPORTUNITY_STAGES),
+            "import_duplicate_policies": _vocab_options(
+                IMPORT_DUPLICATE_POLICIES,
+                {
+                    "create_new": "Create a new canonical CRM person from the approved row.",
+                    "update_existing": "Merge approved fields into the strongest duplicate candidate.",
+                    "skip_duplicate": "Skip canonical create/update for this approved row.",
+                },
+            ),
+            "dream_weaver_room_types": _room_type_options(),
+        },
+        "meta": {
+            "tenant_id": _tenant_scope(user),
+            "source_of_truth": "backend_canonical_crm_vocabulary",
+        },
+    }
+
+
 # ── Models ────────────────────────────────────────────────────────────────────
 
 class ProposalApprovalRequest(BaseModel):
     proposal_id: str
     decision: str = Field(..., pattern="^(approved|rejected|needs_more_info)$")
     notes: str = Field(default="", max_length=2000)
+    field_overrides: dict[str, Any] = Field(default_factory=dict)
+    duplicate_policy: str = Field(default="create_new", pattern="^(create_new|update_existing|skip_duplicate)$")
 
 
 class CreatePersonRequest(BaseModel):
@@ -228,6 +319,139 @@ class ClientDataPatchRequest(BaseModel):
     urgency: str | None = Field(default=None, max_length=64)
 
 
+IMPORT_VALIDATION_FIELDS = (
+    "full_name",
+    "primary_email",
+    "primary_phone",
+    "buyer_type",
+    "budget_band",
+    "project_name",
+)
+
+
+def _clean_import_value(value: Any) -> str | None:
+    if value is None:
+        return None
+    text = str(value).strip()
+    return text or None
+
+
+def _digits_only(value: str | None) -> str:
+    return "".join(ch for ch in (value or "") if ch.isdigit())
+
+
+def _validate_import_canonical(canonical: dict[str, Any]) -> list[dict[str, Any]]:
+    issues: list[dict[str, Any]] = []
+    full_name = _clean_import_value(canonical.get("full_name"))
+    email = _clean_import_value(canonical.get("primary_email"))
+    phone = _clean_import_value(canonical.get("primary_phone"))
+
+    if not full_name:
+        issues.append({
+            "field": "full_name",
+            "severity": "error",
+            "message": "Full name is required before commit.",
+        })
+    elif len(full_name) < 2:
+        issues.append({
+            "field": "full_name",
+            "severity": "warning",
+            "message": "Full name is unusually short.",
+        })
+
+    if email and ("@" not in email or "." not in email.split("@")[-1]):
+        issues.append({
+            "field": "primary_email",
+            "severity": "error",
+            "message": "Email must look like a valid address.",
+        })
+
+    if phone and len(_digits_only(phone)) < 7:
+        issues.append({
+            "field": "primary_phone",
+            "severity": "error",
+            "message": "Phone number must contain at least 7 digits.",
+        })
+
+    if not email and not phone:
+        issues.append({
+            "field": "primary_phone",
+            "severity": "warning",
+            "message": "No phone or email is present; future dedupe and outreach quality will be limited.",
+        })
+
+    return issues
+
+
+def _crm_person_candidate(row: Any) -> dict[str, Any]:
+    return {
+        "person_id": str(row["person_id"]),
+        "full_name": row["full_name"],
+        "primary_email": row["primary_email"],
+        "primary_phone": row["primary_phone"],
+        "buyer_type": row["buyer_type"],
+        "source_confidence": float(row["source_confidence"]) if row["source_confidence"] is not None else None,
+        "created_at": row["created_at"].isoformat() if row["created_at"] else None,
+        "updated_at": row["updated_at"].isoformat() if row["updated_at"] else None,
+        "match_reason": row["match_reason"],
+        "match_score": int(row["match_score"]),
+    }
+
+
+def _proposal_field_diff(canonical: dict[str, Any], existing: dict[str, Any] | None) -> list[dict[str, Any]]:
+    fields = sorted(set(IMPORT_VALIDATION_FIELDS).union(canonical.keys()))
+    diffs: list[dict[str, Any]] = []
+    for field in fields:
+        proposed = _clean_import_value(canonical.get(field))
+        current = _clean_import_value(existing.get(field)) if existing else None
+        if proposed is None and current is None:
+            continue
+        diffs.append({
+            "field": field,
+            "proposed": proposed,
+            "existing": current,
+            "changed": proposed != current,
+        })
+    return diffs
+
+
+async def _find_duplicate_person(conn, tenant_id: str, canonical: dict[str, Any]) -> Any | None:
+    email = _clean_import_value(canonical.get("primary_email"))
+    phone = _clean_import_value(canonical.get("primary_phone"))
+    full_name = _clean_import_value(canonical.get("full_name"))
+    return await conn.fetchrow(
+        """
+        SELECT p.person_id, p.full_name, p.primary_email, p.primary_phone, p.buyer_type,
+               p.source_confidence, p.created_at, p.updated_at,
+               CASE
+                 WHEN $2::text IS NOT NULL AND LOWER(p.primary_email) = LOWER($2::text) THEN 'email'
+                 WHEN $3::text IS NOT NULL AND regexp_replace(COALESCE(p.primary_phone, ''), '\\D', '', 'g') = regexp_replace($3::text, '\\D', '', 'g') THEN 'phone'
+                 WHEN $4::text IS NOT NULL AND LOWER(p.full_name) = LOWER($4::text) THEN 'name'
+                 ELSE 'fuzzy'
+               END AS match_reason,
+               CASE
+                 WHEN $2::text IS NOT NULL AND LOWER(p.primary_email) = LOWER($2::text) THEN 100
+                 WHEN $3::text IS NOT NULL AND regexp_replace(COALESCE(p.primary_phone, ''), '\\D', '', 'g') = regexp_replace($3::text, '\\D', '', 'g') THEN 95
+                 WHEN $4::text IS NOT NULL AND LOWER(p.full_name) = LOWER($4::text) THEN 70
+                 ELSE 50
+               END AS match_score
+        FROM crm_people p
+        WHERE p.tenant_id = $1
+          AND (
+            ($2::text IS NOT NULL AND LOWER(p.primary_email) = LOWER($2::text))
+            OR ($3::text IS NOT NULL AND regexp_replace(COALESCE(p.primary_phone, ''), '\\D', '', 'g') = regexp_replace($3::text, '\\D', '', 'g'))
+            OR ($4::text IS NOT NULL AND LOWER(p.full_name) = LOWER($4::text))
+          )
+        ORDER BY match_score DESC, p.updated_at DESC NULLS LAST
+        LIMIT 1
+        """,
+        tenant_id,
+        email,
+        phone,
+        full_name,
+    )
+
+
 class UpdateReminderRequest(BaseModel):
     status: str = Field(..., min_length=1, max_length=32)
     due_at: str | None = None
@@ -419,6 +643,121 @@ async def get_import_batch(
     }
 
 
+@router.get("/crm/imports/{batch_id}/workbench", tags=["CRM Imports"])
+async def get_import_workbench(
+    request: Request,
+    batch_id: str,
+    user: UserPrincipal = Depends(get_current_user),
+) -> dict[str, Any]:
+    """
+    Return enterprise review diagnostics for an import batch.
+
+    This endpoint is intentionally read-only. It lets iPad/WebOS operators see
+    per-field validation, duplicate candidates, and row-level diffs before
+    committing approved proposals into canonical CRM tables.
+    """
+    pool = await _get_pool(request)
+    tenant_id = _tenant_scope(user)
+    async with pool.acquire() as conn:
+        batch_exists = await conn.fetchval(
+            "SELECT EXISTS (SELECT 1 FROM workflow_import_batches WHERE batch_id = $1::uuid AND tenant_id = $2)",
+            batch_id,
+            tenant_id,
+        )
+        if not batch_exists:
+            raise HTTPException(status_code=404, detail=f"Import batch '{batch_id}' not found.")
+
+        proposal_rows = await conn.fetch(
+            """
+            SELECT action_id, proposal_payload, confidence, status, approval_required, created_at
+            FROM workflow_actions
+            WHERE tenant_id = $1
+              AND action_type = 'import_proposal'
+              AND proposal_payload->>'batch_id' = $2
+            ORDER BY (proposal_payload->>'row_number')::int ASC
+            LIMIT 200
+            """,
+            tenant_id,
+            batch_id,
+        )
+
+        rows: list[dict[str, Any]] = []
+        duplicate_count = 0
+        validation_error_count = 0
+        validation_warning_count = 0
+
+        for proposal in proposal_rows:
+            payload = dict(proposal["proposal_payload"] or {})
+            canonical = dict(payload.get("canonical_payload") or {})
+            email = _clean_import_value(canonical.get("primary_email"))
+            phone = _clean_import_value(canonical.get("primary_phone"))
+            full_name = _clean_import_value(canonical.get("full_name"))
+
+            duplicate_candidates = await conn.fetch(
+                """
+                SELECT p.person_id, p.full_name, p.primary_email, p.primary_phone, p.buyer_type,
+                       p.source_confidence, p.created_at, p.updated_at,
+                       CASE
+                         WHEN $2::text IS NOT NULL AND LOWER(p.primary_email) = LOWER($2::text) THEN 'email'
+                         WHEN $3::text IS NOT NULL AND regexp_replace(COALESCE(p.primary_phone, ''), '\\D', '', 'g') = regexp_replace($3::text, '\\D', '', 'g') THEN 'phone'
+                         WHEN $4::text IS NOT NULL AND LOWER(p.full_name) = LOWER($4::text) THEN 'name'
+                         ELSE 'fuzzy'
+                       END AS match_reason,
+                       CASE
+                         WHEN $2::text IS NOT NULL AND LOWER(p.primary_email) = LOWER($2::text) THEN 100
+                         WHEN $3::text IS NOT NULL AND regexp_replace(COALESCE(p.primary_phone, ''), '\\D', '', 'g') = regexp_replace($3::text, '\\D', '', 'g') THEN 95
+                         WHEN $4::text IS NOT NULL AND LOWER(p.full_name) = LOWER($4::text) THEN 70
+                         ELSE 50
+                       END AS match_score
+                FROM crm_people p
+                WHERE p.tenant_id = $1
+                  AND (
+                    ($2::text IS NOT NULL AND LOWER(p.primary_email) = LOWER($2::text))
+                    OR ($3::text IS NOT NULL AND regexp_replace(COALESCE(p.primary_phone, ''), '\\D', '', 'g') = regexp_replace($3::text, '\\D', '', 'g'))
+                    OR ($4::text IS NOT NULL AND LOWER(p.full_name) = LOWER($4::text))
+                  )
+                ORDER BY match_score DESC, p.updated_at DESC NULLS LAST
+                LIMIT 5
+                """,
+                tenant_id,
+                email,
+                phone,
+                full_name,
+            )
+            candidates = [_crm_person_candidate(row) for row in duplicate_candidates]
+            existing = candidates[0] if candidates else None
+            validation = _validate_import_canonical(canonical)
+            validation_error_count += sum(1 for issue in validation if issue["severity"] == "error")
+            validation_warning_count += sum(1 for issue in validation if issue["severity"] == "warning")
+            if candidates:
+                duplicate_count += 1
+
+            rows.append({
+                "proposal_id": str(proposal["action_id"]),
+                "row_number": payload.get("row_number"),
+                "status": proposal["status"],
+                "confidence": float(proposal["confidence"]) if proposal["confidence"] else 0.0,
+                "validation": validation,
+                "duplicate_candidates": candidates,
+                "duplicate_policy": payload.get("duplicate_policy") or ("update_existing" if candidates else "create_new"),
+                "field_diffs": _proposal_field_diff(canonical, existing),
+            })
+
+    return {
+        "status": "ok",
+        "data": {
+            "batch_id": batch_id,
+            "summary": {
+                "proposal_count": len(rows),
+                "duplicate_count": duplicate_count,
+                "validation_error_count": validation_error_count,
+                "validation_warning_count": validation_warning_count,
+            },
+            "rows": rows,
+        },
+    }
+
+
 @router.put("/crm/imports/{batch_id}/review-proposal", tags=["CRM Imports"])
 async def review_proposal(
     request: Request,
@@ -435,7 +774,7 @@ async def review_proposal(
     async with pool.acquire() as conn:
         action = await conn.fetchrow(
             """
-            SELECT action_id, confidence, approval_required
+            SELECT action_id, confidence, approval_required, proposal_payload
             FROM workflow_actions
             WHERE action_id = $1::uuid
               AND tenant_id = $2
@@ -449,7 +788,41 @@ async def review_proposal(
             raise HTTPException(status_code=404, detail="Proposal not found.")
 
         decision_id = str(uuid.uuid4())
-        new_status = "approved" if body.decision == "approved" else "rejected"
+        new_status = {
+            "approved": "approved",
+            "rejected": "rejected",
+            "needs_more_info": "review_required",
+        }[body.decision]
+        proposal_payload = dict(action["proposal_payload"] or {})
+        if body.field_overrides:
+            canonical_payload = dict(proposal_payload.get("canonical_payload") or {})
+            cleaned_overrides = {
+                key: value
+                for key, value in body.field_overrides.items()
+                if key and value is not None and str(value).strip()
+            }
+            canonical_payload.update(cleaned_overrides)
+            if cleaned_overrides:
+                missing_required = [
+                    field for field in proposal_payload.get("missing_required", [])
+                    if field not in cleaned_overrides
+                ]
+                unresolved_fields = [
+                    field for field in proposal_payload.get("unresolved_fields", [])
+                    if field not in cleaned_overrides
+                ]
+                proposal_payload["canonical_payload"] = canonical_payload
+                proposal_payload["missing_required"] = missing_required
+                proposal_payload["unresolved_fields"] = unresolved_fields
+                proposal_payload["remediation"] = {
+                    "source": "ipad_import_workbench",
+                    "field_overrides": cleaned_overrides,
+                    "updated_at": _now(),
+                    "updated_by": user.user_id,
+                }
+        proposal_payload["duplicate_policy"] = body.duplicate_policy
+        proposal_payload["duplicate_policy_updated_at"] = _now()
+        proposal_payload["duplicate_policy_updated_by"] = user.user_id
 
         await conn.execute(
             """
@@ -465,11 +838,12 @@ async def review_proposal(
         await conn.execute(
             """
             UPDATE workflow_actions
-            SET status = $1::wf_status, updated_at = NOW()
-            WHERE action_id = $2::uuid
-              AND tenant_id = $3
+            SET status = $1::wf_status, proposal_payload = $2::jsonb, updated_at = NOW()
+            WHERE action_id = $3::uuid
+              AND tenant_id = $4
             """,
             new_status,
+            json.dumps(proposal_payload),
             body.proposal_id,
             _tenant_scope(user),
         )
@@ -519,30 +893,91 @@ async def commit_approved_proposals(
             try:
                 payload = row["proposal_payload"]
                 canonical = payload.get("canonical_payload", {})
-                if not canonical.get("full_name"):
+                validation_errors = [
+                    issue for issue in _validate_import_canonical(canonical)
+                    if issue["severity"] == "error"
+                ]
+                if validation_errors:
                     skipped += 1
+                    errors.append(
+                        f"Proposal {row['action_id']}: "
+                        + "; ".join(issue["message"] for issue in validation_errors)
+                    )
                     continue
 
-                person_id = str(uuid.uuid4())
-                await conn.execute(
-                    """
-                    INSERT INTO crm_people (
-                        person_id, tenant_id, full_name, primary_email, primary_phone,
-                        buyer_type, source_confidence, metadata_json, created_at, updated_at
-                    ) VALUES (
-                        $1::uuid, $2, $3, $4, $5, $6, $7, $8::jsonb, NOW(), NOW()
+                duplicate_policy = payload.get("duplicate_policy") or "create_new"
+                if duplicate_policy not in IMPORT_DUPLICATE_POLICIES:
+                    duplicate_policy = "create_new"
+
+                duplicate_person = await _find_duplicate_person(conn, _tenant_scope(user), canonical)
+                if duplicate_person and duplicate_policy == "skip_duplicate":
+                    skipped += 1
+                    await conn.execute(
+                        """
+                        UPDATE workflow_actions
+                        SET status = 'executed'::wf_status, updated_at = NOW()
+                        WHERE action_id = $1::uuid
+                          AND tenant_id = $2
+                        """,
+                        row["action_id"],
+                        _tenant_scope(user),
+                    )
+                    continue
+
+                if duplicate_person and duplicate_policy == "update_existing":
+                    person_id = str(duplicate_person["person_id"])
+                    await conn.execute(
+                        """
+                        UPDATE crm_people
+                        SET full_name = COALESCE($3, full_name),
+                            primary_email = COALESCE($4, primary_email),
+                            primary_phone = COALESCE($5, primary_phone),
+                            buyer_type = COALESCE($6, buyer_type),
+                            source_confidence = GREATEST(COALESCE(source_confidence, 0), $7),
+                            metadata_json = COALESCE(metadata_json, '{}'::jsonb) || $8::jsonb,
+                            updated_at = NOW()
+                        WHERE person_id = $1::uuid
+                          AND tenant_id = $2
+                        """,
+                        person_id,
+                        _tenant_scope(user),
+                        canonical.get("full_name"),
+                        canonical.get("primary_email"),
+                        canonical.get("primary_phone"),
+                        canonical.get("buyer_type"),
+                        payload.get("confidence", 0.5),
+                        json.dumps({
+                            "source_batch": batch_id,
+                            "import_row": payload.get("row_number"),
+                            "duplicate_policy": duplicate_policy,
+                            "merged_from_import": True,
+                        }),
+                    )
+                else:
+                    person_id = str(uuid.uuid4())
+                    await conn.execute(
+                        """
+                        INSERT INTO crm_people (
+                            person_id, tenant_id, full_name, primary_email, primary_phone,
+                            buyer_type, source_confidence, metadata_json, created_at, updated_at
+                        ) VALUES (
+                            $1::uuid, $2, $3, $4, $5, $6, $7, $8::jsonb, NOW(), NOW()
+                        )
+                        ON CONFLICT DO NOTHING
+                        """,
+                        person_id,
+                        _tenant_scope(user),
+                        canonical.get("full_name"),
+                        canonical.get("primary_email"),
+                        canonical.get("primary_phone"),
+                        canonical.get("buyer_type"),
+                        payload.get("confidence", 0.5),
+                        json.dumps({
+                            "source_batch": batch_id,
+                            "import_row": payload.get("row_number"),
+                            "duplicate_policy": duplicate_policy,
+                        }),
                     )
-                    ON CONFLICT DO NOTHING
-                    """,
-                    person_id,
-                    _tenant_scope(user),
-                    canonical.get("full_name"),
-                    canonical.get("primary_email"),
-                    canonical.get("primary_phone"),
-                    canonical.get("buyer_type"),
-                    payload.get("confidence", 0.5),
-                    json.dumps({"source_batch": batch_id, "import_row": payload.get("row_number")}),
-                )
 
                 if canonical.get("status") or canonical.get("budget_band"):
                     lead_id = str(uuid.uuid4())
@@ -985,6 +1420,11 @@ async def create_task(
     """Create a reminder / follow-up task."""
     pool = await _get_pool(request)
     reminder_id = str(uuid.uuid4())
+    next_priority = _normalize_choice(
+        body.priority,
+        allowed=CANONICAL_TASK_PRIORITIES,
+        field_name="task priority",
+    )
 
     async with pool.acquire() as conn:
         due_dt = _parse_optional_datetime(body.due_at, field_name="due_at")
@@ -1035,7 +1475,7 @@ async def create_task(
             body.title,
             body.notes,
             due_dt,
-            body.priority,
+            next_priority,
         )
 
     return {"status": "ok", "data": {"reminder_id": reminder_id, "title": body.title}}
@@ -1362,13 +1802,26 @@ async def list_client_data(
     offset: int = Query(default=0, ge=0),
 ) -> dict[str, Any]:
     pool = await _get_pool(request)
-    params: list[Any] = []
+    filter_params: list[Any] = []
     where = "1=1"
     if search:
-        params.append(f"%{search.lower()}%")
-        where = f"(lower(p.full_name) LIKE ${len(params)} OR lower(COALESCE(p.primary_phone,'')) LIKE ${len(params)} OR lower(COALESCE(p.primary_email,'')) LIKE ${len(params)} OR lower(COALESCE(pi.projects,'')) LIKE ${len(params)})"
-    params.extend([limit, offset])
+        filter_params.append(f"%{search.lower()}%")
+        where = f"(lower(p.full_name) LIKE ${len(filter_params)} OR lower(COALESCE(p.primary_phone,'')) LIKE ${len(filter_params)} OR lower(COALESCE(p.primary_email,'')) LIKE ${len(filter_params)} OR lower(COALESCE(pi.projects,'')) LIKE ${len(filter_params)})"
+    row_params = [*filter_params, limit, offset]
     async with pool.acquire() as conn:
+        total_count = await conn.fetchval(
+            f"""
+            WITH interests AS (
+                SELECT person_id, string_agg(DISTINCT project_name, ', ') AS projects
+                FROM crm_property_interests GROUP BY person_id
+            )
+            SELECT COUNT(*)::int
+            FROM crm_people p
+            LEFT JOIN interests pi ON pi.person_id = p.person_id
+            WHERE {where}
+            """,
+            *filter_params,
+        )
         rows = await conn.fetch(
             f"""
             WITH interests AS (
@@ -1397,11 +1850,21 @@ async def list_client_data(
             LEFT JOIN read_next_best_action nba ON nba.person_id = p.person_id
             WHERE {where}
             ORDER BY lc.last_contact_at DESC NULLS LAST, qd_score DESC, p.full_name ASC
-            LIMIT ${len(params)-1} OFFSET ${len(params)}
+            LIMIT ${len(row_params)-1} OFFSET ${len(row_params)}
             """,
-            *params,
+            *row_params,
         )
-    return {"status": "ok", "data": [dict(r) for r in rows], "meta": {"count": len(rows), "limit": limit, "offset": offset}}
+    return {
+        "status": "ok",
+        "data": [dict(r) for r in rows],
+        "meta": {
+            "count": len(rows),
+            "total_count": total_count or 0,
+            "has_more": offset + len(rows) < (total_count or 0),
+            "limit": limit,
+            "offset": offset,
+        },
+    }
 
 
 @router.get("/crm/client-data/{person_id}", tags=["CRM Client Data"])
@@ -1454,26 +1917,66 @@ async def get_client_data(request: Request, person_id: str) -> dict[str, Any]:
 
 
 @router.patch("/crm/client-data/{person_id}", tags=["CRM Client Data"])
-async def patch_client_data(request: Request, person_id: str, body: ClientDataPatchRequest) -> dict[str, Any]:
+async def patch_client_data(
+    request: Request,
+    person_id: str,
+    body: ClientDataPatchRequest,
+    user: UserPrincipal = Depends(get_current_user),
+) -> dict[str, Any]:
     pool = await _get_pool(request)
     payload = body.model_dump(exclude_unset=True)
+    tenant_id = _tenant_scope(user)
+    if "lead_status" in payload and payload["lead_status"] is not None:
+        payload["lead_status"] = _normalize_choice(
+            payload["lead_status"],
+            allowed=CANONICAL_LEAD_STAGES,
+            field_name="lead status",
+        )
+    if "urgency" in payload and payload["urgency"] is not None:
+        payload["urgency"] = _normalize_choice(
+            payload["urgency"],
+            allowed=CANONICAL_URGENCY_VALUES,
+            field_name="urgency",
+        )
     person_fields = {k: payload[k] for k in ("full_name", "primary_email", "primary_phone", "buyer_type", "communication_preference", "best_contact_time") if k in payload}
     lead_fields = {k: payload[k] for k in ("budget_band", "urgency") if k in payload}
     async with pool.acquire() as conn:
         async with conn.transaction():
+            person_exists = await conn.fetchval(
+                "SELECT EXISTS (SELECT 1 FROM crm_people WHERE person_id = $1::uuid AND tenant_id = $2)",
+                person_id,
+                tenant_id,
+            )
+            if not person_exists:
+                raise HTTPException(status_code=404, detail=f"Client '{person_id}' not found.")
             if person_fields:
                 sets = ", ".join(f"{key} = ${idx}" for idx, key in enumerate(person_fields, start=1))
-                await conn.execute(f"UPDATE crm_people SET {sets}, updated_at = NOW() WHERE person_id = ${len(person_fields)+1}::uuid", *person_fields.values(), person_id)
+                await conn.execute(
+                    f"UPDATE crm_people SET {sets}, updated_at = NOW() WHERE person_id = ${len(person_fields)+1}::uuid AND tenant_id = ${len(person_fields)+2}",
+                    *person_fields.values(),
+                    person_id,
+                    tenant_id,
+                )
             if lead_fields:
-                lead_id = await conn.fetchval("SELECT lead_id FROM crm_leads WHERE person_id = $1::uuid ORDER BY updated_at DESC LIMIT 1", person_id)
+                lead_id = await conn.fetchval(
+                    "SELECT lead_id FROM crm_leads WHERE person_id = $1::uuid AND tenant_id = $2 ORDER BY updated_at DESC LIMIT 1",
+                    person_id,
+                    tenant_id,
+                )
                 if lead_id:
                     sets = ", ".join(f"{key} = ${idx}" for idx, key in enumerate(lead_fields, start=1))
-                    await conn.execute(f"UPDATE crm_leads SET {sets}, updated_at = NOW() WHERE lead_id = ${len(lead_fields)+1}::uuid", *lead_fields.values(), str(lead_id))
+                    await conn.execute(
+                        f"UPDATE crm_leads SET {sets}, updated_at = NOW() WHERE lead_id = ${len(lead_fields)+1}::uuid AND tenant_id = ${len(lead_fields)+2}",
+                        *lead_fields.values(),
+                        str(lead_id),
+                        tenant_id,
+                    )
             if "lead_status" in payload:
                 await conn.execute(
-                    "UPDATE crm_leads SET status = $1::crm_lead_status, updated_at = NOW() WHERE person_id = $2::uuid",
+                    "UPDATE crm_leads SET status = $1::crm_lead_status, updated_at = NOW() WHERE person_id = $2::uuid AND tenant_id = $3",
                     payload["lead_status"],
                     person_id,
+                    tenant_id,
                 )
     return {"status": "ok", "data": {"person_id": person_id, "updated": sorted(payload.keys())}}
 
@@ -1487,9 +1990,14 @@ async def get_client_data_timeline(request: Request, person_id: str, limit: int
 
 
 @router.post("/crm/client-data/{person_id}/tasks", status_code=201, tags=["CRM Client Data"])
-async def create_client_data_task(request: Request, person_id: str, body: CreateReminderRequest) -> dict[str, Any]:
+async def create_client_data_task(
+    request: Request,
+    person_id: str,
+    body: CreateReminderRequest,
+    user: UserPrincipal = Depends(get_current_user),
+) -> dict[str, Any]:
     patched = body.model_copy(update={"person_id": person_id})
-    return await create_task(request, patched)
+    return await create_task(request, patched, user)
 
 
 async def _client_timeline(conn: Any, person_id: str, limit: int) -> list[dict[str, Any]]:

backend/api/routes_mobile_edge.py

@@ -5,6 +5,7 @@ Mobile Edge API — serves iPhone Edge and Android Phone Edge apps.
 
 Surfaces:
   GET  /mobile-edge/events            — communication events for a lead
+  GET  /mobile-edge/bulk              — coordinated iPad refresh bundle
   POST /mobile-edge/events            — log a new communication event
   GET  /mobile-edge/memory            — memory facts for a lead
   POST /mobile-edge/imports           — operator-assisted import of a recording/note
@@ -54,6 +55,22 @@ def _tenant_scope(user) -> str:
     return user.tenant_id
 
 
+def _normalise_lead_ids(raw_value: Optional[str], max_items: int = 24) -> list[str]:
+    if not raw_value:
+        return []
+    seen: set[str] = set()
+    lead_ids: list[str] = []
+    for part in raw_value.split(","):
+        lead_id = part.strip()
+        if not lead_id or lead_id in seen:
+            continue
+        seen.add(lead_id)
+        lead_ids.append(lead_id)
+        if len(lead_ids) >= max_items:
+            break
+    return lead_ids
+
+
 # ── Pydantic models ───────────────────────────────────────────────────────────
 
 VALID_CHANNELS = {
@@ -135,6 +152,12 @@ class SessionHeartbeat(BaseModel):
     metadata: dict = Field(default_factory=dict)
 
 
+class MobileEdgeBulkRequest(BaseModel):
+    lead_ids: list[str] = Field(default_factory=list, max_length=100)
+    events_limit_per_lead: int = Field(default=4, ge=1, le=25)
+    calendar_limit: int = Field(default=50, ge=1, le=200)
+
+
 # ── Communication Events ───────────────────────────────────────────────────────
 
 @router.get("/events", summary="List communication events for a lead")
@@ -173,6 +196,134 @@ async def list_events(
     }
 
 
+@router.get("/bulk", summary="Bulk mobile-edge refresh bundle")
+async def bulk_mobile_edge(
+    request: Request,
+    lead_ids: Optional[str] = Query(None, description="Comma-separated lead IDs to hydrate timeline events for"),
+    events_limit_per_lead: int = Query(4, ge=1, le=25),
+    calendar_limit: int = Query(50, ge=1, le=200),
+    user=Depends(get_current_user),
+):
+    """
+    Returns a single coordinated payload for native surface refreshes.
+
+    The iPad app uses this endpoint to avoid one request for alerts, one request
+    for calendar, and then one request per lead timeline.
+    """
+    return await _bulk_mobile_edge_payload(
+        request=request,
+        user=user,
+        selected_lead_ids=_normalise_lead_ids(lead_ids),
+        events_limit_per_lead=events_limit_per_lead,
+        calendar_limit=calendar_limit,
+    )
+
+
+@router.post("/bulk", summary="Bulk mobile-edge refresh bundle")
+async def bulk_mobile_edge_post(
+    request: Request,
+    body: MobileEdgeBulkRequest,
+    user=Depends(get_current_user),
+):
+    """POST variant for native clients that need a larger explicit lead set."""
+    seen: set[str] = set()
+    selected_lead_ids = []
+    for lead_id in body.lead_ids:
+        normalized = lead_id.strip()
+        if normalized and normalized not in seen:
+            seen.add(normalized)
+            selected_lead_ids.append(normalized)
+    return await _bulk_mobile_edge_payload(
+        request=request,
+        user=user,
+        selected_lead_ids=selected_lead_ids[:100],
+        events_limit_per_lead=body.events_limit_per_lead,
+        calendar_limit=body.calendar_limit,
+    )
+
+
+async def _bulk_mobile_edge_payload(
+    *,
+    request: Request,
+    user,
+    selected_lead_ids: list[str],
+    events_limit_per_lead: int,
+    calendar_limit: int,
+):
+    tenant_id = _tenant_scope(user)
+    pool = _pool(request)
+
+    async with pool.acquire() as conn:
+        calendar_rows = await conn.fetch(
+            """
+            SELECT calendar_event_id, lead_id, title, description, start_at, end_at,
+                   all_day, status, reminder_minutes, created_by, location, metadata, created_at
+            FROM user_calendar_events
+            WHERE tenant_id=$1 AND owner_user_id=$2
+              AND status <> 'cancelled'
+            ORDER BY start_at ASC LIMIT $3
+            """,
+            tenant_id, user.user_id, calendar_limit,
+        )
+
+        if selected_lead_ids:
+            event_rows = await conn.fetch(
+                """
+                SELECT event_id, lead_id, channel, direction, provider, capture_mode,
+                       consent_state, timestamp, duration_seconds, summary, raw_reference,
+                       recording_ref, provider_metadata, created_at
+                FROM (
+                    SELECT event_id, lead_id, channel, direction, provider, capture_mode,
+                           consent_state, timestamp, duration_seconds, summary, raw_reference,
+                           recording_ref, provider_metadata, created_at,
+                           ROW_NUMBER() OVER (PARTITION BY lead_id ORDER BY timestamp DESC) AS row_number
+                    FROM edge_communication_events
+                    WHERE tenant_id=$1 AND lead_id = ANY($2::text[])
+                ) ranked_events
+                WHERE row_number <= $3
+                ORDER BY lead_id ASC, timestamp DESC
+                """,
+                tenant_id, selected_lead_ids, events_limit_per_lead,
+            )
+        else:
+            event_rows = []
+
+        pending_insights = await conn.fetchval(
+            "SELECT COUNT(*) FROM insight_recommendations WHERE tenant_id=$1 AND status='pending'",
+            tenant_id,
+        )
+        upcoming_events = await conn.fetchval(
+            """
+            SELECT COUNT(*) FROM user_calendar_events
+            WHERE tenant_id=$1 AND owner_user_id=$2
+              AND status='confirmed'
+              AND start_at BETWEEN NOW() AND NOW() + INTERVAL '24 hours'
+            """,
+            tenant_id, user.user_id,
+        )
+        pending_transcriptions = await conn.fetchval(
+            "SELECT COUNT(*) FROM edge_transcription_jobs WHERE tenant_id=$1 AND status='pending'",
+            tenant_id,
+        )
+
+    events_by_lead_id: dict[str, list[dict[str, Any]]] = {lead_id: [] for lead_id in selected_lead_ids}
+    for row in event_rows:
+        event = dict(row)
+        events_by_lead_id.setdefault(event["lead_id"], []).append(event)
+
+    return {
+        "calendar_events": [dict(r) for r in calendar_rows],
+        "lead_events": events_by_lead_id,
+        "alerts": {
+            "pending_insights": pending_insights,
+            "upcoming_calendar_events_24h": upcoming_events,
+            "pending_transcriptions": pending_transcriptions,
+            "generated_at": _now(),
+        },
+        "generated_at": _now(),
+    }
+
+
 @router.post("/events", status_code=status.HTTP_201_CREATED, summary="Log a communication event")
 async def create_event(
     request: Request,