| """ |
| Policy Copilot over HTTP: the third path of this Space, next to /detect and /file. |
| |
| POST /copilot/parse {"text", "context", "answers"} -> ParseResponse (copilot/pipeline.py) |
| GET /copilot/examples sample requests for the dashboard's empty state |
| |
| Same `Authorization: Bearer <API_SECRET>` as /detect. The CLOAK server is the only intended |
| caller (server/src/modules/org/routes/copilot.ts): it checks the admin's session, builds the |
| tenant context from its database, and validates the answer again before the dashboard sees it. |
| |
| What this path does and does not do: |
| - stateless; nothing is stored, nothing is written, no network calls |
| - the admin's text is never logged (they paste real values into it): only the request id, |
| the outcome and the latency |
| - bounded: text <= MAX_TEXT_CHARS, whole body <= MAX_BODY_BYTES (enforced in app.py before the |
| body is read), and the context is trimmed again by copilot/platform/context.py |
| - isolated: if the copilot cannot load, only /copilot/* answers 503; /detect and /file are |
| unaffected, exactly like docclass.py |
| """ |
|
|
| from __future__ import annotations |
|
|
| import logging |
| import time |
| from typing import Any, Callable |
|
|
| from fastapi import APIRouter, HTTPException, Request |
| from pydantic import BaseModel, ConfigDict, Field |
|
|
| log = logging.getLogger("cloak.copilot") |
|
|
| |
| MAX_TEXT_CHARS = 2_000 |
| |
| |
| MAX_BODY_BYTES = 2_000_000 |
|
|
| _parse: Callable[..., dict] | None = None |
| _examples: Callable[[], list[str]] | None = None |
| _snapshot: str | None = None |
| LOAD_ERROR: str | None = None |
|
|
|
|
| def load() -> None: |
| """Import the copilot and warm its vocabulary. Never raises; a failure is reported by /health.""" |
| global _parse, _examples, _snapshot, LOAD_ERROR |
| try: |
| from copilot.pipeline import examples, parse |
| from copilot.platform import catalog |
|
|
| warm = parse("block aadhaar in chatgpt", {}) |
| if warm.get("status") == "error": |
| raise RuntimeError("warm-up request failed") |
| _parse, _examples = parse, examples |
| _snapshot = catalog.snapshot()["_source"]["syncedAt"] |
| LOAD_ERROR = None |
| log.info("policy copilot ready (platform snapshot %s)", _snapshot) |
| except Exception as exc: |
| _parse = _examples = None |
| LOAD_ERROR = f"{type(exc).__name__}: {exc}" |
| log.exception("policy copilot failed to load") |
|
|
|
|
| def status() -> dict: |
| return {"loaded": _parse is not None, "snapshot": _snapshot, "error": LOAD_ERROR, |
| "max_text_chars": MAX_TEXT_CHARS} |
|
|
|
|
| class ParseRequest(BaseModel): |
| |
| model_config = ConfigDict(extra="forbid") |
|
|
| text: str = Field(..., min_length=1, max_length=MAX_TEXT_CHARS) |
| context: dict[str, Any] = Field(default_factory=dict) |
| answers: dict[str, Any] = Field(default_factory=dict) |
|
|
|
|
| def _ready() -> Callable[..., dict]: |
| if _parse is None: |
| raise HTTPException(status_code=503, detail="Policy Copilot is not loaded") |
| return _parse |
|
|
|
|
| def build_router(verify: Callable[[Request], None]) -> APIRouter: |
| router = APIRouter(prefix="/copilot") |
|
|
| @router.post("/parse") |
| def parse_route(body: ParseRequest, request: Request): |
| verify(request) |
| parse = _ready() |
| t0 = time.perf_counter() |
| out = parse(body.text, body.context, body.answers) |
| log.info("copilot parse %s status=%s drafts=%d %.1fms", out.get("requestId"), out.get("status"), |
| len(out.get("drafts") or []), (time.perf_counter() - t0) * 1000) |
| return out |
|
|
| @router.get("/examples") |
| def examples_route(request: Request): |
| verify(request) |
| _ready() |
| return {"examples": _examples() if _examples else []} |
|
|
| return router |
|
|
