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

"""
Production pipeline (docs/06): sourcing -> extraction -> structuring ->
enrichment/fusion -> QA. Compliant connectors only; every field gets provenance;
rows below the order's confidence target are dropped (deliver fewer, never fabricate).

P010 additions (both flagged in BEFORE_STATE.md, both additive — neither
changes which rows get delivered by the Python filter above):

  * Stage 6 (advisory, feature-flagged): after QA, `delivered` rows are
    optionally handed to the TS QualityService via
    `workers.quality_bridge.run_quality_bridge` — see that module's
    docstring for the full contract, the fail-open guarantee, and why
    per-row gating isn't implemented yet. Disabled unless
    `OPT_QUALITY_BRIDGE_ENABLED=true`.
  * Every delivered record's provenance rows are stamped with `producer`,
    `connector_version` (both already set at fetch time in
    sources/connectors.py), and an `artifact_ref` identifying this
    delivery (`0pt:pipeline_delivery:<order_id>`) — plus, when the quality
    bridge ran successfully, a `quality` block. BEFORE_STATE.md §3 found
    these fields absent from the actual dataflow entirely.
"""
from __future__ import annotations

import logging
from typing import Any

from workers.resolve import resolve
from workers.quality_bridge import run_quality_bridge

from sources.connectors import (
    CompanyResult,
    SourceBlockedError,
    brreg_by_orgnr,
    companies_house_by_number,
    cvr_by_number,
    gleif_by_lei,
    prh_by_businessid,
)
from workers.adaptive_router import default_router, plan_for_market

logger = logging.getLogger(__name__)

PIPELINE_VERSION = "0pt-pipeline@1"

# Minimal built-in seed so a job is demonstrable without customer input.
DEMO_SEEDS: dict[str, list[str]] = {
    "NO": ["923609016"],            # brreg orgnr (example)
    "GB": ["00000006"],             # companies house number (example)
}


async def _fetch(source_id: str, key: str) -> CompanyResult | None:
    if source_id == "gleif":
        return await gleif_by_lei(key)
    if source_id == "brreg":
        return await brreg_by_orgnr(key)
    if source_id == "companies_house":
        return await companies_house_by_number(key)
    if source_id == "prh":
        return await prh_by_businessid(key)
    if source_id == "cvr":
        return await cvr_by_number(key)
    return None


async def run_job(order_id: str, config: dict[str, Any]) -> dict[str, Any]:
    markets: list[str] = config.get("markets", []) or ["NO"]
    threshold = float(config.get("confidence", 92)) / 100.0
    # Use caller-supplied seeds when present (an explicit empty {} is honoured so a
    # real job is never silently backfilled with DEMO_SEEDS), but treat an explicit
    # null the same as "missing" -- `config.get("seeds") is not None` avoids a 500
    # (None.get) on client-controlled `{"seeds": null}`, which the old `... or ...`
    # idiom tolerated and the `"seeds" in config` form regressed.
    seeds: dict[str, list[str]] = config["seeds"] if config.get("seeds") is not None else DEMO_SEEDS

    records: list[dict[str, Any]] = []
    per_market: dict[str, int] = {}

    # 1) sourcing + 2) extraction
    for market in markets:
        keys = seeds.get(market, [])
        got = 0
        for key in keys:
            # Recomputed per key (not hoisted above this loop): a source
            # blocked while fetching an earlier key in this same market must
            # be excluded from the plan for every key still to come in this
            # run, not just retried forever with no memory (P014 -- see
            # adaptive_router.py's module docstring).
            plan = plan_for_market(market)
            for source_id in plan:
                try:
                    res = await _fetch(source_id, key)
                except SourceBlockedError:
                    default_router.record_blocked(market, source_id)
                    res = None
                except Exception:
                    res = None
                if res and res.name:
                    records.append({
                        "market": market, "name": res.name, "status": res.status,
                        "lei": res.lei, "org_nr": res.org_nr, "source": res.source,
                        "provenance": [p.__dict__ for p in res.provenance],
                        "confidence": max((p.confidence for p in res.provenance), default=0.0),
                    })
                    got += 1
                    break  # first compliant source that answers wins
        per_market[market] = got

    # 3) structuring — entity resolution (blocking + fuzzy match + corroboration)
    deduped = resolve(records)

    # 4) enrichment/fusion already folded into confidence above.
    # 5) QA — drop rows below the order's confidence threshold. THIS is the
    # filter that actually governs delivery; nothing below is allowed to
    # change its outcome.
    delivered = [r for r in deduped if r["confidence"] >= threshold]
    dropped = len(deduped) - len(delivered)
    avg_conf = round(sum(r["confidence"] for r in delivered) / len(delivered), 3) if delivered else 0.0

    # Stamp the delivery artifact reference on every provenance row of every
    # delivered record (BEFORE_STATE.md §3: `artifact_ref` did not exist
    # anywhere). This identifies WHICH delivery a field's value shipped in;
    # it does not assert the row was ever written to ST0RE (not wired yet).
    artifact_ref = f"0pt:pipeline_delivery:{order_id}"
    for r in delivered:
        for p in r.get("provenance", []):
            p["artifact_ref"] = artifact_ref

    # 6) quality bridge (advisory, feature-flagged, fail-open) — see
    # workers/quality_bridge.py's docstring for the full contract. Runs
    # AFTER the QA gate above and can never change `delivered`.
    tenant_id = config.get("tenant_id")
    quality_bridge_metrics: dict[str, Any]
    if not tenant_id:
        # The engine has no tenant context yet for orders created through
        # this API (orchestrator/main.py never threads tenant_id into
        # `config` — a pre-existing, separately-tracked gap). Rather than
        # fabricate a tenant to make the bridge call "work", skip it
        # honestly and say why.
        quality_bridge_metrics = {
            "status": "not_configured",
            "detail": "config.tenant_id missing — engine has no tenant context for this order",
        }
    else:
        bridge_result = await run_quality_bridge(
            tenant_id=tenant_id,
            project_id=config.get("project_id", "p_default"),
            dataset_id=f"0pt-order-{order_id}",
            rows=delivered,
            correlation_id=config.get("correlation_id"),
        )
        quality_bridge_metrics = bridge_result.as_metrics()
        if bridge_result.status == "ok":
            quality_info = {
                "score": bridge_result.score,
                "run_id": bridge_result.run_id,
                "run_status": bridge_result.run_status,
                "checked_by": "0pt-quality-bridge",
            }
            for r in delivered:
                for p in r.get("provenance", []):
                    p["quality"] = quality_info
        elif bridge_result.status != "disabled":
            logger.warning(
                "0pt.quality_bridge did not complete for order %s: %s (%s)",
                order_id, bridge_result.status, bridge_result.detail,
            )

    return {
        "stage": "qa", "pct": 100, "state": "done",
        "metrics": {
            "rows": len(delivered), "dropped_below_threshold": dropped,
            "per_market": per_market, "avg_confidence": avg_conf,
            "pipeline_version": PIPELINE_VERSION,
            "quality_bridge": quality_bridge_metrics,
        },
        "records": delivered,
    }