feat(engine+edge): live vehicles + screening endpoints, entity-resolution + job-queue + financials modules

Edge /api/vehicles (RDW NL, live now) + /api/screen (OpenSanctions, live now), both defer to engine when ENGINE_URL set. Engine: /vehicles/{plate} + /screen endpoints; workers/resolve (blocking+fuzzy entity resolution, now used by pipeline); workers/queue (in-process job queue); sources/financials (Companies House accounts + XBRL path). py_compile + tsc + next build green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

app/api/screen/route.ts

@@ -0,0 +1,27 @@
+import { NextRequest, NextResponse } from "next/server";
+import { engineFetch } from "@/lib/engine";
+
+// GET /api/screen?q=Name — OpenSanctions screening (sanctions/PEP). VERIFY
+// commercial terms before production volume. Defers to engine when running.
+export async function GET(req: NextRequest) {
+  const q = req.nextUrl.searchParams.get("q") ?? "";
+  if (!q) return NextResponse.json({ matches: [] });
+
+  const eng = await engineFetch(`/screen?q=${encodeURIComponent(q)}`);
+  if (eng && eng.ok) return NextResponse.json(await eng.json());
+
+  try {
+    const r = await fetch(`https://api.opensanctions.org/search/default?q=${encodeURIComponent(q)}&limit=5`, {
+      headers: { "User-Agent": "0pt/0.1 (+https://0pt.app; SIAX)" },
+      cache: "no-store",
+    });
+    if (!r.ok) return NextResponse.json({ matches: [], source: "opensanctions" });
+    const data = (await r.json()) as { results?: { caption?: string; score?: number; schema?: string }[] };
+    return NextResponse.json({
+      source: "opensanctions",
+      matches: (data.results ?? []).map((x) => ({ name: x.caption ?? null, score: x.score ?? null, schema: x.schema ?? null })),
+    });
+  } catch {
+    return NextResponse.json({ matches: [], source: "opensanctions" });
+  }
+}

app/api/vehicles/route.ts

@@ -0,0 +1,34 @@
+import { NextRequest, NextResponse } from "next/server";
+import { engineFetch } from "@/lib/engine";
+
+// GET /api/vehicles?plate=XX-99-YY — RDW (NL) open vehicle data (technical/fleet
+// only, no owner PII). Defers to the engine when running; else calls RDW directly.
+export async function GET(req: NextRequest) {
+  const plate = (req.nextUrl.searchParams.get("plate") ?? "").toUpperCase().replace(/[-\s]/g, "");
+  if (!plate) return NextResponse.json({ error: { code: "bad_request", message: "plate krävs" } }, { status: 400 });
+
+  const eng = await engineFetch(`/vehicles/${plate}`);
+  if (eng && eng.ok) return NextResponse.json(await eng.json());
+
+  try {
+    const r = await fetch(`https://opendata.rdw.nl/resource/m9d7-ebf2.json?kenteken=${plate}`, {
+      headers: { "User-Agent": "0pt/0.1 (+https://0pt.app; SIAX)" },
+      cache: "no-store",
+    });
+    if (!r.ok) return NextResponse.json({ vehicle: null }, { status: 200 });
+    const arr = (await r.json()) as Record<string, string>[];
+    if (!arr.length) return NextResponse.json({ vehicle: null });
+    const d = arr[0];
+    return NextResponse.json({
+      source: "rdw",
+      vehicle: {
+        reg_plate: plate, market: "NL",
+        make: d.merk ?? null, model: d.handelsbenaming ?? null,
+        year: (d.datum_eerste_toelating ?? "").slice(0, 4) || null,
+      },
+      provenance: { level: 3, source: "rdw" },
+    });
+  } catch {
+    return NextResponse.json({ vehicle: null, error: "fetch_failed" }, { status: 200 });
+  }
+}

services/api/orchestrator/main.py

@@ -16,6 +16,7 @@ from fastapi import FastAPI, HTTPException
 from fastapi.middleware.cors import CORSMiddleware
 from pydantic import BaseModel
 
+from sources.connectors import opensanctions_screen, rdw_vehicle
 from workers.pipeline import run_job
 
 app = FastAPI(title="0pt data engine", version="0.1")
@@ -87,6 +88,17 @@ async def sources_for(market: str):
     return [s for s in _SOURCES if s.get("market") in (market, "global")]
 
 
+@app.get("/vehicles/{plate}")
+async def vehicle(plate: str):
+    v = await rdw_vehicle(plate)
+    return v or {"vehicle": None}
+
+
+@app.get("/screen")
+async def screen(q: str):
+    return await opensanctions_screen(q)
+
+
 @app.post("/estimates")
 async def estimates(body: EstimateIn):
     return compute_estimate(body)

services/api/sources/financials.py

@@ -0,0 +1,34 @@
+"""
+Structured financials (docs/06). XBRL/iXBRL is the fast, exact lane (not OCR).
+Companies House publishes accounts as iXBRL; this fetches the accounts filing
+history (real, key-gated) and documents the XBRL extraction step.
+"""
+from __future__ import annotations
+
+import os
+from typing import Any
+
+import httpx
+
+USER_AGENT = "0pt-data-engine/0.1 (+https://0pt.app; SIAX)"
+
+
+async def companies_house_filings(number: str) -> dict[str, Any]:
+    key = os.getenv("COMPANIES_HOUSE_API_KEY")
+    if not key:
+        return {"company": number, "filings": [], "note": "level0: kräver COMPANIES_HOUSE_API_KEY"}
+    url = f"https://api.company-information.service.gov.uk/company/{number}/filing-history?category=accounts"
+    async with httpx.AsyncClient(headers={"User-Agent": USER_AGENT}, timeout=15) as c:
+        r = await c.get(url, auth=(key, ""))
+        if r.status_code != 200:
+            return {"company": number, "filings": []}
+        items = r.json().get("items", [])
+        return {
+            "company": number,
+            "source": "companies_house",
+            "filings": [{"date": i.get("date"), "type": i.get("type"), "description": i.get("description")} for i in items[:10]],
+        }
+
+# XBRL extraction (extension): with `arelle` installed, download each accounts
+# filing's iXBRL document and pull tagged facts (turnover, profit, employees)
+# into fact_financials with provenance. Structured, exact — never OCR for XBRL.

services/api/workers/pipeline.py

@@ -7,7 +7,7 @@ from __future__ import annotations
 
 from typing import Any
 
-from rapidfuzz import fuzz
+from workers.resolve import resolve
 
 from sources.connectors import (
     CompanyResult,
@@ -70,14 +70,8 @@ async def run_job(order_id: str, config: dict[str, Any]) -> dict[str, Any]:
                     break  # first compliant source that answers wins
         per_market[market] = got
 
-    # 3) structuring — dedup by fuzzy name within market
-    deduped: list[dict[str, Any]] = []
-    for r in records:
-        dup = next((d for d in deduped if d["market"] == r["market"] and fuzz.ratio(d["name"], r["name"]) > 92), None)
-        if dup:
-            dup["confidence"] = min(0.99, dup["confidence"] + 0.1)  # corroboration bump
-        else:
-            deduped.append(r)
+    # 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.

services/api/workers/queue.py

@@ -0,0 +1,31 @@
+"""
+Lightweight in-process job queue (docs/02). Adequate for single-worker dev; swap
+the backend for Redis/Celery at scale without changing callers.
+"""
+from __future__ import annotations
+
+import asyncio
+from typing import Any, Awaitable, Callable
+
+Job = Callable[..., Awaitable[Any]]
+
+
+class JobQueue:
+    def __init__(self) -> None:
+        self._q: "asyncio.Queue[tuple[Job, tuple[Any, ...]]]" = asyncio.Queue()
+
+    async def enqueue(self, fn: Job, *args: Any) -> None:
+        await self._q.put((fn, args))
+
+    async def worker(self) -> None:
+        while True:
+            fn, args = await self._q.get()
+            try:
+                await fn(*args)
+            except Exception as e:  # keep the worker alive on a failed job
+                print("job failed:", repr(e))
+            finally:
+                self._q.task_done()
+
+
+queue = JobQueue()

services/api/workers/resolve.py

@@ -0,0 +1,33 @@
+"""
+Entity resolution (docs/06): block, score, and link records across sources so the
+same company from GLEIF + Brreg + CVR collapses to one record with corroborated
+confidence. rapidfuzz only; deterministic; no external calls.
+"""
+from __future__ import annotations
+
+from typing import Any
+
+from rapidfuzz import fuzz
+
+
+def block_key(rec: dict[str, Any]) -> tuple[str, str]:
+    """Cheap blocking key — same market + first initial — to avoid O(n^2)."""
+    return (rec.get("market") or "", (rec.get("name") or "")[:1].upper())
+
+
+def resolve(records: list[dict[str, Any]], threshold: int = 92) -> list[dict[str, Any]]:
+    out: list[dict[str, Any]] = []
+    for r in records:
+        match = None
+        for o in out:
+            if block_key(o) == block_key(r) and fuzz.token_sort_ratio(o.get("name", ""), r.get("name", "")) >= threshold:
+                match = o
+                break
+        if match:
+            # corroboration: independent source agreeing bumps confidence + level
+            match["confidence"] = min(0.99, match.get("confidence", 0.0) + 0.1)
+            match["agreed_sources"] = match.get("agreed_sources", 1) + 1
+            match.setdefault("merged_from", []).append(r.get("source"))
+        else:
+            out.append({**r, "agreed_sources": 1})
+    return out