feat: Oracle Canvas Component Schema and Qwen 3.6 integration (#31)

Co-authored-by: Sagnik <sagnik7896@gmail.com>
Reviewed-on: sagnik/Project_Velocity#31

backend/api/routes_oracle.py

@@ -7,6 +7,7 @@ from backend.oracle.action_service import oracle_action_service
 from backend.oracle.persona_service import persona_service
 from backend.services.mcp_registry import mcp_registry
 from backend.services.nemoclaw_runtime import nemoclaw_runtime
+from backend.services.runtime_llm_service import runtime_llm_service
 
 router = APIRouter()
 
@@ -38,6 +39,7 @@ async def oracle_health() -> dict:
         "status": "ok",
         "persona": await persona_service.health(),
         "mcp_tools": mcp_registry.list_tools(),
+        "runtime_llm": await runtime_llm_service.list_providers(),
     }
 
 

backend/api/routes_oracle_templates.py

@@ -22,6 +22,7 @@ from __future__ import annotations
 
 import json
 import logging
+import os
 from typing import Any, Optional
 
 from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
@@ -32,6 +33,7 @@ from backend.auth.dependencies import get_current_user
 logger = logging.getLogger("velocity.oracle_templates")
 
 router = APIRouter()
+_DEFAULT_TENANT_ID = os.getenv("ORACLE_DEFAULT_TENANT_ID", "tenant_velocity")
 
 
 # ── Helpers ───────────────────────────────────────────────────────────────────
@@ -43,6 +45,10 @@ def _pool(request: Request):
     return pool
 
 
+def _tenant_id() -> str:
+    return _DEFAULT_TENANT_ID
+
+
 # ── Models ────────────────────────────────────────────────────────────────────
 
 class ChapterCreate(BaseModel):
@@ -112,7 +118,7 @@ async def list_template_chapters(
             GROUP BY ch.chapter_id
             ORDER BY ch.sort_order ASC
             """,
-            user.role,
+            _tenant_id(),
         )
     return {"chapters": [dict(r) for r in rows]}
 
@@ -132,7 +138,7 @@ async def create_template_chapter(
             VALUES ($1,$2,$3,$4)
             RETURNING chapter_id, created_at
             """,
-            user.role, body.name, body.description, body.sort_order,
+            _tenant_id(), body.name, body.description, body.sort_order,
         )
     return {"chapter_id": str(row["chapter_id"]), "created_at": str(row["created_at"])}
 
@@ -149,7 +155,7 @@ async def list_template_subchapters(
     pool = _pool(request)
     async with pool.acquire() as conn:
         where = "WHERE sub.tenant_id=$1"
-        params: list[Any] = [user.role]
+        params: list[Any] = [_tenant_id()]
         idx = 2
         if not include_inactive:
             where += " AND sub.is_active=TRUE"
@@ -186,7 +192,7 @@ async def create_template_subchapter(
         # Verify chapter exists and belongs to tenant
         ch_exists = await conn.fetchval(
             "SELECT 1 FROM oracle_template_chapters WHERE chapter_id=$1 AND tenant_id=$2",
-            body.chapter_id, user.role,
+            body.chapter_id, _tenant_id(),
         )
         if not ch_exists:
             raise HTTPException(404, "Chapter not found")
@@ -198,7 +204,7 @@ async def create_template_subchapter(
             VALUES ($1,$2,$3,$4,$5)
             RETURNING subchapter_id, created_at
             """,
-            body.chapter_id, user.role, body.name, body.description, body.sort_order,
+            body.chapter_id, _tenant_id(), body.name, body.description, body.sort_order,
         )
     return {"subchapter_id": str(row["subchapter_id"]), "created_at": str(row["created_at"])}
 
@@ -218,7 +224,7 @@ async def list_component_templates(
 ):
     pool = _pool(request)
     where = "WHERE t.tenant_id=$1"
-    params: list[Any] = [user.role]
+    params: list[Any] = [_tenant_id()]
     idx = 2
 
     if chapter_id:
@@ -270,7 +276,7 @@ async def create_component_template(
             ) VALUES ($1,$2,$3,$4,$5,$6,$7::jsonb,$8,$9,$10,'draft')
             RETURNING template_id, created_at
             """,
-            user.role, body.name, body.category, body.chapter_id, body.subchapter_id,
+            _tenant_id(), body.name, body.category, body.chapter_id, body.subchapter_id,
             body.accepted_shapes,
             json.dumps(body.json_template) if body.json_template else None,
             body.description, body.origin, body.version,
@@ -294,7 +300,7 @@ async def get_component_template(
             LEFT JOIN oracle_template_subchapters sub ON sub.subchapter_id = t.subchapter_id
             WHERE t.template_id=$1 AND t.tenant_id=$2
             """,
-            template_id, user.role,
+            template_id, _tenant_id(),
         )
     if not row:
         raise HTTPException(404, "Template not found")
@@ -315,7 +321,7 @@ async def add_seed_example(
     async with pool.acquire() as conn:
         exists = await conn.fetchval(
             "SELECT 1 FROM oracle_component_templates WHERE template_id=$1 AND tenant_id=$2",
-            template_id, user.role,
+            template_id, _tenant_id(),
         )
         if not exists:
             raise HTTPException(404, "Template not found")
@@ -371,7 +377,7 @@ async def trigger_synthetic_job(
     async with pool.acquire() as conn:
         exists = await conn.fetchval(
             "SELECT 1 FROM oracle_component_templates WHERE template_id=$1 AND tenant_id=$2",
-            body.template_id, user.role,
+            body.template_id, _tenant_id(),
         )
         if not exists:
             raise HTTPException(404, "Template not found")
@@ -384,7 +390,7 @@ async def trigger_synthetic_job(
             ) VALUES ($1,$2,$3,$4,$5,$6,$7)
             RETURNING job_id, status, created_at
             """,
-            user.role, body.template_id, body.chapter_id, body.subchapter_id,
+            _tenant_id(), body.template_id, body.chapter_id, body.subchapter_id,
             body.model, body.requested_count, user.user_id,
         )
     logger.info(

backend/api/routes_runtime_llm.py

@@ -0,0 +1,140 @@
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+from pydantic import BaseModel, Field
+
+from backend.auth.dependencies import UserPrincipal, get_current_user
+from backend.services.runtime_llm_service import runtime_llm_service
+
+router = APIRouter()
+
+
+class ChatMessage(BaseModel):
+    role: str = Field(..., pattern="^(system|user|assistant)$")
+    content: str = Field(..., min_length=1)
+
+
+class RuntimeChatRequest(BaseModel):
+    provider: str | None = None
+    model: str | None = None
+    system_prompt: str | None = None
+    messages: list[ChatMessage]
+    temperature: float = Field(default=0.2, ge=0.0, le=2.0)
+    response_format: str | None = Field(default=None, pattern="^(json|text)$")
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+
+class BatchItemRequest(BaseModel):
+    request_id: str
+    messages: list[ChatMessage]
+    system_prompt: str | None = None
+    temperature: float = Field(default=0.2, ge=0.0, le=2.0)
+    response_format: str | None = Field(default=None, pattern="^(json|text)$")
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+
+class RuntimeBatchRequest(BaseModel):
+    provider: str | None = None
+    model: str | None = None
+    job_type: str = Field(..., min_length=1, max_length=128)
+    metadata: dict[str, Any] = Field(default_factory=dict)
+    items: list[BatchItemRequest] = Field(..., min_length=1, max_length=128)
+
+
+def _normalize_user(user: UserPrincipal) -> dict[str, str]:
+    return {
+        "user_id": user.user_id,
+        "role": user.role,
+    }
+
+
+@router.get("/providers", summary="List configured runtime LLM providers and models")
+async def list_runtime_providers(_: UserPrincipal = Depends(get_current_user)) -> dict:
+    return {"status": "ok", "data": await runtime_llm_service.list_providers()}
+
+
+@router.post("/chat", summary="Execute a single runtime LLM chat completion")
+async def runtime_chat(
+    payload: RuntimeChatRequest,
+    user: UserPrincipal = Depends(get_current_user),
+) -> dict:
+    response = await runtime_llm_service.chat(
+        provider_id=payload.provider,
+        model=payload.model,
+        system_prompt=payload.system_prompt,
+        messages=[message.model_dump() for message in payload.messages],
+        temperature=payload.temperature,
+        response_format=payload.response_format,
+        metadata={
+            **payload.metadata,
+            "requested_by": _normalize_user(user),
+        },
+    )
+    return {"status": "ok", "data": response}
+
+
+@router.post("/batch", status_code=status.HTTP_202_ACCEPTED, summary="Submit a persisted runtime LLM batch job")
+async def runtime_batch(
+    payload: RuntimeBatchRequest,
+    request: Request,
+    user: UserPrincipal = Depends(get_current_user),
+) -> dict:
+    pool = getattr(request.app.state, "db_pool", None)
+    result = await runtime_llm_service.submit_batch(
+        provider_id=payload.provider,
+        model=payload.model,
+        job_type=payload.job_type,
+        items=[item.model_dump() for item in payload.items],
+        metadata={
+            **payload.metadata,
+            "requested_by": _normalize_user(user),
+        },
+        pool=pool,
+        actor_id=user.user_id,
+    )
+    return {"status": "ok", "data": result}
+
+
+@router.get("/jobs/{job_id}", summary="Get runtime LLM batch job status")
+async def get_runtime_job(
+    job_id: str,
+    request: Request,
+    _: UserPrincipal = Depends(get_current_user),
+) -> dict:
+    pool = getattr(request.app.state, "db_pool", None)
+    job = await runtime_llm_service.get_job(job_id, pool=pool)
+    if not job:
+        raise HTTPException(status_code=404, detail=f"Runtime LLM job '{job_id}' not found.")
+
+    return {
+        "status": "ok",
+        "data": {
+            "job_id": job["job_id"],
+            "status": job["status"],
+            "provider": job["provider"],
+            "model": job["model"],
+            "job_type": job["job_type"],
+            "submitted_count": job["submitted_count"],
+            "completed_count": job["completed_count"],
+            "failed_count": job["failed_count"],
+            "created_at": job["created_at"],
+            "started_at": job["started_at"],
+            "completed_at": job["completed_at"],
+            "metadata": job.get("metadata") or {},
+        },
+    }
+
+
+@router.get("/jobs/{job_id}/results", summary="Get runtime LLM batch job item results")
+async def get_runtime_job_results(
+    job_id: str,
+    request: Request,
+    _: UserPrincipal = Depends(get_current_user),
+) -> dict:
+    pool = getattr(request.app.state, "db_pool", None)
+    results = await runtime_llm_service.list_job_results(job_id, pool=pool)
+    if results is None:
+        raise HTTPException(status_code=404, detail=f"Runtime LLM job '{job_id}' not found.")
+    return {"status": "ok", "data": results, "meta": {"count": len(results)}}

backend/main.py

@@ -63,6 +63,7 @@ from backend.api.routes_inventory import router as inventory_router
 from backend.api.routes_admin_surface import router as admin_surface_router
 from backend.api.routes_oracle_templates import router as oracle_templates_router
 from backend.api.routes_crm_imports import router as crm_imports_router
+from backend.api.routes_runtime_llm import router as runtime_llm_router
 from backend.auth.dependencies import (
     create_access_token, verify_password, get_current_user, UserPrincipal
 )
@@ -146,6 +147,7 @@ app.include_router(mobile_edge_router, prefix="/api/mobile-edge", tags=["Mobile
 app.include_router(inventory_router, prefix="/api/inventory", tags=["Inventory"])
 app.include_router(admin_surface_router, prefix="/api/admin-surface", tags=["Admin Surface"])
 app.include_router(crm_imports_router, prefix="/api", tags=["CRM Canonical"])
+app.include_router(runtime_llm_router, prefix="/api/runtime/llm", tags=["Runtime LLM"])
 
 # Public vault link (no /api prefix — shared externally with prospects)
 from backend.routers.vault import router as public_vault_router

backend/oracle/action_service.py

@@ -17,8 +17,23 @@ except Exception:  # pragma: no cover
 _DB_URL = os.getenv("DATABASE_URL", "")
 
 
-def _now() -> str:
-    return datetime.now(timezone.utc).isoformat()
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _iso(value: datetime | None) -> str | None:
+    return value.isoformat() if value else None
+
+
+def _coerce_datetime(value: datetime | str | None) -> datetime | None:
+    if value is None or isinstance(value, datetime):
+        return value
+    if isinstance(value, str) and value.strip():
+        try:
+            return datetime.fromisoformat(value)
+        except ValueError:
+            return None
+    return None
 
 
 def _db_ready() -> bool:
@@ -314,8 +329,8 @@ class OracleActionService:
                 json.dumps(action.get("componentIds") or []),
                 json.dumps(action.get("writebackPayload") or {}),
                 json.dumps(action.get("resultPayload") or {}),
-                action["createdAt"],
-                action["updatedAt"],
+                _coerce_datetime(action["createdAt"]),
+                _coerce_datetime(action["updatedAt"]),
             )
         finally:
             await conn.close()
@@ -338,8 +353,8 @@ class OracleActionService:
             "componentIds": row["component_ids"] or [],
             "writebackPayload": row["writeback_payload"] or {},
             "resultPayload": row["result_payload"] or {},
-            "createdAt": row["created_at"].isoformat() if row["created_at"] else None,
-            "updatedAt": row["updated_at"].isoformat() if row["updated_at"] else None,
+            "createdAt": _iso(row["created_at"]),
+            "updatedAt": _iso(row["updated_at"]),
         }
 
 

backend/oracle/canvas_service.py

@@ -57,13 +57,37 @@ def _stringify(value: Any) -> str:
     return str(value) if value is not None else ""
 
 
+def _json_object(value: Any) -> dict[str, Any]:
+    if isinstance(value, dict):
+        return value
+    if isinstance(value, str) and value.strip():
+        try:
+            parsed = json.loads(value)
+            if isinstance(parsed, dict):
+                return parsed
+        except Exception:
+            logger.warning("canvas_service: failed to parse JSON object field; using empty object")
+    return {}
+
+
 def _normalize_component(component: dict[str, Any]) -> dict[str, Any]:
     normalized = deepcopy(component)
     normalized["componentId"] = _stringify(normalized.get("componentId"))
-    descriptor = normalized.get("dataSourceDescriptor") or {}
+    descriptor = _json_object(normalized.get("dataSourceDescriptor"))
     if descriptor.get("descriptorId") is not None:
         descriptor["descriptorId"] = _stringify(descriptor["descriptorId"])
     normalized["dataSourceDescriptor"] = descriptor
+    for field in (
+        "visualizationParameters",
+        "dataBindings",
+        "provenance",
+        "renderingHints",
+        "layout",
+        "accessControls",
+        "styleSignature",
+        "validationState",
+    ):
+        normalized[field] = _json_object(normalized.get(field))
     return normalized
 
 
@@ -105,7 +129,7 @@ def _deserialize_page_row(row: Any, components: list[dict[str, Any]]) -> dict[st
         "isShared": bool(row["is_shared"]),
         "headRevision": head_revision,
         "baseRevision": int(row["base_revision"]),
-        "sharingPolicy": row["sharing_policy"] or {
+        "sharingPolicy": _json_object(row["sharing_policy"]) or {
             "shareMode": "direct_fork_only",
             "allowReshare": False,
             "defaultForkVisibility": "private",

backend/oracle/codebook_service.py

@@ -0,0 +1,340 @@
+"""
+oracle/codebook_service.py
+Loads, normalizes, and retrieves Oracle Canvas codebook examples from the
+expanded GPT and Claude seed packs delivered in Sprint 1.
+
+The runtime treats the GPT pack as the primary normalized corpus and uses the
+Claude pack as a supplement when it adds unique examples or metadata.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import re
+from dataclasses import dataclass
+from functools import lru_cache
+from pathlib import Path
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+_TOKEN_RE = re.compile(r"[a-z0-9]+")
+_STOPWORDS = {
+    "a", "an", "and", "as", "at", "build", "canvas", "chart", "client", "clients",
+    "for", "from", "get", "give", "in", "into", "is", "list", "me", "of", "on",
+    "or", "oracle", "please", "render", "show", "surface", "that", "the", "this",
+    "to", "view", "with",
+}
+
+
+@dataclass(frozen=True)
+class CodebookExample:
+    example_id: str
+    chapter_id: str
+    chapter_name: str
+    subchapter_id: str
+    subchapter_name: str
+    title: str
+    template_name: str
+    component_type: str
+    accepted_shapes: tuple[str, ...]
+    example_json: dict[str, Any]
+    quality_notes: str
+    is_canonical: bool
+    source_pack: str
+    surface_targets: tuple[str, ...]
+    policy_tags: tuple[str, ...]
+    backend_contract_hints: dict[str, Any]
+    score_terms: tuple[str, ...]
+
+
+def _repo_root() -> Path:
+    return Path(__file__).resolve().parents[2]
+
+
+def _safe_load_json(path: Path) -> dict[str, Any]:
+    with path.open("r", encoding="utf-8") as handle:
+        return json.load(handle)
+
+
+def _tokenize(value: str) -> list[str]:
+    lowered = value.lower()
+    return [tok for tok in _TOKEN_RE.findall(lowered) if tok not in _STOPWORDS and len(tok) > 1]
+
+
+def _make_template_id(example: dict[str, Any]) -> str:
+    base = "|".join(
+        [
+            example.get("chapter_id", ""),
+            example.get("subchapter_id", ""),
+            example.get("template_name", ""),
+            example.get("component_type", ""),
+        ]
+    )
+    return hashlib.sha1(base.encode("utf-8")).hexdigest()[:16]
+
+
+def _chapter_maps(payload: dict[str, Any]) -> tuple[dict[str, str], dict[str, str]]:
+    chapters: dict[str, str] = {}
+    subchapters: dict[str, str] = {}
+    for chapter in payload.get("chapters", []):
+        chapter_id = str(chapter.get("chapter_id", "")).strip()
+        if chapter_id:
+            chapters[chapter_id] = str(chapter.get("name", "")).strip()
+        for subchapter in chapter.get("subchapters", []):
+            sub_id = str(subchapter.get("subchapter_id", "")).strip()
+            if sub_id:
+                subchapters[sub_id] = str(subchapter.get("name", "")).strip()
+    return chapters, subchapters
+
+
+def _normalize_examples(payload: dict[str, Any], source_pack: str) -> list[CodebookExample]:
+    chapter_names, subchapter_names = _chapter_maps(payload)
+    raw_examples = payload.get("seed_examples") or payload.get("examples") or []
+    normalized: list[CodebookExample] = []
+    for raw in raw_examples:
+        chapter_id = str(raw.get("chapter_id", "")).strip()
+        subchapter_id = str(raw.get("subchapter_id", "")).strip()
+        title = str(raw.get("title") or raw.get("template_name") or "Oracle Component").strip()
+        template_name = str(raw.get("template_name") or title).strip()
+        component_type = str(raw.get("component_type") or "summary_card").strip()
+        example_json = raw.get("example_json") or {}
+        terms = _tokenize(
+            " ".join(
+                [
+                    title,
+                    template_name,
+                    component_type.replace("_", " "),
+                    chapter_names.get(chapter_id, ""),
+                    subchapter_names.get(subchapter_id, ""),
+                    str(raw.get("quality_notes", "")),
+                    " ".join(raw.get("policy_tags", []) or []),
+                ]
+            )
+        )
+        normalized.append(
+            CodebookExample(
+                example_id=str(raw.get("example_id") or _make_template_id(raw)),
+                chapter_id=chapter_id,
+                chapter_name=chapter_names.get(chapter_id, chapter_id),
+                subchapter_id=subchapter_id,
+                subchapter_name=subchapter_names.get(subchapter_id, subchapter_id),
+                title=title,
+                template_name=template_name,
+                component_type=component_type,
+                accepted_shapes=tuple(raw.get("accepted_shapes") or []),
+                example_json=example_json,
+                quality_notes=str(raw.get("quality_notes") or ""),
+                is_canonical=bool(raw.get("is_canonical")),
+                source_pack=source_pack,
+                surface_targets=tuple(raw.get("surface_targets") or []),
+                policy_tags=tuple(raw.get("policy_tags") or []),
+                backend_contract_hints=dict(raw.get("backend_contract_hints") or {}),
+                score_terms=tuple(terms),
+            )
+        )
+    return normalized
+
+
+class OracleCodebookService:
+    def __init__(self) -> None:
+        root = _repo_root()
+        self.runtime_merged_path = root / "backend" / "oracle" / "oracle_runtime_codebook_merged.json"
+        self.primary_path = root / ".Agent Context" / "Sprint 1" / "Sayan Multi-Surface and Oracle Delivery Pack" / "Sample JSON Schema" / "GPT 5.4" / "oracle_canvas_json_expansion_pack" / "db" / "oracle_template_seed_db_expanded_v1.pretty.json"
+        self.secondary_path = root / ".Agent Context" / "Sprint 1" / "Sayan Multi-Surface and Oracle Delivery Pack" / "Sample JSON Schema" / "Claude Sonnet 4.6" / "oracle_template_expansion" / "oracle_template_seed_db_expanded.json"
+        self.fallback_path = root / "backend" / "oracle" / "oracle_template_seed_db.json"
+
+    @lru_cache(maxsize=1)
+    def load(self) -> dict[str, Any]:
+        corpora: list[CodebookExample] = []
+        sources_loaded: list[str] = []
+        source_paths: list[tuple[Path, str]]
+        if self.runtime_merged_path.exists():
+            source_paths = [
+                (self.runtime_merged_path, "runtime_merged"),
+                (self.fallback_path, "runtime_seed_fallback"),
+            ]
+        else:
+            source_paths = [
+                (self.primary_path, "gpt_5_4"),
+                (self.secondary_path, "claude_sonnet_4_6"),
+                (self.fallback_path, "runtime_seed_fallback"),
+            ]
+
+        for path, label in source_paths:
+            if not path.exists():
+                continue
+            payload = _safe_load_json(path)
+            examples = _normalize_examples(payload, label)
+            if examples:
+                corpora.extend(examples)
+                sources_loaded.append(f"{label}:{len(examples)}")
+
+        deduped: dict[tuple[str, str, str], CodebookExample] = {}
+        for example in corpora:
+            key = (example.subchapter_id, example.template_name.lower(), example.title.lower())
+            existing = deduped.get(key)
+            if existing is None:
+                deduped[key] = example
+                continue
+            # Prefer canonical GPT examples, then canonical examples, then richer source pack.
+            if example.source_pack == "gpt_5_4" and existing.source_pack != "gpt_5_4":
+                deduped[key] = example
+            elif example.is_canonical and not existing.is_canonical:
+                deduped[key] = example
+
+        examples = list(deduped.values())
+        logger.info("Oracle codebook loaded from %s", ", ".join(sources_loaded) or "no sources")
+        return {
+            "examples": examples,
+            "source_summary": sources_loaded,
+            "template_count": len({(e.chapter_id, e.subchapter_id, e.template_name, e.component_type) for e in examples}),
+        }
+
+    def stats(self) -> dict[str, Any]:
+        data = self.load()
+        examples: list[CodebookExample] = data["examples"]
+        return {
+            "example_count": len(examples),
+            "template_count": data["template_count"],
+            "source_summary": data["source_summary"],
+        }
+
+    def list_templates(
+        self,
+        *,
+        category: str | None = None,
+        status: str | None = None,
+        search: str | None = None,
+        limit: int = 50,
+        offset: int = 0,
+    ) -> dict[str, Any]:
+        del status  # runtime codebook templates are always active catalog entries
+        examples: list[CodebookExample] = self.load()["examples"]
+        templates: dict[str, dict[str, Any]] = {}
+        for example in examples:
+            if category and category.lower() not in {example.chapter_name.lower(), example.subchapter_name.lower()}:
+                continue
+            if search:
+                terms = set(example.score_terms)
+                if not set(_tokenize(search)).intersection(terms):
+                    continue
+            template_id = _make_template_id(
+                {
+                    "chapter_id": example.chapter_id,
+                    "subchapter_id": example.subchapter_id,
+                    "template_name": example.template_name,
+                    "component_type": example.component_type,
+                }
+            )
+            record = templates.get(template_id)
+            if record is None:
+                templates[template_id] = {
+                    "templateId": template_id,
+                    "tenantId": "_system",
+                    "name": example.template_name,
+                    "category": example.chapter_name,
+                    "status": "catalog_active",
+                    "origin": "premade",
+                    "version": "codebook-v2",
+                    "acceptedShapes": list(example.accepted_shapes),
+                    "description": f"{example.subchapter_name} · {example.title}",
+                    "chapterId": example.chapter_id,
+                    "subchapterId": example.subchapter_id,
+                    "componentType": example.component_type,
+                    "sourcePack": example.source_pack,
+                    "useCount": 0,
+                    "updatedAt": None,
+                    "createdAt": None,
+                }
+        ordered = list(templates.values())
+        ordered.sort(key=lambda item: (item["category"], item["name"]))
+        total = len(ordered)
+        return {
+            "total": total,
+            "templates": ordered[offset: offset + limit],
+        }
+
+    def search_examples(self, prompt: str, *, limit: int = 8) -> list[CodebookExample]:
+        prompt_terms = set(_tokenize(prompt))
+        if not prompt_terms:
+            prompt_terms = set(_tokenize(prompt.replace("_", " ")))
+
+        scored: list[tuple[int, CodebookExample]] = []
+        for example in self.load()["examples"]:
+            score = 0
+            term_set = set(example.score_terms)
+            overlap = prompt_terms.intersection(term_set)
+            score += len(overlap) * 6
+            lowered_prompt = prompt.lower()
+            if example.template_name.lower() in lowered_prompt:
+                score += 24
+            if example.subchapter_name.lower() in lowered_prompt:
+                score += 20
+            if example.chapter_name.lower() in lowered_prompt:
+                score += 14
+            if example.component_type.replace("_", " ") in lowered_prompt:
+                score += 12
+            if example.is_canonical:
+                score += 8
+            if "live_data_first" in example.policy_tags:
+                score += 4
+            if score > 0:
+                scored.append((score, example))
+
+        scored.sort(key=lambda item: (-item[0], item[1].chapter_id, item[1].subchapter_id, item[1].title))
+        selected: list[CodebookExample] = []
+        seen: set[tuple[str, str]] = set()
+        for _, example in scored:
+            dedupe_key = (example.subchapter_id, example.template_name)
+            if dedupe_key in seen:
+                continue
+            seen.add(dedupe_key)
+            selected.append(example)
+            if len(selected) >= limit:
+                break
+        return selected
+
+    def synthesize_template(self, prompt: str, data_shapes: list[str] | None = None) -> dict[str, Any]:
+        match = next(iter(self.search_examples(prompt, limit=1)), None)
+        shapes = data_shapes or []
+        if match is None:
+            return {
+                "templateId": hashlib.sha1(prompt.encode("utf-8")).hexdigest()[:16],
+                "tenantId": "_system",
+                "name": "Oracle Synthesized Draft",
+                "category": "Custom",
+                "status": "tenant_draft",
+                "origin": "synthesized",
+                "version": "1.0.0",
+                "acceptedShapes": shapes,
+                "description": f"Draft synthesized from prompt: {prompt[:120]}",
+            }
+
+        return {
+            "templateId": _make_template_id(
+                {
+                    "chapter_id": match.chapter_id,
+                    "subchapter_id": match.subchapter_id,
+                    "template_name": match.template_name,
+                    "component_type": match.component_type,
+                }
+            ),
+            "tenantId": "_system",
+            "name": match.template_name,
+            "category": match.chapter_name,
+            "status": "catalog_active",
+            "origin": "premade",
+            "version": "codebook-v2",
+            "acceptedShapes": list(match.accepted_shapes or shapes),
+            "description": f"Best codebook match · {match.subchapter_name}",
+            "componentType": match.component_type,
+            "chapterId": match.chapter_id,
+            "subchapterId": match.subchapter_id,
+            "sourcePack": match.source_pack,
+            "exampleJson": match.example_json,
+        }
+
+
+codebook_service = OracleCodebookService()

backend/oracle/data_access_gateway.py

@@ -236,6 +236,86 @@ class DataAccessGateway:
             """
             return sql, [ctx.tenant_id, row_limit]
 
+        if dataset == "crm_contacts_overview":
+            sql = """
+                SELECT
+                    p.person_id::text AS id,
+                    p.full_name AS name,
+                    COALESCE(p.primary_email, '') AS email,
+                    COALESCE(p.primary_phone, '') AS phone,
+                    COALESCE(p.city, '') AS city,
+                    COALESCE(p.buyer_type, 'unclassified') AS buyer_type,
+                    COALESCE(q.qd_score, 0)::float AS qd_score
+                FROM crm_people p
+                LEFT JOIN LATERAL (
+                    SELECT qd_score
+                    FROM intel_qd_scores q
+                    WHERE q.person_id = p.person_id
+                    ORDER BY q.scored_at DESC
+                    LIMIT 1
+                ) q ON TRUE
+                ORDER BY qd_score DESC, p.full_name ASC
+                LIMIT $1
+            """
+            return sql, [row_limit]
+
+        if dataset == "crm_opportunity_pipeline":
+            sql = """
+                SELECT
+                    o.stage::text AS stage,
+                    COUNT(*)::int AS count,
+                    COALESCE(SUM(o.value), 0)::float AS value,
+                    COALESCE(
+                        json_agg(
+                            json_build_object(
+                                'id', o.opportunity_id,
+                                'name', p.full_name,
+                                'company', COALESCE(a.account_name, ''),
+                                'value', COALESCE(o.value, 0),
+                                'nextAction', COALESCE(o.next_action, '')
+                            )
+                            ORDER BY o.value DESC NULLS LAST
+                        ) FILTER (WHERE o.opportunity_id IS NOT NULL),
+                        '[]'::json
+                    ) AS leads
+                FROM crm_opportunities o
+                JOIN crm_leads l ON l.lead_id = o.lead_id
+                JOIN crm_people p ON p.person_id = l.person_id
+                LEFT JOIN crm_accounts a ON a.account_id = l.account_id
+                GROUP BY o.stage
+                ORDER BY COALESCE(SUM(o.value), 0) DESC, o.stage::text ASC
+                LIMIT $1
+            """
+            return sql, [row_limit]
+
+        if dataset == "crm_property_interest_rollup":
+            sql = """
+                SELECT
+                    project_name AS category,
+                    COUNT(*)::int AS value,
+                    ROUND(AVG(COALESCE((budget_min + budget_max) / 2.0, budget_max, budget_min, 0)), 2)::float AS average_budget
+                FROM crm_property_interests
+                GROUP BY project_name
+                ORDER BY value DESC, project_name ASC
+                LIMIT $1
+            """
+            return sql, [row_limit]
+
+        if dataset == "crm_interaction_timeline":
+            sql = """
+                SELECT
+                    i.interaction_type AS type,
+                    COALESCE(i.summary, i.interaction_type) AS title,
+                    CONCAT(p.full_name, ' · ', i.channel::text) AS summary,
+                    p.full_name AS actor,
+                    TO_CHAR(i.happened_at, 'YYYY-MM-DD HH24:MI') AS date
+                FROM intel_interactions i
+                JOIN crm_people p ON p.person_id = i.person_id
+                ORDER BY i.happened_at DESC
+                LIMIT $1
+            """
+            return sql, [row_limit]
+
         raise ValueError(f"Dataset '{dataset}' is not whitelisted for Oracle execution.")
 
 

backend/oracle/prompt_orchestrator.py

@@ -17,6 +17,8 @@ from .policy_service import PolicyContext, PolicyService
 from .canvas_service import canvas_service
 from .data_access_gateway import data_access_gateway
 from .persona_service import persona_service
+from .codebook_service import codebook_service, CodebookExample
+from backend.services.runtime_llm_service import runtime_llm_service
 from backend.services.nemoclaw_runtime import nemoclaw_runtime
 
 try:
@@ -26,15 +28,30 @@ except Exception:  # pragma: no cover
 
 logger = logging.getLogger(__name__)
 
-_NEMOCLAW_URL = os.getenv("NEMOCLAW_API_URL", "")
-_NEMOCLAW_API_KEY = os.getenv("NEMOCLAW_API_KEY", "")
 _DB_URL = os.getenv("DATABASE_URL", "")
 
 policy_svc = PolicyService()
 
 
-def _now() -> str:
-    return datetime.now(timezone.utc).isoformat()
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _iso(value: datetime | None) -> str | None:
+    if value is None:
+        return None
+    return value.isoformat()
+
+
+def _coerce_datetime(value: datetime | str | None) -> datetime | None:
+    if value is None or isinstance(value, datetime):
+        return value
+    if isinstance(value, str) and value.strip():
+        try:
+            return datetime.fromisoformat(value)
+        except ValueError:
+            return None
+    return None
 
 
 # ── Execution store ───────────────────────────────────────────────────────────
@@ -52,10 +69,10 @@ _INTENT_KEYWORDS: dict[str, list[str]] = {
     "pipeline_board": ["pipeline", "stage", "kanban", "deal", "funnel"],
     "bar_chart": ["bar", "compare", "source", "channel", "distribution", "ranked", "lead", "whale"],
     "geo_map": ["map", "geographic", "location", "district", "region", "area", "dubai"],
-    "table": ["table", "list", "broker", "performance", "leaderboard", "rank", "top"],
+    "table": ["table", "list", "broker", "performance", "leaderboard", "rank", "top", "contact", "client", "account", "crm"],
     "line_chart": ["trend", "time", "monthly", "weekly", "absorption", "forecast"],
     "kpi_tile": ["kpi", "total", "summary", "attainment", "quota", "how many"],
-    "activity_stream": ["timeline", "activity", "history", "follow-up", "queue", "contact"],
+    "activity_stream": ["timeline", "activity", "history", "follow-up", "queue", "contact", "interaction", "message", "call", "email"],
 }
 
 
@@ -109,6 +126,129 @@ _DATASET_MAP: dict[str, str] = {
     "activity_stream": "lead_activity_log",
 }
 
+_CODEBOOK_COMPONENT_MAP: dict[str, str] = {
+    "summary_card": "kpi_tile",
+    "summary_strip": "kpi_tile",
+    "metric_card_group": "kpi_tile",
+    "compact_alert_card": "kpi_tile",
+    "gauge_stack": "kpi_tile",
+    "lead_profile_card": "table",
+    "property_card": "table",
+    "data_table": "table",
+    "leaderboard_table": "table",
+    "matrix_grid": "table",
+    "interaction_timeline": "activity_stream",
+    "message_thread_summary": "activity_stream",
+    "timeline": "activity_stream",
+    "bar_chart": "bar_chart",
+    "line_chart": "line_chart",
+    "heatmap": "geo_map",
+    "geo_map": "geo_map",
+    "pipeline_board": "pipeline_board",
+}
+
+
+def _component_plan_type_from_codebook(example: CodebookExample) -> str:
+    return _CODEBOOK_COMPONENT_MAP.get(example.component_type, "table")
+
+
+def _dataset_for_codebook(example: CodebookExample, prompt: str, component_plan_type: str | None = None) -> str:
+    chapter = example.chapter_name.lower()
+    subchapter = example.subchapter_name.lower()
+    component_plan_type = component_plan_type or _component_plan_type_from_codebook(example)
+    lowered_prompt = prompt.lower()
+
+    if component_plan_type == "activity_stream":
+        return "crm_interaction_timeline"
+    if component_plan_type == "pipeline_board":
+        return "crm_opportunity_pipeline"
+    if component_plan_type == "line_chart" and any(term in lowered_prompt for term in ("trend", "time", "history", "growth")):
+        return "crm_property_interest_rollup"
+
+    if any(term in lowered_prompt for term in ("contact", "client 360", "crm", "account", "lead")):
+        if "timeline" in lowered_prompt or "message" in lowered_prompt or "call" in lowered_prompt or "email" in lowered_prompt:
+            return "crm_interaction_timeline"
+        if "pipeline" in lowered_prompt or "opportunit" in lowered_prompt:
+            return "crm_opportunity_pipeline"
+        if "interest" in lowered_prompt or "project" in lowered_prompt or "property" in lowered_prompt:
+            return "crm_property_interest_rollup"
+        return "crm_contacts_overview"
+
+    if "client" in chapter or "client" in subchapter or "contact" in subchapter:
+        return "crm_contacts_overview"
+    if "opportun" in chapter or "pipeline" in subchapter:
+        return "crm_opportunity_pipeline"
+    if "interaction" in chapter or "communication" in chapter or "timeline" in subchapter:
+        return "crm_interaction_timeline"
+    if "property" in chapter or "inventory" in chapter or "interest" in subchapter:
+        return "crm_property_interest_rollup"
+    return _DATASET_MAP.get(component_plan_type, "oracle_aggregated_metric")
+
+
+def _build_codebook_retrieval_plan(
+    prompt: str,
+    tenant_id: str,
+    actor_role: str,
+    matches: list[CodebookExample],
+) -> dict[str, Any]:
+    row_limit = 50 if actor_role in ("senior_broker", "junior_broker") else 200
+    desired_types = _detect_component_types(prompt)
+    if not desired_types:
+        desired_types = [_component_plan_type_from_codebook(matches[0])] if matches else ["table"]
+
+    title_hints: dict[str, str] = {}
+    for example in matches:
+        mapped = _component_plan_type_from_codebook(example)
+        title_hints.setdefault(mapped, example.title)
+
+    components: list[dict[str, Any]] = []
+    exemplar = matches[0]
+    for component_plan_type in desired_types[:4]:
+        dataset = _dataset_for_codebook(exemplar, prompt, component_plan_type)
+        components.append(
+            {
+                "suggestedType": component_plan_type,
+                "dataset": dataset,
+                "privacyTier": "standard",
+                "rowLimit": row_limit,
+                "joins": [],
+                "queryTemplate": f"SELECT * FROM {dataset} WHERE tenant_id = :tenant_id LIMIT :limit",
+                "queryParameters": {"tenant_id": tenant_id, "limit": row_limit},
+                "templateRef": {
+                    "exampleId": exemplar.example_id,
+                    "templateName": exemplar.template_name,
+                    "componentType": exemplar.component_type,
+                    "chapterName": exemplar.chapter_name,
+                    "subchapterName": exemplar.subchapter_name,
+                    "sourcePack": exemplar.source_pack,
+                },
+                "titleHint": title_hints.get(component_plan_type, exemplar.title),
+            }
+        )
+
+    return {
+        "planId": str(uuid.uuid4()),
+        "components": components,
+        "semanticModelVersion": "oracle_codebook_v2026_04_19_01",
+        "intentClass": "analytical",
+        "planner": "codebook_retrieval",
+    }
+
+
+_RUNTIME_ALLOWED_DATASETS = {
+    "deals",
+    "lead_daily_snapshot",
+    "lead_geo_interest_rollup",
+    "broker_performance",
+    "inventory_absorption",
+    "oracle_aggregated_metric",
+    "lead_activity_log",
+    "crm_contacts_overview",
+    "crm_opportunity_pipeline",
+    "crm_property_interest_rollup",
+    "crm_interaction_timeline",
+}
+
 
 class PromptOrchestrator:
     """
@@ -155,18 +295,35 @@ class PromptOrchestrator:
             "prompt": prompt,
             "intentClass": "analytical",
             "status": "planning",
-            "modelRuntime": "nemoclaw_hosted" if _NEMOCLAW_URL else "deterministic_fallback",
+            "modelRuntime": "runtime_llm" if runtime_llm_service._provider_catalog() else "deterministic_fallback",
             "semanticModelVersion": "oracle_semantic_v2026_04_08_01",
             "warnings": warnings,
             "componentsCreated": [],
             "clientRequestId": client_request_id,
             "createdAt": now,
+            "codebookMatches": [],
         }
         _DEMO_EXECUTIONS[execution_id] = execution
         await self._persist_execution(execution)
 
         # ── Step 1: Build retrieval plan ──────────────────────────────────────
-        if _NEMOCLAW_URL and _NEMOCLAW_API_KEY:
+        codebook_matches = codebook_service.search_examples(prompt, limit=4)
+        execution["codebookMatches"] = [
+            {
+                "exampleId": match.example_id,
+                "templateName": match.template_name,
+                "componentType": match.component_type,
+                "chapterName": match.chapter_name,
+                "subchapterName": match.subchapter_name,
+                "sourcePack": match.source_pack,
+            }
+            for match in codebook_matches
+        ]
+
+        if codebook_matches:
+            retrieval_plan = _build_codebook_retrieval_plan(prompt, tenant_id, actor_role, codebook_matches)
+            execution["status"] = "validated"
+        elif runtime_llm_service._provider_catalog():
             try:
                 retrieval_plan = await self._call_nemoclaw(prompt, conversation_context or [], ctx)
                 execution["status"] = "validated"
@@ -298,7 +455,7 @@ class PromptOrchestrator:
             comp: dict[str, Any] = {
                 "componentId": component_id,
                 "type": mapped_type,
-                "title": self._generate_title(prompt, ctype),
+                "title": str(plan.get("titleHint") or self._generate_title(prompt, ctype)),
                 "description": f"Generated from: \"{prompt[:80]}\"",
                 "dataSourceDescriptor": {
                     "descriptorId": str(uuid.uuid4()),
@@ -321,7 +478,7 @@ class PromptOrchestrator:
                     "promptExecutionId": execution_id,
                     "sourceBranchId": branch_id,
                     "createdBy": actor_id,
-                    "createdAt": _now(),
+                    "createdAt": _iso(_now()),
                 },
                 "renderingHints": self._rendering_hints(ctype),
                 "layout": {
@@ -413,7 +570,7 @@ class PromptOrchestrator:
                 "promptExecutionId": execution_id,
                 "sourceBranchId": branch_id,
                 "createdBy": actor_id,
-                "createdAt": _now(),
+                "createdAt": _iso(_now()),
             },
             "renderingHints": {"estimatedHeightPx": 180, "skeletonVariant": "text", "virtualizationPriority": 4},
             "layout": {
@@ -560,7 +717,7 @@ class PromptOrchestrator:
                 "promptExecutionId": execution_id,
                 "sourceBranchId": branch_id,
                 "createdBy": actor_id,
-                "createdAt": _now(),
+                "createdAt": _iso(_now()),
             },
             "renderingHints": {"estimatedHeightPx": 140, "skeletonVariant": "generic", "virtualizationPriority": 5},
             "layout": {
@@ -601,24 +758,80 @@ class PromptOrchestrator:
         ctx: PolicyContext,
     ) -> dict[str, Any]:
         """
-        Calls the Nemoclaw hosted model endpoint.
-        Raises on failure so the orchestrator can fall back to demo.
+        Uses the shared runtime LLM service to propose a retrieval plan.
+        Raises on malformed output so the orchestrator can fall back safely.
         """
-        import httpx  # type: ignore
-        async with httpx.AsyncClient(timeout=30.0) as client:
-            resp = await client.post(
-                f"{_NEMOCLAW_URL}/v1/oracle/plan",
-                headers={"Authorization": f"Bearer {_NEMOCLAW_API_KEY}"},
-                json={
-                    "prompt": prompt,
-                    "conversationContext": context,
-                    "tenantId": ctx.tenant_id,
-                    "actorRole": ctx.actor_role,
-                    "semanticModelVersion": "oracle_semantic_v2026_04_08_01",
+        row_limit = 50 if ctx.actor_role in ("senior_broker", "junior_broker") else 200
+        system_prompt = (
+            "You are the Oracle planner for Project Velocity. "
+            "Return JSON only. "
+            "Choose up to 4 analytical components for the prompt. "
+            "Allowed component types: pipeline_board, bar_chart, geo_map, table, line_chart, kpi_tile, activity_stream. "
+            "Allowed datasets: deals, lead_daily_snapshot, lead_geo_interest_rollup, broker_performance, inventory_absorption, "
+            "oracle_aggregated_metric, lead_activity_log, crm_contacts_overview, crm_opportunity_pipeline, "
+            "crm_property_interest_rollup, crm_interaction_timeline. "
+            "Return an object with keys semanticModelVersion, intentClass, components. "
+            "Each component must include suggestedType, dataset, and titleHint. "
+            "Do not emit SQL. Do not invent datasets outside the allowlist."
+        )
+        response = await runtime_llm_service.chat(
+            provider_id=None,
+            model=None,
+            system_prompt=system_prompt,
+            messages=[
+                *context,
+                {
+                    "role": "user",
+                    "content": json.dumps(
+                        {
+                            "prompt": prompt,
+                            "tenantId": ctx.tenant_id,
+                            "actorRole": ctx.actor_role,
+                            "rowLimit": row_limit,
+                        }
+                    ),
                 },
+            ],
+            temperature=0.1,
+            response_format="json",
+            metadata={"planner": "oracle_canvas"},
+        )
+        payload = response.get("message", {}).get("parsedJson") or {}
+        components_payload = payload.get("components")
+        if not isinstance(components_payload, list) or not components_payload:
+            raise ValueError("Runtime LLM planner returned no components.")
+
+        normalized_components: list[dict[str, Any]] = []
+        for raw_component in components_payload[:4]:
+            if not isinstance(raw_component, dict):
+                continue
+            suggested_type = str(raw_component.get("suggestedType", "")).strip()
+            dataset = str(raw_component.get("dataset", "")).strip()
+            if suggested_type not in _DATASET_MAP or dataset not in _RUNTIME_ALLOWED_DATASETS:
+                continue
+            normalized_components.append(
+                {
+                    "suggestedType": suggested_type,
+                    "dataset": dataset,
+                    "privacyTier": "standard",
+                    "rowLimit": row_limit,
+                    "joins": [],
+                    "queryTemplate": f"SELECT * FROM {dataset} WHERE tenant_id = :tenant_id LIMIT :limit",
+                    "queryParameters": {"tenant_id": ctx.tenant_id, "limit": row_limit},
+                    "titleHint": str(raw_component.get("titleHint", "")).strip() or self._generate_title(prompt, suggested_type),
+                }
             )
-            resp.raise_for_status()
-            return resp.json()  # type: ignore[no-any-return]
+
+        if not normalized_components:
+            raise ValueError("Runtime LLM planner returned no valid whitelisted components.")
+
+        return {
+            "planId": str(uuid.uuid4()),
+            "components": normalized_components,
+            "semanticModelVersion": str(payload.get("semanticModelVersion") or "oracle_runtime_llm_v2026_04_19_01"),
+            "intentClass": str(payload.get("intentClass") or "analytical"),
+            "planner": "runtime_llm",
+        }
 
     async def get_execution(self, execution_id: str) -> dict[str, Any] | None:
         return _DEMO_EXECUTIONS.get(execution_id)
@@ -668,8 +881,8 @@ class PromptOrchestrator:
                 execution.get("summary"),
                 execution.get("componentsCreated", []),
                 execution.get("clientRequestId"),
-                execution["createdAt"],
-                execution.get("completedAt"),
+                _coerce_datetime(execution["createdAt"]),
+                _coerce_datetime(execution.get("completedAt")),
             )
         finally:
             await conn.close()

backend/oracle/router_v1.py

@@ -26,20 +26,23 @@ import uuid
 from datetime import datetime, timezone
 from typing import Any, Set
 
-from fastapi import APIRouter, HTTPException, Request, WebSocket, WebSocketDisconnect, status
+from fastapi import APIRouter, Depends, HTTPException, Request, WebSocket, WebSocketDisconnect, status
 from pydantic import BaseModel, Field
 
+from backend.auth.dependencies import UserPrincipal, get_current_user
 from .canvas_service import canvas_service
 from .collaboration_service import collaboration_service
 from .action_service import oracle_action_service
 from .persona_service import persona_service
 from .prompt_orchestrator import prompt_orchestrator
 from .policy_service import PolicyService, PolicyContext
+from .codebook_service import codebook_service
 
 logger = logging.getLogger(__name__)
 
 router = APIRouter()
 policy_svc = PolicyService()
+_DEFAULT_TENANT_ID = os.getenv("ORACLE_DEFAULT_TENANT_ID", "tenant_velocity")
 
 # ── Helpers ───────────────────────────────────────────────────────────────────
 
@@ -51,13 +54,32 @@ def _now() -> str:
     return datetime.now(timezone.utc).isoformat()
 
 
-def _build_user_profile(default_page_id: str) -> dict[str, Any]:
+def _normalize_oracle_role(role: str) -> str:
+    mapping = {
+        "JUNIOR_BROKER": "junior_broker",
+        "SENIOR_BROKER": "senior_broker",
+        "SALES_DIRECTOR": "sales_director",
+        "ADMIN": "platform_admin",
+    }
+    return mapping.get(role.strip().upper(), "sales_director")
+
+
+def _build_user_profile(
+    *,
+    user_id: str,
+    email: str,
+    display_name: str,
+    role: str,
+    avatar_url: str | None,
+    default_page_id: str,
+) -> dict[str, Any]:
     return {
-        "userId": os.getenv("ORACLE_DEFAULT_USER_ID", "oracle_operator"),
-        "tenantId": os.getenv("ORACLE_DEFAULT_TENANT_ID", "tenant_velocity"),
-        "email": os.getenv("ORACLE_DEFAULT_EMAIL", "oracle@velocity.local"),
-        "displayName": os.getenv("ORACLE_DEFAULT_DISPLAY_NAME", "Oracle Operator"),
-        "role": os.getenv("ORACLE_DEFAULT_ROLE", "sales_director"),
+        "userId": user_id,
+        "tenantId": _DEFAULT_TENANT_ID,
+        "email": email,
+        "displayName": display_name,
+        "role": _normalize_oracle_role(role),
+        "avatarUrl": avatar_url,
         "timezone": os.getenv("ORACLE_DEFAULT_TIMEZONE", "Asia/Dubai"),
         "locale": os.getenv("ORACLE_DEFAULT_LOCALE", "en-AE"),
         "defaultPageId": default_page_id,
@@ -72,17 +94,39 @@ def _build_user_profile(default_page_id: str) -> dict[str, Any]:
     }
 
 
-async def _get_current_user() -> dict[str, Any]:
+async def _get_current_user_profile(request: Request, user: UserPrincipal) -> dict[str, Any]:
     seed_page = await canvas_service.ensure_default_page(
-        tenant_id=os.getenv("ORACLE_DEFAULT_TENANT_ID", "tenant_velocity"),
-        owner_id=os.getenv("ORACLE_DEFAULT_USER_ID", "oracle_operator"),
+        tenant_id=_DEFAULT_TENANT_ID,
+        owner_id=user.user_id,
         title=os.getenv("ORACLE_DEFAULT_PAGE_TITLE", "Oracle Main Canvas"),
     )
-    return _build_user_profile(seed_page["pageId"])
+    pool = getattr(request.app.state, "db_pool", None)
+    if pool is None:
+        raise HTTPException(status_code=503, detail="Database unavailable.")
+    async with pool.acquire() as conn:
+        row = await conn.fetchrow(
+            """
+            SELECT
+                COALESCE(full_name, split_part(email, '@', 1), id::text) AS display_name,
+                COALESCE(email, id::text || '@velocity.local') AS email,
+                avatar_url
+            FROM users_and_roles
+            WHERE id = $1::uuid
+            """,
+            user.user_id,
+        )
+    return _build_user_profile(
+        user_id=user.user_id,
+        email=row["email"] if row else f"{user.user_id}@velocity.local",
+        display_name=row["display_name"] if row else user.user_id,
+        role=user.role,
+        avatar_url=row["avatar_url"] if row else None,
+        default_page_id=seed_page["pageId"],
+    )
 
 
-async def _ctx_from_me() -> PolicyContext:
-    me = await _get_current_user()
+async def _ctx_from_request(request: Request, user: UserPrincipal) -> PolicyContext:
+    me = await _get_current_user_profile(request, user)
     return PolicyContext(
         tenant_id=me["tenantId"],
         actor_id=me["userId"],
@@ -143,13 +187,13 @@ class PersonaRenderRequest(BaseModel):
 # ── Endpoints ─────────────────────────────────────────────────────────────────
 
 @router.get("/me", summary="Get current user profile")
-async def get_me() -> dict:
-    return _ok(await _get_current_user())
+async def get_me(request: Request, user: UserPrincipal = Depends(get_current_user)) -> dict:
+    return _ok(await _get_current_user_profile(request, user))
 
 
 @router.get("/canvas-pages/{page_id}", summary="Get canvas page by ID")
-async def get_canvas_page(page_id: str) -> dict:
-    ctx = await _ctx_from_me()
+async def get_canvas_page(page_id: str, request: Request, user: UserPrincipal = Depends(get_current_user)) -> dict:
+    ctx = await _ctx_from_request(request, user)
     page = await canvas_service.get_page(page_id, ctx.tenant_id)
     if not page:
         raise HTTPException(status_code=404, detail=f"Canvas page '{page_id}' not found.")
@@ -157,8 +201,13 @@ async def get_canvas_page(page_id: str) -> dict:
 
 
 @router.post("/canvas-pages/{page_id}/prompts", summary="Submit a prompt to generate canvas components")
-async def submit_prompt(page_id: str, payload: PromptSubmitRequest) -> dict:
-    ctx = await _ctx_from_me()
+async def submit_prompt(
+    page_id: str,
+    payload: PromptSubmitRequest,
+    request: Request,
+    user: UserPrincipal = Depends(get_current_user),
+) -> dict:
+    ctx = await _ctx_from_request(request, user)
     execution = await prompt_orchestrator.execute(
         tenant_id=ctx.tenant_id,
         page_id=page_id,
@@ -198,8 +247,13 @@ async def submit_prompt(page_id: str, payload: PromptSubmitRequest) -> dict:
 
 
 @router.post("/canvas-pages/{page_id}/forks", summary="Create a fork (share) from a canvas page")
-async def create_fork(page_id: str, payload: ForkCreateRequest) -> dict:
-    ctx = await _ctx_from_me()
+async def create_fork(
+    page_id: str,
+    payload: ForkCreateRequest,
+    request: Request,
+    user: UserPrincipal = Depends(get_current_user),
+) -> dict:
+    ctx = await _ctx_from_request(request, user)
     page = await canvas_service.get_page(page_id, ctx.tenant_id)
     if not page:
         raise HTTPException(status_code=404, detail="Source page not found.")
@@ -214,8 +268,13 @@ async def create_fork(page_id: str, payload: ForkCreateRequest) -> dict:
 
 
 @router.post("/canvas-pages/{page_id}/rollback", summary="Rollback canvas to a prior revision")
-async def rollback_canvas(page_id: str, payload: RollbackRequest) -> dict:
-    ctx = await _ctx_from_me()
+async def rollback_canvas(
+    page_id: str,
+    payload: RollbackRequest,
+    request: Request,
+    user: UserPrincipal = Depends(get_current_user),
+) -> dict:
+    ctx = await _ctx_from_request(request, user)
     result = await canvas_service.rollback(
         page_id=page_id,
         tenant_id=ctx.tenant_id,
@@ -232,38 +291,44 @@ async def rollback_canvas(page_id: str, payload: RollbackRequest) -> dict:
 
 
 @router.get("/canvas-pages/{page_id}/revisions", summary="List revision history for a canvas page")
-async def list_revisions(page_id: str) -> dict:
-    ctx = await _ctx_from_me()
+async def list_revisions(page_id: str, request: Request, user: UserPrincipal = Depends(get_current_user)) -> dict:
+    ctx = await _ctx_from_request(request, user)
     revisions = await canvas_service.list_revisions(page_id, ctx.tenant_id)
     return _ok(revisions, meta={"count": len(revisions)})
 
 
 @router.get("/component-templates", summary="List component templates")
-async def list_templates(category: str | None = None, status: str | None = None) -> dict:
-    templates = PREMADE_TEMPLATES
-    if category:
-        templates = [t for t in templates if t["category"] == category]
-    if status:
-        templates = [t for t in templates if t["status"] == status]
-    return _ok(templates, meta={"count": len(templates)})
+async def list_templates(
+    category: str | None = None,
+    status: str | None = None,
+    search: str | None = None,
+    limit: int = 50,
+    offset: int = 0,
+) -> dict:
+    result = codebook_service.list_templates(
+        category=category,
+        status=status,
+        search=search,
+        limit=limit,
+        offset=offset,
+    )
+    return _ok(result["templates"], meta={"count": result["total"], "limit": limit, "offset": offset})
 
 
 @router.post("/component-templates/synthesize", summary="Synthesize a new component template from a prompt")
-async def synthesize_template(payload: TemplateSynthesizeRequest) -> dict:
-    me = await _get_current_user()
-    # Stub — full implementation requires Nemoclaw model runtime
-    template = {
-        "templateId": str(uuid.uuid4()),
-        "tenantId": me["tenantId"],
-        "name": "Synthesized Component",
-        "category": "custom",
-        "status": "tenant_draft",
-        "origin": "synthesized",
-        "version": "1.0.0",
-        "acceptedShapes": payload.dataShape,
-        "createdAt": _now(),
-        "updatedAt": _now(),
-    }
+async def synthesize_template(
+    payload: TemplateSynthesizeRequest,
+    request: Request,
+    user: UserPrincipal = Depends(get_current_user),
+) -> dict:
+    me = await _get_current_user_profile(request, user)
+    template = codebook_service.synthesize_template(
+        prompt=payload.prompt,
+        data_shapes=payload.dataShape,
+    )
+    template["tenantId"] = me["tenantId"]
+    template.setdefault("createdAt", _now())
+    template.setdefault("updatedAt", _now())
     return _ok(template)
 
 
@@ -293,8 +358,12 @@ async def list_merge_requests(targetPageId: str | None = None, status: str | Non
 
 
 @router.post("/merge-requests", summary="Open a merge request")
-async def create_merge_request(payload: MergeRequestCreateRequest) -> dict:
-    ctx = await _ctx_from_me()
+async def create_merge_request(
+    payload: MergeRequestCreateRequest,
+    request: Request,
+    user: UserPrincipal = Depends(get_current_user),
+) -> dict:
+    ctx = await _ctx_from_request(request, user)
     source_page = await canvas_service.get_page(payload.sourcePageId, ctx.tenant_id)
     target_page = await canvas_service.get_page(payload.targetPageId, ctx.tenant_id)
     if not source_page or not target_page:
@@ -319,8 +388,13 @@ async def create_merge_request(payload: MergeRequestCreateRequest) -> dict:
 
 
 @router.post("/merge-requests/{mr_id}/review", summary="Submit a merge request review")
-async def review_merge_request(mr_id: str, payload: MergeReviewRequest) -> dict:
-    ctx = await _ctx_from_me()
+async def review_merge_request(
+    mr_id: str,
+    payload: MergeReviewRequest,
+    request: Request,
+    user: UserPrincipal = Depends(get_current_user),
+) -> dict:
+    ctx = await _ctx_from_request(request, user)
     mr = await collaboration_service.review_merge_request(
         mr_id=mr_id,
         decision=payload.decision,
@@ -382,17 +456,3 @@ async def oracle_canvas_ws(ws: WebSocket, page_id: str) -> None:
 
 # ── Pre-made templates seed ───────────────────────────────────────────────────
 
-PREMADE_TEMPLATES = [
-    {"templateId": "tpl_kpi_pipeline_health_v1", "tenantId": "_system", "name": "Pipeline Health KPI", "category": "Executive overview", "status": "catalog_active", "origin": "premade", "version": "1.0.0", "acceptedShapes": ["scalar", "trend_scalar"]},
-    {"templateId": "tpl_bar_source_quality_v3", "tenantId": "_system", "name": "Lead Source Quality Bar", "category": "Lead quality", "status": "catalog_active", "origin": "premade", "version": "3.0.0", "acceptedShapes": ["categorical_aggregate"]},
-    {"templateId": "tpl_geo_investor_heat_v2", "tenantId": "_system", "name": "Investor Geography Heat Map", "category": "Geographic demand", "status": "catalog_active", "origin": "premade", "version": "2.0.0", "acceptedShapes": ["geospatial_aggregate"]},
-    {"templateId": "tpl_pipeline_board_v2", "tenantId": "_system", "name": "Deal Pipeline Board", "category": "Pipeline management", "status": "catalog_active", "origin": "premade", "version": "2.0.0", "acceptedShapes": ["categorical_records"]},
-    {"templateId": "tpl_broker_performance_v1", "tenantId": "_system", "name": "Broker Performance Ranked", "category": "Broker performance", "status": "catalog_active", "origin": "premade", "version": "1.0.0", "acceptedShapes": ["ranked_records"]},
-    {"templateId": "tpl_followup_queue_v1", "tenantId": "_system", "name": "Follow-up Queue", "category": "Operational queues", "status": "catalog_active", "origin": "premade", "version": "1.0.0", "acceptedShapes": ["task_records"]},
-    {"templateId": "tpl_investor_timeline_v1", "tenantId": "_system", "name": "Investor Timeline", "category": "Investor timelines", "status": "catalog_active", "origin": "premade", "version": "1.0.0", "acceptedShapes": ["chronological_events"]},
-    {"templateId": "tpl_absorption_trend_v1", "tenantId": "_system", "name": "Project Absorption Trend", "category": "Inventory and project analytics", "status": "catalog_active", "origin": "premade", "version": "1.0.0", "acceptedShapes": ["time_series"]},
-    {"templateId": "tpl_quota_gauge_v1", "tenantId": "_system", "name": "Quota Attainment Gauge", "category": "Executive overview", "status": "catalog_active", "origin": "premade", "version": "1.0.0", "acceptedShapes": ["scalar"]},
-    {"templateId": "tpl_campaign_lead_line_v1", "tenantId": "_system", "name": "Campaign-to-Lead Quality Timeline", "category": "Marketing analytics", "status": "catalog_active", "origin": "premade", "version": "1.0.0", "acceptedShapes": ["time_series"]},
-    {"templateId": "tpl_followup_gap_v1", "tenantId": "_system", "name": "Follow-up Gap Report", "category": "Operational queues", "status": "catalog_active", "origin": "premade", "version": "1.0.0", "acceptedShapes": ["task_records"]},
-    {"templateId": "tpl_qd_source_compare_v1", "tenantId": "_system", "name": "QD-Weighted Source Comparison", "category": "Lead quality", "status": "catalog_active", "origin": "premade", "version": "1.0.0", "acceptedShapes": ["categorical_aggregate"]},
-]

backend/scripts/build_oracle_runtime_codebook.py

@@ -0,0 +1,103 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from backend.oracle.codebook_service import (
+    _repo_root,
+    _safe_load_json,
+    _normalize_examples,
+)
+
+
+def main() -> None:
+    root = _repo_root()
+    primary_path = root / ".Agent Context" / "Sprint 1" / "Sayan Multi-Surface and Oracle Delivery Pack" / "Sample JSON Schema" / "GPT 5.4" / "oracle_canvas_json_expansion_pack" / "db" / "oracle_template_seed_db_expanded_v1.pretty.json"
+    secondary_path = root / ".Agent Context" / "Sprint 1" / "Sayan Multi-Surface and Oracle Delivery Pack" / "Sample JSON Schema" / "Claude Sonnet 4.6" / "oracle_template_expansion" / "oracle_template_seed_db_expanded.json"
+    fallback_path = root / "backend" / "oracle" / "oracle_template_seed_db.json"
+    output_path = root / "backend" / "oracle" / "oracle_runtime_codebook_merged.json"
+
+    corpora = []
+    for path, label in (
+        (primary_path, "gpt_5_4"),
+        (secondary_path, "claude_sonnet_4_6"),
+        (fallback_path, "runtime_seed_fallback"),
+    ):
+        if path.exists():
+            corpora.extend(_normalize_examples(_safe_load_json(path), label))
+
+    deduped = {}
+    for example in corpora:
+        key = (example.subchapter_id, example.template_name.lower(), example.title.lower())
+        existing = deduped.get(key)
+        if existing is None:
+            deduped[key] = example
+            continue
+        if example.source_pack == "gpt_5_4" and existing.source_pack != "gpt_5_4":
+            deduped[key] = example
+        elif example.is_canonical and not existing.is_canonical:
+            deduped[key] = example
+
+    examples = list(deduped.values())
+    chapters: dict[str, dict] = {}
+    for example in examples:
+        chapter = chapters.setdefault(
+            example.chapter_id,
+            {
+                "chapter_id": example.chapter_id,
+                "name": example.chapter_name,
+                "subchapters": {},
+            },
+        )
+        chapter["subchapters"].setdefault(
+            example.subchapter_id,
+            {
+                "subchapter_id": example.subchapter_id,
+                "name": example.subchapter_name,
+            },
+        )
+
+    payload = {
+        "_meta": {
+            "generated_by": "backend/scripts/build_oracle_runtime_codebook.py",
+            "source_priority": ["gpt_5_4", "claude_sonnet_4_6", "runtime_seed_fallback"],
+            "example_count": len(examples),
+        },
+        "chapters": [
+            {
+                "chapter_id": chapter["chapter_id"],
+                "name": chapter["name"],
+                "subchapters": list(chapter["subchapters"].values()),
+            }
+            for chapter in sorted(chapters.values(), key=lambda item: item["chapter_id"])
+        ],
+        "seed_examples": [
+            {
+                "example_id": example.example_id,
+                "chapter_id": example.chapter_id,
+                "subchapter_id": example.subchapter_id,
+                "title": example.title,
+                "template_name": example.template_name,
+                "component_type": example.component_type,
+                "accepted_shapes": list(example.accepted_shapes),
+                "example_json": example.example_json,
+                "quality_notes": example.quality_notes,
+                "is_canonical": example.is_canonical,
+                "source_pack": example.source_pack,
+                "surface_targets": list(example.surface_targets),
+                "policy_tags": list(example.policy_tags),
+                "backend_contract_hints": example.backend_contract_hints,
+            }
+            for example in sorted(
+                examples,
+                key=lambda item: (item.chapter_id, item.subchapter_id, item.template_name.lower(), item.title.lower()),
+            )
+        ],
+    }
+
+    output_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
+    print(f"Wrote merged Oracle runtime codebook to {output_path}")
+
+
+if __name__ == "__main__":
+    main()

backend/services/runtime_llm_service.py

@@ -0,0 +1,461 @@
+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")
+
+OLLAMA_BASE_URL = os.getenv("OLLAMA_BASE_URL", "http://127.0.0.1:11434").rstrip("/")
+OLLAMA_CHAT_URL = os.getenv("OLLAMA_CHAT_URL", f"{OLLAMA_BASE_URL}/v1/chat/completions")
+OLLAMA_TAGS_URL = os.getenv("OLLAMA_TAGS_URL", f"{OLLAMA_BASE_URL}/api/tags")
+OLLAMA_DEFAULT_MODEL = os.getenv("OLLAMA_MODEL", "qwen3.5:27b")
+
+NEMOCLAW_BASE_URL = os.getenv("NEMOCLAW_BASE_URL", "").rstrip("/")
+NEMOCLAW_CHAT_URL = (os.getenv("NEMOCLAW_CHAT_URL") or f"{NEMOCLAW_BASE_URL}/v1/chat/completions").rstrip("/") if NEMOCLAW_BASE_URL else ""
+NEMOCLAW_DEFAULT_MODEL = os.getenv("NEMOCLAW_MODEL", "nvidia/nemotron-3-super-120b-a12b")
+NEMOCLAW_API_TOKEN = os.getenv("NEMOCLAW_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]:
+        providers: list[RuntimeProvider] = []
+        if OLLAMA_CHAT_URL:
+            providers.append(
+                RuntimeProvider(
+                    provider_id="ollama",
+                    base_url=OLLAMA_BASE_URL,
+                    chat_url=OLLAMA_CHAT_URL,
+                    default_model=OLLAMA_DEFAULT_MODEL,
+                )
+            )
+        if NEMOCLAW_CHAT_URL:
+            providers.append(
+                RuntimeProvider(
+                    provider_id="nemoclaw",
+                    base_url=NEMOCLAW_BASE_URL,
+                    chat_url=NEMOCLAW_CHAT_URL,
+                    default_model=NEMOCLAW_DEFAULT_MODEL,
+                    auth_token=NEMOCLAW_API_TOKEN or None,
+                )
+            )
+        return providers
+
+    def get_provider(self, provider_id: str | None) -> RuntimeProvider:
+        providers = {provider.provider_id: provider for provider in self._provider_catalog()}
+        if provider_id:
+            provider = providers.get(provider_id)
+            if provider is None:
+                raise ValueError(f"Unknown provider '{provider_id}'.")
+            return provider
+
+        if "nemoclaw" in providers:
+            return providers["nemoclaw"]
+        if "ollama" in providers:
+            return providers["ollama"]
+        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:
+                if provider.provider_id == "ollama":
+                    async with httpx.AsyncClient(timeout=10.0) as client:
+                        response = await client.get(OLLAMA_TAGS_URL)
+                        response.raise_for_status()
+                    payload = response.json()
+                    models = [str(item.get("name", "")).strip() for item in payload.get("models", []) if item.get("name")]
+                    if provider.default_model not in models:
+                        models.insert(0, provider.default_model)
+                    status = "online"
+                else:
+                    async with httpx.AsyncClient(timeout=10.0) as client:
+                        response = await client.post(
+                            provider.chat_url,
+                            json={
+                                "model": provider.default_model,
+                                "messages": [{"role": "user", "content": "ping"}],
+                                "max_tokens": 4,
+                            },
+                            headers=provider.headers,
+                        )
+                        response.raise_for_status()
+                    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()