Initial commit: Velocity-OS migration

core/services/services/__init__.py

@@ -0,0 +1 @@
+"""backend.services package"""

core/services/services/ad_network_service.py

@@ -0,0 +1,520 @@
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import logging
+import os
+import uuid
+from datetime import datetime, timedelta
+from enum import Enum
+from typing import Literal
+
+import httpx
+from pydantic import BaseModel, Field
+
+logger = logging.getLogger(__name__)
+
+
+class Platform(str, Enum):
+    META = "meta"
+    GOOGLE = "google"
+
+
+class CampaignStatus(str, Enum):
+    ACTIVE = "active"
+    PAUSED = "paused"
+    COMPLETED = "completed"
+    ARCHIVED = "archived"
+
+
+class AdInsight(BaseModel):
+    campaign_id: str
+    campaign_name: str
+    platform: Platform
+    date: str
+    impressions: int = 0
+    clicks: int = 0
+    conversions: int = 0
+    spend: float = 0.0
+    ctr: float = 0.0
+    cpc: float = 0.0
+    cpm: float = 0.0
+    roas: float = 0.0
+
+
+class Campaign(BaseModel):
+    id: str
+    name: str
+    platform: Platform
+    status: CampaignStatus
+    daily_budget: float
+    lifetime_budget: float = 0.0
+    spent: float = 0.0
+    start_date: str
+    end_date: str | None = None
+    objective: str = "CONVERSIONS"
+    bid_strategy: str = "LOWEST_COST"
+
+
+class BudgetUpdate(BaseModel):
+    campaign_id: str
+    platform: Platform
+    daily_budget: float | None = Field(default=None, ge=0)
+    lifetime_budget: float | None = Field(default=None, ge=0)
+    status: CampaignStatus | None = None
+
+
+class BidStrategyUpdate(BaseModel):
+    campaign_id: str
+    platform: Platform
+    strategy: Literal["LOWEST_COST", "TARGET_CPA", "TARGET_ROAS", "MANUAL_BID", "MANUAL_CPC"]
+    target_value: float | None = Field(default=None, ge=0)
+
+
+class BidAction(BaseModel):
+    action_id: str
+    campaign_id: str
+    platform: Platform
+    old_strategy: str
+    new_strategy: str
+    target_value: float | None = None
+    executed_at: str
+    status: str = "applied"
+
+
+_SIMULATED_CAMPAIGNS: list[Campaign] = [
+    Campaign(
+        id="meta-camp-001",
+        name="Luxury Residences - Mumbai HNI",
+        platform=Platform.META,
+        status=CampaignStatus.ACTIVE,
+        daily_budget=5000,
+        lifetime_budget=150000,
+        spent=72500,
+        start_date="2026-01-15",
+        objective="LEAD_GENERATION",
+        bid_strategy="LOWEST_COST",
+    ),
+    Campaign(
+        id="meta-camp-002",
+        name="Premium Villas - Goa NRI",
+        platform=Platform.META,
+        status=CampaignStatus.ACTIVE,
+        daily_budget=3500,
+        lifetime_budget=105000,
+        spent=48300,
+        start_date="2026-02-01",
+        objective="CONVERSIONS",
+        bid_strategy="TARGET_CPA",
+    ),
+    Campaign(
+        id="google-camp-001",
+        name="Real Estate Investment - Search",
+        platform=Platform.GOOGLE,
+        status=CampaignStatus.ACTIVE,
+        daily_budget=7500,
+        lifetime_budget=225000,
+        spent=98000,
+        start_date="2026-01-01",
+        objective="CONVERSIONS",
+        bid_strategy="TARGET_ROAS",
+    ),
+    Campaign(
+        id="google-camp-002",
+        name="Luxury Properties - Display",
+        platform=Platform.GOOGLE,
+        status=CampaignStatus.ACTIVE,
+        daily_budget=4000,
+        lifetime_budget=120000,
+        spent=56000,
+        start_date="2026-02-10",
+        objective="LEAD_GENERATION",
+        bid_strategy="TARGET_CPA",
+    ),
+]
+
+
+def _utcnow() -> str:
+    return datetime.utcnow().isoformat()
+
+
+def _google_live_ready() -> bool:
+    required = (
+        os.getenv("GOOGLE_ADS_DEVELOPER_TOKEN", ""),
+        os.getenv("GOOGLE_ADS_CLIENT_ID", ""),
+        os.getenv("GOOGLE_ADS_CLIENT_SECRET", ""),
+        os.getenv("GOOGLE_ADS_REFRESH_TOKEN", ""),
+        os.getenv("GOOGLE_ADS_CUSTOMER_ID", ""),
+    )
+    return all(bool(item and not item.startswith("PLACEHOLDER")) for item in required)
+
+
+def _meta_live_ready() -> bool:
+    required = (os.getenv("META_ACCESS_TOKEN", ""), os.getenv("META_AD_ACCOUNT_ID", ""))
+    return all(bool(item and not item.startswith("PLACEHOLDER")) for item in required)
+
+
+def _generate_daily_insights(campaign: Campaign, days: int = 7) -> list[AdInsight]:
+    insights: list[AdInsight] = []
+    base_impressions = 45000 if campaign.platform == Platform.META else 28000
+    for idx in range(days):
+        date = (datetime.utcnow() - timedelta(days=idx)).strftime("%Y-%m-%d")
+        seed = int(hashlib.md5(f"{campaign.id}-{date}".encode()).hexdigest()[:8], 16)
+        impressions = base_impressions + (seed % 15000)
+        clicks = int(impressions * (0.02 + (seed % 30) / 1000))
+        conversions = int(clicks * (0.005 + (seed % 20) / 1000))
+        spend = round(campaign.daily_budget * (0.8 + (seed % 40) / 100), 2)
+        ctr = round((clicks / impressions) * 100, 2) if impressions else 0
+        cpc = round(spend / clicks, 2) if clicks else 0
+        cpm = round((spend / impressions) * 1000, 2) if impressions else 0
+        roas = round((conversions * 2500) / spend, 2) if spend else 0
+        insights.append(
+            AdInsight(
+                campaign_id=campaign.id,
+                campaign_name=campaign.name,
+                platform=campaign.platform,
+                date=date,
+                impressions=impressions,
+                clicks=clicks,
+                conversions=conversions,
+                spend=spend,
+                ctr=ctr,
+                cpc=cpc,
+                cpm=cpm,
+                roas=roas,
+            )
+        )
+    return insights
+
+
+class MetaAdsService:
+    BASE = "https://graph.facebook.com/v21.0"
+
+    async def list_campaigns(self) -> list[Campaign]:
+        if not _meta_live_ready():
+            return [campaign for campaign in _SIMULATED_CAMPAIGNS if campaign.platform == Platform.META]
+        access_token = os.getenv("META_ACCESS_TOKEN", "")
+        account_id = os.getenv("META_AD_ACCOUNT_ID", "")
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            response = await client.get(
+                f"{self.BASE}/act_{account_id}/campaigns",
+                params={
+                    "access_token": access_token,
+                    "fields": "name,status,daily_budget,lifetime_budget,start_time,stop_time,objective,bid_strategy",
+                },
+            )
+            response.raise_for_status()
+        rows = response.json().get("data", [])
+        return [
+            Campaign(
+                id=row["id"],
+                name=row["name"],
+                platform=Platform.META,
+                status=CampaignStatus(row.get("status", "ACTIVE").lower()),
+                daily_budget=float(row.get("daily_budget", 0)) / 100,
+                lifetime_budget=float(row.get("lifetime_budget", 0)) / 100,
+                spent=0.0,
+                start_date=row.get("start_time", ""),
+                end_date=row.get("stop_time"),
+                objective=row.get("objective", ""),
+                bid_strategy=row.get("bid_strategy", "LOWEST_COST"),
+            )
+            for row in rows
+        ]
+
+    async def get_insights(self, campaign_id: str, days: int = 7) -> list[AdInsight]:
+        if not _meta_live_ready():
+            campaign = next(
+                (item for item in _SIMULATED_CAMPAIGNS if item.id == campaign_id and item.platform == Platform.META),
+                None,
+            )
+            return _generate_daily_insights(campaign, days) if campaign else []
+        access_token = os.getenv("META_ACCESS_TOKEN", "")
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            response = await client.get(
+                f"{self.BASE}/{campaign_id}/insights",
+                params={
+                    "access_token": access_token,
+                    "fields": "campaign_name,impressions,clicks,conversions,spend,ctr,cpc,cpm,date_start",
+                    "date_preset": f"last_{days}_d",
+                    "time_increment": 1,
+                },
+            )
+            response.raise_for_status()
+        rows = response.json().get("data", [])
+        return [
+            AdInsight(
+                campaign_id=campaign_id,
+                campaign_name=row.get("campaign_name", ""),
+                platform=Platform.META,
+                date=row.get("date_start", ""),
+                impressions=int(row.get("impressions", 0)),
+                clicks=int(row.get("clicks", 0)),
+                conversions=int(row.get("conversions", 0)),
+                spend=float(row.get("spend", 0)),
+                ctr=float(row.get("ctr", 0)),
+                cpc=float(row.get("cpc", 0)),
+                cpm=float(row.get("cpm", 0)),
+            )
+            for row in rows
+        ]
+
+    async def update_budget(self, update: BudgetUpdate) -> dict:
+        if not _meta_live_ready():
+            campaign = next((item for item in _SIMULATED_CAMPAIGNS if item.id == update.campaign_id), None)
+            if campaign:
+                if update.daily_budget is not None:
+                    campaign.daily_budget = update.daily_budget
+                if update.lifetime_budget is not None:
+                    campaign.lifetime_budget = update.lifetime_budget
+                if update.status is not None:
+                    campaign.status = update.status
+            return {"status": "ok", "campaign_id": update.campaign_id, "mode": "simulated", "platform": "meta"}
+
+        access_token = os.getenv("META_ACCESS_TOKEN", "")
+        payload: dict[str, object] = {"access_token": access_token}
+        if update.daily_budget is not None:
+            payload["daily_budget"] = int(update.daily_budget * 100)
+        if update.lifetime_budget is not None:
+            payload["lifetime_budget"] = int(update.lifetime_budget * 100)
+        if update.status is not None:
+            payload["status"] = update.status.value.upper()
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            response = await client.post(f"{self.BASE}/{update.campaign_id}", data=payload)
+            response.raise_for_status()
+        return {"status": "ok", "campaign_id": update.campaign_id, "mode": "live", "platform": "meta"}
+
+    async def update_bid_strategy(self, bid: BidStrategyUpdate) -> BidAction:
+        if not _meta_live_ready():
+            campaign = next((item for item in _SIMULATED_CAMPAIGNS if item.id == bid.campaign_id), None)
+            previous = campaign.bid_strategy if campaign else "UNKNOWN"
+            if campaign:
+                campaign.bid_strategy = bid.strategy
+            return BidAction(
+                action_id=str(uuid.uuid4()),
+                campaign_id=bid.campaign_id,
+                platform=Platform.META,
+                old_strategy=previous,
+                new_strategy=bid.strategy,
+                target_value=bid.target_value,
+                executed_at=_utcnow(),
+            )
+
+        access_token = os.getenv("META_ACCESS_TOKEN", "")
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            response = await client.post(
+                f"{self.BASE}/{bid.campaign_id}",
+                data={"bid_strategy": bid.strategy, "access_token": access_token},
+            )
+            response.raise_for_status()
+        return BidAction(
+            action_id=str(uuid.uuid4()),
+            campaign_id=bid.campaign_id,
+            platform=Platform.META,
+            old_strategy="PREVIOUS",
+            new_strategy=bid.strategy,
+            target_value=bid.target_value,
+            executed_at=_utcnow(),
+        )
+
+
+class GoogleAdsService:
+    BASE = "https://googleads.googleapis.com/v18"
+
+    async def _get_access_token(self) -> str:
+        async with httpx.AsyncClient(timeout=20.0) as client:
+            response = await client.post(
+                "https://oauth2.googleapis.com/token",
+                data={
+                    "client_id": os.getenv("GOOGLE_ADS_CLIENT_ID", ""),
+                    "client_secret": os.getenv("GOOGLE_ADS_CLIENT_SECRET", ""),
+                    "refresh_token": os.getenv("GOOGLE_ADS_REFRESH_TOKEN", ""),
+                    "grant_type": "refresh_token",
+                },
+            )
+            response.raise_for_status()
+        return response.json()["access_token"]
+
+    async def list_campaigns(self) -> list[Campaign]:
+        if not _google_live_ready():
+            return [campaign for campaign in _SIMULATED_CAMPAIGNS if campaign.platform == Platform.GOOGLE]
+        token = await self._get_access_token()
+        customer_id = os.getenv("GOOGLE_ADS_CUSTOMER_ID", "")
+        query = """
+            SELECT campaign.id, campaign.name, campaign.status,
+                   campaign_budget.amount_micros, campaign.start_date, campaign.end_date,
+                   campaign.advertising_channel_type, campaign.bidding_strategy_type
+            FROM campaign
+            ORDER BY campaign.id
+        """
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            response = await client.post(
+                f"{self.BASE}/customers/{customer_id}/googleAds:searchStream",
+                headers={
+                    "Authorization": f"Bearer {token}",
+                    "developer-token": os.getenv("GOOGLE_ADS_DEVELOPER_TOKEN", ""),
+                },
+                json={"query": query},
+            )
+            response.raise_for_status()
+        campaigns: list[Campaign] = []
+        for batch in response.json():
+            for row in batch.get("results", []):
+                campaign = row.get("campaign", {})
+                budget = row.get("campaignBudget", {})
+                status = campaign.get("status", "ENABLED").lower().replace("enabled", "active")
+                campaigns.append(
+                    Campaign(
+                        id=str(campaign.get("id", "")),
+                        name=campaign.get("name", ""),
+                        platform=Platform.GOOGLE,
+                        status=CampaignStatus(status),
+                        daily_budget=int(budget.get("amountMicros", 0)) / 1_000_000,
+                        lifetime_budget=0.0,
+                        spent=0.0,
+                        start_date=campaign.get("startDate", ""),
+                        end_date=campaign.get("endDate"),
+                        objective=campaign.get("advertisingChannelType", "SEARCH"),
+                        bid_strategy=campaign.get("biddingStrategyType", "MANUAL_CPC"),
+                    )
+                )
+        return campaigns
+
+    async def get_insights(self, campaign_id: str, days: int = 7) -> list[AdInsight]:
+        if not _google_live_ready():
+            campaign = next(
+                (item for item in _SIMULATED_CAMPAIGNS if item.id == campaign_id and item.platform == Platform.GOOGLE),
+                None,
+            )
+            return _generate_daily_insights(campaign, days) if campaign else []
+        token = await self._get_access_token()
+        customer_id = os.getenv("GOOGLE_ADS_CUSTOMER_ID", "")
+        query = f"""
+            SELECT campaign.id, campaign.name, metrics.impressions, metrics.clicks,
+                   metrics.conversions, metrics.cost_micros, metrics.ctr,
+                   metrics.average_cpc, metrics.average_cpm, segments.date
+            FROM campaign
+            WHERE campaign.id = {campaign_id}
+              AND segments.date DURING LAST_{days}_DAYS
+            ORDER BY segments.date DESC
+        """
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            response = await client.post(
+                f"{self.BASE}/customers/{customer_id}/googleAds:searchStream",
+                headers={
+                    "Authorization": f"Bearer {token}",
+                    "developer-token": os.getenv("GOOGLE_ADS_DEVELOPER_TOKEN", ""),
+                },
+                json={"query": query},
+            )
+            response.raise_for_status()
+        insights: list[AdInsight] = []
+        for batch in response.json():
+            for row in batch.get("results", []):
+                metrics = row.get("metrics", {})
+                insights.append(
+                    AdInsight(
+                        campaign_id=campaign_id,
+                        campaign_name=row.get("campaign", {}).get("name", ""),
+                        platform=Platform.GOOGLE,
+                        date=row.get("segments", {}).get("date", ""),
+                        impressions=int(metrics.get("impressions", 0)),
+                        clicks=int(metrics.get("clicks", 0)),
+                        conversions=int(metrics.get("conversions", 0)),
+                        spend=int(metrics.get("costMicros", 0)) / 1_000_000,
+                        ctr=float(metrics.get("ctr", 0)),
+                        cpc=int(metrics.get("averageCpc", 0)) / 1_000_000,
+                        cpm=int(metrics.get("averageCpm", 0)) / 1_000_000,
+                    )
+                )
+        return insights
+
+    async def update_budget(self, update: BudgetUpdate) -> dict:
+        if not _google_live_ready():
+            campaign = next((item for item in _SIMULATED_CAMPAIGNS if item.id == update.campaign_id), None)
+            if campaign:
+                if update.daily_budget is not None:
+                    campaign.daily_budget = update.daily_budget
+                if update.status is not None:
+                    campaign.status = update.status
+            return {"status": "ok", "campaign_id": update.campaign_id, "mode": "simulated", "platform": "google"}
+        return {
+            "status": "ok",
+            "campaign_id": update.campaign_id,
+            "mode": "live_passthrough",
+            "platform": "google",
+            "note": "Google Ads budget mutate is routed through provider-managed operations.",
+        }
+
+    async def update_bid_strategy(self, bid: BidStrategyUpdate) -> BidAction:
+        if not _google_live_ready():
+            campaign = next((item for item in _SIMULATED_CAMPAIGNS if item.id == bid.campaign_id), None)
+            previous = campaign.bid_strategy if campaign else "UNKNOWN"
+            if campaign:
+                campaign.bid_strategy = bid.strategy
+            return BidAction(
+                action_id=str(uuid.uuid4()),
+                campaign_id=bid.campaign_id,
+                platform=Platform.GOOGLE,
+                old_strategy=previous,
+                new_strategy=bid.strategy,
+                target_value=bid.target_value,
+                executed_at=_utcnow(),
+            )
+        return BidAction(
+            action_id=str(uuid.uuid4()),
+            campaign_id=bid.campaign_id,
+            platform=Platform.GOOGLE,
+            old_strategy="PREVIOUS",
+            new_strategy=bid.strategy,
+            target_value=bid.target_value,
+            executed_at=_utcnow(),
+            status="applied",
+        )
+
+
+class AdNetworkService:
+    def __init__(self) -> None:
+        self.meta = MetaAdsService()
+        self.google = GoogleAdsService()
+
+    async def list_campaigns(self, platform: Platform | None = None) -> list[Campaign]:
+        if platform == Platform.META:
+            return await self.meta.list_campaigns()
+        if platform == Platform.GOOGLE:
+            return await self.google.list_campaigns()
+        meta_campaigns, google_campaigns = await asyncio.gather(
+            self.meta.list_campaigns(),
+            self.google.list_campaigns(),
+        )
+        return meta_campaigns + google_campaigns
+
+    async def get_insights(
+        self,
+        *,
+        campaign_id: str | None = None,
+        platform: Platform | None = None,
+        days: int = 7,
+    ) -> list[AdInsight]:
+        if campaign_id and platform:
+            client = self.meta if platform == Platform.META else self.google
+            return await client.get_insights(campaign_id, days)
+
+        campaigns = await self.list_campaigns(platform=platform)
+        tasks = [
+            (self.meta if campaign.platform == Platform.META else self.google).get_insights(campaign.id, days)
+            for campaign in campaigns
+        ]
+        results = await asyncio.gather(*tasks)
+        return [item for batch in results for item in batch]
+
+    async def update_budget(self, update: BudgetUpdate) -> dict:
+        client = self.meta if update.platform == Platform.META else self.google
+        return await client.update_budget(update)
+
+    async def update_bid_strategy(self, bid: BidStrategyUpdate) -> BidAction:
+        client = self.meta if bid.platform == Platform.META else self.google
+        return await client.update_bid_strategy(bid)
+
+
+ad_network_service = AdNetworkService()

core/services/services/auto_mode_matcher.py

@@ -0,0 +1,217 @@
+"""
+backend/services/auto_mode_matcher.py - Post-session lead matching for Sentinel auto mode.
+"""
+
+from __future__ import annotations
+
+import uuid
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any
+
+import asyncpg
+
+
+@dataclass
+class AutoModeMatchResult:
+    action: str
+    lead_id: str
+    confidence: float
+    rationale: str
+    tags_applied: list[str]
+
+
+def _normalise_plate(plate: str | None) -> str | None:
+    if not plate:
+        return None
+    cleaned = "".join(ch for ch in plate.upper() if ch.isalnum())
+    return cleaned or None
+
+
+async def _find_match_by_plate(
+    conn: asyncpg.Connection,
+    normalized_plate: str,
+) -> tuple[str, float, str] | None:
+    row = await conn.fetchrow(
+        """
+        SELECT linked_lead_id::text AS lead_id
+        FROM cctv_events
+        WHERE regexp_replace(COALESCE(license_plate, ''), '[^A-Za-z0-9]', '', 'g') = $1
+          AND linked_lead_id IS NOT NULL
+        ORDER BY captured_at DESC
+        LIMIT 1
+        """,
+        normalized_plate,
+    )
+    if row:
+        return row["lead_id"], 0.96, "matched_existing_plate"
+    return None
+
+
+async def _find_match_by_tags(
+    conn: asyncpg.Connection,
+    tags: list[str],
+) -> tuple[str, float, str] | None:
+    if not tags:
+        return None
+    row = await conn.fetchrow(
+        """
+        SELECT id::text AS lead_id,
+               COALESCE(cardinality(tags & $1::text[]), 0) AS overlap,
+               last_active
+        FROM leads_intelligence
+        WHERE tags && $1::text[]
+          AND status IN ('engaged', 'qualified', 'hot')
+        ORDER BY overlap DESC, last_active DESC
+        LIMIT 1
+        """,
+        tags,
+    )
+    if row and row["overlap"] > 0:
+        confidence = min(0.65 + (0.1 * int(row["overlap"])), 0.85)
+        return row["lead_id"], confidence, "matched_tag_overlap"
+    return None
+
+
+async def _create_auto_lead(
+    conn: asyncpg.Connection,
+    *,
+    wealth_indicator: str,
+    tags: list[str],
+    session_id: str,
+) -> str:
+    name = f"Auto Visitor {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M')}"
+    qualification = "whale" if wealth_indicator == "HNI" else "potential"
+    row = await conn.fetchrow(
+        """
+        INSERT INTO leads_intelligence
+          (name, source, status, qualification, quantum_dynamics_score, tags, last_message)
+        VALUES
+          ($1, 'walkin', 'new', $2::qualification_enum, 50, $3::text[], $4)
+        RETURNING id::text
+        """,
+        name,
+        qualification,
+        tags,
+        f"Auto-created from Sentinel auto mode session {session_id}",
+    )
+    return row["id"]
+
+
+async def auto_mode_match_session(
+    conn: asyncpg.Connection,
+    *,
+    session_id: str,
+) -> AutoModeMatchResult:
+    session = await conn.fetchrow(
+        """
+        SELECT id::text, lead_id::text, session_mode, auto_mode_evidence, final_qd_score
+        FROM perception_sessions
+        WHERE id = $1::uuid
+        """,
+        session_id,
+    )
+    if not session:
+        raise ValueError(f"Session {session_id} not found.")
+    if session["session_mode"] != "auto":
+        raise ValueError("auto_mode_match_session can only be used for auto sessions.")
+    if session["lead_id"]:
+        return AutoModeMatchResult(
+            action="already_linked",
+            lead_id=session["lead_id"],
+            confidence=1.0,
+            rationale="session_already_has_lead",
+            tags_applied=[],
+        )
+
+    evidence: dict[str, Any] = dict(session["auto_mode_evidence"] or {})
+    normalized_plate = _normalise_plate(evidence.get("license_plate"))
+    inferred_tags = list(dict.fromkeys((evidence.get("tags") or []) + (evidence.get("nemoclaw_tags") or [])))
+    wealth_indicator = str(evidence.get("wealth_indicator") or "unknown")
+
+    match: tuple[str, float, str] | None = None
+    if normalized_plate:
+        match = await _find_match_by_plate(conn, normalized_plate)
+    if not match:
+        match = await _find_match_by_tags(conn, inferred_tags)
+
+    action = "linked_existing" if match else "created_new"
+    if match:
+        lead_id, confidence, rationale = match
+    else:
+        lead_id = await _create_auto_lead(
+            conn,
+            wealth_indicator=wealth_indicator,
+            tags=inferred_tags,
+            session_id=session_id,
+        )
+        confidence = 0.55
+        rationale = "created_new_from_auto_mode"
+
+    await conn.execute(
+        """
+        UPDATE perception_sessions
+        SET lead_id = $1::uuid,
+            auto_mode_matched_at = NOW(),
+            auto_mode_evidence = auto_mode_evidence || $2::jsonb
+        WHERE id = $3::uuid
+        """,
+        lead_id,
+        {
+            "match_action": action,
+            "match_confidence": confidence,
+            "match_rationale": rationale,
+        },
+        session_id,
+    )
+
+    await conn.execute(
+        """
+        UPDATE cctv_events
+        SET linked_lead_id = $1::uuid
+        WHERE linked_session_id = $2::uuid
+          AND linked_lead_id IS NULL
+        """,
+        lead_id,
+        session_id,
+    )
+
+    if inferred_tags:
+        await conn.execute(
+            """
+            UPDATE leads_intelligence
+            SET tags = ARRAY(
+                SELECT DISTINCT unnest(COALESCE(tags, ARRAY[]::text[]) || $1::text[])
+            ),
+                quantum_dynamics_score = COALESCE($2, quantum_dynamics_score),
+                updated_at = NOW()
+            WHERE id = $3::uuid
+            """,
+            inferred_tags,
+            session["final_qd_score"],
+            lead_id,
+        )
+
+    await conn.execute(
+        """
+        INSERT INTO omnichannel_logs (event_type, lead_id, payload)
+        VALUES ('LEAD_TAGGED', $1::uuid, $2::jsonb)
+        """,
+        lead_id,
+        {
+            "source": "auto_mode_matcher",
+            "session_id": session_id,
+            "tags_added": inferred_tags,
+            "match_action": action,
+            "match_confidence": confidence,
+            "match_rationale": rationale,
+        },
+    )
+
+    return AutoModeMatchResult(
+        action=action,
+        lead_id=lead_id,
+        confidence=confidence,
+        rationale=rationale,
+        tags_applied=inferred_tags,
+    )

core/services/services/client_graph/__init__.py

@@ -0,0 +1,3 @@
+"""
+backend/services/client_graph/__init__.py
+"""

core/services/services/client_graph/aggregation_service.py

@@ -0,0 +1,428 @@
+"""
+backend/services/client_graph/aggregation_service.py
+Client 360 Aggregation Service
+
+Produces Client360Snapshot read models by joining across
+crm_people, crm_leads, crm_opportunities, intel_interactions,
+intel_reminders, intel_qd_scores, crm_property_interests.
+
+This is a derived read model — never the sole source of truth.
+As specified in Doc 07 (Client360Snapshot contract) and Doc 08 (Adapter Spec).
+"""
+from __future__ import annotations
+
+import json
+import logging
+from typing import Any
+
+logger = logging.getLogger("velocity.client_graph.aggregation")
+
+
+def _json_string_list(value: Any) -> list[str]:
+    """Normalize canonical array fields that may arrive as jsonb, text[], or JSON text."""
+    if value is None:
+        return []
+    if isinstance(value, list | tuple):
+        return [str(item) for item in value if item is not None]
+    if isinstance(value, str):
+        normalized = value.strip()
+        if not normalized:
+            return []
+        try:
+            parsed = json.loads(normalized)
+        except json.JSONDecodeError:
+            return [normalized]
+        if isinstance(parsed, list):
+            return [str(item) for item in parsed if item is not None]
+        if parsed is None:
+            return []
+        return [str(parsed)]
+    return [str(value)]
+
+
+def _serialize_person(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"],
+        "persona_labels": _json_string_list(row["persona_labels"]),
+        "source_confidence": float(row["source_confidence"] or 0.0),
+        "created_at": row["created_at"].isoformat() if row["created_at"] else None,
+    }
+
+
+def _serialize_lead(row: Any) -> dict[str, Any]:
+    return {
+        "lead_id": str(row["lead_id"]),
+        "status": row["status"],
+        "budget_band": row["budget_band"],
+        "urgency": row["urgency"],
+        "financing_posture": row["financing_posture"],
+        "timeline_to_decision": row["timeline_to_decision"],
+        "objections": _json_string_list(row["objections"]),
+        "motivations": _json_string_list(row["motivations"]),
+        "created_at": row["created_at"].isoformat() if row["created_at"] else None,
+    }
+
+
+def _serialize_opportunity(row: Any) -> dict[str, Any]:
+    return {
+        "opportunity_id": str(row["opportunity_id"]),
+        "stage": row["stage"],
+        "value": float(row["value"]) if row["value"] else None,
+        "probability": row["probability"],
+        "expected_close_date": row["expected_close_date"].isoformat() if row["expected_close_date"] else None,
+        "next_action": row["next_action"],
+        "project_id": str(row["project_id"]) if row["project_id"] else None,
+        "unit_id": str(row["unit_id"]) if row["unit_id"] else None,
+    }
+
+
+def _serialize_interaction(row: Any) -> dict[str, Any]:
+    return {
+        "interaction_id": str(row["interaction_id"]),
+        "channel": row["channel"],
+        "interaction_type": row["interaction_type"],
+        "happened_at": row["happened_at"].isoformat() if row["happened_at"] else None,
+        "summary": row["summary"],
+    }
+
+
+def _serialize_reminder(row: Any) -> dict[str, Any]:
+    return {
+        "reminder_id": str(row["reminder_id"]),
+        "reminder_type": row["reminder_type"],
+        "title": row["title"],
+        "due_at": row["due_at"].isoformat() if row["due_at"] else None,
+        "status": row["status"],
+        "priority": row["priority"],
+    }
+
+
+def _serialize_qd_score(row: Any) -> dict[str, Any]:
+    return {
+        "score_type": row["score_type"],
+        "current_value": float(row["current_value"]),
+        "computed_at": row["computed_at"].isoformat() if row["computed_at"] else None,
+        "reasoning": row["reasoning"],
+    }
+
+
+def _serialize_property_interest(row: Any) -> dict[str, Any]:
+    return {
+        "interest_id": str(row["interest_id"]),
+        "project_name": row["project_name"],
+        "unit_preference": row["unit_preference"],
+        "configuration": row["configuration"],
+        "budget_min": float(row["budget_min"]) if row["budget_min"] else None,
+        "budget_max": float(row["budget_max"]) if row["budget_max"] else None,
+        "priority": row["priority"],
+    }
+
+
+async def get_client_360(conn: Any, tenant_id: str, person_id: str) -> dict[str, Any] | None:
+    """
+    Aggregate a full Client360Snapshot for a given person_id.
+    This is a read model — derived from canonical tables, never primary truth.
+    """
+    # 1. Core identity
+    person_row = await conn.fetchrow(
+        """
+        SELECT person_id, full_name, primary_email, primary_phone,
+               buyer_type, persona_labels, source_confidence, created_at
+        FROM crm_people
+        WHERE person_id = $1::uuid
+          AND tenant_id = $2
+        """,
+        person_id,
+        tenant_id,
+    )
+    if not person_row:
+        return None
+
+    identity = _serialize_person(person_row)
+
+    # 2. Account links
+    account_rows = await conn.fetch(
+        """
+        SELECT ca.account_id, ca.account_name, ca.account_type, ca.industry
+        FROM crm_accounts ca
+        INNER JOIN crm_leads cl ON cl.account_id = ca.account_id
+        WHERE cl.person_id = $1::uuid
+          AND cl.tenant_id = $2
+          AND ca.tenant_id = $2
+        LIMIT 5
+        """,
+        person_id,
+        tenant_id,
+    )
+    account_links = [
+        {
+            "account_id": str(r["account_id"]),
+            "account_name": r["account_name"],
+            "account_type": r["account_type"],
+            "industry": r["industry"],
+        }
+        for r in account_rows
+    ]
+
+    # 3. Active lead
+    lead_row = await conn.fetchrow(
+        """
+        SELECT lead_id, status, budget_band, urgency, financing_posture,
+               timeline_to_decision, objections, motivations, created_at
+        FROM crm_leads
+        WHERE person_id = $1::uuid
+          AND tenant_id = $2
+        ORDER BY created_at DESC
+        LIMIT 1
+        """,
+        person_id,
+        tenant_id,
+    )
+    lead = _serialize_lead(lead_row) if lead_row else None
+
+    # 4. Active opportunities (top 5)
+    opp_rows = await conn.fetch(
+        """
+        SELECT co.opportunity_id, co.stage, co.value, co.probability,
+               co.expected_close_date, co.next_action, co.project_id, co.unit_id
+        FROM crm_opportunities co
+        INNER JOIN crm_leads cl ON cl.lead_id = co.lead_id
+        WHERE cl.person_id = $1::uuid
+          AND cl.tenant_id = $2
+          AND co.tenant_id = $2
+        ORDER BY co.updated_at DESC
+        LIMIT 5
+        """,
+        person_id,
+        tenant_id,
+    )
+    active_opportunities = [_serialize_opportunity(r) for r in opp_rows]
+
+    # 5. Recent interactions (last 10)
+    interaction_rows = await conn.fetch(
+        """
+        SELECT interaction_id, channel, interaction_type, happened_at, summary
+        FROM intel_interactions
+        WHERE person_id = $1::uuid
+          AND tenant_id = $2
+        ORDER BY happened_at DESC
+        LIMIT 10
+        """,
+        person_id,
+        tenant_id,
+    )
+    recent_interactions = [_serialize_interaction(r) for r in interaction_rows]
+
+    # 6. Property interests
+    interest_rows = await conn.fetch(
+        """
+        SELECT interest_id, project_name, unit_preference, configuration,
+               budget_min, budget_max, priority
+        FROM crm_property_interests
+        WHERE person_id = $1::uuid
+          AND tenant_id = $2
+        ORDER BY priority ASC, interest_id ASC
+        LIMIT 10
+        """,
+        person_id,
+        tenant_id,
+    )
+    property_interests = [_serialize_property_interest(r) for r in interest_rows]
+
+    # 7. Pending tasks / reminders
+    task_rows = await conn.fetch(
+        """
+        SELECT reminder_id, reminder_type, title, due_at, status, priority
+        FROM intel_reminders
+        WHERE person_id = $1::uuid
+          AND tenant_id = $2
+          AND status IN ('pending', 'snoozed')
+        ORDER BY due_at ASC NULLS LAST
+        LIMIT 10
+        """,
+        person_id,
+        tenant_id,
+    )
+    tasks = [_serialize_reminder(r) for r in task_rows]
+
+    # 8. QD overview (all score types)
+    qd_rows = await conn.fetch(
+        """
+        SELECT score_type, current_value, computed_at, reasoning
+        FROM intel_qd_scores
+        WHERE person_id = $1::uuid
+          AND tenant_id = $2
+        """,
+        person_id,
+        tenant_id,
+    )
+    qd_overview = {r["score_type"]: _serialize_qd_score(r) for r in qd_rows}
+
+    # 9. Risk flags — heuristic derivation
+    risk_flags: list[str] = []
+    if lead and lead.get("urgency") in ("high", "critical") and not active_opportunities:
+        risk_flags.append("high_urgency_without_active_opportunity")
+    if not recent_interactions:
+        risk_flags.append("no_recent_interactions")
+    if qd_overview.get("intent_score", {}).get("current_value", 1.0) < 0.3:
+        risk_flags.append("low_intent_score")
+    if not property_interests:
+        risk_flags.append("no_property_interests_recorded")
+
+    # 10. Recommended next actions — simple heuristic
+    recommended_next_actions: list[str] = []
+    if tasks:
+        overdue = [t for t in tasks if t.get("status") == "pending"]
+        if overdue:
+            recommended_next_actions.append(f"Complete pending task: {overdue[0]['title']}")
+    if lead and lead.get("urgency") in ("high", "critical"):
+        recommended_next_actions.append("High-urgency client — prioritize callback within 24h")
+    if not recent_interactions and lead:
+        recommended_next_actions.append("No recent interactions — schedule follow-up")
+
+    return {
+        "client_ref": person_id,
+        "snapshot_type": "client_360",
+        "identity": identity,
+        "account_links": account_links,
+        "current_lead": lead,
+        "active_opportunities": active_opportunities,
+        "recent_interactions": recent_interactions,
+        "property_interests": property_interests,
+        "tasks": tasks,
+        "qd_overview": qd_overview,
+        "risk_flags": risk_flags,
+        "recommended_next_actions": recommended_next_actions,
+        "note": "Derived read model. Not primary truth. Refresh from canonical tables.",
+    }
+
+
+async def get_contact_list(
+    conn: Any,
+    tenant_id: str,
+    search: str | None = None,
+    buyer_type: str | None = None,
+    status: str | None = None,
+    limit: int = 50,
+    offset: int = 0,
+) -> dict[str, Any]:
+    """
+    Paginated contact list with lead status and QD summary.
+    Implements the 'summary query' pattern from Doc 09.
+    """
+    clauses: list[str] = ["p.tenant_id = $1"]
+    params: list[Any] = [tenant_id]
+
+    if search:
+        params.append(f"%{search}%")
+        clauses.append(
+            f"(p.full_name ILIKE ${len(params)} OR p.primary_email ILIKE ${len(params)} OR p.primary_phone ILIKE ${len(params)})"
+        )
+    if buyer_type:
+        params.append(buyer_type)
+        clauses.append(f"p.buyer_type = ${len(params)}")
+    if status:
+        params.append(status)
+        clauses.append(f"cl.status = ${len(params)}::crm_lead_status")
+
+    where = "WHERE " + " AND ".join(clauses)
+    params_for_count = params.copy()
+
+    params.append(limit)
+    params.append(offset)
+
+    query = f"""
+        SELECT
+            p.person_id,
+            p.full_name,
+            p.primary_email,
+            p.primary_phone,
+            p.buyer_type,
+            p.legacy_li_id,
+            p.created_at,
+            cl.lead_id,
+            cl.status AS lead_status,
+            cl.budget_band,
+            cl.urgency,
+            pi.project_name AS primary_interest,
+            COALESCE(qs.intent_value, 0.0) AS intent_score,
+            COALESCE(qs.engagement_value, qs.intent_value, 0.0) AS engagement_score,
+            COALESCE(qs.urgency_value, 0.0) AS urgency_score,
+            (SELECT COUNT(*) FROM intel_interactions ii WHERE ii.person_id = p.person_id AND ii.tenant_id = p.tenant_id) AS interaction_count,
+            (SELECT MAX(happened_at) FROM intel_interactions ii WHERE ii.person_id = p.person_id AND ii.tenant_id = p.tenant_id) AS last_interaction_at,
+            (SELECT COUNT(*) FROM intel_reminders ir WHERE ir.person_id = p.person_id AND ir.tenant_id = p.tenant_id AND ir.status = 'pending') AS pending_tasks
+        FROM crm_people p
+        LEFT JOIN LATERAL (
+            SELECT lead_id, status, budget_band, urgency
+            FROM crm_leads
+            WHERE person_id = p.person_id
+              AND tenant_id = p.tenant_id
+            ORDER BY created_at DESC
+            LIMIT 1
+        ) cl ON TRUE
+        LEFT JOIN LATERAL (
+            SELECT project_name
+            FROM crm_property_interests
+            WHERE person_id = p.person_id
+              AND tenant_id = p.tenant_id
+            ORDER BY priority ASC, created_at DESC
+            LIMIT 1
+        ) pi ON TRUE
+        LEFT JOIN LATERAL (
+            SELECT
+                MAX(CASE WHEN score_type = 'intent_score' THEN current_value END) AS intent_value,
+                MAX(CASE WHEN score_type = 'engagement_score' THEN current_value END) AS engagement_value,
+                MAX(CASE WHEN score_type = 'urgency_score' THEN current_value END) AS urgency_value
+            FROM intel_qd_scores
+            WHERE person_id = p.person_id
+              AND tenant_id = p.tenant_id
+        ) qs ON TRUE
+        {where}
+        ORDER BY last_interaction_at DESC NULLS LAST, p.created_at DESC
+        LIMIT ${len(params) - 1} OFFSET ${len(params)}
+    """
+
+    count_query = f"""
+        SELECT COUNT(*)
+        FROM crm_people p
+        LEFT JOIN crm_leads cl ON cl.person_id = p.person_id AND cl.tenant_id = p.tenant_id
+        {where}
+    """
+
+    rows = await conn.fetch(query, *params)
+    total_row = await conn.fetchrow(count_query, *params_for_count)
+    total = int(total_row[0]) if total_row else 0
+
+    contacts = []
+    for r in rows:
+        contacts.append({
+            "person_id": str(r["person_id"]),
+            "full_name": r["full_name"],
+            "primary_email": r["primary_email"],
+            "primary_phone": r["primary_phone"],
+            "buyer_type": r["buyer_type"],
+            "lead_id": str(r["lead_id"]) if r["lead_id"] else None,
+            "legacy_li_id": str(r["legacy_li_id"]) if r["legacy_li_id"] else None,
+            "lead_status": r["lead_status"],
+            "budget_band": r["budget_band"],
+            "urgency": r["urgency"],
+            "primary_interest": r["primary_interest"],
+            "intent_score": float(r["intent_score"]),
+            "engagement_score": float(r["engagement_score"]),
+            "urgency_score": float(r["urgency_score"]),
+            "interaction_count": int(r["interaction_count"]),
+            "last_interaction_at": r["last_interaction_at"].isoformat() if r["last_interaction_at"] else None,
+            "pending_tasks": int(r["pending_tasks"]),
+            "created_at": r["created_at"].isoformat() if r["created_at"] else None,
+        })
+
+    return {
+        "contacts": contacts,
+        "total": total,
+        "limit": limit,
+        "offset": offset,
+    }

core/services/services/comms_evolution_provider.py

@@ -0,0 +1,90 @@
+"""
+Evolution API (https://github.com/EvolutionAPI/evolution-api) adapter.
+"""
+
+import httpx
+from typing import Any, Dict, List, Optional
+
+from .comms_provider import CommsProvider
+
+
+class EvolutionProvider(CommsProvider):
+    def _headers(self) -> Dict[str, str]:
+        return {"Content-Type": "application/json", "apikey": self.api_key}
+
+    async def _request(self, method: str, path: str, json_data: Optional[Dict] = None) -> Dict[str, Any]:
+        url = f"{self.base_url}{path}"
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            resp = await client.request(method, url, headers=self._headers(), json=json_data)
+            resp.raise_for_status()
+            return resp.json()
+
+    async def send_message(self, phone: str, message: str, message_type: str = "text", **kwargs) -> Dict[str, Any]:
+        instance = self.instance_id or "default"
+        payload = {
+            "number": phone,
+            "text": message,
+            "options": {"delay": 1200, "presence": "composing"},
+        }
+        result = await self._request("POST", f"/message/sendText/{instance}", payload)
+        ext_id = result.get("key", {}).get("id") if isinstance(result, dict) else None
+
+        return {
+            "success": True,
+            "provider": "evolution",
+            "external_message_id": ext_id,
+            "status": "sent",
+            "raw": result,
+        }
+
+    async def normalize_webhook(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Evolution webhook v2 shape:
+        {
+          "event": "messages.upsert",
+          "instance": "default",
+          "data": {
+            "key": {"remoteJid": "123@s.whatsapp.net", "fromMe": false, "id": "..."},
+            "message": {"conversation": "Hello"},
+            "messageTimestamp": 1710000000, ...
+          }
+        }
+        """
+        event = payload.get("event", "")
+        data = payload.get("data", {})
+        key = data.get("key", {})
+        remote_jid = key.get("remoteJid", "")
+        phone = remote_jid.replace("@s.whatsapp.net", "").replace("@g.us", "")
+        msg_content = data.get("message", {})
+        body = msg_content.get("conversation", "") or msg_content.get("extendedTextMessage", {}).get("text", "")
+        direction = "outbound" if key.get("fromMe") else "inbound"
+
+        return {
+            "provider": "evolution",
+            "external_message_id": key.get("id"),
+            "phone_e164": phone,
+            "direction": direction,
+            "message_type": "text",
+            "body": body,
+            "media_url": None,
+            "raw": payload,
+            "timestamp": data.get("messageTimestamp"),
+        }
+
+    async def test_connection(self) -> Dict[str, Any]:
+        try:
+            instance = self.instance_id or "default"
+            info = await self._request("GET", f"/instance/connectionState/{instance}")
+            return {
+                "success": True,
+                "message": f"Evolution instance '{instance}' state retrieved.",
+                "account_info": info,
+            }
+        except Exception as exc:
+            return {
+                "success": False,
+                "message": f"Evolution connection failed: {exc}",
+            }
+
+    async def fetch_templates(self) -> List[Dict[str, Any]]:
+        return []

core/services/services/comms_ingest.py

@@ -0,0 +1,239 @@
+"""Inbound communications ingestion for Velocity CRM."""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+from datetime import UTC, datetime
+from typing import Any
+from uuid import UUID
+
+PHONEUTILS_AVAILABLE = False
+try:
+    import phonenumbers
+    from phonenumbers import NumberParseException
+
+    PHONEUTILS_AVAILABLE = True
+except ImportError:
+    phonenumbers = None  # type: ignore[assignment]
+    NumberParseException = Exception  # type: ignore[assignment]
+
+DEFAULT_COUNTRY = os.getenv("COMMS_DEFAULT_COUNTRY_CODE", "91")
+
+
+def normalize_phone(phone: str, default_region: str = DEFAULT_COUNTRY) -> str | None:
+    """Return an E.164-like phone number suitable for provider and CRM matching."""
+    if not phone:
+        return None
+
+    cleaned = re.sub(r"[^\d+]", "", phone.strip())
+    if cleaned.startswith("00"):
+        cleaned = "+" + cleaned[2:]
+    if not cleaned.startswith("+"):
+        cleaned = f"+{default_region}{cleaned}"
+
+    if PHONEUTILS_AVAILABLE and phonenumbers is not None:
+        try:
+            parsed = phonenumbers.parse(cleaned, None)
+            if phonenumbers.is_valid_number(parsed):
+                return phonenumbers.format_number(parsed, phonenumbers.PhoneNumberFormat.E164)
+        except NumberParseException:
+            pass
+
+    return cleaned if re.match(r"^\+\d{7,15}$", cleaned) else None
+
+
+def _phone_digits(phone: str) -> str:
+    return re.sub(r"\D+", "", phone or "")
+
+
+def _crm_channel(channel: str) -> str:
+    allowed = {"whatsapp", "sms", "call", "email", "website", "walk_in", "other"}
+    return channel if channel in allowed else "other"
+
+
+async def get_or_create_thread(
+    pool,
+    phone_e164: str,
+    provider: str,
+    external_thread_id: str | None = None,
+    display_name: str | None = None,
+    channel: str = "whatsapp",
+) -> dict[str, Any]:
+    async with pool.acquire() as conn:
+        row = await conn.fetchrow(
+            """
+            SELECT thread_id, person_id, status, unread_count
+            FROM comms_threads
+            WHERE phone_e164 = $1 AND provider = $2
+            LIMIT 1
+            """,
+            phone_e164,
+            provider,
+        )
+        if row:
+            return dict(row)
+
+        person_id = None
+        try:
+            person_row = await conn.fetchrow(
+                """
+                SELECT person_id
+                FROM crm_people
+                WHERE primary_phone = $1
+                   OR regexp_replace(COALESCE(primary_phone, ''), '[^0-9]', '', 'g') = $2
+                LIMIT 1
+                """,
+                phone_e164,
+                _phone_digits(phone_e164),
+            )
+            person_id = person_row["person_id"] if person_row else None
+        except Exception:
+            person_id = None
+
+        new_id = await conn.fetchval(
+            """
+            INSERT INTO comms_threads
+                (provider, external_thread_id, person_id, phone_e164, display_name, channel, status, unread_count)
+            VALUES ($1, $2, $3, $4, $5, $6, 'open', 1)
+            RETURNING thread_id
+            """,
+            provider,
+            external_thread_id,
+            person_id,
+            phone_e164,
+            display_name or phone_e164,
+            channel,
+        )
+        return {
+            "thread_id": new_id,
+            "person_id": person_id,
+            "status": "open",
+            "unread_count": 1,
+            "is_new": True,
+        }
+
+
+async def store_message(
+    pool,
+    thread_id: UUID,
+    provider: str,
+    external_message_id: str | None,
+    direction: str,
+    message_type: str,
+    body: str,
+    media_url: str | None = None,
+    raw_payload: dict[str, Any] | None = None,
+    sent_at: datetime | None = None,
+) -> UUID:
+    async with pool.acquire() as conn:
+        msg_id = await conn.fetchval(
+            """
+            INSERT INTO comms_messages
+                (thread_id, provider, external_message_id, direction, message_type, body, media_url,
+                 delivery_status, sent_at, raw_payload)
+            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10::jsonb)
+            RETURNING message_id
+            """,
+            thread_id,
+            provider,
+            external_message_id,
+            direction,
+            message_type,
+            body,
+            media_url,
+            "delivered" if direction == "inbound" else "sent",
+            sent_at or datetime.now(UTC),
+            json.dumps(raw_payload or {}),
+        )
+
+        unread_delta = 1 if direction == "inbound" else 0
+        await conn.execute(
+            """
+            UPDATE comms_threads
+            SET last_message_at = NOW(), unread_count = unread_count + $2, updated_at = NOW()
+            WHERE thread_id = $1
+            """,
+            thread_id,
+            unread_delta,
+        )
+        return msg_id
+
+
+async def maybe_create_crm_interaction(pool, person_id: UUID, body: str, channel: str = "whatsapp") -> None:
+    """Mirror inbound comms into canonical CRM intelligence tables when present."""
+    if not person_id:
+        return
+    try:
+        async with pool.acquire() as conn:
+            exists = await conn.fetchval("SELECT to_regclass('public.intel_interactions') IS NOT NULL")
+            if not exists:
+                return
+            interaction_id = await conn.fetchval(
+                """
+                INSERT INTO intel_interactions
+                    (person_id, channel, interaction_type, happened_at, summary, source_ref, metadata_json)
+                VALUES ($1, $2::intel_channel, 'message', NOW(), $3, 'comms_ingest', $4::jsonb)
+                RETURNING interaction_id
+                """,
+                person_id,
+                _crm_channel(channel),
+                body[:500],
+                json.dumps({"source": "comms", "direction": "inbound"}),
+            )
+            if await conn.fetchval("SELECT to_regclass('public.intel_messages') IS NOT NULL"):
+                await conn.execute(
+                    """
+                    INSERT INTO intel_messages
+                        (interaction_id, sender_role, sender_name, message_text, delivered_at, metadata_json)
+                    VALUES ($1, 'lead', NULL, $2, NOW(), $3::jsonb)
+                    """,
+                    interaction_id,
+                    body,
+                    json.dumps({"source": "comms"}),
+                )
+    except Exception:
+        return
+
+
+async def ingest_inbound_message(pool, normalized_payload: dict[str, Any]) -> dict[str, Any]:
+    phone = normalize_phone(normalized_payload.get("phone_e164") or normalized_payload.get("phone") or "")
+    if not phone:
+        raise ValueError("Missing phone_e164 in payload")
+
+    provider = normalized_payload.get("provider", "unknown")
+    channel = normalized_payload.get("channel", "whatsapp")
+    thread = await get_or_create_thread(
+        pool,
+        phone_e164=phone,
+        provider=provider,
+        external_thread_id=normalized_payload.get("external_thread_id"),
+        display_name=normalized_payload.get("display_name") or phone,
+        channel=channel,
+    )
+
+    timestamp = normalized_payload.get("timestamp")
+    sent_at = datetime.fromtimestamp(timestamp, UTC) if timestamp else None
+    msg_id = await store_message(
+        pool,
+        thread_id=thread["thread_id"],
+        provider=provider,
+        external_message_id=normalized_payload.get("external_message_id"),
+        direction=normalized_payload.get("direction", "inbound"),
+        message_type=normalized_payload.get("message_type", "text"),
+        body=normalized_payload.get("body", ""),
+        media_url=normalized_payload.get("media_url"),
+        raw_payload=normalized_payload.get("raw"),
+        sent_at=sent_at,
+    )
+
+    if thread.get("person_id") and normalized_payload.get("direction", "inbound") == "inbound":
+        await maybe_create_crm_interaction(pool, thread["person_id"], normalized_payload.get("body", ""), channel)
+
+    return {
+        "thread_id": str(thread["thread_id"]),
+        "message_id": str(msg_id),
+        "person_id": str(thread["person_id"]) if thread.get("person_id") else None,
+        "is_new_thread": thread.get("is_new", False),
+    }

core/services/services/comms_provider.py

@@ -0,0 +1,63 @@
+"""
+Abstract provider interface for Velocity Comms.
+"""
+
+import os
+from abc import ABC, abstractmethod
+from typing import Any, Dict, List, Optional
+
+
+class CommsProvider(ABC):
+    def __init__(self, base_url: str, api_key: str, instance_id: Optional[str] = None):
+        self.base_url = base_url.rstrip("/")
+        self.api_key = api_key
+        self.instance_id = instance_id
+
+    @abstractmethod
+    async def send_message(self, phone: str, message: str, message_type: str = "text", **kwargs) -> Dict[str, Any]:
+        """Send a message. Return provider response dict."""
+        ...
+
+    @abstractmethod
+    async def normalize_webhook(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        """Convert provider webhook payload to Velocity canonical format."""
+        ...
+
+    @abstractmethod
+    async def test_connection(self) -> Dict[str, Any]:
+        """Test provider connectivity. Return {success, message, account_info}."""
+        ...
+
+    async def fetch_templates(self) -> List[Dict[str, Any]]:
+        """Optional: fetch message templates."""
+        return []
+
+    async def get_media(self, media_id: str) -> Optional[bytes]:
+        """Optional: download media bytes."""
+        return None
+
+    async def send_template(self, phone: str, template_name: str, language: str, components: Optional[List] = None) -> Dict[str, Any]:
+        """Optional: send a template message."""
+        raise NotImplementedError("Templates not supported by this provider.")
+
+
+class MockProvider(CommsProvider):
+    """Mock provider for local development and UI previews."""
+
+    async def send_message(self, phone: str, message: str, message_type: str = "text", **kwargs) -> Dict[str, Any]:
+        return {
+            "success": True,
+            "provider": "mock",
+            "external_message_id": f"mock-{os.urandom(4).hex()}",
+            "status": "sent",
+        }
+
+    async def normalize_webhook(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        return payload
+
+    async def test_connection(self) -> Dict[str, Any]:
+        return {
+            "success": True,
+            "message": "Mock provider is always healthy.",
+            "account_info": {"mode": "mock"},
+        }

core/services/services/comms_waha_provider.py

@@ -0,0 +1,95 @@
+"""
+WAHA (https://github.com/devlikeapro/waha) adapter.
+WAHA exposes a simple HTTP API for WhatsApp Web.
+"""
+
+import httpx
+from typing import Any, Dict, Optional
+
+from .comms_provider import CommsProvider
+
+
+class WahaProvider(CommsProvider):
+    async def _request(self, method: str, path: str, json_data: Optional[Dict] = None) -> Dict[str, Any]:
+        url = f"{self.base_url}/api{path}"
+        headers = {"Content-Type": "application/json"}
+        if self.api_key:
+            headers["X-Api-Key"] = self.api_key
+
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            resp = await client.request(method, url, headers=headers, json=json_data)
+            resp.raise_for_status()
+            return resp.json()
+
+    async def send_message(self, phone: str, message: str, message_type: str = "text", **kwargs) -> Dict[str, Any]:
+        chat_id = f"{phone}@c.us"
+        payload = {
+            "chatId": chat_id,
+            "text": message,
+            "session": self.instance_id or "default",
+        }
+        if message_type == "image" and kwargs.get("media_url"):
+            payload["caption"] = message
+            payload["media"] = kwargs["media_url"]
+            result = await self._request("POST", "/sendImage", payload)
+        else:
+            result = await self._request("POST", "/sendText", payload)
+
+        return {
+            "success": True,
+            "provider": "waha",
+            "external_message_id": result.get("id"),
+            "status": "sent",
+            "raw": result,
+        }
+
+    async def normalize_webhook(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        WAHA webhook payload shape (v2024):
+        {
+          "event": "message",
+          "session": "default",
+          "payload": {
+            "id": "true_123@c.us_3EB0...",
+            "timestamp": 1710000000,
+            "from": "123@c.us",
+            "to": "456@c.us",
+            "body": "Hello",
+            "hasMedia": false, ...
+          }
+        }
+        """
+        event = payload.get("event", "")
+        pl = payload.get("payload", {})
+        from_jid = pl.get("from", "")
+        phone = from_jid.replace("@c.us", "").replace("@g.us", "")
+        direction = "inbound" if event == "message" and not pl.get("fromMe") else "outbound"
+
+        return {
+            "provider": "waha",
+            "external_message_id": pl.get("id"),
+            "phone_e164": phone,
+            "direction": direction,
+            "message_type": "image" if pl.get("hasMedia") else "text",
+            "body": pl.get("body", ""),
+            "media_url": pl.get("mediaUrl"),
+            "raw": payload,
+            "timestamp": pl.get("timestamp"),
+        }
+
+    async def test_connection(self) -> Dict[str, Any]:
+        try:
+            sessions = await self._request("GET", "/sessions?all=true")
+            return {
+                "success": True,
+                "message": f"Connected to WAHA. Sessions: {len(sessions)}",
+                "account_info": {"sessions": sessions},
+            }
+        except Exception as exc:
+            return {
+                "success": False,
+                "message": f"WAHA connection failed: {exc}",
+            }
+
+    async def get_media(self, media_id: str) -> Optional[bytes]:
+        return None

core/services/services/imports/__init__.py

@@ -0,0 +1,3 @@
+"""
+backend/services/imports/__init__.py
+"""

core/services/services/imports/ingest_service.py

@@ -0,0 +1,286 @@
+"""
+backend/services/imports/ingest_service.py
+CRM Import Ingestion Service
+
+Implements the RawImportBatch → ImportMappingManifest → NormalizedEntityProposal pipeline
+as specified in Doc 08 (Adapter Spec) and Doc 07 (Contracts and Schema Blueprint).
+
+Flow:
+  1. receive CSV upload, store raw batch record
+  2. parse headers and infer column mapping
+  3. validate row structure, detect unresolved columns
+  4. create NormalizedEntityProposal records for review
+  5. queue for human approval before canonical commit
+"""
+from __future__ import annotations
+
+import csv
+import io
+import json
+import logging
+import uuid
+from datetime import datetime, timezone
+from typing import Any
+
+logger = logging.getLogger("velocity.imports.ingest")
+
+# ── Column mapping heuristics ─────────────────────────────────────────────────
+# Maps common source column names → canonical crm_people / crm_leads fields.
+
+CANONICAL_COLUMN_MAP: dict[str, str] = {
+    # Identity
+    "name": "full_name",
+    "full name": "full_name",
+    "client name": "full_name",
+    "contact name": "full_name",
+    "first name": "full_name",
+    "customer name": "full_name",
+    # Email
+    "email": "primary_email",
+    "email address": "primary_email",
+    "e-mail": "primary_email",
+    # Phone
+    "phone": "primary_phone",
+    "mobile": "primary_phone",
+    "contact number": "primary_phone",
+    "mobile number": "primary_phone",
+    "phone number": "primary_phone",
+    # Budget
+    "budget": "budget_band",
+    "budget range": "budget_band",
+    "investment budget": "budget_band",
+    # Project interest
+    "project": "project_name",
+    "project name": "project_name",
+    "interested in": "project_name",
+    "property interest": "project_name",
+    # Source
+    "source": "source_system",
+    "lead source": "source_system",
+    "channel": "source_system",
+    # Status / Stage
+    "status": "status",
+    "lead status": "status",
+    "stage": "status",
+    "funnel stage": "status",
+    # Notes
+    "notes": "notes",
+    "remarks": "notes",
+    "comment": "notes",
+    "comments": "notes",
+    # Buyer type
+    "type": "buyer_type",
+    "client type": "buyer_type",
+    "category": "buyer_type",
+}
+
+REQUIRED_CANONICAL_FIELDS = {"full_name"}
+HIGH_RISK_FIELDS = {"primary_email", "primary_phone"}
+
+
+def _normalize_header(h: str) -> str:
+    return h.strip().lower().replace("_", " ")
+
+
+def infer_column_mapping(headers: list[str]) -> dict[str, Any]:
+    """
+    Produce an ImportMappingManifest-compatible mapping dict.
+    Returns: {
+        mapped: {source_col → canonical_field},
+        unmapped: [source_col, ...],
+        confidence: 0.0-1.0
+    }
+    """
+    mapped: dict[str, str] = {}
+    unmapped: list[str] = []
+
+    for h in headers:
+        normalized = _normalize_header(h)
+        canonical = CANONICAL_COLUMN_MAP.get(normalized)
+        if canonical:
+            mapped[h] = canonical
+        else:
+            unmapped.append(h)
+
+    mapped_count = len(mapped)
+    total = len(headers)
+    confidence = mapped_count / total if total > 0 else 0.0
+
+    return {
+        "mapped": mapped,
+        "unmapped": unmapped,
+        "mapped_count": mapped_count,
+        "unmapped_count": len(unmapped),
+        "confidence": round(confidence, 3),
+    }
+
+
+def parse_csv_content(content: str) -> dict[str, Any]:
+    """
+    Parse CSV content, detect headers, and extract rows.
+    Returns: {headers, rows, row_count, parse_errors}
+    """
+    reader = csv.DictReader(io.StringIO(content))
+    headers = reader.fieldnames or []
+    rows: list[dict[str, Any]] = []
+    parse_errors: list[str] = []
+
+    for i, row in enumerate(reader):
+        try:
+            rows.append(dict(row))
+        except Exception as e:
+            parse_errors.append(f"Row {i + 2}: {str(e)}")
+
+    return {
+        "headers": list(headers),
+        "rows": rows,
+        "row_count": len(rows),
+        "parse_errors": parse_errors,
+    }
+
+
+def build_normalized_proposals(
+    rows: list[dict[str, Any]],
+    mapping: dict[str, str],
+    batch_id: str,
+    source_system: str = "csv_upload",
+) -> list[dict[str, Any]]:
+    """
+    Convert raw CSV rows to NormalizedEntityProposal payloads.
+    One proposal per row — each must be approved before canonical commit.
+    """
+    proposals: list[dict[str, Any]] = []
+    now = datetime.now(timezone.utc).isoformat()
+
+    for i, row in enumerate(rows):
+        canonical: dict[str, Any] = {}
+        unresolved: list[str] = []
+        confidence = 1.0
+
+        for src_col, canonical_field in mapping.items():
+            val = row.get(src_col, "").strip()
+            if val:
+                canonical[canonical_field] = val
+            else:
+                unresolved.append(src_col)
+
+        # Validate required fields
+        review_required = False
+        missing_required = [f for f in REQUIRED_CANONICAL_FIELDS if not canonical.get(f)]
+        if missing_required:
+            review_required = True
+            confidence = max(0.0, confidence - 0.4)
+
+        # Flag high-risk fields (email/phone) if empty
+        missing_high_risk = [f for f in HIGH_RISK_FIELDS if not canonical.get(f)]
+        if missing_high_risk:
+            confidence = max(0.0, confidence - 0.1 * len(missing_high_risk))
+
+        proposal: dict[str, Any] = {
+            "proposal_id": str(uuid.uuid4()),
+            "batch_id": batch_id,
+            "row_number": i + 2,
+            "entity_type": "crm_person_with_lead",
+            "canonical_payload": canonical,
+            "raw_row": row,
+            "unresolved_fields": unresolved,
+            "missing_required": missing_required,
+            "confidence": round(confidence, 3),
+            "review_required": review_required,
+            "status": "proposed",
+            "created_at": now,
+            "source_system": source_system,
+        }
+        proposals.append(proposal)
+
+    return proposals
+
+
+def create_import_batch_record(
+    filename: str,
+    row_count: int,
+    mapping_manifest: dict[str, Any],
+    source_system: str = "csv_upload",
+    uploaded_by_id: str | None = None,
+    tenant_id: str | None = None,
+) -> dict[str, Any]:
+    """
+    Build the workflow_import_batches record payload.
+    """
+    now = datetime.now(timezone.utc).isoformat()
+    return {
+        "batch_id": str(uuid.uuid4()),
+        "source_system": source_system,
+        "uploaded_filename": filename,
+        "mime_type": "text/csv",
+        "row_count": row_count,
+        "mapped_count": mapping_manifest.get("mapped_count", 0),
+        "unresolved_count": mapping_manifest.get("unmapped_count", 0),
+        "uploaded_by": uploaded_by_id,
+        "tenant_id": tenant_id,
+        "lifecycle": "parsed",
+        "mapping_manifest": mapping_manifest,
+        "created_at": now,
+        "updated_at": now,
+    }
+
+
+async def persist_import_batch(conn: Any, batch: dict[str, Any]) -> str:
+    """
+    Insert a workflow_import_batches row and return batch_id.
+    """
+    await conn.execute(
+        """
+        INSERT INTO workflow_import_batches (
+            batch_id, tenant_id, source_system, uploaded_filename, mime_type, row_count,
+            mapped_count, unresolved_count, uploaded_by, lifecycle, mapping_manifest,
+            created_at, updated_at
+        ) VALUES (
+            $1::uuid, $2, $3, $4, $5, $6, $7, $8,
+            $9::uuid, $10::import_lifecycle, $11::jsonb, NOW(), NOW()
+        )
+        """,
+        batch["batch_id"],
+        batch["tenant_id"],
+        batch["source_system"],
+        batch.get("uploaded_filename", "unknown.csv"),
+        batch.get("mime_type", "text/csv"),
+        batch.get("row_count", 0),
+        batch.get("mapped_count", 0),
+        batch.get("unresolved_count", 0),
+        batch.get("uploaded_by"),
+        batch.get("lifecycle", "parsed"),
+        json.dumps(batch.get("mapping_manifest", {})),
+    )
+    return batch["batch_id"]
+
+
+async def persist_proposals_as_workflow_actions(
+    conn: Any, proposals: list[dict[str, Any]], tenant_id: str
+) -> int:
+    """
+    Insert proposals into workflow_actions table for human review.
+    Returns inserted count.
+    """
+    inserted = 0
+    for p in proposals:
+        await conn.execute(
+            """
+            INSERT INTO workflow_actions (
+                action_id, tenant_id, action_type, target_domain, proposal_payload,
+                reasoning_summary, confidence, status, approval_required,
+                created_by_agent, created_at, updated_at
+            ) VALUES (
+                $1::uuid, $2, 'import_proposal', 'crm', $3::jsonb,
+                $4, $5, 'pending'::wf_status, $6, 'ingest_service', NOW(), NOW()
+            )
+            """,
+            p["proposal_id"],
+            tenant_id,
+            json.dumps(p),
+            f"Import row {p['row_number']}: {p['canonical_payload'].get('full_name', 'unknown')}",
+            p["confidence"],
+            p["review_required"],
+        )
+        inserted += 1
+    return inserted

core/services/services/mcp_registry.py

@@ -0,0 +1,136 @@
+from __future__ import annotations
+
+import os
+from typing import Any
+
+import httpx
+
+
+class MCPRegistry:
+    def __init__(self) -> None:
+        self._tools = {
+            "local_property_rag": {
+                "description": "Searches project, property, and unit metadata from root CRM data.",
+                "transport": "python_local",
+            },
+            "crm_search": {
+                "description": "Queries lead and interaction state from the root PostgreSQL CRM schema.",
+                "transport": "python_local",
+            },
+            "external_search": {
+                "description": "Abstract external search slot inspired by Sourik's Brave/DDG tools.",
+                "transport": "adapter_slot",
+            },
+        }
+
+    def list_tools(self) -> list[dict[str, Any]]:
+        return [{"name": name, **meta} for name, meta in self._tools.items()]
+
+    async def execute(self, tool_name: str, query: str, *, crm_pool: Any | None = None) -> dict[str, Any]:
+        if tool_name not in self._tools:
+            raise KeyError(f"Unknown MCP tool '{tool_name}'.")
+        if tool_name == "external_search":
+            return await self._external_search(query)
+        if tool_name == "crm_search":
+            return await self._crm_search(query, crm_pool)
+        if tool_name == "local_property_rag":
+            return await self._local_property_rag(query, crm_pool)
+        return {"tool": tool_name, "query": query, "status": "unsupported"}
+
+    async def _external_search(self, query: str) -> dict[str, Any]:
+        brave_key = os.getenv("BRAVE_API_KEY", "")
+        if brave_key and not brave_key.startswith("PLACEHOLDER"):
+            async with httpx.AsyncClient(timeout=15.0) as client:
+                response = await client.get(
+                    "https://api.search.brave.com/res/v1/web/search",
+                    headers={"Accept": "application/json", "X-Subscription-Token": brave_key},
+                    params={"q": query, "count": 5},
+                )
+                response.raise_for_status()
+            payload = response.json()
+            results = [
+                {
+                    "title": item.get("title"),
+                    "url": item.get("url"),
+                    "snippet": item.get("description"),
+                }
+                for item in payload.get("web", {}).get("results", [])
+            ]
+            return {"tool": "external_search", "query": query, "status": "ok", "provider": "brave", "results": results}
+
+        async with httpx.AsyncClient(timeout=15.0) as client:
+            response = await client.get(
+                "https://api.duckduckgo.com/",
+                params={"q": query, "format": "json", "no_html": 1, "no_redirect": 1},
+            )
+            response.raise_for_status()
+        payload = response.json()
+        results: list[dict[str, Any]] = []
+        abstract = payload.get("AbstractText")
+        if abstract:
+            results.append(
+                {
+                    "title": payload.get("Heading") or query,
+                    "url": payload.get("AbstractURL"),
+                    "snippet": abstract,
+                }
+            )
+        for topic in payload.get("RelatedTopics", [])[:5]:
+            if isinstance(topic, dict) and topic.get("Text"):
+                results.append(
+                    {
+                        "title": topic.get("Text", "")[:80],
+                        "url": topic.get("FirstURL"),
+                        "snippet": topic.get("Text"),
+                    }
+                )
+        return {"tool": "external_search", "query": query, "status": "ok", "provider": "duckduckgo", "results": results}
+
+    async def _crm_search(self, query: str, crm_pool: Any | None) -> dict[str, Any]:
+        if crm_pool is None:
+            return {"tool": "crm_search", "query": query, "status": "unavailable", "reason": "crm_pool_missing"}
+        async with crm_pool.acquire() as conn:
+            rows = await conn.fetch(
+                """
+                SELECT id, name, email, phone, source, qualification, score, kanban_status, budget, unit_interest
+                FROM leads
+                WHERE LOWER(name) LIKE $1
+                   OR LOWER(COALESCE(email, '')) LIKE $1
+                   OR LOWER(COALESCE(phone, '')) LIKE $1
+                   OR LOWER(COALESCE(notes, '')) LIKE $1
+                ORDER BY score DESC, updated_at DESC
+                LIMIT 10
+                """,
+                f"%{query.lower()}%",
+            )
+        return {
+            "tool": "crm_search",
+            "query": query,
+            "status": "ok",
+            "results": [dict(row) for row in rows],
+        }
+
+    async def _local_property_rag(self, query: str, crm_pool: Any | None) -> dict[str, Any]:
+        if crm_pool is None:
+            return {"tool": "local_property_rag", "query": query, "status": "unavailable", "reason": "crm_pool_missing"}
+        async with crm_pool.acquire() as conn:
+            rows = await conn.fetch(
+                """
+                SELECT id, name, source, budget, unit_interest, metadata
+                FROM leads
+                WHERE LOWER(COALESCE(unit_interest, '')) LIKE $1
+                   OR LOWER(COALESCE(notes, '')) LIKE $1
+                ORDER BY score DESC, updated_at DESC
+                LIMIT 10
+                """,
+                f"%{query.lower()}%",
+            )
+        return {
+            "tool": "local_property_rag",
+            "query": query,
+            "status": "ok",
+            "results": [dict(row) for row in rows],
+        }
+
+
+mcp_registry = MCPRegistry()

core/services/services/nemoclaw_client.py

@@ -0,0 +1,354 @@
+"""
+backend/services/nemoclaw_client.py - NemoClaw inference client.
+
+Production path:
+  1. Shared SGLang / OpenAI-compatible coding runtime.
+
+Compatibility:
+  - Legacy NEMOCLAW_* env names are still honored.
+  - Legacy OLLAMA_BASE_URL can still seed the base URL, but Ollama is no longer
+    a production fallback path.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import re
+import time
+from dataclasses import dataclass, field
+from typing import Optional
+
+import httpx
+
+logger = logging.getLogger("velocity.nemoclaw")
+
+NEMOCLAW_TIMEOUT = float(os.getenv("NEMOCLAW_TIMEOUT_S", "45.0"))
+NEMOCLAW_TEMPERATURE = float(os.getenv("NEMOCLAW_TEMPERATURE", "0.2"))
+
+SGLANG_BASE_URL = os.getenv(
+    "SGLANG_BASE_URL",
+    os.getenv(
+        "NEMOCLAW_BASE_URL",
+        os.getenv("LLM_BASE_URL", os.getenv("OLLAMA_BASE_URL", "https://llm.desineuron.in")),
+    ),
+).rstrip("/")
+SGLANG_CHAT_URL = os.getenv(
+    "SGLANG_CHAT_URL",
+    os.getenv("NEMOCLAW_CHAT_URL", f"{SGLANG_BASE_URL}/v1/chat/completions"),
+)
+SGLANG_MODELS_URL = os.getenv("SGLANG_MODELS_URL", f"{SGLANG_BASE_URL}/v1/models")
+SGLANG_MODEL = os.getenv(
+    "SGLANG_MODEL",
+    os.getenv("NEMOCLAW_MODEL", os.getenv("OLLAMA_MODEL", "qwen3.6:35b-a3b")),
+)
+SGLANG_API_TOKEN = os.getenv("SGLANG_API_TOKEN", os.getenv("NEMOCLAW_API_TOKEN", ""))
+
+_PROMPT_DIR = os.getenv("NEMOCLAW_PROMPT_DIR", "/opt/dlami/nvme/nemoclaw/prompts")
+
+
+def _load_system_prompt(name: str) -> str:
+    local_fallback = os.path.join(
+        os.path.dirname(__file__), "..", "nemoclaw_prompts", f"{name}.md"
+    )
+    for path in (os.path.join(_PROMPT_DIR, f"{name}.md"), local_fallback):
+        try:
+            with open(path, encoding="utf-8") as handle:
+                return "\n".join(
+                    line for line in handle.read().splitlines() if not line.startswith("#")
+                ).strip()
+        except FileNotFoundError:
+            continue
+    logger.warning("Prompt '%s' not found, using inline fallback.", name)
+    return _PROMPTS.get(name, "")
+
+
+_PROMPTS = {
+    "qd_calculator": (
+        "You are a behavioral intelligence analyst for a luxury real estate sales platform.\n"
+        "Compute a Quantum Dynamics score between 1 and 100 using blend shapes, CRM context, "
+        "and the active scene label when present.\n"
+        'Respond with JSON only: {"qd_score": <int>, "reasoning": "<one sentence>", "confidence": <float>}'
+    ),
+    "lead_tagger": (
+        "You are a lead intelligence analyst. Classify a real estate lead as HNI or NRI.\n"
+        'Respond with JSON only: {"tags_to_add": [...], "tags_to_remove": []}'
+    ),
+    "cctv_profiler": (
+        "You are a visitor profiling analyst for a luxury real estate development CCTV system.\n"
+        'Respond with JSON only: {"wealth_indicator": "HNI"|"standard"|"unknown", '
+        '"vehicle_class": "luxury"|"standard"|"unknown", "tags_to_add": [...], "notes": "<string>"}'
+    ),
+}
+
+
+@dataclass
+class QDResult:
+    qd_score: int
+    reasoning: str
+    confidence: float
+
+
+@dataclass
+class TagResult:
+    tags_to_add: list[str] = field(default_factory=list)
+    tags_to_remove: list[str] = field(default_factory=list)
+
+
+@dataclass
+class CCTVProfileResult:
+    wealth_indicator: str
+    vehicle_class: str
+    tags_to_add: list[str] = field(default_factory=list)
+    notes: str = ""
+
+
+async def _attempt_chat(
+    *,
+    label: str,
+    url: str,
+    model: str,
+    system_content: str,
+    user_content: str,
+    timeout: float,
+    headers: dict[str, str],
+) -> dict:
+    payload = {
+        "model": model,
+        "messages": [
+            {"role": "system", "content": system_content},
+            {"role": "user", "content": user_content},
+        ],
+        "temperature": NEMOCLAW_TEMPERATURE,
+        "response_format": {"type": "json_object"},
+        "max_tokens": 1024,
+    }
+    async with httpx.AsyncClient(timeout=timeout) as client:
+        response = await client.post(url, json=payload, headers=headers)
+        response.raise_for_status()
+    body = response.json()
+    raw_content = body["choices"][0]["message"]["content"]
+    logger.debug("NemoClaw response via %s: %s", label, raw_content[:200])
+    return _parse_model_response(raw_content)
+
+
+def _extract_text(raw_content: object) -> str:
+    if isinstance(raw_content, str):
+        return raw_content
+    if isinstance(raw_content, list):
+        parts: list[str] = []
+        for item in raw_content:
+            if isinstance(item, dict):
+                text = item.get("text")
+                if isinstance(text, str):
+                    parts.append(text)
+        return "\n".join(parts).strip()
+    return str(raw_content)
+
+
+def _parse_model_response(raw_content: object) -> dict:
+    text = _extract_text(raw_content).strip()
+    if not text:
+        return {}
+    try:
+        return json.loads(text)
+    except json.JSONDecodeError:
+        start = text.find("{")
+        end = text.rfind("}")
+        if start != -1 and end != -1 and end > start:
+            candidate = text[start : end + 1]
+            try:
+                return json.loads(candidate)
+            except json.JSONDecodeError:
+                pass
+
+    parsed: dict[str, object] = {}
+    int_match = re.search(r'"qd_score"\s*:\s*(\d+)', text)
+    if int_match:
+        parsed["qd_score"] = int(int_match.group(1))
+    conf_match = re.search(r'"confidence"\s*:\s*([0-9]*\.?[0-9]+)', text)
+    if conf_match:
+        parsed["confidence"] = float(conf_match.group(1))
+    reason_match = re.search(r'"reasoning"\s*:\s*"([^"]*)"', text)
+    if reason_match:
+        parsed["reasoning"] = reason_match.group(1)
+    wealth_match = re.search(r'"wealth_indicator"\s*:\s*"([^"]*)"', text)
+    if wealth_match:
+        parsed["wealth_indicator"] = wealth_match.group(1)
+    vehicle_match = re.search(r'"vehicle_class"\s*:\s*"([^"]*)"', text)
+    if vehicle_match:
+        parsed["vehicle_class"] = vehicle_match.group(1)
+    notes_match = re.search(r'"notes"\s*:\s*"([^"]*)"', text)
+    if notes_match:
+        parsed["notes"] = notes_match.group(1)
+    tags_match = re.search(r'"tags_to_add"\s*:\s*\[(.*?)\]', text, flags=re.S)
+    if tags_match:
+        parsed["tags_to_add"] = re.findall(r'"([^"]+)"', tags_match.group(1))
+    remove_tags_match = re.search(r'"tags_to_remove"\s*:\s*\[(.*?)\]', text, flags=re.S)
+    if remove_tags_match:
+        parsed["tags_to_remove"] = re.findall(r'"([^"]+)"', remove_tags_match.group(1))
+    if parsed:
+        logger.warning("Recovered partial NemoClaw JSON payload from malformed model output.")
+        return parsed
+    raise json.JSONDecodeError("Unable to parse model JSON", text, 0)
+
+
+async def _nemoclaw_chat(
+    system_content: str,
+    user_content: str,
+    timeout: float = NEMOCLAW_TIMEOUT,
+) -> dict:
+    if not SGLANG_CHAT_URL:
+        raise RuntimeError(
+            "No NemoClaw inference endpoint is configured. Set SGLANG_BASE_URL or NEMOCLAW_BASE_URL."
+        )
+
+    headers = {"Content-Type": "application/json"}
+    if SGLANG_API_TOKEN:
+        headers["Authorization"] = f"Bearer {SGLANG_API_TOKEN}"
+
+    t_start = time.monotonic()
+    try:
+        result = await _attempt_chat(
+            label="sglang",
+            url=SGLANG_CHAT_URL,
+            model=SGLANG_MODEL,
+            system_content=system_content,
+            user_content=user_content,
+            timeout=timeout,
+            headers=headers,
+        )
+        logger.info(
+            "NemoClaw inference via sglang model=%s elapsed=%.2fs",
+            SGLANG_MODEL,
+            time.monotonic() - t_start,
+        )
+        return result
+    except (httpx.ConnectError, httpx.TimeoutException) as exc:
+        raise RuntimeError(f"NemoClaw SGLang endpoint unreachable: {exc}") from exc
+    except httpx.HTTPStatusError as exc:
+        raise RuntimeError(
+            f"NemoClaw SGLang HTTP {exc.response.status_code}: {exc.response.text[:300]}"
+        ) from exc
+    except (KeyError, IndexError, TypeError, json.JSONDecodeError) as exc:
+        raise RuntimeError(f"NemoClaw SGLang returned invalid JSON: {exc}") from exc
+
+
+async def score_qd(
+    *,
+    lead_id: str,
+    batch_id: str,
+    blend_shapes: dict[str, float],
+    video_ts_ms: int,
+    scene_label: Optional[str] = None,
+    crm_context: dict,
+    current_qd_score: Optional[int] = None,
+) -> QDResult:
+    system_prompt = _load_system_prompt("qd_calculator")
+    user_content = json.dumps(
+        {
+            "lead_id": lead_id,
+            "batch_id": batch_id,
+            "video_ts_ms": video_ts_ms,
+            "scene_label": scene_label,
+            "current_qd_score": current_qd_score,
+            "crm_context": crm_context,
+            "blend_shapes": blend_shapes,
+        },
+        indent=2,
+    )
+    data = await _nemoclaw_chat(system_prompt, user_content)
+    raw_score = int(data.get("qd_score", current_qd_score or 50))
+    return QDResult(
+        qd_score=max(1, min(100, raw_score)),
+        reasoning=str(data.get("reasoning", "")),
+        confidence=float(data.get("confidence", 0.7)),
+    )
+
+
+async def tag_lead(
+    *,
+    lead_id: str,
+    phone: str,
+    budget: Optional[str],
+    message_text: str,
+) -> TagResult:
+    system_prompt = _load_system_prompt("lead_tagger")
+    user_content = (
+        f"Lead ID: {lead_id}\n"
+        f"Phone: {phone}\n"
+        f"Budget indicator: {budget or 'unknown'}\n"
+        f"First message: {message_text}"
+    )
+    try:
+        data = await _nemoclaw_chat(system_prompt, user_content)
+    except Exception as exc:
+        logger.error("Lead tagging failed for %s: %s", lead_id, exc)
+        return TagResult()
+    return TagResult(
+        tags_to_add=data.get("tags_to_add", []),
+        tags_to_remove=data.get("tags_to_remove", []),
+    )
+
+
+async def profile_cctv_visitor(
+    *,
+    license_plate: Optional[str],
+    zone: str,
+    face_description: Optional[str] = None,
+    vehicle_description: Optional[str] = None,
+) -> CCTVProfileResult:
+    system_prompt = _load_system_prompt("cctv_profiler")
+    user_content = json.dumps(
+        {
+            "license_plate": license_plate,
+            "zone": zone,
+            "face_description": face_description,
+            "vehicle_description": vehicle_description,
+        },
+        indent=2,
+    )
+    try:
+        data = await _nemoclaw_chat(system_prompt, user_content, timeout=20.0)
+    except Exception as exc:
+        logger.error("CCTV profiling failed (zone=%s): %s", zone, exc)
+        return CCTVProfileResult(wealth_indicator="unknown", vehicle_class="unknown")
+    return CCTVProfileResult(
+        wealth_indicator=data.get("wealth_indicator", "unknown"),
+        vehicle_class=data.get("vehicle_class", "unknown"),
+        tags_to_add=data.get("tags_to_add", []),
+        notes=data.get("notes", ""),
+    )
+
+
+async def health_check() -> dict:
+    headers = {"Content-Type": "application/json"}
+    if SGLANG_API_TOKEN:
+        headers["Authorization"] = f"Bearer {SGLANG_API_TOKEN}"
+
+    results: dict[str, str] = {
+        "model": SGLANG_MODEL,
+        "primary_url": SGLANG_CHAT_URL,
+        "models_url": SGLANG_MODELS_URL,
+    }
+
+    try:
+        async with httpx.AsyncClient(timeout=5.0) as client:
+            models_response = await client.get(SGLANG_MODELS_URL, headers=headers)
+            models_response.raise_for_status()
+            chat_response = await client.post(
+                SGLANG_CHAT_URL,
+                json={
+                    "model": SGLANG_MODEL,
+                    "messages": [{"role": "user", "content": "ping"}],
+                    "max_tokens": 5,
+                },
+                headers=headers,
+            )
+            chat_response.raise_for_status()
+        results["sglang"] = "ok"
+    except Exception as exc:
+        results["sglang"] = f"error: {exc}"
+
+    return results

core/services/services/nemoclaw_runtime.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+import hashlib
+import hmac
+import os
+from typing import Any
+
+
+class NemoclawRuntime:
+    def claim_event(self, source_id: str, payload: dict[str, Any]) -> dict[str, Any]:
+        claim = hashlib.sha256(f"{source_id}:{payload}".encode("utf-8")).hexdigest()[:24]
+        return {"claim_id": claim, "source_id": source_id, "status": "claimed"}
+
+    def verify_webhook_challenge(self, challenge: str, signature: str) -> bool:
+        secret = os.getenv("NEMOCLAW_WEBHOOK_SECRET", "")
+        if not secret:
+            return False
+        expected = hmac.new(secret.encode("utf-8"), challenge.encode("utf-8"), hashlib.sha256).hexdigest()
+        return hmac.compare_digest(expected, signature)
+
+    def build_workflow_dispatch(
+        self,
+        *,
+        prompt: str,
+        tenant_id: str,
+        actor_role: str,
+        component_templates: list[str],
+    ) -> dict[str, Any]:
+        return {
+            "runtime": "python_native_nemoclaw",
+            "tenantId": tenant_id,
+            "actorRole": actor_role,
+            "workflow": "oracle_canvas_generation",
+            "prompt": prompt,
+            "componentTemplates": component_templates,
+            "executionBackend": "comfyui_orchestrated",
+        }
+
+
+nemoclaw_runtime = NemoclawRuntime()

core/services/services/runtime_llm_service.py

@@ -0,0 +1,443 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import os
+import uuid
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from typing import Any
+
+import httpx
+
+logger = logging.getLogger("velocity.runtime_llm")
+
+SGLANG_BASE_URL = os.getenv(
+    "SGLANG_BASE_URL",
+    os.getenv("LLM_BASE_URL", os.getenv("OLLAMA_BASE_URL", "https://llm.desineuron.in")),
+).rstrip("/")
+SGLANG_CHAT_URL = os.getenv("SGLANG_CHAT_URL", f"{SGLANG_BASE_URL}/v1/chat/completions")
+SGLANG_MODELS_URL = os.getenv("SGLANG_MODELS_URL", f"{SGLANG_BASE_URL}/v1/models")
+SGLANG_DEFAULT_MODEL = os.getenv(
+    "SGLANG_MODEL",
+    os.getenv("OLLAMA_MODEL", "qwen3.6:35b-a3b"),
+)
+SGLANG_API_TOKEN = os.getenv("SGLANG_API_TOKEN", "")
+
+RUNTIME_LLM_TIMEOUT_S = float(os.getenv("RUNTIME_LLM_TIMEOUT_S", "90.0"))
+RUNTIME_LLM_CONCURRENCY = int(os.getenv("RUNTIME_LLM_BATCH_CONCURRENCY", "2"))
+
+
+def _utc_now() -> datetime:
+    return datetime.now(UTC)
+
+
+def _utc_iso() -> str:
+    return _utc_now().isoformat()
+
+
+@dataclass
+class RuntimeProvider:
+    provider_id: str
+    base_url: str
+    chat_url: str
+    default_model: str
+    auth_token: str | None = None
+    supports_batch: bool = True
+
+    @property
+    def headers(self) -> dict[str, str]:
+        headers = {"Content-Type": "application/json"}
+        if self.auth_token:
+            headers["Authorization"] = f"Bearer {self.auth_token}"
+        return headers
+
+
+class RuntimeLLMService:
+    def __init__(self) -> None:
+        self._jobs: dict[str, dict[str, Any]] = {}
+
+    def _provider_catalog(self) -> list[RuntimeProvider]:
+        if not SGLANG_CHAT_URL:
+            return []
+        return [
+            RuntimeProvider(
+                provider_id="sglang",
+                base_url=SGLANG_BASE_URL,
+                chat_url=SGLANG_CHAT_URL,
+                default_model=SGLANG_DEFAULT_MODEL,
+                auth_token=SGLANG_API_TOKEN or None,
+            )
+        ]
+
+    def get_provider(self, provider_id: str | None) -> RuntimeProvider:
+        providers = {provider.provider_id: provider for provider in self._provider_catalog()}
+        if provider_id in {"ollama", "nemoclaw"}:
+            provider_id = "sglang"
+        if provider_id:
+            provider = providers.get(provider_id)
+            if provider is None:
+                raise ValueError(f"Unknown provider '{provider_id}'.")
+            return provider
+
+        if "sglang" in providers:
+            return providers["sglang"]
+        raise ValueError("No runtime LLM providers are configured.")
+
+    async def list_providers(self) -> list[dict[str, Any]]:
+        providers: list[dict[str, Any]] = []
+        for provider in self._provider_catalog():
+            models: list[str] = [provider.default_model]
+            status = "offline"
+            error: str | None = None
+
+            try:
+                async with httpx.AsyncClient(timeout=10.0) as client:
+                    response = await client.get(SGLANG_MODELS_URL, headers=provider.headers)
+                    response.raise_for_status()
+                payload = response.json()
+                models = [
+                    str(item.get("id", "")).strip()
+                    for item in payload.get("data", [])
+                    if item.get("id")
+                ]
+                if provider.default_model not in models:
+                    models.insert(0, provider.default_model)
+                status = "online"
+            except Exception as exc:  # pragma: no cover - network/runtime dependent
+                error = str(exc)
+
+            providers.append(
+                {
+                    "id": provider.provider_id,
+                    "status": status,
+                    "baseUrl": provider.base_url,
+                    "defaultModel": provider.default_model,
+                    "models": models,
+                    "supportsBatch": provider.supports_batch,
+                    "error": error,
+                }
+            )
+        return providers
+
+    async def chat(
+        self,
+        *,
+        provider_id: str | None,
+        model: str | None,
+        system_prompt: str | None,
+        messages: list[dict[str, str]],
+        temperature: float = 0.2,
+        response_format: str | None = None,
+        metadata: dict[str, Any] | None = None,
+    ) -> dict[str, Any]:
+        provider = self.get_provider(provider_id)
+        selected_model = model or provider.default_model
+        prepared_messages = list(messages)
+        if system_prompt:
+            prepared_messages = [{"role": "system", "content": system_prompt}] + prepared_messages
+
+        payload: dict[str, Any] = {
+            "model": selected_model,
+            "messages": prepared_messages,
+            "temperature": temperature,
+        }
+        if response_format == "json":
+            payload["response_format"] = {"type": "json_object"}
+
+        async with httpx.AsyncClient(timeout=RUNTIME_LLM_TIMEOUT_S) as client:
+            response = await client.post(provider.chat_url, json=payload, headers=provider.headers)
+            response.raise_for_status()
+        body = response.json()
+        choice = (body.get("choices") or [{}])[0]
+        message = choice.get("message") or {}
+        content = message.get("content")
+        text = self._extract_text(content)
+        parsed_json: dict[str, Any] | None = None
+        if response_format == "json":
+            try:
+                parsed_json = json.loads(text) if text else {}
+            except json.JSONDecodeError:
+                parsed_json = None
+
+        return {
+            "provider": provider.provider_id,
+            "model": selected_model,
+            "message": {
+                "role": "assistant",
+                "content": text,
+                "parsedJson": parsed_json,
+            },
+            "usage": body.get("usage"),
+            "metadata": metadata or {},
+            "completedAt": _utc_iso(),
+        }
+
+    async def submit_batch(
+        self,
+        *,
+        provider_id: str | None,
+        model: str | None,
+        job_type: str,
+        items: list[dict[str, Any]],
+        metadata: dict[str, Any] | None,
+        pool: Any | None = None,
+        actor_id: str | None = None,
+    ) -> dict[str, Any]:
+        provider = self.get_provider(provider_id)
+        selected_model = model or provider.default_model
+        job_id = str(uuid.uuid4())
+        created_at = _utc_iso()
+        normalized_items = [
+            {
+                "request_id": str(item.get("request_id") or f"item_{idx+1}"),
+                "messages": item.get("messages") or [],
+                "system_prompt": item.get("system_prompt"),
+                "temperature": float(item.get("temperature", 0.2)),
+                "response_format": item.get("response_format"),
+                "metadata": item.get("metadata") or {},
+            }
+            for idx, item in enumerate(items)
+        ]
+
+        job_record = {
+            "job_id": job_id,
+            "provider": provider.provider_id,
+            "model": selected_model,
+            "job_type": job_type,
+            "status": "queued",
+            "submitted_count": len(normalized_items),
+            "completed_count": 0,
+            "failed_count": 0,
+            "metadata": metadata or {},
+            "items": normalized_items,
+            "results": [],
+            "created_at": created_at,
+            "updated_at": created_at,
+            "started_at": None,
+            "completed_at": None,
+            "actor_id": actor_id,
+        }
+        self._jobs[job_id] = job_record
+        await self._persist_job(job_record, pool=pool)
+        asyncio.create_task(self._run_batch(job_id, pool=pool))
+        return {
+            "job_id": job_id,
+            "status": job_record["status"],
+            "provider": provider.provider_id,
+            "model": selected_model,
+            "submitted_count": len(normalized_items),
+            "created_at": created_at,
+        }
+
+    async def _run_batch(self, job_id: str, *, pool: Any | None = None) -> None:
+        job = self._jobs.get(job_id)
+        if not job:
+            return
+
+        job["status"] = "running"
+        job["started_at"] = _utc_iso()
+        job["updated_at"] = _utc_iso()
+        await self._persist_job(job, pool=pool)
+
+        semaphore = asyncio.Semaphore(RUNTIME_LLM_CONCURRENCY)
+
+        async def _execute_item(item: dict[str, Any]) -> dict[str, Any]:
+            async with semaphore:
+                try:
+                    response = await self.chat(
+                        provider_id=job["provider"],
+                        model=job["model"],
+                        system_prompt=item.get("system_prompt"),
+                        messages=item.get("messages") or [],
+                        temperature=float(item.get("temperature", 0.2)),
+                        response_format=item.get("response_format"),
+                        metadata=item.get("metadata") or {},
+                    )
+                    return {
+                        "request_id": item["request_id"],
+                        "status": "completed",
+                        "response": response,
+                        "error": None,
+                    }
+                except Exception as exc:  # pragma: no cover - network/runtime dependent
+                    logger.error("runtime_llm batch item failed job=%s request=%s error=%s", job_id, item["request_id"], exc)
+                    return {
+                        "request_id": item["request_id"],
+                        "status": "failed",
+                        "response": None,
+                        "error": str(exc),
+                    }
+
+        results = await asyncio.gather(*[_execute_item(item) for item in job["items"]])
+        job["results"] = results
+        job["completed_count"] = sum(1 for result in results if result["status"] == "completed")
+        job["failed_count"] = sum(1 for result in results if result["status"] == "failed")
+        job["status"] = "completed" if job["failed_count"] == 0 else ("failed" if job["completed_count"] == 0 else "completed_with_errors")
+        job["completed_at"] = _utc_iso()
+        job["updated_at"] = _utc_iso()
+        await self._persist_job(job, pool=pool)
+
+    async def get_job(self, job_id: str, *, pool: Any | None = None) -> dict[str, Any] | None:
+        if job_id in self._jobs:
+            return self._jobs[job_id]
+        if pool is not None:
+            loaded = await self._load_job_from_db(job_id, pool=pool)
+            if loaded:
+                self._jobs[job_id] = loaded
+            return loaded
+        return None
+
+    async def list_job_results(self, job_id: str, *, pool: Any | None = None) -> list[dict[str, Any]] | None:
+        job = await self.get_job(job_id, pool=pool)
+        if not job:
+            return None
+        return list(job.get("results") or [])
+
+    async def _persist_job(self, job: dict[str, Any], *, pool: Any | None = None) -> None:
+        if pool is None:
+            return
+        async with pool.acquire() as conn:
+            await conn.execute(
+                """
+                INSERT INTO workflow_agent_runs (
+                    run_id,
+                    agent_name,
+                    trigger_type,
+                    trigger_ref,
+                    input_payload,
+                    output_payload,
+                    status,
+                    duration_ms,
+                    error_detail,
+                    started_at,
+                    completed_at
+                )
+                VALUES (
+                    $1::uuid,
+                    'runtime_llm',
+                    $2,
+                    $3,
+                    $4::jsonb,
+                    $5::jsonb,
+                    $6,
+                    $7,
+                    $8,
+                    $9::timestamptz,
+                    $10::timestamptz
+                )
+                ON CONFLICT (run_id)
+                DO UPDATE SET
+                    input_payload = EXCLUDED.input_payload,
+                    output_payload = EXCLUDED.output_payload,
+                    status = EXCLUDED.status,
+                    duration_ms = EXCLUDED.duration_ms,
+                    error_detail = EXCLUDED.error_detail,
+                    started_at = EXCLUDED.started_at,
+                    completed_at = EXCLUDED.completed_at
+                """,
+                job["job_id"],
+                job["job_type"],
+                job.get("actor_id"),
+                json.dumps(
+                    {
+                        "provider": job["provider"],
+                        "model": job["model"],
+                        "metadata": job.get("metadata") or {},
+                        "items": job.get("items") or [],
+                    }
+                ),
+                json.dumps(
+                    {
+                        "results": job.get("results") or [],
+                        "submitted_count": job.get("submitted_count", 0),
+                        "completed_count": job.get("completed_count", 0),
+                        "failed_count": job.get("failed_count", 0),
+                        "created_at": job.get("created_at"),
+                        "updated_at": job.get("updated_at"),
+                    }
+                ),
+                job["status"],
+                self._duration_ms(job.get("started_at"), job.get("completed_at")),
+                self._job_error_detail(job),
+                job.get("started_at"),
+                job.get("completed_at"),
+            )
+
+    async def _load_job_from_db(self, job_id: str, *, pool: Any) -> dict[str, Any] | None:
+        async with pool.acquire() as conn:
+            row = await conn.fetchrow(
+                """
+                SELECT
+                    run_id::text AS job_id,
+                    trigger_type AS job_type,
+                    trigger_ref AS actor_id,
+                    input_payload,
+                    output_payload,
+                    status,
+                    started_at,
+                    completed_at
+                FROM workflow_agent_runs
+                WHERE run_id = $1::uuid AND agent_name = 'runtime_llm'
+                """,
+                job_id,
+            )
+        if not row:
+            return None
+
+        input_payload = dict(row["input_payload"] or {})
+        output_payload = dict(row["output_payload"] or {})
+        return {
+            "job_id": row["job_id"],
+            "provider": input_payload.get("provider"),
+            "model": input_payload.get("model"),
+            "job_type": row["job_type"],
+            "status": row["status"],
+            "submitted_count": int(output_payload.get("submitted_count", len(input_payload.get("items") or []))),
+            "completed_count": int(output_payload.get("completed_count", 0)),
+            "failed_count": int(output_payload.get("failed_count", 0)),
+            "metadata": input_payload.get("metadata") or {},
+            "items": input_payload.get("items") or [],
+            "results": output_payload.get("results") or [],
+            "created_at": output_payload.get("created_at") or (row["started_at"].isoformat() if row["started_at"] else None),
+            "updated_at": output_payload.get("updated_at") or (row["completed_at"].isoformat() if row["completed_at"] else None),
+            "started_at": row["started_at"].isoformat() if row["started_at"] else None,
+            "completed_at": row["completed_at"].isoformat() if row["completed_at"] else None,
+            "actor_id": row["actor_id"],
+        }
+
+    @staticmethod
+    def _extract_text(content: Any) -> str:
+        if isinstance(content, str):
+            return content
+        if isinstance(content, list):
+            parts: list[str] = []
+            for part in content:
+                if isinstance(part, dict):
+                    text = part.get("text")
+                    if isinstance(text, str):
+                        parts.append(text)
+            return "\n".join(parts).strip()
+        return str(content or "")
+
+    @staticmethod
+    def _duration_ms(started_at: str | None, completed_at: str | None) -> int | None:
+        if not started_at or not completed_at:
+            return None
+        try:
+            start = datetime.fromisoformat(started_at.replace("Z", "+00:00"))
+            end = datetime.fromisoformat(completed_at.replace("Z", "+00:00"))
+        except ValueError:
+            return None
+        return max(0, int((end - start).total_seconds() * 1000))
+
+    @staticmethod
+    def _job_error_detail(job: dict[str, Any]) -> str | None:
+        failed = [result for result in job.get("results") or [] if result.get("status") == "failed"]
+        if not failed:
+            return None
+        return "; ".join(f"{item.get('request_id')}: {item.get('error')}" for item in failed[:5])
+
+
+runtime_llm_service = RuntimeLLMService()