""" backend/routers/sentinel.py - Sentinel WebSocket and biometric endpoints. """ from __future__ import annotations import asyncio import json import logging import math import os import re import uuid from datetime import datetime, timezone from typing import Any, Set import asyncpg from fastapi import APIRouter, Depends, HTTPException, WebSocket, WebSocketDisconnect from pydantic import BaseModel from backend.auth.dependencies import UserPrincipal, get_current_user, require_role from backend.db.pool import get_pool from backend.services.auto_mode_matcher import auto_mode_match_session from backend.services.nemoclaw_client import score_qd, tag_lead logger = logging.getLogger("velocity.sentinel") router = APIRouter() _UUID_RE = re.compile( r"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[1-5][0-9a-fA-F]{3}-[89abAB][0-9a-fA-F]{3}-[0-9a-fA-F]{12}$" ) class SentinelConnectionManager: def __init__(self) -> None: self._channels: dict[str, Set[WebSocket]] = { "notifications": set(), "perception": set(), } async def connect(self, ws: WebSocket, channel: str) -> None: await ws.accept() self._channels.setdefault(channel, set()).add(ws) logger.info("WS connected: channel=%s total=%d", channel, len(self._channels[channel])) def disconnect(self, ws: WebSocket, channel: str) -> None: self._channels.get(channel, set()).discard(ws) async def broadcast(self, payload: dict[str, Any], channel: str = "notifications") -> None: dead: Set[WebSocket] = set() for ws in list(self._channels.get(channel, set())): try: await ws.send_text(json.dumps(payload)) except Exception: dead.add(ws) self._channels[channel] -= dead async def broadcast_all(self, payload: dict[str, Any]) -> None: for channel in self._channels: await self.broadcast(payload, channel) def connection_count(self, channel: str) -> int: return len(self._channels.get(channel, set())) manager = SentinelConnectionManager() _perception_producer_task: asyncio.Task | None = None def _is_uuid(value: str | None) -> bool: return bool(value and _UUID_RE.match(value)) async def _resolve_scene_label( conn: asyncpg.Connection, video_asset_id: str | None, video_ts_ms: int, ) -> str | None: if not video_asset_id: return None row = await conn.fetchrow( """ SELECT room_type, description FROM video_scene_maps WHERE video_asset_id = $1 AND start_ms <= $2 AND end_ms >= $2 ORDER BY start_ms DESC LIMIT 1 """, video_asset_id, video_ts_ms, ) if not row: return None description = row["description"] return f"{row['room_type']} - {description}" if description else str(row["room_type"]) async def _ensure_session_row( conn: asyncpg.Connection, *, session_id: str, session_mode: str, lead_id: str | None, video_asset_id: str | None, ) -> None: await conn.execute( """ INSERT INTO perception_sessions (id, session_mode, lead_id, video_asset_id, auto_mode_evidence) VALUES ($1::uuid, $2::session_mode_enum, $3::uuid, $4, '{}'::jsonb) ON CONFLICT (id) DO UPDATE SET video_asset_id = EXCLUDED.video_asset_id, lead_id = COALESCE(perception_sessions.lead_id, EXCLUDED.lead_id) """, session_id, session_mode, lead_id if _is_uuid(lead_id) else None, video_asset_id or "unknown", ) async def _resolve_canonical_context( conn: asyncpg.Connection, *, person_id: str | None, canonical_lead_id: str | None, legacy_lead_id: str | None, ) -> dict[str, Any] | None: if _is_uuid(person_id): row = await conn.fetchrow( """ SELECT p.person_id::text AS person_id, p.full_name, p.primary_phone, p.buyer_type, p.legacy_li_id::text AS legacy_li_id, cl.lead_id::text AS lead_id, cl.status AS lead_status, cl.budget_band, cl.urgency, COALESCE(( SELECT current_value FROM intel_qd_scores WHERE person_id = p.person_id AND score_type = 'engagement_score' ORDER BY computed_at DESC LIMIT 1 ), 0.50) AS engagement_score FROM crm_people p LEFT JOIN crm_leads cl ON cl.person_id = p.person_id WHERE p.person_id = $1::uuid ORDER BY cl.created_at DESC NULLS LAST LIMIT 1 """, person_id, ) if row: return dict(row) if _is_uuid(canonical_lead_id): row = await conn.fetchrow( """ SELECT p.person_id::text AS person_id, p.full_name, p.primary_phone, p.buyer_type, p.legacy_li_id::text AS legacy_li_id, cl.lead_id::text AS lead_id, cl.status AS lead_status, cl.budget_band, cl.urgency, COALESCE(( SELECT current_value FROM intel_qd_scores WHERE person_id = p.person_id AND score_type = 'engagement_score' ORDER BY computed_at DESC LIMIT 1 ), 0.50) AS engagement_score FROM crm_leads cl INNER JOIN crm_people p ON p.person_id = cl.person_id WHERE cl.lead_id = $1::uuid LIMIT 1 """, canonical_lead_id, ) if row: return dict(row) if _is_uuid(legacy_lead_id): row = await conn.fetchrow( """ SELECT p.person_id::text AS person_id, p.full_name, p.primary_phone, p.buyer_type, p.legacy_li_id::text AS legacy_li_id, cl.lead_id::text AS lead_id, cl.status AS lead_status, cl.budget_band, cl.urgency, COALESCE(( SELECT current_value FROM intel_qd_scores WHERE person_id = p.person_id AND score_type = 'engagement_score' ORDER BY computed_at DESC LIMIT 1 ), 0.50) AS engagement_score FROM crm_people p LEFT JOIN crm_leads cl ON cl.person_id = p.person_id WHERE p.legacy_li_id = $1::uuid ORDER BY cl.created_at DESC NULLS LAST LIMIT 1 """, legacy_lead_id, ) if row: return dict(row) return None async def _ensure_canonical_interaction( conn: asyncpg.Connection, *, person_id: str, canonical_lead_id: str | None, session_id: str, session_mode: str, video_asset_id: str | None, ) -> str: existing = await conn.fetchrow( """ SELECT interaction_id::text FROM intel_interactions WHERE source_ref = $1 AND channel = 'perception_session'::intel_channel ORDER BY created_at DESC LIMIT 1 """, session_id, ) if existing: return existing["interaction_id"] row = await conn.fetchrow( """ INSERT INTO intel_interactions ( person_id, lead_id, channel, interaction_type, source_ref, summary, metadata_json ) VALUES ( $1::uuid, $2::uuid, 'perception_session'::intel_channel, 'sentinel_live_session', $3, $4, $5::jsonb ) ON CONFLICT DO NOTHING RETURNING interaction_id::text """, person_id, canonical_lead_id if _is_uuid(canonical_lead_id) else None, session_id, f"Sentinel live session ({session_mode})", json.dumps( { "session_id": session_id, "session_mode": session_mode, "video_asset_id": video_asset_id, } ), ) return row["interaction_id"] async def _persist_canonical_qd( conn: asyncpg.Connection, *, person_id: str, interaction_id: str, session_id: str, scene_label: str | None, video_ts_ms: int, result_qd_score: int, baseline_score: int, blend_shapes: dict[str, float], session_mode: str, ) -> None: normalized_score = max(0.0, min(1.0, result_qd_score / 100.0)) normalized_delta = max(-1.0, min(1.0, (result_qd_score - baseline_score) / 100.0)) event_type = "engagement_sample" if result_qd_score - baseline_score >= 8: event_type = "engagement_spike" elif result_qd_score - baseline_score <= -8: event_type = "negative_shift" metadata = { "interaction_id": interaction_id, "scene_label": scene_label, "video_ts_ms": video_ts_ms, "qd_before": baseline_score, "qd_after": result_qd_score, "delta": result_qd_score - baseline_score, "session_mode": session_mode, "blend_shapes": blend_shapes, } await conn.execute( """ INSERT INTO intel_perception_events ( person_id, session_ref, event_type, engagement_score, media_ref, happened_at, metadata_json ) VALUES ($1::uuid, $2, $3, $4, $5, NOW(), $6::jsonb) """, person_id, session_id, event_type, normalized_score, scene_label or f"video_ts:{video_ts_ms}", json.dumps(metadata), ) await conn.execute( """ INSERT INTO intel_qd_scores ( person_id, score_type, current_value, computed_at, evidence_refs_json, reasoning, metadata_json ) VALUES ( $1::uuid, 'engagement_score', $2, NOW(), $3::jsonb, $4, $5::jsonb ) ON CONFLICT (person_id, score_type) DO UPDATE SET current_value = EXCLUDED.current_value, computed_at = EXCLUDED.computed_at, evidence_refs_json = EXCLUDED.evidence_refs_json, reasoning = EXCLUDED.reasoning, metadata_json = EXCLUDED.metadata_json """, person_id, normalized_score, json.dumps([session_id, interaction_id]), f"Sentinel session updated engagement to {result_qd_score}/100", json.dumps(metadata), ) await conn.execute( """ INSERT INTO intel_qd_timeseries ( person_id, score_type, signal_source, timestamp, value, delta, evidence_ref, metadata_json ) VALUES ( $1::uuid, 'engagement_score', 'sentinel_live_session', NOW(), $2, $3, $4, $5::jsonb ) """, person_id, normalized_score, normalized_delta, session_id, json.dumps(metadata), ) async def _build_showroom_perception_payload(pool: asyncpg.Pool) -> dict[str, Any]: now = datetime.now(timezone.utc) seconds = now.timestamp() simulated_count = max(1, int(round(9 + 5 * math.sin(seconds / 300.0) + 2 * math.sin(seconds / 53.0)))) simulated_sentiment = max(0.0, min(1.0, 0.62 + 0.18 * math.sin(seconds / 210.0))) try: async with pool.acquire() as conn: row = await conn.fetchrow( """ SELECT COUNT(*) FILTER (WHERE happened_at >= NOW() - INTERVAL '15 minutes')::int AS recent_events, COUNT(DISTINCT session_ref) FILTER (WHERE happened_at >= NOW() - INTERVAL '15 minutes')::int AS active_sessions, COALESCE(AVG(engagement_score) FILTER (WHERE happened_at >= NOW() - INTERVAL '15 minutes'), 0.0)::float AS avg_engagement, COUNT(*) FILTER ( WHERE event_type IN ('engagement_spike', 'positive_shift') AND happened_at >= NOW() - INTERVAL '15 minutes' )::int AS positive_events, COUNT(*) FILTER ( WHERE event_type = 'negative_shift' AND happened_at >= NOW() - INTERVAL '15 minutes' )::int AS negative_events FROM intel_perception_events """ ) active_sessions = int(row["active_sessions"] or 0) recent_events = int(row["recent_events"] or 0) avg_engagement = float(row["avg_engagement"] or 0.0) if recent_events > 0: visitor_count = max(active_sessions, recent_events) sentiment_score = max(0.0, min(1.0, avg_engagement)) source = "canonical_perception_events" else: visitor_count = simulated_count sentiment_score = simulated_sentiment source = "mathematical_simulation" positive_events = int(row["positive_events"] or 0) negative_events = int(row["negative_events"] or 0) except Exception: active_sessions = 0 visitor_count = simulated_count sentiment_score = simulated_sentiment positive_events = 0 negative_events = 0 source = "mathematical_simulation" showroom_heat = max(0.0, min(1.0, (visitor_count / 18.0) * 0.45 + sentiment_score * 0.55)) conversion_intent = max(0.0, min(1.0, sentiment_score * 0.7 + min(visitor_count, 20) / 20.0 * 0.3)) sentiment_label = "positive" if sentiment_score >= 0.66 else ("neutral" if sentiment_score >= 0.42 else "negative") return { "type": "PERCEPTION_ANALYTICS", "data": { "generated_at": now.isoformat(), "source": source, "visitor_count": visitor_count, "active_sessions": active_sessions, "sentiment_score": round(sentiment_score, 3), "sentiment_label": sentiment_label, "positive_events": positive_events, "negative_events": negative_events, "showroom_intelligence": { "showroom_heat": round(showroom_heat, 3), "conversion_intent": round(conversion_intent, 3), "recommended_staffing": "high_touch" if visitor_count >= 12 or conversion_intent >= 0.72 else "standard", "zones": [ { "zone": "model_apartment", "visitors": max(1, int(visitor_count * 0.42)), "dwell_seconds": int(220 + 140 * showroom_heat), }, { "zone": "pricing_desk", "visitors": max(0, int(visitor_count * conversion_intent * 0.35)), "dwell_seconds": int(120 + 180 * conversion_intent), }, { "zone": "amenities_gallery", "visitors": max(0, visitor_count - int(visitor_count * 0.42)), "dwell_seconds": int(90 + 90 * sentiment_score), }, ], }, }, } async def _perception_producer(pool: asyncpg.Pool) -> None: while True: try: if manager.connection_count("perception") == 0: await asyncio.sleep(1.0) continue payload = await _build_showroom_perception_payload(pool) await manager.broadcast(payload, "perception") await asyncio.sleep(float(os.getenv("SENTINEL_PERCEPTION_INTERVAL_SECONDS", "3"))) except asyncio.CancelledError: raise except Exception as exc: logger.exception("Sentinel perception producer failed: %s", exc) await asyncio.sleep(5.0) def _ensure_perception_producer(pool: asyncpg.Pool) -> None: global _perception_producer_task if _perception_producer_task is None or _perception_producer_task.done(): _perception_producer_task = asyncio.create_task(_perception_producer(pool)) @router.websocket("/ws/notifications") async def notifications_ws(ws: WebSocket) -> None: await manager.connect(ws, "notifications") try: while True: data = await ws.receive_text() await ws.send_text(json.dumps({"type": "ack", "data": data})) except WebSocketDisconnect: manager.disconnect(ws, "notifications") @router.websocket("/ws/perception") async def perception_ws(ws: WebSocket) -> None: await manager.connect(ws, "perception") pool: asyncpg.Pool | None = getattr(ws.app.state, "db_pool", None) if pool is None: await ws.send_text(json.dumps({"type": "system", "data": {"error": "Database unavailable"}})) await ws.close(code=1011) return _ensure_perception_producer(pool) try: while True: raw = await ws.receive_text() try: packet = json.loads(raw) except json.JSONDecodeError: continue if packet.get("event") != "BIOMETRIC_PACKET": continue person_id = packet.get("person_id") canonical_lead_id = packet.get("canonical_lead_id") lead_id = packet.get("lead_id") session_id = packet.get("session_id") session_mode = packet.get("session_mode", "assigned") video_ts_ms = int(packet.get("video_ts_ms", 0)) video_asset_id = packet.get("video_asset_id") blend_shapes = packet.get("blend_shapes", {}) if ( not session_id or not _is_uuid(session_id) or session_mode not in {"assigned", "auto"} or not isinstance(blend_shapes, dict) or not blend_shapes ): continue async def _score( sid: str = session_id, lid: str | None = lead_id, mode: str = session_mode, bts: int = video_ts_ms, bs: dict[str, float] = blend_shapes, asset_id: str | None = video_asset_id, ) -> None: try: async with pool.acquire() as conn: async with conn.transaction(): canonical = await _resolve_canonical_context( conn, person_id=person_id, canonical_lead_id=canonical_lead_id, legacy_lead_id=lid, ) await _ensure_session_row( conn, session_id=sid, session_mode=mode, lead_id=(canonical["legacy_li_id"] if canonical and canonical.get("legacy_li_id") else lid), video_asset_id=asset_id, ) lead_row = None legacy_lead_id = canonical["legacy_li_id"] if canonical and canonical.get("legacy_li_id") else lid if _is_uuid(legacy_lead_id): lead_row = await conn.fetchrow( """ SELECT quantum_dynamics_score, budget, interest, tags FROM leads_intelligence WHERE id = $1::uuid """, legacy_lead_id, ) session_row = await conn.fetchrow( """ SELECT final_qd_score, auto_mode_evidence FROM perception_sessions WHERE id = $1::uuid """, sid, ) scene_label = await _resolve_scene_label(conn, asset_id, bts) crm = { "budget": (canonical["budget_band"] if canonical and canonical.get("budget_band") else None) or (lead_row["budget"] if lead_row else None) or "unknown", "interest": (lead_row["interest"] if lead_row else None) or "unknown", "prior_interaction_count": await conn.fetchval( """ SELECT COUNT(*) FROM omnichannel_logs WHERE lead_id = $1::uuid """, legacy_lead_id, ) if _is_uuid(legacy_lead_id) else 0, "tags": list((lead_row["tags"] if lead_row else None) or []), "session_mode": mode, } baseline_score = ( lead_row["quantum_dynamics_score"] if lead_row and lead_row["quantum_dynamics_score"] is not None else ( int(round(float(canonical["engagement_score"]) * 100)) if canonical and canonical.get("engagement_score") is not None else ((session_row["final_qd_score"] if session_row else None) or 50) ) ) result = await score_qd( lead_id=(canonical["person_id"] if canonical else None) or lid or sid, batch_id=sid, blend_shapes=bs, video_ts_ms=bts, scene_label=scene_label, crm_context=crm, current_qd_score=baseline_score, ) evidence = dict((session_row["auto_mode_evidence"] if session_row else {}) or {}) evidence.update( { "last_scene_label": scene_label, "last_video_ts_ms": bts, "person_id": canonical["person_id"] if canonical else None, } ) await conn.execute( """ UPDATE perception_sessions SET final_qd_score = $1, auto_mode_evidence = $2::jsonb WHERE id = $3::uuid """, result.qd_score, evidence, sid, ) if canonical and _is_uuid(canonical["person_id"]): interaction_id = await _ensure_canonical_interaction( conn, person_id=canonical["person_id"], canonical_lead_id=canonical["lead_id"], session_id=sid, session_mode=mode, video_asset_id=asset_id, ) await _persist_canonical_qd( conn, person_id=canonical["person_id"], interaction_id=interaction_id, session_id=sid, scene_label=scene_label, video_ts_ms=bts, result_qd_score=result.qd_score, baseline_score=baseline_score, blend_shapes=bs, session_mode=mode, ) if lead_row and _is_uuid(legacy_lead_id): await conn.execute( """ INSERT INTO omnichannel_logs (event_type, lead_id, payload, video_timestamp_ms) VALUES ('SENTIMENT_SPIKE', $1::uuid, $2::jsonb, $3) """, legacy_lead_id, json.dumps( { "blend_shapes": bs, "scene_label": scene_label, "qd_before": baseline_score, "qd_after": result.qd_score, "confidence": result.confidence, "session_id": sid, "person_id": canonical["person_id"] if canonical else None, } ), bts, ) await conn.execute( """ UPDATE leads_intelligence SET quantum_dynamics_score = $1, updated_at = NOW() WHERE id = $2::uuid """, result.qd_score, legacy_lead_id, ) event = { "type": "QD_UPDATED", "data": { "person_id": canonical["person_id"] if canonical else None, "lead_id": legacy_lead_id, "session_id": sid, "qd_score": result.qd_score, "delta": result.qd_score - baseline_score, "reasoning": result.reasoning, "scene_label": scene_label, "timestamp": datetime.now(timezone.utc).isoformat(), }, } await manager.broadcast_all(event) except Exception as exc: logger.exception("QD scoring failed for session %s: %s", sid, exc) asyncio.create_task(_score()) except WebSocketDisconnect: manager.disconnect(ws, "perception") class ConsentRequest(BaseModel): lead_id: str ip_address: str | None = None user_agent: str | None = None class TagLeadRequest(BaseModel): lead_id: str phone: str budget: str | None = None message_text: str class SessionCompleteRequest(BaseModel): session_id: str session_mode: str person_id: str | None = None canonical_lead_id: str | None = None lead_id: str | None = None lead_name: str | None = None final_qd_score: int | None = None @router.post("/consent", status_code=201, summary="Record biometric consent") async def record_consent( body: ConsentRequest, pool: asyncpg.Pool = Depends(get_pool), ) -> dict[str, str]: async with pool.acquire() as conn: await conn.execute( """ INSERT INTO consent_log (lead_id, ip_address, user_agent, action) VALUES ($1::uuid, $2, $3, 'granted') """, body.lead_id, body.ip_address, body.user_agent, ) return {"status": "consent_recorded"} @router.post("/session/complete", summary="Close a perception session and finalize auto mode if needed") async def complete_session( body: SessionCompleteRequest, pool: asyncpg.Pool = Depends(get_pool), ) -> dict[str, Any]: if not _is_uuid(body.session_id): raise HTTPException(status_code=400, detail="session_id must be a UUID.") if body.session_mode not in {"assigned", "auto"}: raise HTTPException(status_code=400, detail="session_mode must be assigned or auto.") async with pool.acquire() as conn: async with conn.transaction(): await _ensure_session_row( conn, session_id=body.session_id, session_mode=body.session_mode, lead_id=body.lead_id, video_asset_id=None, ) canonical = await _resolve_canonical_context( conn, person_id=body.person_id, canonical_lead_id=body.canonical_lead_id, legacy_lead_id=body.lead_id, ) await conn.execute( """ UPDATE perception_sessions SET ended_at = NOW(), final_qd_score = COALESCE($1, final_qd_score) WHERE id = $2::uuid """, body.final_qd_score, body.session_id, ) if canonical and body.final_qd_score is not None: interaction_id = await _ensure_canonical_interaction( conn, person_id=canonical["person_id"], canonical_lead_id=canonical["lead_id"], session_id=body.session_id, session_mode=body.session_mode, video_asset_id=None, ) await _persist_canonical_qd( conn, person_id=canonical["person_id"], interaction_id=interaction_id, session_id=body.session_id, scene_label="session_complete", video_ts_ms=0, result_qd_score=body.final_qd_score, baseline_score=body.final_qd_score, blend_shapes={}, session_mode=body.session_mode, ) if body.session_mode == "auto": result = await auto_mode_match_session(conn, session_id=body.session_id) event = { "type": "LEAD_TAGGED", "data": { "lead_id": result.lead_id, "tags": result.tags_applied, "lead_name": "Auto-matched lead", "session_id": body.session_id, }, } await manager.broadcast(event, "notifications") return { "status": "completed", "session_id": body.session_id, "lead_id": result.lead_id, "match_action": result.action, "match_confidence": result.confidence, "tags_applied": result.tags_applied, } return {"status": "completed", "session_id": body.session_id} @router.post("/tag-lead", summary="Apply NemoClaw lead tagging to a CRM lead") async def tag_lead_route( body: TagLeadRequest, pool: asyncpg.Pool = Depends(get_pool), user: UserPrincipal = Depends(require_role("SENIOR_BROKER")), ) -> dict[str, Any]: result = await tag_lead( lead_id=body.lead_id, phone=body.phone, budget=body.budget, message_text=body.message_text, ) async with pool.acquire() as conn: await conn.execute( """ UPDATE leads_intelligence SET tags = ARRAY( SELECT DISTINCT unnest( COALESCE(tags, ARRAY[]::text[]) || $1::text[] ) ) WHERE id = $2::uuid """, result.tags_to_add, body.lead_id, ) await conn.execute( """ INSERT INTO omnichannel_logs (event_type, lead_id, payload) VALUES ('LEAD_TAGGED', $1::uuid, $2::jsonb) """, body.lead_id, json.dumps( { "tags_added": result.tags_to_add, "tags_removed": result.tags_to_remove, "actor_user_id": user.user_id, } ), ) event = { "type": "LEAD_TAGGED", "data": { "lead_id": body.lead_id, "tags": result.tags_to_add, }, } await manager.broadcast(event, "notifications") return { "lead_id": body.lead_id, "tags_to_add": result.tags_to_add, "tags_to_remove": result.tags_to_remove, } @router.get("/qd-score/{lead_id}", summary="Current Quantum Dynamics score for a lead") async def get_qd_score( lead_id: str, pool: asyncpg.Pool = Depends(get_pool), user: UserPrincipal = Depends(require_role("SENIOR_BROKER")), ) -> dict[str, Any]: del user async with pool.acquire() as conn: row = await conn.fetchrow( "SELECT quantum_dynamics_score, tags FROM leads_intelligence WHERE id = $1::uuid", lead_id, ) if not row: raise HTTPException(status_code=404, detail="Lead not found.") return { "lead_id": lead_id, "qd_score": row["quantum_dynamics_score"], "tags": list(row["tags"] or []), } @router.get("/analytics/live", summary="Live Sentinel perception analytics for native clients") async def live_perception_analytics( pool: asyncpg.Pool = Depends(get_pool), user: UserPrincipal = Depends(get_current_user), ) -> dict[str, Any]: del user async with pool.acquire() as conn: session = await conn.fetchrow( """ SELECT COUNT(*) FILTER (WHERE ended_at IS NULL)::int AS active_sessions, COUNT(*) FILTER (WHERE started_at >= NOW() - INTERVAL '24 hours')::int AS visitor_count_24h, COALESCE(ROUND(AVG(final_qd_score) FILTER (WHERE final_qd_score IS NOT NULL)::numeric, 1), 0)::float AS avg_qd_score FROM perception_sessions """ ) sentiment_rows = await conn.fetch( """ SELECT CASE WHEN event_type IN ('engagement_spike', 'positive_shift') OR COALESCE(engagement_score, 0) >= 0.70 THEN 'positive' WHEN event_type IN ('negative_shift', 'exit_risk') OR COALESCE(engagement_score, 0) <= 0.35 THEN 'negative' ELSE 'neutral' END AS bucket, COUNT(*)::int AS count FROM intel_perception_events WHERE happened_at >= NOW() - INTERVAL '24 hours' GROUP BY bucket """ ) journey_rows = await conn.fetch( """ SELECT perception_id::text, COALESCE(event_type, 'perception') AS event_type, COALESCE(session_ref, '') AS session_ref, COALESCE(media_ref, '') AS scene_label, COALESCE(engagement_score, 0)::float AS engagement_score, happened_at, COALESCE(metadata_json, '{}'::jsonb) AS metadata_json FROM intel_perception_events ORDER BY happened_at DESC LIMIT 12 """ ) sentiment = {"positive": 0, "neutral": 0, "negative": 0} for row in sentiment_rows: sentiment[row["bucket"]] = row["count"] journey = [ { "eventId": row["perception_id"], "eventType": row["event_type"], "sessionRef": row["session_ref"], "sceneLabel": row["scene_label"], "engagementScore": row["engagement_score"], "happenedAt": row["happened_at"].isoformat() if row["happened_at"] else None, "metadata": dict(row["metadata_json"] or {}), } for row in journey_rows ] return { "status": "ok", "data": { "liveStreamPath": "/api/sentinel/ws/perception", "activeSessions": session["active_sessions"] if session else 0, "visitorCount24h": session["visitor_count_24h"] if session else 0, "averageQdScore": session["avg_qd_score"] if session else 0, "sentimentDistribution": sentiment, "journey": journey, }, } async def broadcast_sentinel_event(payload: dict[str, Any]) -> None: await manager.broadcast(payload, "notifications")