""" oracle/router_v1.py FastAPI router for all Oracle v1 endpoints. Mounted at /api/oracle/v1 in main.py. Endpoints (from spec §13.2): GET /me GET /canvas-pages/{pageId} POST /canvas-pages/{pageId}/prompts POST /canvas-pages/{pageId}/forks POST /canvas-pages/{pageId}/rollback GET /canvas-pages/{pageId}/revisions GET /component-templates POST /component-templates/synthesize (stub) GET /merge-requests POST /merge-requests POST /merge-requests/{mrId}/review WS /ws/oracle/canvas/{pageId} """ from __future__ import annotations import json import logging import os import uuid from datetime import datetime, timezone from typing import Any, Set from fastapi import APIRouter, HTTPException, Request, WebSocket, WebSocketDisconnect, status from pydantic import BaseModel, Field 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 logger = logging.getLogger(__name__) router = APIRouter() policy_svc = PolicyService() # ── Helpers ─────────────────────────────────────────────────────────────────── def _ok(data: Any, meta: dict | None = None) -> dict: return {"status": "ok", "data": data, "meta": meta or {}} def _now() -> str: return datetime.now(timezone.utc).isoformat() def _build_user_profile(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"), "timezone": os.getenv("ORACLE_DEFAULT_TIMEZONE", "Asia/Dubai"), "locale": os.getenv("ORACLE_DEFAULT_LOCALE", "en-AE"), "defaultPageId": default_page_id, "canvasPreferences": { "defaultDensity": "comfortable", "defaultPlacementMode": "append_after_last_visible_component", "showLineageBadges": True, }, "policyProfileId": os.getenv("ORACLE_POLICY_PROFILE_ID", "policy_sales_director_standard_v4"), "createdAt": os.getenv("ORACLE_PROFILE_CREATED_AT", _now()), "updatedAt": _now(), } async def _get_current_user() -> 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"), title=os.getenv("ORACLE_DEFAULT_PAGE_TITLE", "Oracle Main Canvas"), ) return _build_user_profile(seed_page["pageId"]) async def _ctx_from_me() -> PolicyContext: me = await _get_current_user() return PolicyContext( tenant_id=me["tenantId"], actor_id=me["userId"], actor_role=me["role"], ) # ── Pydantic Models ─────────────────────────────────────────────────────────── class PromptSubmitRequest(BaseModel): clientRequestId: str = Field(..., description="Client-generated idempotency key") branchId: str prompt: str = Field(..., min_length=1, max_length=4096) conversationContext: list[dict[str, str]] = Field(default_factory=list) placementMode: str = Field("append_after_last_visible_component") targetLeadId: str | None = None plannedWriteback: dict[str, Any] = Field(default_factory=dict) class ForkCreateRequest(BaseModel): recipientUserId: str sourceRevision: int visibility: str = Field("private", pattern="^(private|team)$") message: str = "" class RollbackRequest(BaseModel): targetRevision: int = Field(..., ge=1) clientRequestId: str class MergeRequestCreateRequest(BaseModel): sourcePageId: str sourceBranchId: str targetPageId: str targetBranchId: str title: str = Field(..., min_length=1, max_length=256) description: str = "" class MergeReviewRequest(BaseModel): decision: str = Field(..., pattern="^(approve|reject|changes_requested)$") comment: str = "" resolutions: list[dict[str, Any]] = Field(default_factory=list) class TemplateSynthesizeRequest(BaseModel): prompt: str dataShape: list[str] styleSignatureRef: str | None = None class PersonaRenderRequest(BaseModel): promptName: str = Field(..., pattern="^(qd_calculator|lead_tagger|cctv_profiler)$") variables: dict[str, Any] = Field(default_factory=dict) # ── Endpoints ───────────────────────────────────────────────────────────────── @router.get("/me", summary="Get current user profile") async def get_me() -> dict: return _ok(await _get_current_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() 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.") return _ok(page) @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() execution = await prompt_orchestrator.execute( tenant_id=ctx.tenant_id, page_id=page_id, branch_id=payload.branchId, actor_id=ctx.actor_id, actor_role=ctx.actor_role, prompt=payload.prompt, conversation_context=payload.conversationContext, client_request_id=payload.clientRequestId, placement_mode=payload.placementMode, ) if execution["status"] == "failed": raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail={"errors": execution.get("warnings", [])}, ) page = await canvas_service.get_page(page_id, ctx.tenant_id) action = await oracle_action_service.create_from_execution( execution=execution, target_entity_type="lead" if payload.targetLeadId else "canvas_page", target_entity_id=payload.targetLeadId or page_id, action_type="oracle_prompt_writeback_plan" if payload.targetLeadId else "oracle_canvas_generation", writeback_payload=payload.plannedWriteback, ) return _ok({ "executionId": execution["executionId"], "actionId": action["actionId"], "status": execution["status"], "pageId": page_id, "branchId": payload.branchId, "headRevision": execution.get("headRevision", page.get("headRevision", 0) if page else 0), "componentsCreated": execution.get("componentsCreated", []), "summary": execution.get("summary", ""), "warnings": execution.get("warnings", []), "components": page.get("components", []) if page else [], }) @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() page = await canvas_service.get_page(page_id, ctx.tenant_id) if not page: raise HTTPException(status_code=404, detail="Source page not found.") fork = await collaboration_service.create_fork( source_page=page, recipient_user_id=payload.recipientUserId, created_by=ctx.actor_id, visibility=payload.visibility, message=payload.message, ) return _ok(fork) @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() result = await canvas_service.rollback( page_id=page_id, tenant_id=ctx.tenant_id, actor_id=ctx.actor_id, target_revision=payload.targetRevision, idempotency_key=payload.clientRequestId, ) page = await canvas_service.get_page(page_id, ctx.tenant_id) return _ok({ "pageId": page_id, "headRevision": result.get("revisionNumber", payload.targetRevision), "components": page.get("components", []) if page else [], }) @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() 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)}) @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(), } return _ok(template) @router.get("/persona/health", summary="Health check for Oracle persona prompt loading") async def persona_health() -> dict: return _ok(await persona_service.health()) @router.post("/persona/render", summary="Render a subordinate Oracle persona prompt") async def persona_render(payload: PersonaRenderRequest) -> dict: try: rendered = await persona_service.render_prompt( prompt_name=payload.promptName, variables=payload.variables, ) except FileNotFoundError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc return _ok(rendered) @router.get("/merge-requests", summary="List merge requests for a target page") async def list_merge_requests(targetPageId: str | None = None, status: str | None = None) -> dict: if not targetPageId: raise HTTPException(status_code=400, detail="targetPageId query param required") mrs = await collaboration_service.list_merge_requests(targetPageId, status) return _ok(mrs, meta={"count": len(mrs)}) @router.post("/merge-requests", summary="Open a merge request") async def create_merge_request(payload: MergeRequestCreateRequest) -> dict: ctx = await _ctx_from_me() 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: raise HTTPException(status_code=404, detail="Source or target page not found.") mr = await collaboration_service.open_merge_request( tenant_id=ctx.tenant_id, source_page_id=payload.sourcePageId, source_branch_id=payload.sourceBranchId, source_head_revision=source_page.get("headRevision", 0), target_page_id=payload.targetPageId, target_branch_id=payload.targetBranchId, target_base_revision=target_page.get("headRevision", 0), title=payload.title, description=payload.description, created_by=ctx.actor_id, source_components=source_page.get("components", []), target_components=target_page.get("components", []), base_components=[], # Simplified: empty base for demo ) return _ok(mr) @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() mr = await collaboration_service.review_merge_request( mr_id=mr_id, decision=payload.decision, reviewer_id=ctx.actor_id, comment=payload.comment, resolutions=payload.resolutions, ) return _ok(mr) # ── WebSocket ───────────────────────────────────────────────────────────────── class OracleConnectionManager: def __init__(self) -> None: self.active: dict[str, Set[WebSocket]] = {} async def connect(self, ws: WebSocket, page_id: str) -> None: await ws.accept() self.active.setdefault(page_id, set()).add(ws) def disconnect(self, ws: WebSocket, page_id: str) -> None: page_connections = self.active.get(page_id, set()) page_connections.discard(ws) async def broadcast_page(self, page_id: str, payload: dict) -> None: dead: set[WebSocket] = set() for ws in self.active.get(page_id, set()): try: await ws.send_text(json.dumps(payload)) except Exception: dead.add(ws) if dead: self.active.get(page_id, set()).difference_update(dead) oracle_manager = OracleConnectionManager() @router.websocket("/ws/oracle/canvas/{page_id}") async def oracle_canvas_ws(ws: WebSocket, page_id: str) -> None: """ WebSocket endpoint for real-time Oracle canvas collaboration. Event types: oracle.page.revision.committed, oracle.prompt.received, oracle.presence.updated """ await oracle_manager.connect(ws, page_id) try: while True: data = await ws.receive_text() try: msg = json.loads(data) # Reflect heartbeat if msg.get("type") == "heartbeat": await ws.send_text(json.dumps({"type": "heartbeat.ack", "timestamp": _now()})) except json.JSONDecodeError: pass except WebSocketDisconnect: oracle_manager.disconnect(ws, page_id) # ── 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"]}, ]