fix: restore Velocity-OS pillar data contracts

core/api/api/routes_crm.py

@@ -205,6 +205,8 @@ class SyntheticSeedRequest(BaseModel):
 
 
 def _serialize_lead(row: Any) -> dict[str, Any]:
+    if not hasattr(row, "get"):
+        row = dict(row)
     score = int(row["score"] or 0)
     status_label = _normalize_stage(row["kanban_status"])
     qualification = row["qualification"] or _infer_qualification(score, row.get("source"), row.get("notes"))
@@ -427,6 +429,51 @@ def _merge_lead_sources(
     )
 
 
+def _relative_time_label(value: Any) -> str:
+    if value is None:
+        return "No contact yet"
+    if isinstance(value, str):
+        try:
+            parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+        except ValueError:
+            return value
+    else:
+        parsed = value
+    if parsed.tzinfo is None:
+        parsed = parsed.replace(tzinfo=timezone.utc)
+    delta = _now() - parsed.astimezone(timezone.utc)
+    if delta.days > 0:
+        return f"{delta.days}d ago"
+    hours = int(delta.total_seconds() // 3600)
+    if hours > 0:
+        return f"{hours}h ago"
+    minutes = int(delta.total_seconds() // 60)
+    if minutes > 0:
+        return f"{minutes}m ago"
+    return "Just now"
+
+
+def _kanban_stage_payload(stage_id: str, label: str, emoji: str, leads: list[dict[str, Any]]) -> dict[str, Any]:
+    return {
+        "id": stage_id,
+        "label": label,
+        "emoji": emoji,
+        "leads": [
+            {
+                "id": lead["id"],
+                "name": lead["name"],
+                "location": lead.get("unit_interest") or lead.get("budget") or lead.get("source"),
+                "qdScore": int(lead.get("score") or 0),
+                "qdDelta": int((lead.get("metadata") or {}).get("qd_delta") or 0),
+                "lastContactRelative": _relative_time_label(lead.get("updated_at") or lead.get("created_at")),
+                "lastContactChannel": lead.get("source") or "crm",
+                "isVaultActive": bool((lead.get("metadata") or {}).get("vault_active")),
+            }
+            for lead in leads
+        ],
+    }
+
+
 async def _fetch_legacy_chat_logs(conn: Any, tenant_id: str, lead_id: str | None = None, channel: str | None = None) -> list[dict[str, Any]]:
     clauses: list[str] = ["tenant_id = $1"]
     params: list[Any] = [tenant_id]
@@ -1308,6 +1355,53 @@ async def get_kanban_board(
     return {"status": "ok", "data": board}
 
 
+@crm_router.get("/pipeline/kanban")
+async def get_pipeline_kanban(
+    request: Request,
+    user: UserPrincipal = Depends(get_current_user),
+) -> list[dict[str, Any]]:
+    """Velocity-OS Pipeline board contract.
+
+    The simplified WebOS pillar expects raw stage arrays, while the older CRM
+    board endpoint returns a wrapped administrative payload. Keep this route
+    explicit so the UI gets exactly the shape it renders.
+    """
+    await _ensure_schema(request)
+    pool = await _get_pool(request)
+    tenant_id = _tenant_scope(user)
+    async with pool.acquire() as conn:
+        try:
+            canonical_leads = await _fetch_canonical_leads(conn, tenant_id)
+        except Exception as exc:
+            logger.warning("Canonical CRM pipeline bridge unavailable for tenant %s: %s", tenant_id, exc)
+            canonical_leads = []
+        legacy_rows = await conn.fetch(
+            """
+            SELECT id, name, email, phone, source, notes, qualification, score, kanban_status,
+                   budget, unit_interest, metadata, created_at, updated_at
+            FROM leads
+            WHERE tenant_id = $1
+            ORDER BY score DESC, updated_at DESC, created_at DESC
+            """,
+            tenant_id,
+        )
+    leads = _merge_lead_sources(canonical_leads, [_serialize_lead(row) for row in legacy_rows])
+    stage_defs = [
+        ("new", "New", "N"),
+        ("qualifying", "Qualifying", "Q"),
+        ("site_visit", "Site Visit", "V"),
+        ("negotiation", "Negotiation", "D"),
+        ("closed", "Closed", "C"),
+    ]
+    grouped: dict[str, list[dict[str, Any]]] = {stage_id: [] for stage_id, _, _ in stage_defs}
+    for lead in leads:
+        grouped.setdefault(lead.get("stage") or "new", []).append(lead)
+    return [
+        _kanban_stage_payload(stage_id, label, emoji, grouped.get(stage_id, []))
+        for stage_id, label, emoji in stage_defs
+    ]
+
+
 @crm_router.put("/kanban/move")
 async def move_kanban_card(
     request: Request,