"""Company profile enrichment: Wikidata mapping, Wikipedia/homepage precedence, provenance, grounded LLM text, people/relationship
persistence and idempotency. Fixtures in `fixtures/enrichment/` (trimmed Alphabet/Google entity, its English labels, a Wikipedia summary,
a homepage with a JSON-LD Organization). No live network: the fake fetcher answers the exact API URLs the client builds."""
from __future__ import annotations

import json
from pathlib import Path
from typing import Any
from urllib.parse import parse_qs, urlparse

import pytest
from conftest import FakeFetcher
from factories import (
    cleanup,
    intel_db,  # noqa: F401 — pytest fixture registered by import
    make_company,
)

from companyatlas.db import execute, fetch_all, fetch_one, jsonb, transaction
from companyatlas.services import enrichment as en

FIX = Path(__file__).resolve().parents[1] / "fixtures" / "enrichment"
ENTITY = json.loads((FIX / "wikidata-Q95.json").read_text(encoding="utf-8"))["entities"]["Q95"]
_LABELS = json.loads((FIX / "wikidata-labels.json").read_text(encoding="utf-8"))
LABELS = {q: v["label"] for q, v in _LABELS.items()}
DESCRIPTIONS = {q: v["description"] for q, v in _LABELS.items()}
SUMMARY = json.loads((FIX / "wikipedia-summary.json").read_text(encoding="utf-8"))
HOMEPAGE = (FIX / "homepage.html").read_text(encoding="utf-8")
EXTRA_LABELS: dict[str, str] = {}                      # test-local QIDs (never real ones, so DB tests cannot touch real companies)
CLAIMS = {("Q30", "P297"): ["US"], ("Q4917", "P498"): ["USD"]}
WD_URL = "https://www.wikidata.org/wiki/Q95"
WP_URL = "https://en.wikipedia.org/wiki/Google"


class WikimediaFetcher(FakeFetcher):
    """Answers wbgetentities / wbgetclaims from the fixtures whatever the id order; other URLs use the FakeFetcher routes."""

    async def get(self, url: str, **kw: Any) -> Any:
        p = urlparse(url)
        if p.netloc == "www.wikidata.org" and p.path == "/w/api.php":
            self.calls.append(url)
            assert kw.get("respect_robots") is False and kw.get("rate_per_min")
            q = {k: v[0] for k, v in parse_qs(p.query).items()}
            if q["action"] == "wbgetentities":
                ids = q["ids"].split("|")
                if "claims" in q["props"]:
                    ents = {i: ENTITY for i in ids if i == "Q95"} | {i: {"id": i, "missing": ""} for i in ids if i != "Q95"}
                else:
                    labels = LABELS | EXTRA_LABELS
                    descs = DESCRIPTIONS | {q: "company" for q in EXTRA_LABELS}
                    ents = {i: {"id": i, "labels": ({"en": {"language": "en", "value": labels[i]}} if labels.get(i) else {}),
                                "descriptions": ({"en": {"language": "en", "value": descs[i]}} if descs.get(i) else {})} for i in ids}
                return self._json({"entities": ents})
            if q["action"] == "wbgetclaims":
                vals = CLAIMS.get((q["entity"], q["property"]), [])
                return self._json({"claims": {q["property"]: [{"mainsnak": {"snaktype": "value", "datavalue": {"value": v, "type": "string"}}, "rank": "normal"} for v in vals]}})
        return await super().get(url, **kw)

    def _json(self, payload: dict[str, Any]) -> Any:
        from conftest import make_result

        return make_result("https://www.wikidata.org/w/api.php", json.dumps(payload), content_type="application/json")


def company_row(**over: Any) -> dict[str, Any]:
    base = {"id": "co_test", "slug": "google", "display_name": "Google", "canonical_domain": "google.com", "website": "https://www.google.com",
            "description": "American multinational technology company", "industries": [], "industry_primary": None, "country": None, "hq_city": None,
            "hq_region": None, "founded_year": None, "employees": None, "public_company": False, "ticker": None, "exchange": None, "legal_name": None,
            "lei": None, "sec_cik": None, "logo_url": None, "wikidata_id": "Q95", "source_meta": {"source": "wikidata", "industry_labels": ["technology company"]}}
    base.update(over)
    return base


@pytest.fixture
def fetcher(monkeypatch: pytest.MonkeyPatch) -> WikimediaFetcher:
    async def no_dns(url: str) -> None:            # SSRF validation resolves hosts — never in tests
        return None

    monkeypatch.setattr(en, "validate_destination_async", no_dns)
    f = WikimediaFetcher()
    f.add("https://en.wikipedia.org/api/rest_v1/page/summary/Google", json.dumps(SUMMARY), content_type="application/json")
    f.add("https://www.example-robotics.test/", HOMEPAGE)
    return f


# ================================================================================================================ Wikidata mapping


async def test_wikidata_mapping_and_provenance(fetcher: WikimediaFetcher) -> None:
    res = await en.enrich_company(company_row(), fetcher=fetcher, sources=("wikidata",), use_db=False, llm=False)
    p = res.profile
    assert res.sources_used == ["wikidata"] and not res.errors
    assert p["legal_name"] == "Google LLC" and p["founded_year"] == 1998 and p["legal_form"] == "limited liability company"
    assert p["hq"]["city"] == "Mountain View" and p["hq"]["country"] == "US" and p["hq"]["region"] == "California"
    assert p["hq"]["lat"] == pytest.approx(37.42, abs=0.01) and p["hq"]["lon"] == pytest.approx(-122.08, abs=0.01)
    assert p["employees"] and p["employees_year"] and p["employees_year"] >= 2013           # latest P585 observation wins
    assert p["revenue"] == {"value": 305630000000.0, "currency": "USD", "year": 2023}       # 2023 beats 2021
    assert p["net_income"]["year"] == 2023 and p["total_assets"] is None
    assert "software" in p["industries"] and p["industry_labels"] and "Internet industry" in p["industry_labels"]
    assert "Google Search" in p["products"]
    assert p["exchange"] is None and p["ticker"] is None                                    # both Nasdaq listings ended in 2016 (P582)
    assert p["isin"] == "US02079K3059" and p["lei"] == "7ZW8QJWVPR4P1J1KQY45" and p["sec_cik"] == "0001824723"
    assert p["official_website"] == "https://about.google/" and p["public_company"] is True   # ISIN present
    assert p["logo_url"].startswith("https://commons.wikimedia.org/wiki/Special:FilePath/Google_2026_logo.svg")
    assert p["socials"]["linkedin"] == "https://www.linkedin.com/company/google" and p["socials"]["x"] == "https://x.com/Google"
    assert p["socials"]["github"] == "https://github.com/google" and p["socials"]["crunchbase"] == "https://www.crunchbase.com/organization/google"
    assert p["wikipedia_url"] == WP_URL and p["wikidata_url"] == WD_URL
    assert p["description"] == "American multinational technology company" and p["description_source"] == "wikidata"
    by_field = {s["field"]: s for s in p["sources"]}
    assert by_field["employees"] == {"field": "employees", "source": "wikidata", "url": WD_URL, "retrieved_at": by_field["employees"]["retrieved_at"]}
    assert by_field["industry_labels"]["source"] == "wikidata" and by_field["revenue"]["url"] == WD_URL
    assert p["enriched_at"] and p["version"] == "profile-v1"
    # people: CEO (current) listed, founders listed, an ended CEO tenure → no_longer_listed (never "left")
    people = {x.name: x for x in res.people}
    assert people["Sundar Pichai"].status == "listed" and people["Sundar Pichai"].role_category == "ceo" and people["Sundar Pichai"].is_executive
    assert people["Larry Page"].role_category == "founder" and people["Sergey Brin"].source_url == WD_URL
    assert people["Eric Schmidt"].status == "no_longer_listed" and people["Eric Schmidt"].valid_to is not None
    assert all(x.title in ("Chief Executive Officer", "Founder") for x in res.people)
    # relationships: parent → SUBSIDIARY_OF, subsidiaries → PARENT_OF, owner of → OWNER_OF, capped per property
    kinds = {(r.kind, r.to_qid): r for r in res.relationships}
    assert kinds[("SUBSIDIARY_OF", "Q20800404")].to_name == "Alphabet Inc." and kinds[("SUBSIDIARY_OF", "Q20800404")].valid_from is not None
    assert kinds[("PARENT_OF", "Q1318441")].to_name == "AdMob" and kinds[("PARENT_OF", "Q1318441")].property == "P355"
    assert not any(k == "OWNER_OF" for k, _ in kinds)                    # Google's "owner of" items in the fixture are products → filtered
    assert ("PARENT_OF", "Q1053674") not in kinds                       # deprecated-rank statement (DoubleClick) ignored
    assert en.looks_like_organisation("top-level domain", default=True) is False and en.looks_like_organisation("American advertising company", default=False)
    assert en.looks_like_organisation("provides Internet ad serving services", default=False) is True
    assert en.looks_like_organisation("note-taking service developed by Google", default=False) is False
    assert en.looks_like_organisation(None, default=False) is False and en.looks_like_organisation("something unusual", default=True) is True
    # column back-fills: null columns filled, registry description NOT replaced by the one-line Wikidata description (same rank)
    assert res.column_updates["founded_year"] == 1998 and res.column_updates["hq_city"] == "Mountain View" and res.column_updates["country"] == "US"
    assert res.column_updates["legal_name"] == "Google LLC" and res.column_updates["public_company"] is True
    assert "description" not in res.column_updates and res.industries and "software" in res.industries
    # label lookups are batched (≤ 50 ids per call) and country/currency codes resolved through wbgetclaims
    label_calls = [c for c in fetcher.calls if "props=labels%7Cdescriptions" in c]
    assert label_calls and all(len(parse_qs(urlparse(c).query)["ids"][0].split("|")) <= 50 for c in label_calls)
    assert any("wbgetclaims" in c and "P297" in c for c in fetcher.calls)


def _item(qid: str) -> dict[str, Any]:
    return {"snaktype": "value", "datavalue": {"value": {"entity-type": "item", "id": qid}, "type": "wikibase-entityid"}}


def _string(v: str) -> dict[str, Any]:
    return {"snaktype": "value", "datavalue": {"value": v, "type": "string"}}


async def test_current_listing_and_ticker_qualifier(fetcher: WikimediaFetcher) -> None:
    """A current P414 statement gives the exchange; the ticker comes from its P249 qualifier when there is no top-level P249."""
    entity = {"id": "Q95", "labels": {}, "descriptions": {}, "sitelinks": {},
              "claims": {"P414": [{"mainsnak": {**_item("Q82059"), "property": "P414"}, "rank": "normal", "qualifiers": {"P249": [{**_string("GOOGL"), "property": "P249"}]}},
                                  {"mainsnak": {**_item("Q82059"), "property": "P414"}, "rank": "normal",
                                   "qualifiers": {"P249": [{**_string("OLD"), "property": "P249"}],
                                                  "P582": [{"snaktype": "value", "property": "P582", "datavalue": {"value": {"time": "+2010-01-01T00:00:00Z", "precision": 11}, "type": "time"}}]}}],
                         "P1128": [{"mainsnak": {"snaktype": "value", "property": "P1128", "datavalue": {"value": {"amount": "+10", "unit": "1"}, "type": "quantity"}}, "rank": "normal"},
                                   {"mainsnak": {"snaktype": "value", "property": "P1128", "datavalue": {"value": {"amount": "+12", "unit": "1"}, "type": "quantity"}}, "rank": "normal",
                                    "qualifiers": {"P585": [{"snaktype": "value", "property": "P585", "datavalue": {"value": {"time": "+2020-01-01T00:00:00Z", "precision": 9}, "type": "time"}}]}}]}}
    res = await en.enrich_company(company_row(description=None), fetcher=fetcher, entity=entity, sources=("wikidata",), use_db=False, llm=False)
    assert res.profile["exchange"] == "Nasdaq" and res.profile["ticker"] == "GOOGL" and res.profile["public_company"] is True
    assert res.profile["employees"] == 12 and res.profile["employees_year"] == 2020       # dated observation beats an undated one
    assert res.column_updates["ticker"] == "GOOGL" and res.column_updates["exchange"] == "Nasdaq"


async def test_wikidata_helpers() -> None:
    assert en.wd_time({"time": "+1998-09-04T00:00:00Z", "precision": 11})[0] == 1998
    assert en.wd_time({"time": "+2015-00-00T00:00:00Z", "precision": 9})[1].isoformat() == "2015-01-01"
    assert en.wd_time({"time": "-0050-00-00T00:00:00Z", "precision": 9}) == (None, None, 9)
    assert en.wd_quantity({"amount": "+47756", "unit": "1"}) == (47756.0, None)
    assert en.wd_quantity({"amount": "-3.5", "unit": "http://www.wikidata.org/entity/Q4917"}) == (-3.5, "Q4917")
    assert en.commons_url("Google 2026 logo.svg") == "https://commons.wikimedia.org/wiki/Special:FilePath/Google_2026_logo.svg"
    assert en.commons_url("Éclair (1).png") == "https://commons.wikimedia.org/wiki/Special:FilePath/%C3%89clair_%281%29.png"
    q = en.latest_quantity(ENTITY, "P2139")
    assert q and q[2] == 2023
    url = en.WikidataClient.entities_url(["Q95", "Q3884"])
    assert url.startswith("https://www.wikidata.org/w/api.php?format=json&action=wbgetentities&ids=Q95%7CQ3884&props=labels%7Cdescriptions%7Cclaims%7Csitelinks")
    assert "sitefilter=" in url and "enwiki" in url
    assert en.rank_of("description", "wikipedia") > en.rank_of("description", "llm") > en.rank_of("description", "homepage") > en.rank_of("description", "wikidata")
    assert en.rank_of("employees", "wikidata") > en.rank_of("employees", "homepage") > en.rank_of("employees", "registry")


async def test_wikidata_client_tolerates_failures() -> None:
    f = FakeFetcher()                                    # every URL → 404
    wd = en.WikidataClient(f)
    assert await wd.entities(["Q95"]) == {} and await wd.labels(["Q30"]) == {} and await wd.country_iso("Q30") is None
    assert await wd.currency_code("Q4917") == "USD"      # static table, no request
    res = await en.enrich_company(company_row(), fetcher=f, sources=("wikidata", "wikipedia"), use_db=False, llm=False)
    assert res.errors == ["wikidata: entity unavailable"] and res.profile["description"] == "American multinational technology company"
    assert res.column_updates == {} and res.people == [] and res.relationships == []


# ================================================================================================================ Wikipedia / homepage / precedence


async def test_wikipedia_description_with_attribution(fetcher: WikimediaFetcher) -> None:
    res = await en.enrich_company(company_row(), fetcher=fetcher, sources=("wikidata", "wikipedia"), use_db=False, llm=False)
    p = res.profile
    assert res.sources_used == ["wikidata", "wikipedia"]
    assert p["description"].startswith("Google LLC is an American multinational technology corporation") and len(p["description"]) >= 200
    assert p["description_source"] == "wikipedia" and p["description_url"] == WP_URL and p["description_license"] == "CC BY-SA 4.0"
    assert p["description_attribution"] == "Text from Wikipedia (en), CC BY-SA 4.0"
    assert p["logo_url"].startswith("https://commons.wikimedia.org/")          # Wikidata logo outranks the Wikipedia thumbnail
    assert res.column_updates["description"] == p["description"]               # wikipedia outranks the registry one-liner
    src = {s["field"]: s["source"] for s in p["sources"]}
    assert src["description"] == "wikipedia" and src["wikipedia_url"] in ("wikidata", "wikipedia")


def test_clean_extract() -> None:
    raw = "Acme Corp (pronounced /ˈækmi/) is a company.[1] It makes things.[citation needed]\n\nSecond paragraph here.\nThird.\nFourth is dropped."
    assert en.clean_extract(raw) == "Acme Corp is a company. It makes things. Second paragraph here. Third."
    long = " ".join([f"Sentence number {i} is here." for i in range(200)])
    out = en.clean_extract(long, max_chars=300)
    assert out and len(out) <= 300 and out.endswith(".")
    assert en.clean_extract("") is None


async def test_homepage_facts_and_icon(fetcher: WikimediaFetcher) -> None:
    company = company_row(id="co_home", slug="example-robotics", display_name="Example Robotics", canonical_domain="example-robotics.test",
                          website="https://www.example-robotics.test/", wikidata_id=None, description=None, source_meta={"source": "manual"})
    res = await en.enrich_company(company, fetcher=fetcher, sources=("homepage",), use_db=False, llm=False)
    p = res.profile
    assert res.sources_used == ["homepage"] and not res.errors
    assert p["description"].startswith("Example Robotics designs and builds autonomous mobile robots") and p["description_source"] == "homepage"
    assert p["description_attribution"] == en.HOMEPAGE_ATTRIBUTION and p["description_license"] is None
    assert p["legal_name"] == "Example Robotics Inc." and p["founded_year"] == 2014 and p["employees"] == 420 and p["phone"] == "+1 514-555-0100"
    assert p["hq"] == {"city": "Montréal", "region": "Quebec", "country": "CA", "address": "1200 Rue Example, H2X 1Y4, Montréal, Quebec, CA", "lat": None, "lon": None}
    assert p["logo_url"] == "https://www.example-robotics.test/static/logo.svg"
    assert p["icon_url"] == "https://www.example-robotics.test/static/apple-touch-icon.png"        # apple-touch-icon beats favicon and og:image
    assert p["socials"] == {"linkedin": "https://www.linkedin.com/company/example-robotics", "x": "https://twitter.com/examplerobotics",
                            "github": "https://github.com/example-robotics"}
    assert res.column_updates["description"] == p["description"] and res.column_updates["country"] == "CA" and res.column_updates["employees"] == 420
    facts = en.profile_facts(p)
    keys = {f["key"]: f for f in facts}
    assert keys["headquarters"]["value"] == "Montréal, Quebec, CA" and keys["employees"]["value"] == "420" and keys["founded"]["source"] == "homepage"
    assert keys["founded"]["url"] == "https://www.example-robotics.test/"


def test_parse_homepage_ignores_relative_junk() -> None:
    html = '<html><head><meta name="description" content="short"><link rel="icon" href="javascript:alert(1)"><meta property="og:image" content="//cdn.example.test/og.png">' \
           '<script type="application/ld+json">{"@type":"Organization","name":"X","numberOfEmployees":{"minValue":10,"maxValue":50},"foundingDate":"not a date","address":"12 Main St"}</script></head></html>'
    f = en.parse_homepage(html, "https://www.example.test/")
    assert f.description is None and f.icon == "https://cdn.example.test/og.png" and f.employees is None and f.founded_year is None and f.address == "12 Main St"


async def test_better_source_is_never_overwritten(fetcher: WikimediaFetcher) -> None:
    """A company whose description column already came from Wikipedia keeps it when only the homepage runs; a homepage logo does not
    replace a Wikidata logo, but does replace a registry one."""
    company = company_row(id="co_keep", slug="example-robotics", canonical_domain="example-robotics.test", website="https://www.example-robotics.test/",
                          wikidata_id=None, description="Long encyclopedic text from Wikipedia about the company.", logo_url="https://commons.wikimedia.org/x.svg",
                          employees=400, source_meta={"source": "wikidata", "provenance": {"description": {"source": "wikipedia", "url": WP_URL},
                                                                                            "logo_url": {"source": "wikidata", "url": WD_URL}}})
    res = await en.enrich_company(company, fetcher=fetcher, sources=("homepage",), use_db=False, llm=False)
    assert res.profile["description"] == "Long encyclopedic text from Wikipedia about the company." and res.profile["description_source"] == "wikipedia"
    assert res.profile["logo_url"] == "https://commons.wikimedia.org/x.svg"
    assert res.profile["employees"] == 420                                  # homepage JSON-LD outranks the registry value…
    assert res.column_updates == {"employees": 420, "hq_city": "Montréal", "hq_region": "Quebec", "country": "CA", "founded_year": 2014,
                                  "legal_name": "Example Robotics Inc."}    # …and description / logo_url are left alone
    # registry logo (no provenance) is replaced by the homepage JSON-LD logo
    company2 = company_row(id="co_keep2", slug="example-robotics", canonical_domain="example-robotics.test", website="https://www.example-robotics.test/",
                           wikidata_id=None, logo_url="https://seed.example/logo.png", source_meta={"source": "wikidata"})
    res2 = await en.enrich_company(company2, fetcher=fetcher, sources=("homepage",), use_db=False, llm=False)
    assert res2.column_updates["logo_url"] == "https://www.example-robotics.test/static/logo.svg"


# ================================================================================================================ LLM guard rails


def test_numbers_grounded() -> None:
    src = "Founded in Montréal in 2014, the company employs 420 people and operates 2 sites."
    assert en.numbers_grounded("The company was founded in 2014 and has 420 employees.", src)
    assert not en.numbers_grounded("The company has 1,200 employees.", src)
    assert not en.numbers_grounded("Revenue reached $3.5 billion in 2014.", src)
    assert en.numbers_grounded("No figures here.", src)


async def test_llm_only_without_wikipedia_and_with_enough_text(fetcher: WikimediaFetcher, monkeypatch: pytest.MonkeyPatch) -> None:
    calls: list[dict[str, Any]] = []

    async def fake_llm(company: dict[str, Any], text: str, *, source_url: str) -> tuple[str | None, str | None, dict[str, Any]]:
        calls.append({"company": company["id"], "chars": len(text), "url": source_url})
        return "Example Robotics builds autonomous mobile robots for warehouses and serves retailers in Canada and Europe.", "llm_x", {"status": "done"}

    monkeypatch.setattr(en, "llm_profile_text", fake_llm)
    monkeypatch.setattr(en.settings, "llm_enabled", True)
    monkeypatch.setattr(en.settings, "llm_base_url", "http://llm.test/v1")
    company = company_row(id="co_llm", slug="example-robotics", canonical_domain="example-robotics.test", website="https://www.example-robotics.test/",
                          wikidata_id=None, description=None, source_meta={"source": "manual"})
    res = await en.enrich_company(company, fetcher=fetcher, sources=("homepage", "llm"), use_db=False, llm=True)
    assert calls and calls[0]["chars"] >= en.settings.enrich_llm_min_text_chars and calls[0]["url"] == "https://www.example-robotics.test/"
    assert res.profile["description_source"] == "llm" and res.profile["description_attribution"] == en.LLM_ATTRIBUTION and res.llm_job_id == "llm_x"
    assert res.profile["description"].startswith("Example Robotics builds") and "llm" in res.sources_used
    # with a Wikipedia extract the LLM is never called
    calls.clear()
    res = await en.enrich_company(company_row(), fetcher=fetcher, sources=("wikidata", "wikipedia", "llm"), use_db=False, llm=True)
    assert not calls and res.profile["description_source"] == "wikipedia"


# ================================================================================================================ persistence (Postgres)


def test_entity() -> dict[str, Any]:
    """The Q95 fixture with every company-like target renamed to a test-only QID (real companies in the local DB must never be linked)."""
    import copy

    ent = copy.deepcopy(ENTITY)
    ent["id"] = "QZTEST95"
    for prop in ("P749", "P355", "P127", "P1830"):
        for st in ent["claims"].get(prop) or []:
            v = st["mainsnak"]["datavalue"]["value"]
            old = v["id"]
            v["id"] = "QZTEST" + old[1:]
            EXTRA_LABELS[v["id"]] = LABELS.get(old) or f"ZTest {old}"
    EXTRA_LABELS["QZTEST20800404"] = "ZTest Alphabet"
    return ent


@pytest.mark.usefixtures("intel_db")
async def test_persist_people_relationships_idempotent(fetcher: WikimediaFetcher) -> None:
    entity = test_entity()
    wd_url = "https://www.wikidata.org/wiki/QZTEST95"
    try:
        async with transaction() as conn:
            google = await make_company(conn, name="ZTest Google", country="US")
            alphabet = await make_company(conn, name="ZTest Alphabet", country="US")
            await execute(conn, "update companies set wikidata_id = 'QZTEST95', description = 'American multinational technology company', source_meta = cast(:m as jsonb) where id = :id",
                          id=google["id"], m=jsonb({"source": "wikidata"}))
            await execute(conn, "update companies set wikidata_id = 'QZTEST20800404' where id = :id", id=alphabet["id"])
            # a page-sourced person that Wikidata also knows: must keep its page provenance and title
            await execute(conn, """insert into people (id, company_id, name, name_norm, title, role_category, is_executive, status, source_url)
                                   values ('person_ztest_pichai', :c, 'Sundar Pichai', :nn, 'CEO, Google and Alphabet', 'ceo', true, 'listed', :url)""",
                          c=google["id"], nn=en.norm_name("Sundar Pichai"), url="https://about.google/leadership/")
            row = await fetch_one(conn, "select * from companies where id = :id", id=google["id"])
        res = await en.enrich_company(row, fetcher=fetcher, entity=entity, sources=("wikidata", "wikipedia"), use_db=True, llm=False)
        async with transaction() as conn:
            stored = await en.persist(conn, row, res)
        assert stored["people"] >= 4 and stored["relationships_new"] >= 4
        async with transaction() as conn:
            c = await fetch_one(conn, "select * from companies where id = :id", id=google["id"])
            people = await fetch_all(conn, "select * from people where company_id = :c order by name", c=google["id"])
            rels = await fetch_all(conn, "select * from company_relationships where from_company_id = :c or to_company_id = :c order by kind", c=google["id"])
        meta = c["source_meta"]
        assert meta["profile"]["description_source"] == "wikipedia" and meta["enriched_at"] and meta["enrichment"]["sources"] == ["wikidata", "wikipedia"]
        assert c["description"].startswith("Google LLC is an American") and meta["provenance"]["description"]["source"] == "wikipedia"
        assert meta["provenance"]["description"]["previous"] == "American multinational technology company"
        assert c["founded_year"] == 1998 and c["hq_city"] == "Mountain View" and c["legal_name"] == "Google LLC" and c["public_company"] is True
        assert "software" in c["industries"] and c["industry_primary"] == c["industries"][0]
        by_name = {p["name"]: p for p in people}
        pichai = by_name["Sundar Pichai"]
        assert pichai["id"] == "person_ztest_pichai" and pichai["title"] == "CEO, Google and Alphabet" and pichai["source_url"] == "https://about.google/leadership/"
        assert by_name["Larry Page"]["source_url"] == wd_url and by_name["Larry Page"]["title"] == "Founder" and by_name["Larry Page"]["status"] == "listed"
        assert by_name["Eric Schmidt"]["status"] == "no_longer_listed" and by_name["Eric Schmidt"]["removed_at"] is not None
        sub = [r for r in rels if r["kind"] == "SUBSIDIARY_OF" and r["from_company_id"] == google["id"]]
        assert len(sub) == 1 and sub[0]["to_company_id"] == alphabet["id"] and sub[0]["provenance"]["property"] == "P749" and float(sub[0]["confidence"]) == pytest.approx(0.85)
        inverse = [r for r in rels if r["kind"] == "PARENT_OF" and r["from_company_id"] == alphabet["id"] and r["to_company_id"] == google["id"]]
        assert len(inverse) == 1
        named = [r for r in rels if r["kind"] == "PARENT_OF" and r["from_company_id"] == google["id"]]
        assert named and all(r["to_company_id"] is None and r["to_name"] for r in named)
        # second run: no new people / relationships, profile refreshed, columns unchanged
        res2 = await en.enrich_company(c, fetcher=fetcher, entity=entity, sources=("wikidata", "wikipedia"), use_db=True, llm=False)
        assert res2.column_updates == {} and res2.industries == []
        async with transaction() as conn:
            stored2 = await en.persist(conn, c, res2)
            n_people = await fetch_one(conn, "select count(*) as n from people where company_id = :c", c=google["id"])
            n_rel = await fetch_one(conn, "select count(*) as n from company_relationships where from_company_id = :c or to_company_id = :c", c=google["id"])
            c2 = await fetch_one(conn, "select * from companies where id = :id", id=google["id"])
        assert stored2["relationships_new"] == 0 and stored2["relationships_seen"] == len(rels)
        assert n_people["n"] == len(people) and n_rel["n"] == len(rels)
        assert c2["source_meta"]["profile"]["enriched_at"] >= meta["profile"]["enriched_at"] and c2["description"] == c["description"]
        # pending queue: freshly enriched companies are no longer due
        async with transaction() as conn:
            due = {r["id"] for r in await en.pending_companies(conn, 100000)}
        assert google["id"] not in due and alphabet["id"] in due
        # batch runner end-to-end (fake fetcher: the entity lookup for a test QID is "missing" → recorded, still persisted, never raises)
        stats = await en.enrich_pending(company_keys=[alphabet["slug"]], fetcher=fetcher, sources=("wikidata",), llm=False)
        assert stats["companies"] == 1 and stats["ok"] == 1 and stats["failed"] == 0 and stats["requests"] >= 1
        async with transaction() as conn:
            a = await fetch_one(conn, "select source_meta from companies where id = :id", id=alphabet["id"])
        assert a["source_meta"]["enrichment"]["errors"] == ["wikidata: entity unavailable"] and a["source_meta"]["profile"]["version"] == "profile-v1"
    finally:
        await cleanup()