0pt/services/api/workers/quality_bridge.py

"""
services/api/workers/quality_bridge.py
=========================================
P010: the TS/Python split-brain bridge (BEFORE_STATE.md §2). The production
order pipeline (workers/pipeline.py) has always run its own one-line QA
filter (`[r for r in deduped if r["confidence"] >= threshold]`) and never
touched this repo's sophisticated, versioned TS QualityService (393 lines:
rule registry, GE delegation, OpenLineage) — "noll bryggor mellan sig", per
BEFORE_STATE.md.

This module is the ADDITIVE bridge, called optionally from pipeline.py. It
is NOT a replacement for the Python filter and cannot become a silent one:

  * Feature-flagged OFF by default (OPT_QUALITY_BRIDGE_ENABLED unset/"false").
    A disabled bridge makes zero network calls and returns immediately.
  * FAILS OPEN. Any failure mode (bridge URL unset, connection refused,
    timeout, non-2xx, malformed JSON) is caught here and reported back as
    `{"status": "unavailable", ...}` — it NEVER raises into pipeline.run_job,
    and it never changes which rows get delivered. `run_job` remains the
    single source of truth for delivery; this bridge is read-only/advisory.
  * ADVISORY ONLY, not a second gate — and this is a real limitation, not a
    choice made lightly: ExpectationEngine's execute() returns per-rule
    PASS/FAIL COUNTS and up to 20 unexpected-value SAMPLES
    (src/quality/expectation-engine.ts), not a per-row boolean array. There
    is no way, from a dataset-level score + a handful of samples, to know
    *which* of the N rows in this batch failed which rule — so per-row
    gating against the TS engine's result cannot be implemented honestly
    without first extending that engine's output contract (out of scope for
    this pass; flagged, not worked around with a guess). A future
    `OPT_QUALITY_BRIDGE_GATE_ENABLED` flag is deliberately NOT implemented
    here for that reason — see the docstring on `BridgeResult` below.

Transport: a direct HTTP call to the Next.js edge's
POST /api/0pt/quality/bridge-run (app/api/0pt/quality/bridge-run/route.ts),
reusing the same `X-Internal-Token` shared-secret convention lib/engine.ts
already uses for the (opposite-direction) edge -> engine call, for
consistency. This was chosen over an N0D job dispatch (the other option the
brief raised) because: the TS QualityService instance the pipeline needs to
reach lives in-process inside the Next.js app (src/services/quality.ts, an
in-memory singleton, same pattern as the Source Registry before its Postgres
persistence landed) — there is no durable store an N0D job could read/write
through today, and dispatching a job whose only effect is "call this
in-process singleton" would add a queue hop for no isolation benefit. A
synchronous HTTP call, wrapped in a short timeout and a hard fail-open, is
the simpler and more honest choice for a first cut; nothing here blocks a
later move to N0D once the quality state has real (Postgres) persistence.
"""
from __future__ import annotations

import os
from dataclasses import dataclass
from typing import Any

import httpx

_TIMEOUT = httpx.Timeout(4.0, connect=2.0)  # short — this must never meaningfully slow delivery


def _enabled() -> bool:
    return os.getenv("OPT_QUALITY_BRIDGE_ENABLED", "false").strip().lower() in ("1", "true", "yes", "on")


def _bridge_url() -> str | None:
    base = os.getenv("OPT_QUALITY_BRIDGE_URL") or os.getenv("EDGE_URL")
    if not base:
        return None
    return f"{base.rstrip('/')}/api/0pt/quality/bridge-run"


@dataclass
class BridgeResult:
    """
    Outcome of one bridge-run attempt. `status`:
      - "disabled":    OPT_QUALITY_BRIDGE_ENABLED is not set — no call made.
      - "not_configured": enabled, but no OPT_QUALITY_BRIDGE_URL/EDGE_URL set.
      - "unavailable": enabled + configured, but the call failed (timeout,
                        connection error, non-2xx, malformed body). Never raises.
      - "ok":           a real run completed; `score`/`run_id`/`issue_count` are populated.

    There is deliberately no `gated`/`dropped_rows` field — see the module
    docstring's "ADVISORY ONLY" section for why per-row gating isn't
    implementable against the current TS engine contract.
    """
    status: str
    score: float | None = None
    run_id: str | None = None
    run_status: str | None = None
    issue_count: int | None = None
    detail: str | None = None

    def as_metrics(self) -> dict[str, Any]:
        return {
            "status": self.status,
            "score": self.score,
            "run_id": self.run_id,
            "run_status": self.run_status,
            "issue_count": self.issue_count,
            **({"detail": self.detail} if self.detail else {}),
        }


async def run_quality_bridge(
    *,
    tenant_id: str,
    project_id: str,
    dataset_id: str,
    rows: list[dict[str, Any]],
    correlation_id: str | None = None,
) -> BridgeResult:
    """Ask the TS QualityService to grade `rows`. Never raises — see module docstring."""
    if not _enabled():
        return BridgeResult(status="disabled")

    url = _bridge_url()
    if not url:
        return BridgeResult(status="not_configured", detail="OPT_QUALITY_BRIDGE_URL/EDGE_URL not set")

    payload: dict[str, Any] = {
        "tenant_id": tenant_id,
        "project_id": project_id,
        "dataset_id": dataset_id,
        "rows": rows,
    }
    if correlation_id:
        payload["correlation_id"] = correlation_id

    headers = {
        "Content-Type": "application/json",
        "X-Internal-Token": os.getenv("INTERNAL_TOKEN", ""),
    }

    try:
        async with httpx.AsyncClient(timeout=_TIMEOUT) as client:
            r = await client.post(url, json=payload, headers=headers)
        if r.status_code not in (200, 201):
            return BridgeResult(status="unavailable", detail=f"HTTP {r.status_code}")
        body = r.json()
        run = body.get("run") or {}
        issues = body.get("issues") or []
        return BridgeResult(
            status="ok",
            score=run.get("score"),
            run_id=run.get("run_id"),
            run_status=run.get("status"),
            issue_count=len(issues),
        )
    except httpx.TimeoutException:
        return BridgeResult(status="unavailable", detail="timeout")
    except httpx.HTTPError as exc:
        return BridgeResult(status="unavailable", detail=f"{type(exc).__name__}: {exc}")
    except (ValueError, KeyError, TypeError) as exc:
        # Malformed/unexpected response body — treat exactly like network
        # unavailability, never propagate into the pipeline.
        return BridgeResult(status="unavailable", detail=f"malformed response: {type(exc).__name__}")