diff --git a/services/api/workers/pipeline.py b/services/api/workers/pipeline.py index b6da183..44e6b90 100644 --- a/services/api/workers/pipeline.py +++ b/services/api/workers/pipeline.py @@ -1,23 +1,9 @@ -""" -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). +"""Production pipeline: sourcing -> extraction -> canonical ER -> QA -> quality. -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:`) — plus, when the quality - bridge ran successfully, a `quality` block. BEFORE_STATE.md §3 found - these fields absent from the actual dataflow entirely. +Identity decisions are owned by src/er/entity-resolution-service.ts. The +Python worker only applies canonical `match` decisions returned by its +internal bridge. If tenant context or the canonical bridge is unavailable, +records remain unmerged: false identity merges are never used as fallback. """ from __future__ import annotations @@ -26,7 +12,6 @@ from typing import Any from workers.resolve import resolve from workers.quality_bridge import run_quality_bridge - from sources.connectors import ( CompanyResult, SourceBlockedError, @@ -39,53 +24,38 @@ from sources.connectors import ( from workers.adaptive_router import default_router, plan_for_market logger = logging.getLogger(__name__) +PIPELINE_VERSION = "0pt-pipeline@2" -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) + "NO": ["923609016"], + "GB": ["00000006"], } 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) + 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 + tenant_id = config.get("tenant_id") + project_id = config.get("project_id", "p_default") + correlation_id = config.get("correlation_id") 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: @@ -97,57 +67,55 @@ async def run_job(order_id: str, config: dict[str, Any]) -> dict[str, Any]: 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, + "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 + break per_market[market] = got - # 3) structuring — entity resolution (blocking + fuzzy match + corroboration) - deduped = resolve(records) + # Canonical ER is tenant-bound. Missing tenant context must never be + # papered over with a synthetic tenant or the removed Python scorer. + if tenant_id: + deduped = await resolve( + records, + tenant_id=tenant_id, + project_id=project_id, + correlation_id=correlation_id, + ) + er_mode = "canonical_ts" + else: + deduped = [dict(r) for r in records] + er_mode = "fail_closed_unmerged_missing_tenant" - # 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", + "detail": "config.tenant_id missing — quality bridge not called", } else: bridge_result = await run_quality_bridge( tenant_id=tenant_id, - project_id=config.get("project_id", "p_default"), + project_id=project_id, dataset_id=f"0pt-order-{order_id}", rows=delivered, - correlation_id=config.get("correlation_id"), + correlation_id=correlation_id, ) quality_bridge_metrics = bridge_result.as_metrics() if bridge_result.status == "ok": @@ -167,11 +135,16 @@ async def run_job(order_id: str, config: dict[str, Any]) -> dict[str, Any]: ) return { - "stage": "qa", "pct": 100, "state": "done", + "stage": "qa", + "pct": 100, + "state": "done", "metrics": { - "rows": len(delivered), "dropped_below_threshold": dropped, - "per_market": per_market, "avg_confidence": avg_conf, + "rows": len(delivered), + "dropped_below_threshold": dropped, + "per_market": per_market, + "avg_confidence": avg_conf, "pipeline_version": PIPELINE_VERSION, + "entity_resolution": er_mode, "quality_bridge": quality_bridge_metrics, }, "records": delivered,