treg/tests/test_routing.py

"""Capability routing — first-party routed endpoints (docs/CAPABILITY-ROUTING-PLAN.md).
`treg.people.email.find` picks a child and runs it through the ordinary call use case."""

from __future__ import annotations

import asyncio
import json
from pathlib import Path

import pytest
from httpx import AsyncClient
from sqlmodel import select

from treg import audit
from treg.domain import money as ledger
from treg.application.call import route as call_route
from treg.application.call import service as call_service
from treg.application.call.types import UpstreamResponse
from treg.config import get_settings
from treg.infra.db import session_maker
from treg.domain.catalog import store as catalog_store
from treg.domain.catalog.routing import paths as P
from treg.domain.catalog.routing.contracts import canonical_identity
from treg.domain.catalog.routing.plan import Candidate, cost_at, rank
from treg.infra.catalog_observations import CachedEndpointObservationReader
from treg.models import CallRecord, Hold, LedgerEntry

from test_marketplace_call import _balance, platform_on  # noqa: F401

ROUTED = "treg.people.email.find"


@pytest.fixture
def enrichment_on(monkeypatch, platform_on):
    for p in ("HUNTER", "TOMBA", "LEADMAGIC", "LEADSFORGE", "FINDYMAIL", "AVIATO", "FIBER_AI"):
        monkeypatch.setenv(f"TREG_PLATFORM_KEY_{p}", f"PLATFORM-{p}-KEY")
    monkeypatch.setenv("TREG_PLATFORM_KEY_TOMBA_SECRET", "PLATFORM-TOMBA-SECRET")
    monkeypatch.setenv("TREG_PLATFORM_PROVIDERS", "hunter,tomba,leadmagic,leadsforge,findymail,aviato,fiber-ai")
    get_settings.cache_clear()
    yield
    get_settings.cache_clear()


@pytest.fixture
def enrichment_with_miss_declarers_on(monkeypatch, enrichment_on):
    """enrichment_on plus the two providers whose miss is a 4xx with a body (prospeo 400 NO_MATCH)
    or a plain 4xx (limadata 404) — without them in TREG_PLATFORM_PROVIDERS a prospeo/limadata
    test never runs the child and passes vacuously."""
    for p in ("PROSPEO", "LIMADATA"):
        monkeypatch.setenv(f"TREG_PLATFORM_KEY_{p}", f"PLATFORM-{p}-KEY")
    monkeypatch.setenv("TREG_PLATFORM_PROVIDERS", "hunter,tomba,leadmagic,leadsforge,findymail,aviato,fiber-ai,prospeo,limadata")
    get_settings.cache_clear()
    yield
    get_settings.cache_clear()


@pytest.fixture
def enrichment_with_quickenrich_on(monkeypatch, platform_on):
    """Like enrichment_on but includes QuickEnrich - the cheapest provider for phone/email lookups."""
    for p in ("HUNTER", "TOMBA", "LEADMAGIC", "LEADSFORGE", "FINDYMAIL", "AVIATO", "FIBER_AI", "QUICKENRICH"):
        monkeypatch.setenv(f"TREG_PLATFORM_KEY_{p}", f"PLATFORM-{p}-KEY")
    monkeypatch.setenv("TREG_PLATFORM_KEY_TOMBA_SECRET", "PLATFORM-TOMBA-SECRET")
    monkeypatch.setenv("TREG_PLATFORM_PROVIDERS", "hunter,tomba,leadmagic,leadsforge,findymail,aviato,fiber-ai,quickenrich")
    get_settings.cache_clear()
    yield
    get_settings.cache_clear()


def _relay_by_provider(answers: dict[str, list[tuple[int, dict]]], seen: list):
    """A fake upstream keyed by the vendor host: each provider answers its scripted list in order."""
    async def _relay(request, upstream_url, tool, secrets, client, drop_params=None, force_identity=False):
        provider = next((p for p in answers if p != "*" and p in upstream_url), "*")  # "*": any other vendor
        body = b""
        async for chunk in request.body_stream():
            body += chunk
        seen.append((provider, request.method, dict(request.query_items), json.loads(body) if body else None))
        status, doc = answers[provider].pop(0)
        payload = json.dumps(doc).encode()
        async def _s():
            yield payload
        async def _c():
            return None
        return UpstreamResponse(status, ((b"content-type", b"application/json"),), _s(), _c)
    return _relay


# ---- pure ------------------------------------------------------------------------------------

def test_expression_language():
    doc = {"data": {"email": "a@x.io", "score": 80, "verification": {"status": "valid"}}, "emails": [{"email": "e", "type": "work"}], "none": []}
    assert P.evaluate("data.email", doc) == "a@x.io"
    assert P.evaluate("data.score / 100", doc) == 0.8
    assert P.evaluate("data.verification.status == 'valid'", doc) is True
    assert P.evaluate("data.email == null", doc) is False and P.evaluate("data.missing == null", doc) is True
    assert P.evaluate("none == []", doc) is True and P.evaluate("emails == []", doc) is False
    assert P.evaluate("emails[0].email", doc) == "e" and P.evaluate("emails[3].email", doc) is None
    assert P.evaluate("coalesce(data.missing, data.email)", doc) == "a@x.io"
    assert P.evaluate("split_first(data.name)", {"data": {"name": "Patrick Collison"}}) == "Patrick"
    assert P.evaluate("split_last(data.name)", {"data": {"name": "Patrick"}}) is None
    assert P.evaluate("join(a, b)", {"a": "Patrick", "b": "Collison"}) == "Patrick Collison"
    with pytest.raises(ValueError):
        P.evaluate("nope(a)", doc)


def test_admission_only_contract_verifies_adapters_but_generates_no_routed_row():
    """`routed: false` (contracts.yaml): the influencers.club raw/profile/full tiers are one provider
    at three prices, so their contract exists for cache result admission only — the adapters must
    verify (that is what `has_result_rules` reads), and no `treg.creators.profile` row may appear."""
    cat = catalog_store.load()
    for cap in ("creators.profile", "creators.analytics", "creators.enrich.by_email"):
        assert cat.contracts[cap].routed is False
        assert "treg." + cap not in cat.by_id
    assert cat.contracts["people.email.find"].routed is True
    tiers = ["influencersclub.creators.enrich." + t for t in ("raw", "profile", "full", "analytics", "email")]
    assert all(cat.adapters[eid].verified and not cat.adapters[eid].verify_note for eid in tiers)
    # ≥ 2 verified children of one capability would have generated a row on a routed contract
    assert len([e for e in cat.for_capability("creators.profile") if cat.adapters[e["id"]].verified]) >= 2


def test_every_shipped_adapter_round_trips_its_fixture():
    cat = catalog_store.load()
    bad = {eid: a.verify_note for eid, a in cat.adapters.items() if not a.verified}
    assert bad == {}, bad
    ep = cat.by_id[ROUTED]
    assert ep["kind"] == "routed" and ep["provider"] == "treg" and len(ep["routed_children"]) >= 8
    assert cat.platform_eligible(ep) and ep["cost_range_usd"][0] < ep["cost_range_usd"][1]
    # a hand-verified round trip on the plan's worked example
    ad = cat.adapters["leadsforge.people.email.find"]
    q, b = ad.to_upstream({"first_name": "Patrick", "last_name": "Collison", "domain": "stripe.com", "full_name": "Patrick Collison"})
    assert b == {"firstName": "Patrick", "lastName": "Collison", "companyDomain": "stripe.com"} and q == {}
    assert ad.from_upstream({"email": "p@stripe.com", "status": "succeeded"}) == {"email": "p@stripe.com"}
    assert ad.is_miss({"email": None}) and not ad.is_miss({"email": "x"})


def test_openmart_tools_are_direct_only_not_routed():
    cat = catalog_store.load()
    assert "openmart.companies.search" not in cat.adapters
    assert "openmart.companies.search" not in cat.by_id["treg.companies.search"]["routed_children"]
    assert cat.platform_eligible(cat.by_id["openmart.companies.search"])
    assert "openmart.companies.enrich" not in cat.by_id["treg.companies.enrich"]["routed_children"]


def test_tavily_routes_synchronous_web_tools_and_keeps_crawl_direct():
    cat = catalog_store.load()
    routed = {
        "tavily.web.search": "treg.web.search",
        "tavily.web.extract": "treg.web.extract",
        "tavily.web.map": "treg.web.map",
    }
    for child, parent in routed.items():
        assert cat.adapters[child].verified
        assert child in cat.by_id[parent]["routed_children"]
    assert "tavily.web.crawl" not in cat.adapters
    assert "treg.web.crawl" not in cat.by_id or (
        "tavily.web.crawl" not in cat.by_id["treg.web.crawl"]["routed_children"]
    )
    assert cat.platform_eligible(cat.by_id["tavily.web.crawl"])


def test_serper_routes_search_and_single_page_extract_only():
    cat = catalog_store.load()
    routed = {
        "serper.web.search": "treg.web.search",
        "serper.web.extract": "treg.web.extract",
    }
    for child, parent in routed.items():
        assert cat.adapters[child].verified
        assert child in cat.by_id[parent]["routed_children"]
        assert cat.platform_eligible(cat.by_id[child])
    direct = set(ep["id"] for ep in cat.for_provider("serper")) - set(routed)
    assert not direct & set(cat.adapters)


def test_fetchin_linkedin_adapters_are_verified_and_routed():
    cat = catalog_store.load()
    routed = {
        "fetchinio.linkedin.user.profile": "treg.linkedin.user.profile",
        "fetchinio.linkedin.company.profile": "treg.linkedin.company.profile",
        "fetchinio.linkedin.user.posts": "treg.linkedin.user.posts",
        "fetchinio.linkedin.post.comments": "treg.linkedin.post.comments",
        "fetchinio.linkedin.post.reactions": "treg.linkedin.post.reactions",
    }
    for child, parent in routed.items():
        assert cat.adapters[child].verified
        assert not cat.adapters[child].verify_note
        assert child in cat.by_id[parent]["routed_children"]

    # The new posts route is genuinely comparative, not a synthetic one-provider wrapper.
    assert {
        "aviato.linkedin.user.posts",
        "fetchinio.linkedin.user.posts",
        "harvestapi.linkedin.user.posts",
    } <= set(cat.by_id["treg.linkedin.user.posts"]["routed_children"])

    # Fetchin has no provider-neutral contracts for these provider-native operations.
    assert "fetchinio.linkedin.user.reactions" not in cat.adapters
    assert "fetchinio.linkedin.post.engagement" not in cat.adapters

    profile = cat.adapters["fetchinio.linkedin.user.profile"]
    assert profile.is_miss({"id": "profile-id", "firstName": None, "lastName": None})
    assert not profile.is_miss({"id": "profile-id", "firstName": "Ada", "lastName": None})


async def test_fetchin_member_posts_route_uses_adapter_and_settles(
    clients: AsyncClient, platform_on, monkeypatch,
):
    monkeypatch.setenv("TREG_PLATFORM_KEY_FETCHINIO", "PLATFORM-FETCHINIO")
    monkeypatch.setenv("TREG_PLATFORM_PROVIDERS", "fetchinio")
    get_settings.cache_clear()
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider({
        "fetchin.io": [(200, {
            "posts": [{"id": "urn:li:activity:1", "text": "hello"}],
            "paginationToken": "next-page",
            "hasMore": True,
        })],
    }, seen))

    before = await _balance(clients)
    response = await clients.post(
        "/call/treg.linkedin.user.posts",
        json={"linkedin_handle": "satyanadella", "limit": 1},
        headers={"X-Treg-Route-Prefer": "fetchinio"},
    )
    assert response.status_code == 200, response.text
    data = response.json()
    assert data["_treg"]["served_by"] == "fetchinio.linkedin.user.posts"
    assert data["output"]["posts"] == [{"id": "urn:li:activity:1", "text": "hello"}]
    assert data["output"]["next_cursor"] == "next-page"
    assert data["output"]["has_more"] is True
    assert seen == [("fetchin.io", "GET", {
        "profileUrlOrUrn": "https://www.linkedin.com/in/satyanadella", "count": "1",
    }, None)]
    assert before - await _balance(clients) == 1_500
    get_settings.cache_clear()


async def test_tavily_routed_empty_search_is_a_paid_miss_then_falls_through(
    clients, monkeypatch,
):
    monkeypatch.setenv("TREG_PLATFORM_KEY_TAVILY", "PLATFORM-TAVILY")
    monkeypatch.setenv("TREG_PLATFORM_KEY_EXA", "PLATFORM-EXA")
    monkeypatch.setenv("TREG_PLATFORM_PROVIDERS", "tavily,exa")
    get_settings.cache_clear()
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider({
        "tavily": [(200, {"results": [], "usage": {"credits": 1}})],
        "exa": [(200, {
            "results": [{"title": "Example", "url": "https://example.com"}],
            "costDollars": {"total": 0.007},
        })],
    }, seen))

    before = await _balance(clients)
    response = await clients.post(
        "/call/treg.web.search",
        json={"q": "example query", "limit": 3},
        headers={"X-Treg-Route-Prefer": "tavily,exa"},
    )
    assert response.status_code == 200, response.text
    data = response.json()
    assert data["_treg"]["served_by"] == "exa.web.search"
    assert [attempt["outcome"] for attempt in data["_treg"]["tried"]] == ["miss", "hit"]
    assert [attempt["charged_micro"] for attempt in data["_treg"]["tried"]] == [8_000, 7_000]
    assert data["_treg"]["charged_micro"] == 15_000
    assert before - await _balance(clients) == 15_000
    assert [row[0] for row in seen] == ["tavily", "exa"]
    assert seen[0][3] == {
        "query": "example query", "max_results": 3,
        "search_depth": "basic", "include_usage": True,
    }
    async with session_maker() as db:
        assert (await db.execute(select(Hold))).scalars().all() == []
        entries = (await db.execute(select(LedgerEntry))).scalars().all()
    call_id = response.headers["X-Treg-Call-Id"]
    assert {entry.call_id for entry in entries if entry.kind == "settle"} == {
        call_id + ":r0", call_id + ":r1",
    }
    get_settings.cache_clear()


def test_dropleads_routing_surface_contains_only_verified_single_record_tools():
    catalog = catalog_store.load()
    expected = {
        "dropleads.people.email.find",
        "dropleads.people.phone.find",
        "dropleads.people.email.verify",
        "dropleads.people.search",
        "dropleads.people.enrich",
        "dropleads.companies.search",
        "dropleads.companies.enrich",
    }
    assert {eid for eid in expected if catalog.adapters[eid].verified} == expected
    assert not any("bulk" in eid or eid.endswith(".count") for eid in expected)


def test_dropleads_country_filters_use_each_upstream_schema():
    catalog = catalog_store.load()
    _, people_body = catalog.adapters["dropleads.people.search"].to_upstream({
        "company_domain": "example.com", "country": "US",
    })
    _, company_body = catalog.adapters["dropleads.companies.search"].to_upstream({
        "domain": "example.com", "country": "US",
    })
    assert people_body["filters"]["countries"] == ["United States"]
    assert company_body["filters"]["countries"] == {"include": ["United States"]}


def test_prospeo_routing_surface_uses_fixed_single_record_modes():
    catalog = catalog_store.load()
    expected = {
        "prospeo.people.email.find",
        "prospeo.people.phone.find",
        "prospeo.people.enrich",
        "prospeo.people.search",
        "prospeo.companies.enrich",
        "prospeo.companies.search",
    }
    assert {eid for eid in expected if catalog.adapters[eid].verified} == expected
    assert not any("bulk" in eid or "suggestions" in eid for eid in expected)
    _, email_body = catalog.adapters["prospeo.people.email.find"].to_upstream({
        "full_name": "Jane Doe", "domain": "example.com",
    })
    assert email_body == {
        "data": {"full_name": "Jane Doe", "company_website": "example.com"},
        "only_verified_email": True,
        "enrich_mobile": False,
        "only_verified_mobile": False,
    }
    _, phone_body = catalog.adapters["prospeo.people.phone.find"].to_upstream({
        "linkedin_url": "https://www.linkedin.com/in/example",
    })
    assert phone_body["enrich_mobile"] is True
    assert phone_body["only_verified_mobile"] is True


def test_aiark_routing_surface_uses_verified_bounded_adapters():
    catalog = catalog_store.load()
    expected = {
        "aiark.people.search",
        "aiark.companies.search",
        "aiark.people.email.find",
        "aiark.people.phone.find",
        "aiark.people.enrich",
    }
    assert {eid for eid in expected if catalog.adapters[eid].verified} == expected
    _, people = catalog.adapters["aiark.people.search"].to_upstream({
        "company_domain": "example.com",
    })
    _, companies = catalog.adapters["aiark.companies.search"].to_upstream({
        "domain": "example.com",
    })
    bounded = {
        "account": {"domain": {"any": {"include": ["example.com"]}}},
        "page": 0,
        "size": 1,
    }
    assert people == bounded
    assert companies == bounded


def test_aiark_finders_treat_present_but_empty_outputs_as_misses():
    catalog = catalog_store.load()
    email = catalog.adapters["aiark.people.email.find"]
    phone = catalog.adapters["aiark.people.phone.find"]
    assert email.is_miss({"data": None})
    assert email.is_miss({"data": {"email": {"output": []}}})
    assert not email.is_miss({"data": {"email": {"output": [{
        "address": "jane@example.com",
    }]}}})
    assert phone.is_miss({"data": None})
    assert phone.is_miss({"data": {"data": [[]]}})
    assert not phone.is_miss({"data": {"data": [["+15550101000"]]}})


def test_limadata_routing_surface_contains_only_its_verified_adapters():
    catalog = catalog_store.load()
    expected = {
        "limadata.people.email.find.name",
        "limadata.people.email.find.linkedin",
        "limadata.people.email.verify",
        "limadata.people.phone.find",
        "limadata.companies.enrich",
    }
    assert {
        eid for eid, adapter in catalog.adapters.items()
        if eid.startswith("limadata.") and adapter.verified
    } == expected
    assert "limadata.people.email.find.name" in catalog.by_id[
        "treg.people.email.find"
    ]["routed_children"]
    assert "limadata.people.email.verify" in catalog.by_id[
        "treg.people.email.verify"
    ]["routed_children"]
    assert "limadata.people.phone.find" in catalog.by_id[
        "treg.people.phone.find"
    ]["routed_children"]
    assert "limadata.people.enrich" not in catalog.adapters
    assert "limadata.people.enrich" not in catalog.by_id[
        "treg.people.enrich"
    ]["routed_children"]


def test_wiza_routing_surface_uses_bounded_single_record_searches():
    catalog = catalog_store.load()
    expected = {
        "wiza.people.search",
        "wiza.companies.search",
        "wiza.companies.enrich",
    }
    assert {eid for eid in expected if catalog.adapters[eid].verified} == expected
    _, people_body = catalog.adapters["wiza.people.search"].to_upstream({"title": "Founder"})
    _, company_body = catalog.adapters["wiza.companies.search"].to_upstream({
        "technology": "amazon-web-services",
    })
    assert people_body == {"filters": {"job_title": [{"v": "Founder", "s": "i"}]}, "size": 1}
    assert company_body == {
        "filters": {"technologies": [{"v": "amazon-web-services", "s": "i"}]},
        "size": 1,
    }


def test_identity_variants_derive_and_never_cross():
    contract = catalog_store.load().contracts["people.email.find"]
    ident, variant = canonical_identity(contract, {"full_name": "Patrick Collison", "domain": "stripe.com"})
    assert variant == ("domain", "full_name") and ident["first_name"] == "Patrick" and ident["last_name"] == "Collison"
    ident, variant = canonical_identity(contract, {"first_name": "Patrick", "last_name": "Collison", "domain": "stripe.com"})
    assert ident["full_name"] == "Patrick Collison"
    ident, variant = canonical_identity(contract, {"linkedin_url": "https://www.linkedin.com/in/x"})
    assert variant == ("linkedin_url",) and "domain" not in ident
    assert canonical_identity(contract, {"full_name": "Patrick Collison"})[1] is None


def test_cost_at_and_ranking_math():
    assert cost_at({"usd": 0.0038, "type": "per_result", "per": 1}, {"limit": 10}) == 38_000
    assert cost_at({"usd": 0.0044, "type": "per_result", "per": 25}, {"limit": 10}) == 110_000, "lusha: 1 credit per 25 rows, minimum 1"
    assert cost_at({"usd": 0.0044, "type": "per_result", "per": 25}, {"limit": 40}) == 220_000
    assert cost_at({"usd": 0.005, "type": "per_call"}, {"limit": 10}) == 5_000
    assert cost_at({"usd": None}, {}) is None
    ep = lambda i, t="per_success": {"id": i, "provider": i.split(".")[0], "cost": {"type": t}}
    a = Candidate(ep("a.x"), None, ("domain",), "platform", 24_500, hit_rate=0.4, ok_rate=None, p50_ms=100, last_ok_days=1)
    b = Candidate(ep("b.x", "per_call"), None, ("domain",), "platform", 20_000, hit_rate=0.8, ok_rate=None, p50_ms=100, last_ok_days=1)
    own = Candidate(ep("c.x"), None, ("domain",), "credential", 0, hit_rate=None, ok_rate=None, p50_ms=None, last_ok_days=None)
    anonymous = Candidate(ep("d.x"), None, ("domain",), "anonymous", 0, hit_rate=None, ok_rate=None, p50_ms=None, last_ok_days=None)
    assert a.expected_cost_per_hit == pytest.approx(24_500), "per-success: billed only on a hit → price per hit"
    assert b.expected_cost_per_hit == pytest.approx(25_000), "per-call at 80% hit rate: 20000/0.8"
    assert [c.endpoint["id"] for c in rank([a, b, anonymous, own])] == ["c.x", "d.x", "a.x", "b.x"]
    assert [c.endpoint["id"] for c in rank([a, b], prefer=["b"])] == ["b.x", "a.x"]
    assert [c.endpoint["id"] for c in rank([a, b], exclude=["a"])] == ["b.x"]
    a.exhausted = True
    assert [c.endpoint["id"] for c in rank([a, b])] == ["b.x"]


async def test_concurrent_routed_plans_share_one_cached_observation_refresh(
    clients: AsyncClient, enrichment_on, monkeypatch,
):
    from treg.domain.catalog import stats

    class Source:
        calls = 0

        async def get_many(self, endpoint_ids):
            self.calls += 1
            return {
                endpoint_id: {
                    "samples": 20, "ok_rate": 1.0, "p50_ms": 20, "p95_ms": 40,
                    "last_ok_days": 0, "hit_rate": 0.5, "hit_samples": 20,
                }
                for endpoint_id in endpoint_ids
            }

    async def request_time_aggregate(*args, **kwargs):
        raise AssertionError("routed planning must not aggregate CallRecord on the request path")

    source = Source()
    reader = CachedEndpointObservationReader(source)
    monkeypatch.setattr(call_route, "_endpoint_observation_reader", reader, raising=False)
    monkeypatch.setattr(stats, "observed", request_time_aggregate)
    ep = catalog_store.load().by_id[ROUTED]

    class _Org:
        id = 1

    class _Caller:
        org_id = 1
        org = _Org()

    try:
        plans = await asyncio.gather(*(
            call_route.build_plan(
                ep, {"full_name": "Patrick Collison", "domain": "stripe.com"},
                _Caller(), call_route.RouteOptions.from_headers(lambda key: None),
            )
            for _ in range(20)
        ))
        await reader.wait_for_idle()
    finally:
        await reader.aclose()

    assert all(plan.candidates for plan in plans)
    assert source.calls == 1


async def test_routed_plan_keeps_per_success_hit_fallback_from_the_cache(
    clients: AsyncClient, enrichment_on, monkeypatch,
):
    from treg import api as A
    from treg.domain.catalog import stats

    endpoint_id = "tomba.people.email.find"
    async with session_maker() as db:
        for cost_micro in (8_900, 8_900, 0):
            db.add(CallRecord(
                org_id=1, user_email="a@b.c", tool_name=endpoint_id, method="GET", path="/x",
                status_code=200, endpoint_id=endpoint_id, cost_observed_micro=cost_micro, hit=None,
            ))
        await db.commit()
    monkeypatch.setattr(stats, "MIN_HIT_SAMPLES", 3)
    reader = A.app.state.endpoint_observation_reader
    assert await reader.get_many([endpoint_id]) == {}
    await reader.wait_for_idle()
    warm = await reader.get_many([endpoint_id])
    assert warm[endpoint_id]["hit_rate"] == pytest.approx(2 / 3, abs=1e-3)

    async def request_time_aggregate(*args, **kwargs):
        raise AssertionError("the routed plan must use the warm observation cache")

    monkeypatch.setattr(stats, "observed", request_time_aggregate)
    monkeypatch.setattr(call_route, "_endpoint_observation_reader", reader, raising=False)
    ep = catalog_store.load().by_id[ROUTED]

    class _Org:
        id = 1

    class _Caller:
        org_id = 1
        org = _Org()

    plan = await call_route.build_plan(
        ep, {"full_name": "Patrick Collison", "domain": "stripe.com"},
        _Caller(), call_route.RouteOptions.from_headers(lambda key: None),
    )
    tomba = next(candidate for candidate in plan.candidates if candidate.endpoint["id"] == endpoint_id)
    assert tomba.hit_rate == pytest.approx(2 / 3, abs=1e-3)


# ---- the call path ---------------------------------------------------------------------------

@pytest.mark.parametrize("routed", [False, True])
@pytest.mark.parametrize("verdict", ["valid", "invalid"])
async def test_tomba_verification_keeps_email_in_query_and_settles(
    clients: AsyncClient, enrichment_on, monkeypatch, routed, verdict,
):
    email = "person+tag@example.com"
    payload = {"data": {"email": {"status": verdict, "score": 99}}}
    seen = []

    async def relay(request, upstream_url, tool, secrets, client, drop_params=None, **kwargs):
        seen.append(upstream_url)
        assert upstream_url == "https://api.tomba.io/v1/email-verifier"
        assert request.method == "GET"
        assert dict(request.query_items) == {"email": email}
        assert "email" not in (drop_params or ())

        async def body():
            yield json.dumps(payload).encode()

        async def close():
            pass

        return UpstreamResponse(200, ((b"content-type", b"application/json"),), body(), close)

    monkeypatch.setattr(call_service, "relay", relay)
    before = await _balance(clients)
    if routed:
        response = await clients.post(
            "/call/treg.people.email.verify", json={"email": email},
            headers={"X-Treg-Route-Prefer": "tomba"},
        )
    else:
        response = await clients.get("/call/tomba.people.email.verify", params={"email": email})

    assert response.status_code == 200, response.text
    assert len(seen) == 1
    if routed:
        doc = response.json()
        assert doc["raw"] == payload
        assert doc["output"] == {"valid": verdict == "valid", "status": verdict, "score": 99}
        assert doc["_treg"]["served_by"] == "tomba.people.email.verify"
        assert doc["_treg"]["outcome"] == "hit", "an invalid verdict is still a verification answer"
    else:
        assert response.json() == payload
    assert int(response.headers["X-Treg-Cost-Micro"]) == 8_900
    assert before - await _balance(clients) == 8_900
    async with session_maker() as db:
        assert (await db.execute(select(Hold))).scalars().all() == []


async def test_routed_call_runs_the_cheapest_child_and_returns_output_raw_and_provenance(clients: AsyncClient, enrichment_on, monkeypatch):
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider(
        {"tomba": [(200, {"data": {"email": "patrick@stripe.com", "score": 99, "first_name": "Patrick", "last_name": "Collison",
                                    "verification": {"status": "valid"}}})]}, seen))
    before = await _balance(clients)
    r = await clients.post(f"/call/{ROUTED}", json={"full_name": "Patrick Collison", "domain": "stripe.com"})
    assert r.status_code == 200, r.text
    d = r.json()
    assert d["output"] == {"email": "patrick@stripe.com", "confidence": 0.99, "first_name": "Patrick", "last_name": "Collison", "verified": True}
    assert d["raw"]["data"]["score"] == 99, "the winning provider's body, verbatim"
    assert d["_treg"]["served_by"] == "tomba.people.email.find" and d["_treg"]["outcome"] == "hit"
    assert "advice" not in d["_treg"], "the provider vouched for the mailbox — nothing to add"
    assert r.headers["X-Treg-Served-By"] == "tomba.people.email.find" and r.headers["X-Treg-Providers-Tried"] == "tomba"
    assert seen == [("tomba", "GET", {"domain": "stripe.com", "full_name": "Patrick Collison"}, None)]
    charged = int(r.headers["X-Treg-Cost-Micro"])
    assert charged == 8_900 and before - await _balance(clients) == charged, "tomba's price, nothing else"
    assert d["_treg"]["charged_micro"] == charged
    async with session_maker() as db:
        assert (await db.execute(select(Hold))).scalars().all() == []
        entries = (await db.execute(select(LedgerEntry))).scalars().all()
    assert {e.call_id for e in entries if e.kind == "settle"} == {r.headers["X-Treg-Call-Id"] + ":r0"}
    await audit.drain()
    rows = (await clients.get("/calls")).json()
    kinds = {(x["tool_name"], x.get("credential_tier")) for x in rows}
    assert (ROUTED, "routed") in kinds and ("tomba.people.email.find", "platform") in kinds


async def test_an_unverified_hit_carries_verify_advice(clients: AsyncClient, enrichment_on, monkeypatch):
    """A found address the provider did not vouch for (Tomba's verification status is not `valid`
    — the catch-all shape that bounced for a recruiting team on 2026-09-06) is still a HIT and
    still billed, but the answer says so in `_treg.advice` and points at the verify endpoint. A
    suggestion, not a chained call: the balance moves by the find alone."""
    monkeypatch.setattr(call_service, "relay", _relay_by_provider(
        {"tomba": [(200, {"data": {"email": "alan@pruittstructures.com", "score": 96,
                                    "verification": {"status": "accept_all"}}})]}, []))
    before = await _balance(clients)
    r = await clients.post(f"/call/{ROUTED}", json={"full_name": "Alan Marquez", "domain": "pruittstructures.com"})
    assert r.status_code == 200, r.text
    d = r.json()
    assert d["output"]["email"] == "alan@pruittstructures.com" and d["output"]["verified"] is False
    assert d["_treg"]["outcome"] == "hit"
    assert "treg.people.email.verify" in d["_treg"]["advice"]
    assert before - await _balance(clients) == int(r.headers["X-Treg-Cost-Micro"]) == 8_900, "the find, nothing chained"


async def test_a_people_search_hit_always_carries_verify_advice(clients: AsyncClient, enrichment_on, monkeypatch):
    """Search rows are directory listings: a row's email is found, not confirmed deliverable. The
    contract has no `verified` output, so the advice attaches to every hit — Hunter domain-search
    rows with `verification: null` were 73 of one team's 79 bounces (2026-09-08). Still a
    suggestion: one child call, the find's price, nothing chained."""
    monkeypatch.setattr(call_service, "relay", _relay_by_provider(
        {"hunter": [(200, {"data": {"emails": [{"value": "info@royalfarms.com", "type": "generic", "confidence": 10,
                                                 "verification": {"date": None, "status": None}}]},
                            "meta": {"results": 1}})],
         "*": [(200, {"persons": []})] * 12}, []))
    before = await _balance(clients)
    r = await clients.post("/call/treg.people.search", json={"company_domain": "royalfarms.com", "limit": 10})
    assert r.status_code == 200, r.text
    d = r.json()
    assert d["_treg"]["served_by"] == "hunter.companies.emails" and d["_treg"]["outcome"] == "hit"
    assert d["output"]["people"][0]["verification"]["status"] is None, "the row's own field, untouched"
    assert "treg.people.email.verify" in d["_treg"]["advice"] and "directory" in d["_treg"]["advice"]
    assert before - await _balance(clients) == int(r.headers["X-Treg-Cost-Micro"]), "the find, nothing chained"


async def test_error_on_the_first_child_falls_back_to_the_second(clients: AsyncClient, enrichment_on, monkeypatch):
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider(
        {"tomba": [(503, {"error": "down"})],
         "findymail": [(200, {"contact": {"name": "Patrick Collison", "email": "patrick@stripe.com"}})]}, seen))
    before = await _balance(clients)
    r = await clients.post(f"/call/{ROUTED}", json={"full_name": "Patrick Collison", "domain": "stripe.com"})
    assert r.status_code == 200, r.text
    d = r.json()
    assert [t["outcome"] for t in d["_treg"]["tried"]] == ["error", "hit"]
    assert d["_treg"]["served_by"] == "findymail.search.name" and d["output"]["email"] == "patrick@stripe.com"
    assert r.headers["X-Treg-Providers-Tried"] == "tomba,findymail"
    assert before - await _balance(clients) == 19_800, "the failed child released its hold; only findymail charged"
    assert seen[1] == ("findymail", "POST", {}, {"name": "Patrick Collison", "domain": "stripe.com"})


async def test_child_capability_pin_refusal_falls_back_to_the_pinned_provider(
    clients: AsyncClient, enrichment_on, monkeypatch,
):
    org_id = (await clients.get("/orgs")).json()[0]["org_id"]
    pinned = await clients.post(
        f"/orgs/{org_id}/pins",
        json={"capability": "people.email.find", "provider": "hunter"},
    )
    assert pinned.status_code == 200, pinned.text
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider({
        "hunter": [(200, {"data": {
            "email": "patrick@stripe.com", "score": 90,
            "verification": {"status": "valid"},
        }})],
    }, seen))

    response = await clients.post(
        f"/call/{ROUTED}",
        json={"full_name": "Patrick Collison", "domain": "stripe.com"},
        headers={"X-Treg-Route-Prefer": "tomba,hunter"},
    )

    assert response.status_code == 200, response.text
    doc = response.json()
    assert doc["_treg"]["served_by"] == "hunter.people.email.find"
    assert [attempt["outcome"] for attempt in doc["_treg"]["tried"]] == ["error", "hit"]
    assert doc["_treg"]["tried"][0]["endpoint_id"] == "tomba.people.email.find"
    assert [provider for provider, *_ in seen] == ["hunter"]


async def test_platform_vendor_401_falls_back_to_the_next_provider(
    clients: AsyncClient, enrichment_on, monkeypatch,
):
    findymail = catalog_store.load().by_id["findymail.search.name"]
    monkeypatch.setitem(findymail, "cost", {**findymail["cost"], "type": "per_call"})
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider({
        "tomba": [(401, {"error": "invalid platform key"})],
        "findymail": [(200, {"contact": {
            "name": "Patrick Collison", "email": "patrick@stripe.com",
        }})],
    }, seen))

    response = await clients.post(
        f"/call/{ROUTED}",
        json={"full_name": "Patrick Collison", "domain": "stripe.com"},
        headers={
            "X-Treg-Route-Prefer": "tomba,findymail",
            "X-Treg-Route-Exclude": "hunter,leadmagic,leadsforge,aviato,fiber-ai",
        },
    )

    assert response.status_code == 200, response.text
    doc = response.json()
    assert doc["_treg"]["served_by"] == "findymail.search.name"
    assert [attempt["outcome"] for attempt in doc["_treg"]["tried"]] == ["error", "hit"]
    assert [provider for provider, *_ in seen] == ["tomba", "findymail"]


async def test_routed_insufficient_balance_still_stops_before_fallback(
    clients: AsyncClient, enrichment_on, monkeypatch,
):
    org_id = (await clients.get("/orgs")).json()[0]["org_id"]
    async with session_maker() as db:
        await ledger.reserve(db, org_id, "drain routed balance", 1_000_000)
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider({
        "tomba": [(200, {"data": {"email": "should-not-run@example.com"}})],
    }, seen))

    response = await clients.post(
        f"/call/{ROUTED}",
        json={"full_name": "Patrick Collison", "domain": "stripe.com"},
        headers={"X-Treg-Route-Prefer": "tomba,findymail"},
    )

    assert response.status_code == 402
    assert response.json()["detail"]["error"] == "insufficient_balance"
    assert seen == []


async def test_waterfall_is_on_by_default_can_be_turned_off_and_respects_max_cost(clients: AsyncClient, enrichment_on, monkeypatch):
    miss_tomba = (200, {"data": {"email": None, "score": None, "verification": {"status": None}}})
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider({"tomba": [miss_tomba]}, seen))
    r = await clients.post(f"/call/{ROUTED}", json={"full_name": "Nobody Here", "domain": "stripe.com"},
                           headers={"X-Treg-Route-Waterfall": "0"})
    assert r.status_code == 200 and r.json()["_treg"]["outcome"] == "miss" and r.json()["output"]["email"] is None
    assert r.headers["X-Treg-Route-Outcome"] == "miss" and len(seen) == 1, "waterfall off: stop at the first miss"
    # waterfall (the default): miss → next cheapest → hit; skips a candidate that would breach the ceiling
    seen.clear()
    monkeypatch.setattr(call_service, "relay", _relay_by_provider(
        {"tomba": [miss_tomba], "findymail": [(200, {"contact": {"name": "N H", "email": None}})],
         "hunter": [(200, {"data": {"email": "n@stripe.com", "score": 50, "verification": {"status": "valid"}}})]}, seen))
    r = await clients.post(f"/call/{ROUTED}", json={"full_name": "Nobody Here", "domain": "stripe.com"},
                           headers={"X-Treg-Route-Max-Cost": "0.08"})
    assert r.status_code == 200, r.text
    tried = r.json()["_treg"]["tried"]
    assert [t["outcome"] for t in tried] == ["miss", "miss", "hit"] and r.json()["_treg"]["served_by"] == "hunter.people.email.find"
    assert [p for p, *_ in seen] == ["tomba", "findymail", "hunter"]
    assert r.json()["_treg"]["charged_micro"] == 24_500, "misses on per-success providers are free; only the hit is billed"
    assert [t["charged_micro"] for t in tried] == [0, 0, 24_500]
    # a ceiling the third candidate would breach stops the waterfall there
    seen.clear()
    monkeypatch.setattr(call_service, "relay", _relay_by_provider(
        {"tomba": [miss_tomba], "findymail": [(200, {"contact": {"name": "N H", "email": None}})],
         "hunter": [(200, {"data": {"email": None, "score": None}})]}, seen))
    r = await clients.post(f"/call/{ROUTED}", json={"full_name": "Nobody Here", "domain": "stripe.com"},
                           headers={"X-Treg-Route-Max-Cost": "0.02"})
    assert r.status_code == 200 and r.json()["_treg"]["outcome"] == "miss"
    # free misses do not consume the ceiling, but hunter (2.45¢ > 2¢) and everything dearer is skipped
    assert [p for p, *_ in seen] == ["tomba", "findymail"]
    assert all(t["outcome"] in ("miss", "skipped") for t in r.json()["_treg"]["tried"]) and r.json()["_treg"]["charged_micro"] == 0


async def test_max_cost_below_the_cheapest_refuses_before_any_call(clients: AsyncClient, enrichment_on, monkeypatch):
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider({"tomba": [(200, {})]}, seen))
    r = await clients.post(f"/call/{ROUTED}", json={"full_name": "P C", "domain": "stripe.com"}, headers={"X-Treg-Route-Max-Cost": "0.001"})
    assert r.status_code == 402 and r.json()["detail"]["error"] == "route_max_cost" and seen == []
    async with session_maker() as db:
        assert (await db.execute(select(Hold))).scalars().all() == []


async def test_identity_no_provider_accepts_is_422_naming_variants(clients: AsyncClient, enrichment_on):
    r = await clients.post(f"/call/{ROUTED}", json={"full_name": "Patrick Collison"})
    assert r.status_code == 422 and r.json()["detail"]["error"] == "identity_incomplete"
    assert ["domain", "full_name"] in r.json()["detail"]["variants"]


async def test_caller_fault_on_a_child_stops_and_own_key_ranks_first(clients: AsyncClient, enrichment_on, monkeypatch):
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider(
        {"hunter": [(200, {"data": {"email": "p@stripe.com", "score": 90, "verification": {"status": "valid"}}})]}, seen))
    await clients.post("/secrets", json={"name": "hunter", "value": "MY-HUNTER-KEY"})  # tier 2 for hunter
    before = await _balance(clients)
    r = await clients.post(f"/call/{ROUTED}", json={"full_name": "Patrick Collison", "domain": "stripe.com"})
    assert r.status_code == 200 and r.json()["_treg"]["served_by"] == "hunter.people.email.find"
    assert r.json()["_treg"]["tier"] == "credential" and await _balance(clients) == before, "own key: first, and free"
    # a vendor 4xx on the child goes on ONLY to providers that bill nothing for a rejected request
    # (per_success / free): tomba is per_success, so it is asked; when it rejects too, the caller gets
    # route_caller_fault naming both — and no paid-per-call provider was ever asked.
    seen.clear()
    monkeypatch.setattr(call_service, "relay", _relay_by_provider(
        {"hunter": [(400, {"errors": [{"details": "bad"}]})], "tomba": [(400, {"error": "bad"})], "*": [(400, {"error": "bad"})]}, seen))
    r = await clients.post(f"/call/{ROUTED}", json={"full_name": "Patrick Collison", "domain": "stripe.com"})
    assert r.status_code == 400 and r.json()["detail"]["error"] == "route_caller_fault", r.text
    assert [p for p, *_ in seen][:2] == ["hunter", "tomba"] and len(seen) == 3, "at most two fallbacks, then the 4xx is the caller's"
    outcomes = {t["endpoint_id"]: t["outcome"] for t in r.json()["detail"]["tried"]}
    assert outcomes["hunter.people.email.find"] == "error" and outcomes["tomba.people.email.find"] == "error"
    # a scraper's "please retry" 400 (tikhub, live 2026-08-28) is why: the next free-on-failure provider answers
    seen.clear()
    monkeypatch.setattr(call_service, "relay", _relay_by_provider(
        {"hunter": [(400, {"errors": [{"details": "bad"}]})],
         "tomba": [(200, {"data": {"email": "p@stripe.com", "score": 90, "verification": {"status": "valid"}}})]}, seen))
    r = await clients.post(f"/call/{ROUTED}", json={"full_name": "Patrick Collison", "domain": "stripe.com"})
    assert r.status_code == 200 and r.json()["_treg"]["served_by"] == "tomba.people.email.find", r.text


async def test_a_2xx_without_the_required_core_is_a_miss_not_a_hit(clients: AsyncClient, enrichment_on, monkeypatch):
    """A 200 whose body lacks the contract's required field (a null result under a success envelope)
    is a MISS: the waterfall goes on, and the verdict/hit-rate never counts it as answered."""
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider(
        {"hunter": [(200, {"data": {"email": None, "score": None}})],
         "tomba": [(200, {"data": {"email": "p@stripe.com", "score": 90, "verification": {"status": "valid"}}})]}, seen))
    await clients.post("/secrets", json={"name": "hunter", "value": "MY-HUNTER-KEY"})
    r = await clients.post(f"/call/{ROUTED}", json={"full_name": "Patrick Collison", "domain": "stripe.com"})
    assert r.status_code == 200, r.text
    outcomes = {t["endpoint_id"]: t["outcome"] for t in r.json()["_treg"]["tried"]}
    assert outcomes["hunter.people.email.find"] == "miss" and r.json()["_treg"]["served_by"] == "tomba.people.email.find"


async def test_catalog_get_on_the_routed_endpoint_shows_the_plan(clients: AsyncClient, enrichment_on):
    r = await clients.get(f"/catalog/endpoints/{ROUTED}")
    assert r.status_code == 200, r.text
    d = r.json()
    assert d["endpoint"]["kind"] == "routed" and d["routing"]["contract"]["identity"]
    # the same job from unrouted providers is named here too — the search page points at this row
    also = {a["endpoint_id"] for a in d["routing"]["also"]}
    assert also.isdisjoint(d["endpoint"]["routed_children"]) and all(i.endswith("email.find") or "." in i for i in also)
    plan = d["routing"]["plan"]
    assert plan and plan[0]["usd"] <= plan[-1]["usd"] and plan[0]["accepts"]
    assert "hit_rate" not in plan[0], "unmeasured says nothing rather than nulls"
    assert {c["endpoint_id"] for c in plan} <= set(d["endpoint"]["routed_children"] if "routed_children" in d["endpoint"] else [c["endpoint_id"] for c in plan])


async def test_idempotent_replay_of_a_routed_call_never_calls_a_provider_twice(clients: AsyncClient, enrichment_on, monkeypatch):
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider(
        {"tomba": [(200, {"data": {"email": "p@stripe.com", "score": 90, "verification": {"status": "valid"}}})]}, seen))
    h = {"Idempotency-Key": "route-1"}
    r1 = await clients.post(f"/call/{ROUTED}", json={"full_name": "Patrick Collison", "domain": "stripe.com"}, headers=h)
    r2 = await clients.post(f"/call/{ROUTED}", json={"full_name": "Patrick Collison", "domain": "stripe.com"}, headers=h)
    assert r1.status_code == 200 and r2.status_code == 200 and r2.headers.get("X-Treg-Idempotent-Replay") == "true"
    assert r2.json() == r1.json() and len(seen) == 1


@pytest.mark.parametrize(
    ("terminal", "expected_status", "expected_error"),
    [
        ((400, {"message": "invalid email"}), 400, "route_caller_fault"),
        ((503, {"message": "provider down"}), 502, "route_failed"),
    ],
)
async def test_idempotent_replay_preserves_a_routed_failure_after_partial_charge(
    clients: AsyncClient, enrichment_on, monkeypatch, terminal, expected_status, expected_error,
):
    routed = "treg.people.email.verify"
    tomba_miss = (200, {"data": {"email": {"status": None, "score": None}}})
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider(
        {"tomba": [tomba_miss, tomba_miss], "leadmagic": [terminal, terminal]},
        seen,
    ))
    headers = {
        "Idempotency-Key": "route-partially-charged-failure",
        "X-Treg-Route-Prefer": "tomba,leadmagic",
        "X-Treg-Route-Exclude": "hunter",
    }
    before = await _balance(clients)

    r1 = await clients.post(f"/call/{routed}", json={"email": "bad@example.com"}, headers=headers)
    r2 = await clients.post(f"/call/{routed}", json={"email": "bad@example.com"}, headers=headers)

    assert r1.status_code == expected_status and r1.json()["detail"]["error"] == expected_error
    assert r2.status_code == r1.status_code and r2.json() == r1.json()
    assert r2.headers.get("X-Treg-Idempotent-Replay") == "true"
    assert r1.headers["X-Treg-Cost-Micro"] == r2.headers["X-Treg-Cost-Micro"] == "8900"
    assert before - await _balance(clients) == 8_900
    assert [provider for provider, *_ in seen] == ["tomba", "leadmagic"]


def test_a_per_success_miss_settles_at_zero_when_the_adapter_can_tell():
    """Live 2026-08-28: the first waterfall charged tomba, findymail and leadsforge for misses the
    catalog calls free. The adapter's `miss` predicate is the missing knowledge."""
    from test_marketplace_call import _mk
    from treg.application.call import settle as A
    miss_tomba = b'{"data": {"email": null, "score": null, "first_name": "Z", "verification": {"status": null}}}'
    assert A._observed_cost_micro(_mk("tomba", endpoint_id="tomba.people.email.find", cost_type="per_success"), miss_tomba) == 0
    hit_tomba = b'{"data": {"email": "z@x.io", "score": 90}}'
    assert A._observed_cost_micro(_mk("tomba", endpoint_id="tomba.people.email.find", cost_type="per_success"), hit_tomba) is None, "a hit still settles at the estimate"
    assert A._observed_cost_micro(_mk("findymail", endpoint_id="findymail.search.name", cost_type="per_success"), b'{"contact": {"email": null}}') == 0
    assert A._observed_cost_micro(_mk("leadsforge", endpoint_id="leadsforge.people.email.find", cost_type="per_success"), b'{"email": null, "status": "failed"}') == 0
    assert A._observed_cost_micro(_mk("leadsforge", endpoint_id="leadsforge.people.email.find", cost_type="per_call"), b'{"email": null}') is None, "per_call bills the call"
    assert A._observed_cost_micro(_mk("tomba", endpoint_id="tomba.companies.emails.count", cost_type="per_success"), b'{"data": {}}') is None, "no adapter → no opinion"


async def test_discovery_puts_the_routed_parent_first_and_its_children_under_it(clients: AsyncClient):
    r = await clients.get("/catalog/search", params={"q": "find work email"})
    rows = r.json()["results"]
    ids = [x["id"] for x in rows]
    parent = ids.index(ROUTED)
    kids = [i for i, x in enumerate(rows) if x["capability"] == "people.email.find" and x["id"] != ROUTED]
    assert kids and parent < min(kids), "the routed parent leads its capability group"
    assert kids == list(range(parent + 1, parent + 1 + len(kids))), "children sit right under the parent"
    assert rows[parent]["routed_children"] and any("ROUTED" in h for h in r.json()["hints"])
    p = await clients.get("/catalog/platforms/people")
    group = next(c for c in p.json()["capabilities"] if c["id"] == "people.email.find")
    assert group["endpoints"][0]["id"] == ROUTED
    from treg.domain.catalog.store import group_routed
    plain = [{"id": "a", "capability": "x", "kind": "data"}, {"id": "b", "capability": "y", "kind": "data"}]
    assert group_routed(plain) == plain, "no routed row → order untouched"


async def test_hit_verdict_is_recorded_and_becomes_a_hit_rate(clients: AsyncClient, enrichment_on, monkeypatch):
    from treg.domain.catalog import stats
    from treg.models import CallRecord
    hit = (200, {"data": {"email": "p@stripe.com", "score": 90, "verification": {"status": "valid"}}})
    miss = (200, {"data": {"email": None, "score": None, "verification": {"status": None}}})
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider({"tomba": [hit, miss, hit]}, seen))
    for _ in range(3):
        assert (await clients.get("/call/tomba.people.email.find?full_name=P%20C&domain=stripe.com")).status_code == 200
    await audit.drain()
    async with session_maker() as db:
        rows = (await db.execute(select(CallRecord).where(CallRecord.endpoint_id == "tomba.people.email.find"))).scalars().all()
        assert sorted(r.hit for r in rows) == [False, True, True], "the verdict, never the body"
        # below the floor → None; the floor is about evidence, not a bug
        assert (await stats.observed(db, ["tomba.people.email.find"]))["tomba.people.email.find"]["hit_rate"] is None
        monkeypatch.setattr(stats, "MIN_HIT_SAMPLES", 3)
        s = (await stats.observed(db, ["tomba.people.email.find"], per_success={"tomba.people.email.find"}))["tomba.people.email.find"]
        assert s["hit_rate"] == pytest.approx(2 / 3, abs=1e-3) and s["hit_samples"] == 3
        # historical rows without a verdict: a per-success 2xx with cost_observed 0 is a miss, > 0 a hit
        for r in rows:
            r.hit = None
            r.cost_observed_micro = 8_900 if r.status_code == 200 and "x" else 0
        rows[0].cost_observed_micro = 0
        await db.commit()
        s = (await stats.observed(db, ["tomba.people.email.find"], per_success={"tomba.people.email.find"}))["tomba.people.email.find"]
        assert s["hit_samples"] == 3 and s["hit_rate"] == pytest.approx(2 / 3, abs=1e-3)
        s = (await stats.observed(db, ["tomba.people.email.find"]))["tomba.people.email.find"]
        assert s["hit_samples"] == 0, "the zero-cost fallback applies to per-success endpoints only"
    # the plan reads it: with a measured hit rate the confidence flips from unmeasured to measured
    monkeypatch.setattr(stats, "MIN_HIT_SAMPLES", 3)
    # the catalog reads observations through the process cache: a cold entry answers nothing and
    # refreshes in the background, so warm it the way test_endpoint_stats does
    from treg import api as A
    monkeypatch.setattr(call_route, "_endpoint_observation_reader", A.app.state.endpoint_observation_reader)
    await clients.get(f"/catalog/endpoints/{ROUTED}")
    await A.app.state.endpoint_observation_reader.wait_for_idle()
    r = await clients.get(f"/catalog/endpoints/{ROUTED}")
    tomba = next(c for c in r.json()["routing"]["plan"] if c["endpoint_id"] == "tomba.people.email.find")
    assert tomba["hit_rate"] == pytest.approx(2 / 3, abs=1e-3) and tomba["usd_per_hit"] == pytest.approx(0.0089, abs=1e-4)


async def test_a_registered_tool_for_a_provider_ranks_first_and_is_free(clients: AsyncClient, enrichment_on, monkeypatch):
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider(
        {"hunter": [(200, {"data": {"email": "p@stripe.com", "score": 90, "verification": {"status": "valid"}}})]}, seen))
    sid = (await clients.post("/secrets", json={"name": "my-hunter", "value": "OWN-HUNTER"})).json()["id"]
    r = await clients.post("/tools", json={"name": "our-hunter", "base_url": "https://api.hunter.io/v2", "secret_id": sid})
    assert r.status_code == 200, r.text
    before = await _balance(clients)
    r = await clients.post(f"/call/{ROUTED}", json={"full_name": "Patrick Collison", "domain": "stripe.com"})
    assert r.status_code == 200, r.text
    assert r.json()["_treg"]["served_by"] == "hunter.people.email.find" and r.json()["_treg"]["tier"] == "tool"
    assert await _balance(clients) == before and r.json()["_treg"]["charged_micro"] == 0


async def test_mcp_search_shows_the_routed_parent_first_with_its_children(clients: AsyncClient):
    from treg import mcp as M
    out = await M._catalog_search_impl("find work email", 12, surface=M._TEAM_SURFACE) if "surface" in M._catalog_search_impl.__code__.co_varnames else await M._catalog_search_impl("find work email", 12)
    ids = [r["endpoint_id"] for r in out["results"]]
    parent = ids.index(ROUTED)
    kids = [i for i, r in enumerate(out["results"]) if r["provider"] != "treg" and r["endpoint_id"].split(".", 1)[1] in ("people.email.find", "people.email.find.linkedin", "search.name")]
    assert kids and parent < min(kids)
    assert out["results"][parent]["routed"].startswith("treg picks among")


def test_filters_reach_adapters_through_in_expr_and_array_bodies():
    cat = catalog_store.load()
    contract = cat.contracts["google.keywords.ideas"]
    req, variant = canonical_identity(contract, {"keyword": "coffee"})
    assert variant == ("keyword",) and req["country"] == "us" and req["limit"] == 20, "filter defaults ride with the identity"
    req, _ = canonical_identity(contract, {"keyword": "coffee", "country": "GB", "limit": 5})
    q, b = cat.adapters["dataforseo.google.keywords.ideas"].to_upstream(req)
    assert b == [{"keyword": "coffee", "location_code": 2826, "language_code": "en", "limit": 5}], "task list body, GB → 2826"
    q, b = cat.adapters["seranking.google.keywords.ideas"].to_upstream(req)
    assert q == {"keyword": "coffee", "source": "uk", "limit": "5"}
    q, b = cat.adapters["serpapi.google.keywords.ideas"].to_upstream(req)
    assert q == {"q": "coffee", "gl": "gb", "hl": "en", "engine": "google_autocomplete"}
    q, b = cat.adapters["tomba.people.email.verify"].to_upstream({"email": "a@b.io"})
    assert q == {"email": "a@b.io"}, "Tomba verification requires the email query parameter"
    assert cat.by_id["tomba.people.email.verify"]["path"] == "/v1/email-verifier"
    assert cost_at({"usd": 0.00179, "type": "per_result", "per": 1}, req) == 8_950, "priced at the requested limit"
    ep = cat.by_id["treg.google.keywords.ideas"]
    assert ep["input"]["body"]["country"]["note"].startswith("filter — default 'us'")


async def test_a_keyless_provider_is_dropped_at_planning_not_failed_at_call_time(clients: AsyncClient, enrichment_on, monkeypatch):
    """Live 2026-08-28: exa is platform-eligible but this deployment held no exa key; the child's
    'no credential' 404 aborted the routed call. Planning must drop it and name why."""
    from treg.application.call.route import RouteOptions, build_plan
    cat = catalog_store.load()
    ep = cat.by_id["treg.people.email.find"]
    monkeypatch.setenv("TREG_PLATFORM_KEY_AVIATO", "")  # aviato stays eligible, but keyless
    get_settings.cache_clear()
    class _Org: id = 1
    class _Caller: org_id = 1; org = _Org()
    plan = await build_plan(ep, {"linkedin_url": "https://www.linkedin.com/in/x"}, _Caller(), RouteOptions.from_headers(lambda k: None))
    assert "aviato.people.email.find" not in [c.endpoint["id"] for c in plan.candidates]
    assert any(d["endpoint_id"] == "aviato.people.email.find" and "no aviato key" in d["why"] for d in plan.dropped)


def test_a_contract_may_set_its_own_default_ceiling():
    from treg.application.call.route import RouteOptions, DEFAULT_MAX_COST_MICRO
    cat = catalog_store.load()
    assert cat.contracts["people.search"].default_max_cost_usd is None, "the $1 default covers every current ladder"
    assert RouteOptions.from_headers(lambda k: None, 500_000).max_cost_micro == 500_000
    assert RouteOptions.from_headers(lambda k: None).max_cost_micro == DEFAULT_MAX_COST_MICRO
    assert RouteOptions.from_headers(lambda k: "0.02" if k == "x-treg-route-max-cost" else None, 500_000).max_cost_micro == 20_000


async def test_routed_call_and_access_name_the_providers_dropped_for_this_deployment(clients: AsyncClient, enrichment_on, monkeypatch):
    monkeypatch.setenv("TREG_PLATFORM_KEY_AVIATO", "")
    get_settings.cache_clear()
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider(
        {"tomba": [(200, {"data": {"email": None, "score": None, "verification": {"status": None}}})],
         "findymail": [(200, {"contact": {"name": "x", "email": None}})],
         "leadsforge": [(200, {"email": None, "status": "failed"})]}, seen))
    r = await clients.post(f"/call/{ROUTED}", json={"linkedin_url": "https://www.linkedin.com/in/x"}, headers={"X-Treg-Route-Max-Cost": "0.03"})
    assert r.status_code == 200 and r.json()["_treg"]["outcome"] == "miss"
    dropped = r.json()["_treg"]["dropped"]
    assert any(d["endpoint_id"] == "aviato.people.email.find" and "no aviato key" in d["why"] for d in dropped)
    hunter = next(d for d in dropped if d["endpoint_id"] == "hunter.people.email.find")
    assert hunter == {"endpoint_id": "hunter.people.email.find", "why": "needs {domain, full_name} | {domain, first_name, last_name}"}
    a = await clients.get(f"/catalog/endpoints/{ROUTED}/access")
    assert a.status_code == 200 and a.json()["tier"] == "routed" and a.json()["detail"].startswith("routed — ")
    assert "aviato.people.email.find" in a.json()["detail"]


def test_a_caller_may_send_everything_it_knows_and_each_provider_gets_only_its_variant():
    cat = catalog_store.load()
    contract = cat.contracts["people.phone.find"]
    everything = {"email": "p@stripe.com", "linkedin_url": "https://www.linkedin.com/in/p", "full_name": "Patrick Collison", "domain": "stripe.com"}
    ident, variant = canonical_identity(contract, everything)
    assert ident["first_name"] == "Patrick"
    from treg.domain.catalog.routing.contracts import adapter_accepts
    tomba = cat.adapters["tomba.people.phone.find"]
    v = adapter_accepts(tomba, ident)
    q, b = tomba.to_upstream(ident, v)
    assert q == {"email": "p@stripe.com"}, "tomba insists on exactly one identifier — only the matched variant is sent"
    lf = cat.adapters["leadsforge.people.phone.find"]
    q, b = lf.to_upstream(ident, adapter_accepts(lf, ident))
    assert b == {"firstName": "Patrick", "lastName": "Collison", "companyDomain": "stripe.com"}, "derived names, and not the LinkedIn URL"
    # filters always travel, whatever the variant
    kw = cat.contracts["google.keywords.ideas"]
    req, v = canonical_identity(kw, {"keyword": "coffee", "country": "de"})
    q, b = cat.adapters["seranking.google.keywords.ideas"].to_upstream(req, v)
    assert q == {"keyword": "coffee", "source": "de", "limit": "20"}
    # every adapter still verifies with the change
    assert all(a.verified for a in cat.adapters.values())


def test_rank_prefers_the_candidate_that_uses_more_of_the_identity():
    """Given {company_domain, title}, a title-aware provider outranks a cheaper domain-only one —
    the cheaper answer would be to a different question (the whole company)."""
    from treg.domain.catalog.routing.plan import Candidate, rank
    def cand(eid, variant, price):
        return Candidate(endpoint={"id": eid, "provider": eid.split(".")[0], "cost": {"type": "per_result"}}, adapter=None,
                         variant=variant, tier="platform", price_micro=price, hit_rate=None, ok_rate=None, p50_ms=None, last_ok_days=None)
    free_domain = cand("hunter.x.multi-domain-search", ("company_domain",), 0)
    title_aware = cand("icypeas.people.search", ("company_domain", "title"), 380)
    dearer_title = cand("companyenrich.people.search", ("company_domain", "title"), 19_600)
    given = {"company_domain", "title"}
    assert [c.endpoint["id"] for c in rank([free_domain, dearer_title, title_aware], given=given)] == [
        "icypeas.people.search", "companyenrich.people.search", "hunter.x.multi-domain-search"]
    # a key the caller did NOT send (reached via derive) earns nothing: price decides again
    assert rank([free_domain, title_aware], given={"company_domain"})[0] is free_domain
    # …but a variant DERIVED from what the caller sent covers it: {first,last,domain} from a supplied
    # full_name is as specific as {full_name, domain}, so the cheaper of the two (hunter) leads
    derive = {"first_name": "split_first(full_name)", "last_name": "split_last(full_name)"}
    hunter = cand("hunter.people.email.find", ("first_name", "last_name", "domain"), 4_900)
    apollo = cand("apollo.people.enrich", ("full_name", "domain"), 26_000)
    assert rank([apollo, hunter], given={"full_name", "domain"}, derive=derive)[0] is hunter


def test_a_provider_that_cannot_express_a_supplied_filter_ranks_last_among_equals():
    """Live 2026-08-29: `{q, title, location: London, country: GB}` went to the cheapest candidate,
    which had no place for either geo filter, and returned people in Bengaluru and San Francisco —
    reported as a hit. Cheapness must not buy an answer to a looser question."""
    from treg.domain.catalog.routing.plan import Candidate, ignored_filters, rank
    def cand(eid, price, ignored=()):
        return Candidate(endpoint={"id": eid, "provider": eid.split(".")[0], "cost": {"type": "per_result"}},
                         adapter=None, variant=("q",), tier="platform", price_micro=price, hit_rate=None,
                         ok_rate=None, p50_ms=None, last_ok_days=None, ignored=ignored)
    geo_blind = cand("aviato.people.search", 2_500, ignored=("country", "location"))
    geo_aware = cand("icypeas.people.search", 5_700)
    assert [c.endpoint["id"] for c in rank([geo_blind, geo_aware], given={"q"})] == [
        "icypeas.people.search", "aviato.people.search"], "the dearer provider that honours the filters leads"
    # still reachable when it is the only candidate, and price still decides among equals
    assert rank([geo_blind], given={"q"})[0] is geo_blind
    assert rank([geo_blind, cand("z.people.search", 9_000, ignored=("country", "location"))], given={"q"})[0] is geo_blind

    # and the set itself is read off the adapter's input map, not guessed
    cat = catalog_store.load()
    contract = cat.contracts["people.search"]
    ident = {"q": "backend engineers", "country": "GB", "location": "London, United Kingdom", "limit": 15}
    assert "country" in ignored_filters(cat.adapters["aviato.people.search"], contract, ident)
    assert ignored_filters(cat.adapters["icypeas.people.search"], contract, ident) == (), \
        "icypeas is the only people.search adapter that maps geo — the rule must float it to the top"
    # the full_name variant has exactly two candidates and neither mapped `country` — so a GT search
    # went to New York and was billed (voice-ai-outbound, 2026-09-03). aviato's simple search takes
    # country NAMES (live 2026-09-04: `Guatemala` → 84,145 rows, `GT` → 0), hence country_name().
    simple = cat.adapters["aviato.people.search.simple"]
    by_name, _ = canonical_identity(contract, {"full_name": "Carlos Lopez", "country": "GT", "limit": 5})
    assert ignored_filters(simple, contract, by_name) == ()
    q, _ = simple.to_upstream(by_name, ("full_name",))
    assert q["country"] == "Guatemala", q  # a query value travels as one string, never a list repr
    assert "country" not in simple.to_upstream({**by_name, "country": None}, ("full_name",))[0]


async def test_the_geo_aware_child_wins_a_filtered_search_and_the_answer_says_what_was_dropped(
    clients: AsyncClient, enrichment_on, monkeypatch
):
    """End to end on the real people.search ladder: the caller sends geo, so the child that maps it
    is called even though a CHEAPER one is callable — and when the winner drops a filter, the
    envelope and a header say so, where a caller will see it (not buried in `tried[]`).

    Live 2026-08-29 this went to aviato ($0.0025, maps neither `country` nor `location`) and came
    back with people in Bengaluru and San Francisco for a London brief, reported as a hit."""
    monkeypatch.setenv("TREG_PLATFORM_KEY_ICYPEAS", "PLATFORM-ICYPEAS-KEY")
    monkeypatch.setenv("TREG_PLATFORM_PROVIDERS", "hunter,tomba,leadmagic,leadsforge,findymail,aviato,fiber-ai,icypeas")
    get_settings.cache_clear()
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider(
        {"icypeas": [(200, {"leads": [{"firstname": "Aleksei", "lastname": "Strizhak", "lastJobTitle": "Senior Backend Engineer",
                                       "address": "London Area, United Kingdom", "profileUrl": "https://linkedin.com/in/as"}]})],
         "*": [(200, {"persons": []})] * 12}, seen))
    r = await clients.post("/call/treg.people.search",
                           json={"q": "backend engineer", "title": "Backend Engineer",
                                 "location": "London, United Kingdom", "country": "GB", "limit": 15})
    assert r.status_code == 200, r.text
    d = r.json()
    assert d["_treg"]["served_by"] == "icypeas.people.search", \
        "the child that maps country/location leads, though the cheaper geo-blind aviato is callable"
    assert seen == [("icypeas", "POST", {}, {"query": {"currentJobTitle": {"include": ["Backend Engineer"]},
                                                      "location": {"include": ["London, United Kingdom"]}},
                                             "pagination": {"size": 15}})], \
        "one call, and the geography actually reached the provider"
    assert "ignored_filters" not in d["_treg"] and "X-Treg-Ignored-Filters" not in r.headers

    # …and when the winner cannot express a filter, the answer says which — envelope, header, attempt
    seen2 = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider(
        {"*": [(200, {"items": [{"fullName": "Ada L", "URLs": {"linkedin": "linkedin.com/in/al"}}], "count": {"value": 1}})] * 12}, seen2))
    r2 = await clients.post("/call/treg.people.search", json={"q": "backend engineer", "country": "GB", "limit": 15})
    assert r2.status_code == 200, r2.text
    d2 = r2.json()
    assert d2["_treg"]["served_by"] == "aviato.people.search", "no geo-aware child answers a {q}-only brief here"
    assert d2["_treg"]["ignored_filters"] == ["country"], "the caller sent it; aviato has no place for it"
    assert r2.headers["X-Treg-Ignored-Filters"] == "country"
    assert d2["_treg"]["tried"][-1]["ignored_filters"] == ["country"], "same set in all three places"


async def test_routed_discovery_is_a_runtime_switch(clients: AsyncClient, enrichment_on, monkeypatch):
    """`TREG_ROUTED_DISCOVERY=off` stops search LEADING with `treg.<capability>` — nothing else.
    The endpoints stay callable, priced and reachable by id; only the steering goes away, so a
    deployment can answer "should every agent be pointed at the router by default" with traffic
    instead of an argument, and can undo it without a redeploy."""
    q = {"q": "find someone's work email from their name and company", "limit": 8}
    on = (await clients.get("/catalog/search", params=q)).json()
    ids_on = [r["id"] for r in on["results"]]
    assert any(i.startswith("treg.") for i in ids_on), "steering on: the routed row is in the page"

    monkeypatch.setenv("TREG_ROUTED_DISCOVERY", "off")
    get_settings.cache_clear()
    off = (await clients.get("/catalog/search", params=q)).json()
    ids_off = [r["id"] for r in off["results"]]
    assert not any(i.startswith("treg.") for i in ids_off), \
        "steering off: search looks as it did before routing shipped"
    assert ids_off, "and it still returns the providers themselves"

    # every OTHER discovery surface follows the same switch, or the deployment contradicts itself
    plat = (await clients.get("/catalog/platforms/people")).json()
    flat = json.dumps(plat)
    assert "treg.people." not in flat, "browse view: no routed row while steering is off"
    for path in ("/skill.md", "/llms.txt"):
        body = (await clients.get(path)).text
        assert "Routed endpoints" not in body, f"{path} must not teach what search hides"
        assert "<!--routed" not in body, f"{path} leaked a marker"
        assert "provider_capacity_unavailable" in body, f"{path} lost the unrelated overflow guidance"

    # …but the endpoint is untouched: still callable, still priced, still found by id
    r = await clients.get("/catalog/endpoints/treg.people.email.find")
    assert r.status_code == 200 and r.json()["endpoint"]["kind"] == "routed"
    monkeypatch.setattr(call_service, "relay", _relay_by_provider(
        {"tomba": [(200, {"data": {"email": "p@stripe.com", "score": 99, "first_name": "Patrick",
                                   "last_name": "Collison", "verification": {"status": "valid"}}})]}, []))
    call = await clients.post(f"/call/{ROUTED}", json={"full_name": "Patrick Collison", "domain": "stripe.com"})
    assert call.status_code == 200 and call.json()["_treg"]["served_by"] == "tomba.people.email.find"
    get_settings.cache_clear()

def test_keywords_are_a_filter_so_they_reach_every_provider_that_can_express_them():
    """The brief's SUBSTANCE lives in its keywords. As identity they would be dropped whenever
    another variant matched — icypeas matches {title}, so a `q` carrying "microservices" never
    reached it and the search degenerated to title+location (bench 2026-08-29: "football scouting
    analysts" reached the provider as title="Football Analyst" and scored 0 qualified of 15)."""
    cat = catalog_store.load()
    contract = cat.contracts["people.search"]
    assert "keywords" in contract.filters and "keywords" not in {k for v in contract.identity for k in v}
    ident, variant = canonical_identity(contract, {
        "q": "backend developers with microservices", "title": "Backend Engineer",
        "location": "London, United Kingdom", "country": "GB",
        "keywords": ["microservices", "architecture"], "limit": 15})
    from treg.domain.catalog.routing.contracts import adapter_accepts
    icy = cat.adapters["icypeas.people.search"]
    _, body = icy.to_upstream(ident, adapter_accepts(icy, ident))
    assert body["query"]["keyword"]["include"] == ["microservices", "architecture"], \
        "icypeas takes them natively — the whole point of the contract field"
    exa = cat.adapters["exa.people.search"]
    _, body = exa.to_upstream(ident, adapter_accepts(exa, ident))
    assert "microservices" in body["query"], "a semantic provider gets them folded into the query"
    # and a provider with nowhere to put them says so, which ranks it down (PR #254)
    from treg.domain.catalog.routing.plan import ignored_filters
    assert "keywords" in ignored_filters(cat.adapters["aviato.people.search"], contract, ident)


async def test_a_thin_hit_does_not_end_the_waterfall_when_the_caller_set_min_results(
    clients: AsyncClient, enrichment_on, monkeypatch
):
    """`X-Treg-Route-Min-Results: 3` — one row is not an answer to "find me candidates". The
    router keeps going and the FULLEST answer wins; without it the first non-empty body stops the
    search (bench 2026-08-29: the hand-written policy's `if len(rows) < 3 -> fall through` was the
    single behaviour the routed path could not express)."""
    monkeypatch.setenv("TREG_PLATFORM_KEY_ICYPEAS", "PLATFORM-ICYPEAS-KEY")
    monkeypatch.setenv("TREG_PLATFORM_PROVIDERS", "hunter,tomba,leadmagic,leadsforge,findymail,aviato,fiber-ai,icypeas")
    get_settings.cache_clear()
    thin = {"leads": [{"firstname": "Solo", "lastname": "Row", "profileUrl": "https://linkedin.com/in/s"}]}
    full = {"items": [{"fullName": f"P{i}", "URLs": {"linkedin": f"linkedin.com/in/p{i}"}} for i in range(9)],
            "count": {"value": 9}}
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider(
        {"icypeas": [(200, thin)], "*": [(200, full)] * 12}, seen))
    body = {"q": "backend engineer", "title": "Backend Engineer",
            "location": "London, United Kingdom", "country": "GB", "limit": 15}
    r = await clients.post("/call/treg.people.search", json=body, headers={"X-Treg-Route-Min-Results": "3"})
    assert r.status_code == 200, r.text
    d = r.json()
    assert [t["outcome"] for t in d["_treg"]["tried"]][0] == "weak", "1 row < 3 is not an answer"
    assert d["_treg"]["served_by"] != "icypeas.people.search" and len(d["output"]["people"]) == 9

    # default (min_results 1): the same thin answer ends the search, as before
    seen2 = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider(
        {"icypeas": [(200, thin)], "*": [(200, full)] * 12}, seen2))
    r2 = await clients.post("/call/treg.people.search", json=body)
    assert r2.json()["_treg"]["served_by"] == "icypeas.people.search" and len(seen2) == 1

    # and when NOBODY clears the bar, the fullest weak answer is still returned, not a miss
    seen3 = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider(
        {"icypeas": [(200, thin)], "*": [(200, {"items": [{"fullName": "Two"}, {"fullName": "Rows"}],
                                                "count": {"value": 2}})] * 12}, seen3))
    r3 = await clients.post("/call/treg.people.search", json=body, headers={"X-Treg-Route-Min-Results": "5"})
    assert r3.status_code == 200 and len(r3.json()["output"]["people"]) == 2, "best effort beats nothing"


async def test_the_weak_hit_fallback_is_bounded_like_the_error_fallback(
    clients: AsyncClient, enrichment_on, monkeypatch
):
    """Some briefs HAVE only one right answer ("who runs engineering at X"), so no provider ever
    clears min_results and an unbounded rule pays the whole ladder on every call. Measured on the
    bench's deterministic set: 12.7x ($1.76 -> $22.35 over 28 queries) for answers already correct.
    At most MAX_WEAK_FALLBACKS extra providers are asked, then the fullest answer is returned."""
    monkeypatch.setenv("TREG_PLATFORM_KEY_ICYPEAS", "PLATFORM-ICYPEAS-KEY")
    monkeypatch.setenv("TREG_PLATFORM_PROVIDERS", "hunter,tomba,leadmagic,leadsforge,findymail,aviato,fiber-ai,icypeas")
    get_settings.cache_clear()
    # Every provider answers ONE row in ITS OWN shape — a real thin HIT, not a miss (a miss does
    # not consume the bound, and must not: the bound is about paying for thin answers).
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider(
        {"icypeas":    [(200, {"leads": [{"firstname": "One", "lastname": "Row"}], "total": 1})] * 4,
         "leadsforge": [(200, {"leads": [{"firstName": "One", "lastName": "Row"}]})] * 4,
         "leadmagic":  [(200, {"data": [{"full_name": "One Row"}], "total_count": 1})] * 4,
         "hunter":     [(200, {"data": {"emails": [{"value": "one@stripe.com"}]}, "meta": {"results": 1}})] * 4,
         "*":          [(200, {"items": [{"fullName": "One Row"}], "count": {"value": 1}})] * 8}, seen))
    # {company_domain} has the deepest ladder, so the BOUND is what stops this, not running out
    body = {"company_domain": "stripe.com", "limit": 15}
    r = await clients.post("/call/treg.people.search", json=body,
                           headers={"X-Treg-Route-Min-Results": "5"})   # nothing will ever clear 5
    assert r.status_code == 200, r.text
    d = r.json()
    attempts = [t for t in d["_treg"]["tried"] if t["outcome"] == "weak"]
    assert len(attempts) == call_route.MAX_WEAK_FALLBACKS + 1, \
        f"the first ask plus at most {call_route.MAX_WEAK_FALLBACKS} more, not the whole ladder"
    assert len(seen) == len(attempts), "and no provider beyond the bound was ever called"
    assert len(d["output"]["people"]) == 1, "and the caller still gets the answer that exists"


async def test_merge_unions_the_rows_the_caller_already_paid_for(
    clients: AsyncClient, enrichment_on, monkeypatch
):
    """`X-Treg-Route-Merge: 1` — a list answer is the one shape a union makes sense for, and the
    caller is charged for EVERY attempt already (`charged_micro` sums them), so returning only the
    winner's rows throws away results the team bought. Bench 2026-08-29: a people.search that fell
    through returned the fullest single provider's rows, never icypeas' 5 plus exa's 10."""
    monkeypatch.setenv("TREG_PLATFORM_KEY_ICYPEAS", "PLATFORM-ICYPEAS-KEY")
    monkeypatch.setenv("TREG_PLATFORM_PROVIDERS", "hunter,tomba,leadmagic,leadsforge,findymail,aviato,fiber-ai,icypeas")
    get_settings.cache_clear()
    icy = {"leads": [{"firstname": "Ada", "lastname": "L", "profileUrl": "https://linkedin.com/in/ada"},
                     {"firstname": "Bo", "lastname": "M", "profileUrl": "https://linkedin.com/in/bo"}], "total": 2}
    # one row OVERLAPS on the profile url (different casing/scheme), one is new
    other = {"items": [{"fullName": "Ada L", "URLs": {"linkedin": "www.linkedin.com/in/Ada/"}},
                       {"fullName": "Cy N", "URLs": {"linkedin": "linkedin.com/in/cy"}}], "count": {"value": 2}}
    body = {"q": "backend engineer", "title": "Backend Engineer", "limit": 15}

    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider(
        {"icypeas": [(200, icy)], "*": [(200, other)] * 8}, seen))
    r = await clients.post("/call/treg.people.search", json=body,
                           headers={"X-Treg-Route-Min-Results": "5", "X-Treg-Route-Merge": "1"})
    assert r.status_code == 200, r.text
    d = r.json()
    people = d["output"]["people"]
    keys = sorted(call_route._row_key(p) for p in people)
    assert keys == ["linkedin.com/in/ada", "linkedin.com/in/bo", "linkedin.com/in/cy"], \
        "the union of both providers, and Ada — who both returned, spelled differently — appears ONCE"
    assert len(people) == 3, "3 distinct people from two answers of 2 rows each"
    assert len(d["_treg"]["merged_from"]) >= 2 and "X-Treg-Merged-From" in r.headers
    assert d["_treg"]["charged_micro"] == int(r.headers["X-Treg-Cost-Micro"]), \
        "merging changes no money: the sum over attempts is what it always was"

    # without the header the winner's rows alone come back — the pre-existing contract
    seen2 = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider(
        {"icypeas": [(200, icy)], "*": [(200, other)] * 8}, seen2))
    r2 = await clients.post("/call/treg.people.search", json=body,
                            headers={"X-Treg-Route-Min-Results": "5"})
    assert "merged_from" not in r2.json()["_treg"] and len(r2.json()["output"]["people"]) == 2


def test_a_per_success_endpoint_with_no_adapter_settles_on_the_providers_own_success_rule():
    """The adapter's `miss` predicate covers routed children; 1330 of 1517 per_success endpoints
    have no adapter, and they are the SCRAPERS — whose failure mode is an HTTP 200 carrying an
    error code. Those providers publish a success rule and the catalog records it as `expect`,
    which until now only `scripts/catalog_verify.py` read.

    Live 2026-08-29: `justoneapi.x.linkedin-search-user-v1` answered
    `{"code": 301, "message": "COLLECT FAILED, SEND REQUEST AGAIN"}` — free on the vendor's own
    published terms ("only a code-0 response is billed") — and treg settled $0.0295 against the
    caller."""
    from test_marketplace_call import _mk
    from treg.application.call import settle as A
    cat = catalog_store.load()
    eid = "justoneapi.x.linkedin-search-user-v1"
    ep = cat.by_id[eid]
    assert (ep.get("cost") or {}).get("type") == "per_success" and cat.adapters.get(eid) is None, \
        "the shape this rule exists for: priced per success, no adapter to ask"
    assert ep.get("expect") == {"json_path": "code", "equals": 0}, "the loader must carry the rule"

    mk = _mk("justoneapi", endpoint_id=eid, cost_type="per_success")
    fail = b'{"code": 301, "data": null, "message": "COLLECT FAILED, SEND REQUEST AGAIN"}'
    assert A._observed_cost_micro(mk, fail) == 0, "a vendor-side failure the vendor does not bill"
    ok = b'{"code": 0, "data": {"users": [{"name": "Ada"}]}}'
    assert A._observed_cost_micro(mk, ok) is None, "a real hit still settles at the estimate"

    # a nested rule form (dataforseo's task envelope) reads the same way — picking one that also
    # has no adapter, since an adapter's own predicate takes precedence when there is one
    dfs = [e for e in cat.by_id.values()
           if (e.get("expect") or {}).get("json_path") == "tasks.0.status_code"
           and cat.adapters.get(e["id"]) is None
           and (e.get("cost") or {}).get("type") == "per_success"]
    if dfs:
        m2 = _mk(dfs[0]["provider"], endpoint_id=dfs[0]["id"], cost_type="per_success")
        assert A._observed_cost_micro(m2, b'{"tasks": [{"status_code": 40501}]}') == 0
        assert A._observed_cost_micro(m2, b'{"tasks": [{"status_code": 20000}]}') is None

async def test_a_declared_miss_status_is_a_miss_not_a_caller_fault(clients: AsyncClient, enrichment_on, monkeypatch):
    """aviato answers HTTP 404 `Not Found` for a person it has no record of. The endpoint's YAML says
    so (`miss: {status: 404}`), and the router must read it: before this a waterfall in which the
    other providers all missed ended in a 502 `route_failed` (live 2026-09-03, voice-ai-outbound —
    768 of 1,824 phone.find 502s in 30 days had no failure but an aviato 404), when the honest
    answer was a 200 miss."""
    routed = "treg.people.phone.find"
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider(
        {"aviato": [(404, {"message": "Not Found"})],
         "tomba": [(200, {"data": {"e164_format": None}})],
         "leadmagic": [(200, {"mobile_number": None, "credits_consumed": 0})],
         "findymail": [(200, {"phone": None})],
         "leadsforge": [(200, {"phoneNumber": None})]}, seen))
    before = await _balance(clients)
    r = await clients.post(f"/call/{routed}", json={"linkedin_url": "https://www.linkedin.com/in/nobody-here"})
    assert r.status_code == 200, r.text
    assert r.headers["X-Treg-Route-Outcome"] == "miss" and r.json()["_treg"]["served_by"] is None
    outcomes = {t["endpoint_id"]: t["outcome"] for t in r.json()["_treg"]["tried"]}
    assert outcomes["aviato.people.phone.find"] == "miss" and len(seen) == 5, "the 404 is a miss; every provider was still asked"
    assert await _balance(clients) == before, "nobody found anything, nothing was charged"
    # an UNDECLARED 4xx keeps its meaning: a vendor rejecting the request is still the caller's fault
    seen.clear()
    monkeypatch.setattr(call_service, "relay", _relay_by_provider(
        {"aviato": [(422, {"message": "bad identifier"})], "*": [(422, {"message": "bad"})] * 4}, seen))
    r = await clients.post(f"/call/{routed}", json={"linkedin_url": "https://www.linkedin.com/in/nobody-here"},
                           headers={"X-Treg-Route-Prefer": "aviato"})
    assert r.status_code == 422 and r.json()["detail"]["error"] == "route_caller_fault", r.text
    # and with the waterfall off, the declared 404 alone is the (free) miss the caller asked for
    seen.clear()
    monkeypatch.setattr(call_service, "relay", _relay_by_provider({"aviato": [(404, {"message": "Not Found"})]}, seen))
    r = await clients.post(f"/call/{routed}", json={"linkedin_url": "https://www.linkedin.com/in/nobody-here"},
                           headers={"X-Treg-Route-Prefer": "aviato", "X-Treg-Route-Waterfall": "0"})
    assert r.status_code == 200 and r.headers["X-Treg-Route-Outcome"] == "miss" and len(seen) == 1, r.text
    assert r.json()["output"]["phone"] is None


def test_every_declared_miss_status_names_its_meaning():
    """`miss: {status, means}` is agent-facing (`endpoint_view`) and router-facing: both halves
    are required. The router honours a 4xx only — a `status: 200` block is documentation for the
    agent (tikhub answers 200 with a null body for an unknown id) and the adapter's own predicate
    decides that case, so `_miss_status` must never turn a success into a miss."""
    cat = catalog_store.load()
    declared = {e["id"]: e["miss"] for e in cat.by_id.values() if e.get("miss")}
    assert "aviato.people.phone.find" in declared and "hunter.people.enrich" in declared
    for eid, m in declared.items():
        assert isinstance(m, dict) and m.get("status") is not None and m.get("means"), eid
        assert int(m["status"]) < 500, f"{eid}: a 5xx is never 'asked and answered'"
    assert call_route._miss_status(cat.by_id["aviato.people.phone.find"]) == 404
    assert call_route._miss_status(cat.by_id["tikhub.x.reddit-app-fetch-post-comments"]) is None
    assert call_route._miss_status({"id": "x"}) is None


async def test_lusha_is_the_last_rung_of_the_phone_waterfall_and_settles_on_its_own_bill(clients: AsyncClient, enrichment_on, monkeypatch):
    """Guatemala, 2026-09-03: 7 phones in 44 across tomba/aviato/leadmagic/findymail/leadsforge.
    Lusha's native direct-dial data remains the last rung — dearest per hit (6 credits), so it ranks
    after AI Ark, Dropleads, Prospeo, and the cheaper providers; a miss is free and a matched profile
    with no number costs the 1-credit search, both read off `billing.creditsCharged`."""
    monkeypatch.setenv("TREG_PLATFORM_KEY_LUSHA", "PLATFORM-LUSHA-KEY")
    monkeypatch.setenv("TREG_PLATFORM_PROVIDERS", "hunter,tomba,leadmagic,leadsforge,findymail,aviato,fiber-ai,lusha")
    get_settings.cache_clear()
    routed = "treg.people.phone.find"
    plan = (await clients.get(f"/catalog/endpoints/{routed}")).json()["routing"]["plan"]
    assert plan[-1]["endpoint_id"] == "lusha.people.phone.find" and len(plan) == 13, [c["endpoint_id"] for c in plan]
    def misses():
        return {"aviato": [(404, {"message": "Not Found"})], "tomba": [(200, {"data": {"e164_format": None}})],
                "leadmagic": [(200, {"mobile_number": None, "credits_consumed": 0})],
                "findymail": [(200, {"phone": None})], "leadsforge": [(200, {"phoneNumber": None})]}
    seen = []
    hit = {"requestId": "r", "results": [{"id": "v1.x", "fullName": "Ana Perez",
                                          "phones": [{"number": "+502 5555 0100", "type": "mobile", "doNotCall": False, "countryIso2": "GT"}]}],
           "billing": {"creditsCharged": 6, "resultsReturned": 1}}
    monkeypatch.setattr(call_service, "relay", _relay_by_provider({**misses(), "lusha": [(200, hit)]}, seen))
    # a Lusha attempt RESERVES the 6-credit hit price (~$0.75): on the $1.00 signup grant a team gets
    # one attempt, so fund the second call here rather than let the reserve mask the miss rule
    org_id = (await clients.get("/orgs")).json()[0]["org_id"]
    async with session_maker() as db:
        await ledger.grant(db, org_id, amount_micro=5_000_000, kind="test-funding", once=False)
        await db.commit()
    before = await _balance(clients)
    r = await clients.post(f"/call/{routed}", json={"full_name": "Ana Perez", "domain": "acme.gt"})
    assert r.status_code == 200 and r.json()["_treg"]["served_by"] == "lusha.people.phone.find", r.text
    assert r.json()["output"] == {"phone": "+502 5555 0100", "line_type": "mobile", "country_code": "GT"}
    # {full_name, domain} is accepted by two rungs only (leadsforge, lusha); the four that need a
    # LinkedIn URL or an email are not candidates for this identity at all
    assert [p for p, *_ in seen] == ["leadsforge", "lusha"], "asked last, after every cheaper candidate missed"
    body = seen[-1][3]
    assert body == {"contacts": [{"firstName": "Ana", "lastName": "Perez", "companyDomain": "acme.gt"}], "reveal": ["phones"]}
    rate = catalog_store.load().credit_rates["lusha"]
    assert before - await _balance(clients) == int(6 * rate * 1_000_000 + 0.5), "the bill is Lusha's own creditsCharged"
    # a matched profile with no number is a MISS that still cost the 1-credit search
    seen.clear()
    no_number = {"requestId": "r", "results": [{"id": "v1.x", "fullName": "Ana Perez", "partialProfile": False}],
                 "billing": {"creditsCharged": 1, "resultsReturned": 1}, "status": "partial", "statusReason": "WATERFALL_NOT_CONFIGURED"}
    monkeypatch.setattr(call_service, "relay", _relay_by_provider({**misses(), "lusha": [(200, no_number)]}, seen))
    before = await _balance(clients)
    r = await clients.post(f"/call/{routed}", json={"full_name": "Ana Perez", "domain": "acme.gt"})
    assert r.status_code == 200 and r.headers["X-Treg-Route-Outcome"] == "miss" and r.json()["output"]["phone"] is None, r.text
    assert before - await _balance(clients) == int(1 * rate * 1_000_000 + 0.5)
    get_settings.cache_clear()


async def test_quickenrich_is_cheapest_phone_provider_and_respects_max_cost(clients: AsyncClient, enrichment_with_quickenrich_on, monkeypatch):
    """QuickEnrich is the cheapest phone provider (~$0.0048) and must be considered when max-cost is
    set above its price. Regression for feedback #136/#128: customers got 402s because the router
    was treating a more expensive provider as cheapest when QuickEnrich was not in PLATFORM_PROVIDERS."""
    routed = "treg.people.phone.find"
    plan = (await clients.get(f"/catalog/endpoints/{routed}")).json()["routing"]["plan"]
    prices = [(c["endpoint_id"], c["usd"]) for c in plan]
    quickenrich_entry = next((p for p in prices if "quickenrich" in p[0]), None)
    assert quickenrich_entry is not None, f"QuickEnrich must be in the phone waterfall: {prices}"
    assert quickenrich_entry[1] == min(p[1] for p in prices if p[1]), f"QuickEnrich must be the cheapest: {prices}"

    seen = []
    hit = {'success': True, 'data': {'employee_phone': '+15550100100', 'employee_phone_type': 'mobile'},
           'meta': {'credits_used': 1}}
    monkeypatch.setattr(call_service, "relay", _relay_by_provider({'quickenrich': [(200, hit)]}, seen))

    before = await _balance(clients)
    # max_cost=$0.01 is above QuickEnrich (~$0.0048) but below every other provider
    r = await clients.post(f"/call/{routed}", json={"linkedin_url": "https://www.linkedin.com/in/example"},
                           headers={"X-Treg-Route-Max-Cost": "0.01"})
    assert r.status_code == 200, f"Should succeed with QuickEnrich: {r.text}"
    assert r.json()["_treg"]["served_by"] == "quickenrich.people.phone.find"
    assert r.json()["output"]["phone"] == "+15550100100"
    assert [p for p, *_ in seen] == ["quickenrich"], "Only QuickEnrich should be called"
    charged = before - await _balance(clients)
    assert charged == 4834, f"QuickEnrich should charge 1 credit = $0.004834 = 4834 micro: got {charged}"


async def test_max_cost_below_cheapest_refuses_before_any_call_for_phone(clients: AsyncClient, enrichment_with_quickenrich_on, monkeypatch):
    """When max-cost is below even the cheapest provider (QuickEnrich), the call is refused with
    route_max_cost error before any provider is asked, naming the cheapest candidate."""
    routed = "treg.people.phone.find"
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider({}, seen))
    r = await clients.post(f"/call/{routed}", json={"linkedin_url": "https://www.linkedin.com/in/example"},
                           headers={"X-Treg-Route-Max-Cost": "0.001"})  # $0.001 < QuickEnrich's $0.0048
    assert r.status_code == 402, r.text
    d = r.json()["detail"]
    assert d["error"] == "route_max_cost"
    assert "quickenrich" in d["message"], f"Error should name QuickEnrich as cheapest: {d['message']}"
    assert seen == [], "No provider should be called when max-cost is below the cheapest"


async def test_capped_signal_when_max_cost_truncates_waterfall(clients: AsyncClient, enrichment_with_quickenrich_on, monkeypatch):
    """Feedback #131: When max-cost stops the waterfall early, the result should indicate that more
    expensive providers were skipped (capped=true). This lets callers distinguish an exhaustive miss
    from one truncated by budget - they can raise their ceiling if they need to try all providers."""
    routed = "treg.people.phone.find"
    miss = {'success': False, 'message': 'No data found for this profile'}
    monkeypatch.setattr(call_service, "relay", _relay_by_provider({'quickenrich': [(200, miss)]}, []))
    r = await clients.post(f"/call/{routed}", json={"linkedin_url": "https://www.linkedin.com/in/example"},
                           headers={"X-Treg-Route-Max-Cost": "0.01"})  # ~$0.01 allows QuickEnrich only
    assert r.status_code == 200, r.text
    treg = r.json()["_treg"]
    assert treg["outcome"] == "miss"
    assert treg.get("capped") is True, f"Expected capped=true when waterfall truncated: {treg}"
    assert r.headers.get("X-Treg-Route-Capped") == "true", "Expected X-Treg-Route-Capped header"
    skipped = [t for t in treg["tried"] if t["outcome"] == "skipped" and "would exceed" in t.get("detail", "")]
    assert skipped, f"Expected some providers skipped due to cost: {treg['tried']}"


async def test_strict_filters_refuses_a_looser_answer_instead_of_billing_it(clients: AsyncClient, enrichment_on, monkeypatch):
    """voice-ai-outbound, 2026-09-03: `{full_name, country: GT}` went to a candidate that ignored
    the country and was billed for people in New York. Opt-in, the caller is refused instead —
    unbilled, told which filter, and what identity a filter-aware provider would take."""
    monkeypatch.setenv("TREG_PLATFORM_KEY_CRUSTDATA", "PLATFORM-CRUSTDATA-KEY")
    monkeypatch.setenv("TREG_PLATFORM_PROVIDERS", "hunter,tomba,leadmagic,leadsforge,findymail,aviato,fiber-ai,crustdata")
    get_settings.cache_clear()
    # This regression compares Crustdata with Aviato, independently of other catalog additions.
    cat = catalog_store.load()
    for eid in cat.by_id["treg.people.search"]["routed_children"]:
        if not eid.startswith(("crustdata.", "aviato.")):
            monkeypatch.delitem(cat.adapters, eid)
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider({"*": [(200, {"profiles": [{"name": "Someone"}], "total_count": 1})] * 3}, seen))
    before = await _balance(clients)
    # crustdata's people.search takes full_name and nothing geographic: with the header it is dropped
    r = await clients.post("/call/treg.people.search", json={"full_name": "Carlos Lopez", "country": "GT", "limit": 3},
                           headers={"X-Treg-Route-Strict-Filters": "1", "X-Treg-Route-Exclude": "aviato"})
    assert r.status_code == 422 and r.json()["detail"]["error"] == "no_route_candidate", r.text
    d = r.json()["detail"]
    assert seen == [] and await _balance(clients) == before, "refused before any provider was asked; nothing billed"
    assert any(x["endpoint_id"] == "crustdata.people.search" and x.get("strict") and "country" in x["why"] for x in d["dropped"]), d
    assert "X-Treg-Route-Strict-Filters" in d["message"] and "full_name" in d["message"]
    # without the header the same call goes out, is billed, and says what it ignored
    r = await clients.post("/call/treg.people.search", json={"full_name": "Carlos Lopez", "country": "GT", "limit": 3},
                           headers={"X-Treg-Route-Exclude": "aviato"})
    assert r.status_code == 200 and r.headers["X-Treg-Ignored-Filters"] == "country" and r.json()["_treg"]["ignored_filters"] == ["country"], r.text
    assert len(seen) == 1
    # and a candidate that CAN express the filter is unaffected by the header (aviato's simple search maps country)
    seen.clear()
    monkeypatch.setattr(call_service, "relay", _relay_by_provider({"aviato": [(200, {"items": [{"fullName": "Carlos Lopez", "location": "Guatemala"}], "totalResults": 1})]}, seen))
    r = await clients.post("/call/treg.people.search", json={"full_name": "Carlos Lopez", "country": "GT", "limit": 3},
                           headers={"X-Treg-Route-Strict-Filters": "1"})
    assert r.status_code == 200 and r.json()["_treg"]["served_by"] == "aviato.people.search.simple", r.text
    assert "X-Treg-Ignored-Filters" not in r.headers and seen[0][2]["country"] == "Guatemala"
    get_settings.cache_clear()


@pytest.mark.parametrize("capability", ["people.email.find", "people.phone.find"])
def test_leadsforge_requires_both_name_parts_or_linkedin(capability):
    from treg.domain.catalog.routing.contracts import adapter_accepts
    cat = catalog_store.load()
    adapter = cat.adapters["leadsforge." + capability]
    contract = cat.contracts[capability]
    incomplete, _ = canonical_identity(contract, {"full_name": "Jason", "domain": "example.com"})
    assert adapter_accepts(adapter, incomplete) is None, "Do not send a company-only request"
    complete, _ = canonical_identity(contract, {"full_name": "Test Person", "domain": "example.com"})
    query, body = adapter.to_upstream(complete, adapter_accepts(adapter, complete))
    assert query == {}
    assert body == {"firstName": "Test", "lastName": "Person", "companyDomain": "example.com"}
    linkedin, _ = canonical_identity(contract, {"linkedin_url": "https://www.linkedin.com/in/test-person"})
    assert adapter.to_upstream(linkedin, adapter_accepts(adapter, linkedin))[1] == {
        "linkedinURL": "https://www.linkedin.com/in/test-person"}


def test_successful_contact_lookup_is_not_mailbox_verification():
    cat = catalog_store.load()
    for endpoint, response in [
        ("leadsforge.people.email.find", {"email": "test@example.com", "status": "succeeded"}),
        ("fiber-ai.people.contacts.reveal", {"output": {"profile": {
            "success": True, "emails": [{"email": "test@example.com"}]}}}),
    ]:
        output = cat.adapters[endpoint].from_upstream(response)
        assert output["email"] == "test@example.com"
        assert "verified" not in output, "Successful enrichment is not a deliverability verdict"

@pytest.mark.parametrize("result,valid,miss", [
    ("ok", True, False), ("invalid", False, False), ("disposable", False, False),
    ("catch_all", False, False), ("unknown", False, False), ("unverified", False, False),
])
def test_millionverifier_verdicts(result, valid, miss):
    cat = catalog_store.load()
    eid = "millionverifier.people.email.verify"
    assert eid in cat.by_id["treg.people.email.verify"]["routed_children"]
    assert cat.platform_eligible(cat.by_id[eid])
    assert not cat.platform_eligible(cat.by_id["millionverifier.account.usage"])
    adapter = cat.adapters[eid]
    assert adapter.verified
    doc = {"result": result, "quality": "good" if valid else "bad", "error": ""}
    assert adapter.from_upstream(doc) == {"valid": valid, "status": result}
    assert adapter.is_miss(doc) is miss
    assert adapter.is_miss({"result": "error", "error": "invalid_api_key"})
    assert adapter.is_miss({})


@pytest.mark.parametrize("result,valid", [
    ("deliverable", True),
    ("risky", False),
    ("undeliverable", False),
    ("unknown", False),
])
def test_bounceban_verdicts_join_existing_email_verification_route(result, valid):
    cat = catalog_store.load()
    eid = "bounceban.people.email.verify"
    routed = cat.by_id["treg.people.email.verify"]["routed_children"]
    assert eid in routed
    assert "bounceban.people.email.verify.waterfall" not in routed
    assert cat.platform_eligible(cat.by_id[eid])
    for blocked in (
        "bounceban.people.email.verify.waterfall",
        "bounceban.account.usage",
    ):
        assert not cat.platform_eligible(cat.by_id[blocked])
    adapter = cat.adapters[eid]
    assert adapter.verified
    doc = {"status": "success", "result": result, "score": 99}
    assert adapter.from_upstream(doc) == {"valid": valid, "status": result, "score": 99}
    assert not adapter.is_miss(doc)
    assert adapter.is_miss({"id": "task", "status": "verifying"})


def test_zerobounce_verdicts_join_existing_email_verification_route():
    cat = catalog_store.load()
    eid = "zerobounce.people.email.verify"
    assert eid in cat.by_id["treg.people.email.verify"]["routed_children"]
    assert cat.platform_eligible(cat.by_id[eid])
    adapter = cat.adapters[eid]
    assert adapter.verified
    assert adapter.is_miss({"status": "unknown"})
    assert adapter.is_miss({})
    for status, valid in (("valid", True), ("invalid", False), ("catch-all", False),
                          ("spamtrap", False), ("abuse", False), ("do_not_mail", False)):
        doc = {"status": status}
        assert not adapter.is_miss(doc)
        assert adapter.from_upstream(doc) == {"valid": valid, "status": status}


def test_zerobounce_expensive_discovery_tools_stay_out_of_automatic_routing():
    cat = catalog_store.load()
    assert "zerobounce.people.email.find" not in cat.by_id["treg.people.email.find"]["routed_children"]
    assert "zerobounce.people.email.find" not in cat.adapters
    assert "zerobounce.companies.email_pattern" not in cat.adapters


async def test_zerobounce_serves_existing_email_verification_route(clients, monkeypatch):
    monkeypatch.setenv("TREG_PLATFORM_KEY_ZEROBOUNCE", "PLATFORM-ZEROBOUNCE")
    monkeypatch.setenv("TREG_PLATFORM_PROVIDERS", "zerobounce")
    get_settings.cache_clear()
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider({
        "zerobounce": [(200, {"status": "valid"})],
    }, seen))
    try:
        response = await clients.post(
            "/call/treg.people.email.verify", json={"email": "valid@example.com"})
        assert response.status_code == 200, response.text
        body = response.json()
        assert body["output"] == {"valid": True, "status": "valid"}
        assert body["_treg"]["served_by"] == "zerobounce.people.email.verify"
        assert [row[0] for row in seen] == ["zerobounce"]
    finally:
        get_settings.cache_clear()


async def test_bounceban_serves_existing_email_verification_route(clients, monkeypatch):
    monkeypatch.setenv("TREG_PLATFORM_KEY_BOUNCEBAN", "PLATFORM-BOUNCEBAN")
    monkeypatch.setenv("TREG_PLATFORM_PROVIDERS", "bounceban")
    get_settings.cache_clear()
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider({
        "bounceban": [(200, {
            "id": "task", "status": "success", "result": "risky", "score": 62,
            "credits_consumed": 1, "credits_remaining": 9996,
        })],
    }, seen))
    before = await _balance(clients)
    response = await clients.post(
        "/call/treg.people.email.verify", json={"email": "dev@bounceban.com"})
    assert response.status_code == 200, response.text
    data = response.json()
    assert data["output"]["valid"] is False
    assert data["output"]["status"] == "risky"
    assert data["output"]["score"] == 62
    assert data["_treg"]["served_by"] == "bounceban.people.email.verify"
    assert before - await _balance(clients) == 4_000
    assert [row[0] for row in seen] == ["bounceban"]
    get_settings.cache_clear()


async def test_bounceban_routed_pending_result_is_a_paid_miss_then_falls_through(
    clients, monkeypatch,
):
    monkeypatch.setenv("TREG_PLATFORM_KEY_BOUNCEBAN", "PLATFORM-BOUNCEBAN")
    monkeypatch.setenv("TREG_PLATFORM_KEY_TOMBA", "PLATFORM-TOMBA-KEY")
    monkeypatch.setenv("TREG_PLATFORM_KEY_TOMBA_SECRET", "PLATFORM-TOMBA-SECRET")
    monkeypatch.setenv("TREG_PLATFORM_PROVIDERS", "bounceban,tomba")
    get_settings.cache_clear()
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider({
        "bounceban": [(200, {
            "id": "task", "status": "verifying", "try_again_at": 1789516800,
        })],
        "tomba": [(200, {
            "data": {"email": {"status": "valid", "score": 99}},
        })],
    }, seen))

    before = await _balance(clients)
    response = await clients.post(
        "/call/treg.people.email.verify",
        json={"email": "dev@bounceban.com"},
        headers={"X-Treg-Route-Prefer": "bounceban,tomba"},
    )
    assert response.status_code == 200, response.text
    data = response.json()
    assert data["_treg"]["served_by"] == "tomba.people.email.verify"
    assert [attempt["outcome"] for attempt in data["_treg"]["tried"]] == ["miss", "hit"]
    assert [attempt["charged_micro"] for attempt in data["_treg"]["tried"]] == [4_000, 8_900]
    assert data["_treg"]["charged_micro"] == 12_900
    assert before - await _balance(clients) == 12_900
    assert [row[0] for row in seen] == ["bounceban", "tomba"]
    await audit.drain()
    async with session_maker() as db:
        rows = (await db.execute(select(CallRecord).where(
            CallRecord.provider == "bounceban"))).scalars().all()
        assert len(rows) == 1
        assert rows[0].hit is False
        assert rows[0].cost_charged_micro == 4_000
    get_settings.cache_clear()


async def test_millionverifier_error_falls_through_unbilled(clients, enrichment_on, monkeypatch):
    monkeypatch.setenv("TREG_PLATFORM_KEY_MILLIONVERIFIER", "PLATFORM-MV-KEY")
    monkeypatch.setenv("TREG_PLATFORM_PROVIDERS", "millionverifier,leadmagic")
    get_settings.cache_clear()
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider({
        "millionverifier": [(200, {"result": "error", "error": "Apikey not found"})],
        "leadmagic": [(200, {"email_status": "valid", "credits_consumed": 0.25})],
    }, seen))
    response = await clients.post("/call/treg.people.email.verify", json={"email": "support@millionverifier.com"},
                                  headers={"X-Treg-Route-Prefer": "millionverifier,leadmagic"})
    assert response.status_code == 200, response.text
    data = response.json()
    assert data["_treg"]["served_by"] == "leadmagic.people.email.verify"
    assert data["_treg"]["tried"][0]["outcome"] == "miss"
    await audit.drain()
    async with session_maker() as db:
        rows = (await db.execute(select(CallRecord).where(
            CallRecord.provider == "millionverifier"))).scalars().all()
        assert len(rows) == 1
        assert rows[0].cost_observed_micro == 0


async def test_millionverifier_own_key_precedes_platform_and_is_free(clients, enrichment_on, monkeypatch):
    monkeypatch.setenv("TREG_PLATFORM_KEY_MILLIONVERIFIER", "PLATFORM-MV-KEY")
    monkeypatch.setenv("TREG_PLATFORM_PROVIDERS", "millionverifier,leadmagic")
    get_settings.cache_clear()
    await clients.post("/secrets", json={"name": "millionverifier", "value": "OWN-MV-KEY"})
    before = await _balance(clients)
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider({
        "millionverifier": [(200, {"result": "ok", "quality": "good", "error": ""})],
    }, seen))
    response = await clients.post("/call/treg.people.email.verify", json={"email": "support@millionverifier.com"})
    assert response.status_code == 200, response.text
    assert response.json()["_treg"]["served_by"] == "millionverifier.people.email.verify"
    assert response.json()["_treg"]["tier"] == "credential"
    assert await _balance(clients) == before


async def test_millionverifier_account_usage_requires_own_key(clients, enrichment_on, monkeypatch):
    monkeypatch.setenv("TREG_PLATFORM_KEY_MILLIONVERIFIER", "PLATFORM-MV-KEY")
    monkeypatch.setenv("TREG_PLATFORM_PROVIDERS", "millionverifier")
    get_settings.cache_clear()
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider({
        "millionverifier": [(200, {"credits": 123})],
    }, seen))
    before = await _balance(clients)
    response = await clients.get("/call/millionverifier.account.usage")
    assert response.status_code == 404, response.text
    assert seen == []
    assert await _balance(clients) == before

    await clients.post("/secrets", json={"name": "millionverifier", "value": "OWN-MV-KEY"})
    response = await clients.get("/call/millionverifier.account.usage")
    assert response.status_code == 200, response.text
    assert response.json() == {"credits": 123}
    assert len(seen) == 1
    assert await _balance(clients) == before


@pytest.mark.parametrize("result,free,charged", [
    ("ok", False, True), ("ok", True, True), ("invalid", False, True),
    ("disposable", False, True), ("catch_all", False, False), ("unknown", False, False),
])
async def test_millionverifier_platform_billing(clients, enrichment_on, monkeypatch, result, free, charged):
    """Definitive verdicts cost one credit; risky returns are free, unrelated to free-email flags."""
    monkeypatch.setenv("TREG_PLATFORM_KEY_MILLIONVERIFIER", "PLATFORM-MV-KEY")
    monkeypatch.setenv("TREG_PLATFORM_PROVIDERS", "millionverifier")
    get_settings.cache_clear()
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider({
        "millionverifier": [(200, {"result": result, "quality": "risky" if not charged else "good",
                                   "error": "", "free": free, "credits": 497})],
    }, seen))
    before = await _balance(clients)
    response = await clients.get("/call/millionverifier.people.email.verify", params={"email": "support@millionverifier.com"})
    assert response.status_code == 200, response.text
    delta = before - await _balance(clients)
    assert delta == (1780 if charged else 0)
    assert response.json()["result"] == result


@pytest.mark.parametrize('endpoint,given,query,body,field,data,credits', [
    ('people.email.find', {'full_name': 'Example Person', 'domain': 'example.com'},
     {'first_name': 'Example', 'last_name': 'Person', 'company_url': 'example.com'}, None,
     'email', {'email': 'person@example.com'}, 1),
    ('people.phone.find', {'linkedin_url': 'https://linkedin.com/in/example'},
     {'linkedin_url': 'https://linkedin.com/in/example'}, None,
     'phone', {'employee_phone': '+15550101000'}, 1),
    ('people.enrich', {'email': 'person@example.com'}, {'email': 'person@example.com'}, None,
     'full_name', {'first_name': 'Example', 'last_name': 'Person'}, 1),
    ('people.search', {'company_domain': 'example.com', 'title': 'CEO', 'country': 'us', 'limit': 2}, {},
     {'company_url': {'include': ['example.com'], 'exclude': []},
      'title': {'include': ['CEO'], 'exclude': []}, 'country_code': {'include': ['US'], 'exclude': []}, 'per_page': 2},
     'people', [{'first_name': 'Example', 'has_email': True}], 0),
    ('companies.search', {'domain': 'example.com', 'limit': 2}, {},
     {'company_url': 'example.com', 'per_page': 2},
     'companies', [{'company_name': 'Example'}], 1),
])
async def test_routed_enrichment_adapter_requests_and_usage(
        clients, platform_on, monkeypatch, endpoint, given, query, body, field, data, credits):
    monkeypatch.setenv('TREG_PLATFORM_KEY_QUICKENRICH', 'PLATFORM-QUICKENRICH-KEY')
    monkeypatch.setenv('TREG_PLATFORM_PROVIDERS', 'quickenrich')
    get_settings.cache_clear()
    cat = catalog_store.load()
    child = 'quickenrich.' + endpoint
    parent = 'treg.' + cat.by_id[child]['capability']
    assert child in cat.by_id[parent]['routed_children']
    raw = {'success': True, 'data': data, 'meta': {'credits_used': credits, 'next_cursor': 'next'}}
    seen = []
    monkeypatch.setattr(call_service, 'relay', _relay_by_provider({'quickenrich': [(200, raw)]}, seen))
    before = await _balance(clients)
    response = await clients.post('/call/' + parent, json=given)
    assert response.status_code == 200, response.text
    result = response.json()
    assert result['_treg']['served_by'] == child
    assert result['output'][field]
    assert result['raw'] == raw
    assert seen[0][2:] == (query, body)
    assert before - await _balance(clients) == credits * 4834
    if endpoint == 'people.search':
        assert 'email' not in result['output']['people'][0]
        assert result['output']['next_cursor'] == 'next'


@pytest.mark.parametrize('value', [None, '', 'N/A', ' n/a ', 'null', 'none'])
@pytest.mark.parametrize('endpoint,field', [('people.email.find', 'email'), ('people.phone.find', 'employee_phone')])
def test_contact_adapters_reject_empty_markers(endpoint, field, value):
    ad = catalog_store.load().adapters['quickenrich.' + endpoint]
    assert ad.is_miss({'success': True, 'data': {field: value}})
    assert ad.is_miss({'success': True, 'data': []})
    assert not ad.is_miss({'success': True, 'data': {field: 'contact-value'}})


def test_quickenrich_paid_adapter_fixtures_verify_success_outputs():
    cat = catalog_store.load()
    expected = {
        'quickenrich.people.email.find': ('email', 'person@example.com'),
        'quickenrich.people.phone.find': ('phone', '+15550101000'),
        'quickenrich.people.enrich': ('full_name', 'Example Person'),
    }
    for endpoint_id, (field, value) in expected.items():
        adapter = cat.adapters[endpoint_id]
        assert adapter.verified is True
        assert adapter.verify_note == ''
        example = json.loads(
            (Path(__file__).resolve().parents[1] / 'src/treg/catalog/examples'
             / cat.by_id[endpoint_id]['example_file']).read_text()
        )
        assert not adapter.is_miss(example)
        assert adapter.from_upstream(example)[field] == value


def test_search_adapters_preserve_filters_and_fixed_page_quote():
    from treg.domain.catalog.routing.contracts import adapter_accepts
    cat = catalog_store.load()
    ad = cat.adapters['quickenrich.people.search.domain']
    for title, expected in [(None, 4834), ('CEO', 96680)]:
        given = {'company_domain': 'example.com', 'limit': 1}
        if title:
            given['title'] = title
        ident, _ = canonical_identity(cat.contracts['people.search'], given)
        q, b = ad.to_upstream(ident, adapter_accepts(ad, ident))
        assert q == {'company_url': 'example.com', **({'title': title} if title else {})}
        assert b == {}
        assert cost_at(cat.cost_view(cat.by_id[ad.endpoint_id]['cost'], 'quickenrich'), ident, ad) == expected
    assert 'quickenrich.people.search.domain' in cat.by_id['treg.people.search']['routed_children']
    reverse = cat.adapters['quickenrich.people.enrich']
    assert adapter_accepts(reverse, {'linkedin_url': 'https://linkedin.com/in/example'}) is None
    discovery = cat.adapters['quickenrich.people.search']
    q, b = discovery.to_upstream({'title': 'CEO', 'limit': 3}, ('title',))
    assert b == {'title': {'include': ['CEO'], 'exclude': []}, 'per_page': 3}
    company = cat.adapters['quickenrich.companies.search']
    q, b = company.to_upstream({'industry': 'Software', 'country': 'us', 'limit': 3}, ('industry',))
    assert b == {'industry': {'include': ['Software'], 'exclude': []},
                 'country_code': {'include': ['US'], 'exclude': []}, 'per_page': 3}
    assert adapter_accepts(company, {'technology': 'Python'}) is None


async def test_routed_fixed_page_price_respects_ceiling(clients, platform_on, monkeypatch):
    monkeypatch.setenv('TREG_PLATFORM_KEY_QUICKENRICH', 'PLATFORM-QUICKENRICH-KEY')
    monkeypatch.setenv('TREG_PLATFORM_PROVIDERS', 'quickenrich')
    get_settings.cache_clear()
    seen = []
    # Free discovery misses. The paid title search needs a 20-credit ceiling, even with limit=1.
    miss = {'success': True, 'data': [], 'meta': {'credits_used': 0}}
    monkeypatch.setattr(call_service, 'relay', _relay_by_provider({'quickenrich': [(200, miss)]}, seen))
    response = await clients.post('/call/treg.people.search',
        json={'company_domain': 'example.com', 'title': 'CEO', 'limit': 1},
        headers={'X-Treg-Route-Max-Cost': '0.01'})
    assert response.status_code == 200, response.text
    assert len(seen) == 1 and seen[0][1] == 'POST'


async def test_routed_discovery_miss_tries_domain_search(clients, platform_on, monkeypatch):
    monkeypatch.setenv('TREG_PLATFORM_KEY_QUICKENRICH', 'PLATFORM-QUICKENRICH-KEY')
    monkeypatch.setenv('TREG_PLATFORM_PROVIDERS', 'quickenrich')
    get_settings.cache_clear()
    seen = []
    miss = {'success': True, 'data': [], 'meta': {'credits_used': 0}}
    hit = {'success': True, 'data': [{'first_name': 'Example', 'email': 'person@example.com'}],
           'meta': {'credits_used': 1}}
    monkeypatch.setattr(call_service, 'relay', _relay_by_provider({'quickenrich': [(200, miss), (200, hit)]}, seen))
    before = await _balance(clients)
    response = await clients.post('/call/treg.people.search', json={'company_domain': 'example.com', 'limit': 1})
    assert response.status_code == 200, response.text
    assert response.json()['_treg']['served_by'] == 'quickenrich.people.search.domain'
    assert [s[1] for s in seen] == ['POST', 'GET']
    assert before - await _balance(clients) == 4834


async def test_routed_contact_miss_uses_next_provider(clients, enrichment_on, monkeypatch):
    monkeypatch.setenv('TREG_PLATFORM_KEY_QUICKENRICH', 'PLATFORM-QUICKENRICH-KEY')
    monkeypatch.setenv('TREG_PLATFORM_PROVIDERS', 'quickenrich,tomba')
    get_settings.cache_clear()
    seen = []
    monkeypatch.setattr(call_service, 'relay', _relay_by_provider({
        'quickenrich': [(200, {'success': True, 'data': {'email': 'N/A'}, 'meta': {'credits_used': 0}})],
        'tomba': [(200, {'data': {'email': 'person@example.com', 'verification': {'status': 'valid'}}})],
    }, seen))
    before = await _balance(clients)
    response = await clients.post('/call/treg.people.email.find',
        json={'first_name': 'Example', 'last_name': 'Person', 'domain': 'example.com'})
    assert response.status_code == 200, response.text
    assert response.json()['_treg']['served_by'] == 'tomba.people.email.find'
    assert [s[0] for s in seen] == ['quickenrich', 'tomba']
    assert before - await _balance(clients) == 8900


@pytest.mark.parametrize('expression,expected', [('0', 0), ('2', 9668), ('-1', None), ('true', None), ("'2'", None)])
def test_adapter_unit_quote_requires_nonnegative_integer(expression, expected):
    from dataclasses import replace
    ad = replace(catalog_store.load().adapters['quickenrich.people.search.domain'], cost_units=expression)
    assert cost_at({'usd': 0.004834}, {}, ad) == expected


@pytest.mark.parametrize('endpoint,given,expected', [
    ('people.email.find', {'first_name': 'Example', 'last_name': 'Person', 'domain': 'example.com'}, 4834),
    ('people.phone.find', {'linkedin_url': 'https://linkedin.com/in/example'}, 4834),
    ('people.enrich', {'email': 'person@example.com'}, 4834),
    ('people.search', {'company_domain': 'example.com', 'limit': 10}, 0),
    ('people.search.domain', {'company_domain': 'example.com', 'limit': 1}, 4834),
    ('people.search.domain', {'company_domain': 'example.com', 'title': 'CEO', 'limit': 1}, 96680),
    ('companies.search', {'domain': 'example.com'}, 48340),
    ('companies.search', {'domain': 'example.com', 'limit': 1}, 4834),
    ('companies.search', {'domain': 'example.com', 'limit': 100}, 483400),
])
def test_enrichment_route_quote_matches_direct_reservation(endpoint, given, expected):
    from treg.application.call.resolve import _marketplace_pricing
    from treg.domain.catalog.routing.contracts import adapter_accepts
    cat = catalog_store.load()
    ep = cat.by_id['quickenrich.' + endpoint]
    ad = cat.adapters[ep['id']]
    ident, _ = canonical_identity(cat.contracts[ep['capability']], given)
    query, body = ad.to_upstream(ident, adapter_accepts(ad, ident))
    cost = cat.cost_view(ep['cost'], ep['provider'])
    direct, _ = _marketplace_pricing(ep['provider'], ep['id'], cost, query, json.dumps(body).encode())
    assert direct == cost_at(cost, ident, ad) == expected


@pytest.mark.parametrize('cap,identity,doc,expected',[
    ('find',{'first_name':'Erol','last_name':'Toker','domain':'trykitt.ai'}, {'email':'erol@trykitt.ai','validity':'valid','credits':{'jobCredits':.005}},5000),
    ('verify',{'email':'erol@trykitt.ai'}, {'validity':'unknown','credits':{'jobCredits':.0015}},1500),
])
async def test_trykitt_routed_calls(clients,monkeypatch,kitt_on,cap,identity,doc,expected):
    seen=[]
    monkeypatch.setattr(call_service,'relay',_relay_by_provider({'trykitt':[(200,doc)]},seen))
    before=await _balance(clients)
    r=await clients.post('/call/treg.people.email.'+cap,json=identity)
    assert r.status_code==200,r.text
    assert seen[0][3]['realtime'] is True
    if cap=='find':
        assert seen[0][3]['fullName']=='Erol Toker'
        assert r.json()['output']['verified'] is True
    else: assert r.json()['output']['status']=='unknown'
    assert await _balance(clients)==before-expected



async def test_trykitt_throttle_releases_and_routes_to_next_provider(clients,monkeypatch,kitt_on):
    monkeypatch.setenv('TREG_PLATFORM_PROVIDERS','trykitt,leadmagic')
    monkeypatch.setenv('TREG_PLATFORM_KEY_LEADMAGIC','TEST-LEADMAGIC')
    get_settings.cache_clear()
    seen=[]
    monkeypatch.setattr(call_service,'relay',_relay_by_provider({'trykitt':[(418,{'message': 'temporarily throttled', 'response_code': 418})],'leadmagic':[(200,{'email':'a@example.com','status':'valid','credits_consumed':1})]},seen))
    before=await _balance(clients)
    r=await clients.post('/call/treg.people.email.find',json={'full_name':'A B','domain':'example.com'})
    assert r.status_code==200,r.text
    assert [row[0] for row in seen]==['trykitt','leadmagic']
    assert await _balance(clients)==before-25000


@pytest.mark.parametrize("verdict", ["valid", "invalid", "accept_all", "disposable", "unknown"])
async def test_contactout_routed_verification_preserves_verdict_and_is_free(
    clients, contactout_platform, monkeypatch, verdict,
):
    cat = catalog_store.load()
    assert cat.adapters["contactout.people.email.verify"].verified
    assert "contactout.people.email.verify" in cat.by_id["treg.people.email.verify"]["routed_children"]
    payload = {"status_code": 200, "data": {"status": verdict}}
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider({"contactout": [(200, payload)]}, seen))
    before = await _balance(clients)
    response = await clients.post("/call/treg.people.email.verify", json={"email": "person+tag@example.test"})
    assert response.status_code == 200, response.text
    doc = response.json()
    assert doc["output"] == {"valid": verdict == "valid", "status": verdict}
    assert doc["raw"] == payload
    assert doc["_treg"]["served_by"] == "contactout.people.email.verify"
    assert doc["_treg"]["outcome"] == "hit"
    assert seen == [("contactout", "GET", {"email": "person+tag@example.test"}, None)]
    assert int(response.headers["X-Treg-Cost-Micro"]) == 0
    assert before == await _balance(clients)
    async with session_maker() as db:
        assert not (await db.execute(select(Hold))).scalars().all()
        entries = (await db.execute(select(LedgerEntry).where(LedgerEntry.kind.in_(["reserve", "settle"])))).scalars().all()
        assert all(e.amount_micro == 0 for e in entries)


@pytest.mark.parametrize("payload", [
    {}, {"status_code": 200, "data": {}}, {"status_code": 200, "data": {"status": ""}},
    {"status_code": 403, "message": "No access", "data": {"status": "valid"}},
])
async def test_contactout_verifier_missing_verdict_and_embedded_errors_fall_back(
    clients, enrichment_on, monkeypatch, payload,
):
    monkeypatch.setenv("TREG_PLATFORM_KEY_CONTACTOUT", "PLATFORM-TEST")
    monkeypatch.setenv("TREG_PLATFORM_PROVIDERS", "contactout,tomba")
    get_settings.cache_clear()
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider({
        "contactout": [(200, payload)],
        "tomba": [(200, {"data": {"email": {"status": "invalid", "score": 0}}})],
    }, seen))
    response = await clients.post("/call/treg.people.email.verify", json={"email": "person@example.test"})
    assert response.status_code == 200, response.text
    assert [r[0] for r in seen] == ["contactout", "tomba"]
    assert response.json()["_treg"]["served_by"] == "tomba.people.email.verify"
    assert response.json()["output"]["status"] == "invalid"


async def test_own_verifier_key_precedes_free_contactout_platform_candidate(
    clients, contactout_platform, monkeypatch,
):
    await clients.post('/secrets', json={'name': 'hunter', 'value': 'OWN-TEST'})
    seen = []
    monkeypatch.setattr(call_service, 'relay', _relay_by_provider({
        'hunter': [(200, {'data': {'status': 'valid'}})],
    }, seen))
    before = await _balance(clients)
    response = await clients.post('/call/treg.people.email.verify', json={'email': 'person@example.test'})
    assert response.status_code == 200, response.text
    assert response.json()['_treg']['served_by'] == 'hunter.people.email.verify'
    assert [row[0] for row in seen] == ['hunter']
    assert before == await _balance(clients)


@pytest.mark.parametrize('value,expected', [
    ({'a.example': {'name': 'A'}, 'b.example': {'name': 'B'}}, [{'name': 'A'}, {'name': 'B'}]),
    ([{'name': 'A'}], [{'name': 'A'}]), ({}, []), ([], []), (None, None), ('bad', None),
])
def test_row_values_and_nested_lookup_expressions(value, expected):
    assert P.evaluate('values(rows)', {'rows': value}) == expected
    assert P.evaluate("get(values(rows), '[0].name')", {'rows': value}) == (
        'A' if expected else None)


_CONTACTOUT_DISCOVERY = [
    ('people.email.find', 'people.contact.work',
     {'linkedin_url': 'https://www.linkedin.com/in/example'},
     'GET', {'profile': 'https://www.linkedin.com/in/example',
             'email_type': 'work', 'include_phone': False}, None,
     {'status_code': 200, 'profile': {'work_email': ['work@example.test']}},
     'email', 'work@example.test', 150_000),
    ('people.phone.find', 'people.contact.phone',
     {'linkedin_url': 'https://www.linkedin.com/in/example'},
     'GET', {'profile': 'https://www.linkedin.com/in/example',
             'email_type': 'none', 'include_phone': True}, None,
     {'status_code': 200, 'profile': {'phone': ['+10000000000']}},
     'phone', '+10000000000', 250_000),
    ('companies.search', 'companies.search', {'domain': 'example.test'},
     'POST', {}, {'domain': ['example.test']},
     {'status_code': 200, 'companies': [{'name': 'Example'}]}, 'companies', [{'name': 'Example'}], 20_000),
    ('companies.enrich', 'companies.enrich', {'domain': 'example.test'},
     'POST', {}, {'domains': ['example.test']},
     {'status_code': 200, 'companies': {'example.test': {'name': 'Example', 'domain': 'example.test'}}},
     'name', 'Example', 20_000),
]


@pytest.mark.parametrize('cap,child,identity,method,query,body,payload,field,expected,charge', _CONTACTOUT_DISCOVERY)
async def test_contactout_discovery_routes_preserve_selectors_and_settle(
    clients, contactout_platform, monkeypatch,
    cap, child, identity, method, query, body, payload, field, expected, charge,
):
    cat = catalog_store.load()
    eid = 'contactout.' + child
    assert cat.adapters[eid].verified and not cat.adapters[eid].verify_note
    assert eid in cat.by_id['treg.' + cap]['routed_children']
    seen = []
    monkeypatch.setattr(call_service, 'relay', _relay_by_provider({'contactout': [(200, payload)]}, seen))
    before = await _balance(clients)
    response = await clients.post('/call/treg.' + cap, json=identity)
    assert response.status_code == 200, response.text
    doc = response.json()
    assert doc['raw'] == payload and doc['output'][field] == expected
    assert doc['_treg']['served_by'] == eid and doc['_treg']['outcome'] == 'hit'
    assert seen == [('contactout', method, query, body)]
    assert int(response.headers['X-Treg-Cost-Micro']) == charge
    assert before - await _balance(clients) == charge
    async with session_maker() as db:
        assert not (await db.execute(select(Hold))).scalars().all()


@pytest.mark.parametrize('cap,child,identity,method,query,body,payload,field,expected,charge', _CONTACTOUT_DISCOVERY)
@pytest.mark.parametrize('failed', [False, True])
async def test_contactout_discovery_empty_or_error_response_is_not_a_hit(
    clients, contactout_platform, monkeypatch,
    cap, child, identity, method, query, body, payload, field, expected, charge, failed,
):
    payload = {'status_code': 403, **{k: v for k, v in payload.items() if k != 'status_code'}} if failed else {'status_code': 200}
    seen = []
    monkeypatch.setattr(call_service, 'relay', _relay_by_provider({'contactout': [(200, payload)]}, seen))
    before = await _balance(clients)
    response = await clients.post('/call/treg.' + cap, json=identity)
    assert response.status_code == 200, response.text
    assert response.json()['_treg']['outcome'] == 'miss'
    assert before == await _balance(clients)
    async with session_maker() as db:
        assert not (await db.execute(select(Hold))).scalars().all()


def test_contactout_unverified_pii_routes_stay_direct_only():
    cat = catalog_store.load()
    for cap, child in [('people.search', 'people.search'), ('people.enrich', 'people.enrich'),
                       ('linkedin.user.profile', 'people.linkedin.enrich')]:
        eid = 'contactout.' + child
        assert eid not in cat.adapters
        assert eid not in cat.by_id['treg.' + cap]['routed_children']
    assert 'contactout.people.contact.personal' not in cat.adapters
    assert cat.by_id['contactout.people.contact.personal']['platform'] == 'people'


# ---- regression: adapter exceptions after child success must not crash the parent (2026-09) ----


def _make_throwing_adapter(real, throw_on: str):
    """Create a wrapper adapter that throws on the specified method."""
    class ThrowingAdapter:
        def __init__(self, real):
            self._real = real
            # Copy ALL attributes from the real Adapter dataclass
            self.endpoint_id = real.endpoint_id
            self.accepts = real.accepts
            self.in_map = real.in_map
            self.out_map = real.out_map
            self.miss = real.miss
            self.const = getattr(real, 'const', {})
            self.in_expr = getattr(real, 'in_expr', {})
            self.body_array = getattr(real, 'body_array', False)
            self.test_identity = getattr(real, 'test_identity', {})
            self.cost_units = getattr(real, 'cost_units', '')
            self.additional_capabilities = getattr(real, 'additional_capabilities', ())
            self.verified_capabilities = getattr(real, 'verified_capabilities', ())
            self.verified = real.verified
            self.verify_note = getattr(real, 'verify_note', '')
            self._filter_keys = getattr(real, '_filter_keys', ())

        def to_upstream(self, identity, variant):
            if throw_on == 'to_upstream':
                raise KeyError("simulated to_upstream failure")
            return self._real.to_upstream(identity, variant)

        def from_upstream(self, provider_body):
            if throw_on == 'from_upstream':
                raise ValueError("simulated from_upstream failure")
            return self._real.from_upstream(provider_body)

        def is_miss(self, provider_body):
            if throw_on == 'is_miss':
                raise TypeError("simulated is_miss failure")
            return self._real.is_miss(provider_body)

    return ThrowingAdapter(real)


def _patched_catalog_with_throwing_adapter(original_cat, endpoint_id: str, throw_on: str):
    """Return a new Catalog with one adapter replaced by a throwing wrapper."""
    from dataclasses import replace
    new_adapters = dict(original_cat.adapters)
    new_adapters[endpoint_id] = _make_throwing_adapter(original_cat.adapters[endpoint_id], throw_on)
    return replace(original_cat, adapters=new_adapters)


def _patched_catalog_all_email_find_throw(original_cat):
    """Return a new Catalog where all email.find adapters throw on from_upstream."""
    from dataclasses import replace
    new_adapters = {}
    for eid, adapter in original_cat.adapters.items():
        if 'email.find' in eid:
            new_adapters[eid] = _make_throwing_adapter(adapter, 'from_upstream')
        else:
            new_adapters[eid] = adapter
    return replace(original_cat, adapters=new_adapters)


async def test_adapter_from_upstream_throws_after_child_200_waterfall_continues(
    clients: AsyncClient, enrichment_on, monkeypatch,
):
    """Regression for 2026-09 bug: adapter.from_upstream throwing after a child returned 200 used
    to crash the parent with a bare 502, leaving children audited OK but parent failed. Now the
    adapter failure is recorded as an error and the waterfall continues to the next provider."""
    original_cat = catalog_store.load()
    # Patch tomba's adapter to throw (tomba is first in price order for this identity)
    patched_cat = _patched_catalog_with_throwing_adapter(original_cat, "tomba.people.email.find", "from_upstream")

    monkeypatch.setattr(catalog_store, 'load', lambda: patched_cat)

    seen = []
    # Tomba returns 200 but adapter throws; hunter returns 200 and works fine
    # '*' catches other providers in waterfall (findymail, etc) returning miss
    monkeypatch.setattr(call_service, 'relay', _relay_by_provider({
        '*': [(200, {'data': None})] * 10,  # Other providers return miss-like response
        'tomba': [(200, {'data': {'email': 'bad@format.test', 'unexpectedField': True}})],
        'hunter': [(200, {'data': {'email': 'found@example.test', 'score': 80, 'verification': {'status': 'valid'}}})],
    }, seen))

    r = await clients.post(f'/call/{ROUTED}', json={'full_name': 'Example Person', 'domain': 'example.com'})
    assert r.status_code == 200, r.text
    doc = r.json()
    # Waterfall continued to hunter after tomba's adapter failed
    assert doc['_treg']['served_by'] == 'hunter.people.email.find'
    assert doc['_treg']['outcome'] == 'hit'
    # The tomba error should be recorded in `tried`
    tried = {t['endpoint_id']: t for t in doc['_treg']['tried']}
    assert 'tomba.people.email.find' in tried
    assert tried['tomba.people.email.find']['outcome'] == 'error'
    assert 'adapter.from_upstream failed' in tried['tomba.people.email.find']['detail']
    # Hunter succeeded
    assert tried['hunter.people.email.find']['outcome'] == 'hit'


async def test_adapter_throws_on_all_children_returns_structured_502_with_tried(
    clients: AsyncClient, enrichment_on, monkeypatch,
):
    """When ALL adapters throw on 200 responses, the parent must return a structured 502
    with proper error details and `tried` list - not a bare 500 or empty 502."""
    original_cat = catalog_store.load()
    patched_cat = _patched_catalog_all_email_find_throw(original_cat)

    monkeypatch.setattr(catalog_store, 'load', lambda: patched_cat)

    seen = []
    # All providers return 200 but adapters throw
    # '*' wildcard catches all providers - return data that adapters will parse
    monkeypatch.setattr(call_service, 'relay', _relay_by_provider({
        '*': [(200, {'data': {'email': 'x@test.test'}})] * 15,
    }, seen))

    r = await clients.post(f'/call/{ROUTED}', json={'full_name': 'Example Person', 'domain': 'example.com'})
    # Should be 502 route_failed, not 500 or empty body
    assert r.status_code == 502, r.text
    doc = r.json()
    assert doc['detail']['error'] == 'route_failed'
    # The tried list should have some error outcomes (adapters that threw)
    tried = doc['detail']['tried']
    assert len(tried) > 0
    error_outcomes = [t for t in tried if t['outcome'] == 'error']
    # At least one adapter should have thrown (those with email.find in name)
    assert len(error_outcomes) > 0, f"Expected at least one error outcome, got: {tried}"
    # Check that error details mention adapter failure
    for t in error_outcomes:
        if 'detail' in t and t['detail']:
            assert 'adapter' in t['detail'] or 'failed' in t['detail'], f"Unexpected error detail: {t}"


async def test_adapter_to_upstream_throws_records_error_and_continues(
    clients: AsyncClient, enrichment_on, monkeypatch,
):
    """If adapter.to_upstream throws (before the child call), the error is recorded
    and the waterfall continues to the next candidate."""
    original_cat = catalog_store.load()
    # Patch tomba's adapter to throw on to_upstream
    patched_cat = _patched_catalog_with_throwing_adapter(original_cat, "tomba.people.email.find", "to_upstream")

    monkeypatch.setattr(catalog_store, 'load', lambda: patched_cat)

    seen = []
    # Tomba's to_upstream will throw before relay is called
    # '*' catches other providers, returning miss-like response
    monkeypatch.setattr(call_service, 'relay', _relay_by_provider({
        '*': [(200, {'data': None})] * 10,  # Other providers return miss-like response
        'hunter': [(200, {'data': {'email': 'found@example.test', 'score': 80, 'verification': {'status': 'valid'}}})],
    }, seen))

    r = await clients.post(f'/call/{ROUTED}', json={'full_name': 'Example Person', 'domain': 'example.com'})
    assert r.status_code == 200, r.text
    doc = r.json()
    # Tomba's to_upstream failed, waterfall continued to hunter
    assert doc['_treg']['served_by'] == 'hunter.people.email.find'
    tried = {t['endpoint_id']: t for t in doc['_treg']['tried']}
    assert 'tomba.people.email.find' in tried
    assert tried['tomba.people.email.find']['outcome'] == 'error'
    assert 'adapter.to_upstream failed' in tried['tomba.people.email.find']['detail']


async def test_adapter_is_miss_throws_records_error_and_continues(
    clients: AsyncClient, enrichment_on, monkeypatch,
):
    """If adapter.is_miss throws after parsing the response, the error is recorded
    and the waterfall continues."""
    original_cat = catalog_store.load()
    # Patch tomba's adapter to throw on is_miss
    patched_cat = _patched_catalog_with_throwing_adapter(original_cat, "tomba.people.email.find", "is_miss")

    monkeypatch.setattr(catalog_store, 'load', lambda: patched_cat)

    seen = []
    # '*' catches other providers, returning miss-like response
    monkeypatch.setattr(call_service, 'relay', _relay_by_provider({
        '*': [(200, {'data': None})] * 10,  # Other providers return miss-like response
        'tomba': [(200, {'data': {'email': 'tomba@test.test', 'score': 99}})],
        'hunter': [(200, {'data': {'email': 'found@example.test', 'score': 80, 'verification': {'status': 'valid'}}})],
    }, seen))

    r = await clients.post(f'/call/{ROUTED}', json={'full_name': 'Example Person', 'domain': 'example.com'})
    assert r.status_code == 200, r.text
    doc = r.json()
    # Tomba's is_miss failed, waterfall continued to hunter
    assert doc['_treg']['served_by'] == 'hunter.people.email.find'
    tried = {t['endpoint_id']: t for t in doc['_treg']['tried']}
    assert 'tomba.people.email.find' in tried
    assert tried['tomba.people.email.find']['outcome'] == 'error'
    assert 'adapter.is_miss failed' in tried['tomba.people.email.find']['detail']


async def test_prospeo_no_match_400_is_treated_as_miss_not_error(
    clients: AsyncClient, enrichment_with_miss_declarers_on, monkeypatch,
):
    """Prospeo returns 400 with error_code=NO_MATCH for 'no result' — this is a semantic miss,
    not a caller fault. The waterfall should continue and the parent should not 502."""
    seen = []
    # Prospeo returns 400 NO_MATCH (semantic miss), tomba returns 200 hit
    monkeypatch.setattr(call_service, 'relay', _relay_by_provider({
        '*': [(200, {'data': None})] * 10,  # Other providers miss
        'prospeo': [(400, {'error': True, 'error_code': 'NO_MATCH'})],
        'tomba': [(200, {'data': {'email': 'found@example.test', 'score': 99, 'verification': {'status': 'valid'}}})],
    }, seen))

    r = await clients.post(f'/call/{ROUTED}', json={'full_name': 'Example Person', 'domain': 'example.com'},
                           headers={'X-Treg-Route-Prefer': 'prospeo'})  # ask it first; otherwise tomba's hit ends the waterfall before it runs
    assert r.status_code == 200, r.text
    doc = r.json()
    # Waterfall continued past Prospeo's NO_MATCH
    assert doc['_treg']['outcome'] == 'hit'
    tried = {t['endpoint_id']: t for t in doc['_treg']['tried']}
    # Prospeo should be recorded as miss, not error
    prospeo_attempts = [t for t in doc['_treg']['tried'] if t['provider'] == 'prospeo']
    assert prospeo_attempts, doc['_treg']['tried']
    for attempt in prospeo_attempts:
        assert attempt['outcome'] == 'miss', f"Prospeo NO_MATCH should be miss, not {attempt['outcome']}"


async def test_limadata_404_is_treated_as_miss_not_error(
    clients: AsyncClient, enrichment_with_miss_declarers_on, monkeypatch,
):
    """LimaData returns 404 for 'no email found' — with the miss status declared, this should
    be treated as a miss and the waterfall should continue."""
    seen = []
    # LimaData returns 404 (declared miss), tomba returns 200 hit
    monkeypatch.setattr(call_service, 'relay', _relay_by_provider({
        '*': [(200, {'data': None})] * 10,  # Other providers miss
        'limadata': [(404, {})],
        'tomba': [(200, {'data': {'email': 'found@example.test', 'score': 99, 'verification': {'status': 'valid'}}})],
    }, seen))

    r = await clients.post(f'/call/{ROUTED}', json={'full_name': 'Example Person', 'domain': 'example.com'},
                           headers={'X-Treg-Route-Prefer': 'limadata'})  # ask it first; otherwise tomba's hit ends the waterfall before it runs
    assert r.status_code == 200, r.text
    doc = r.json()
    # Waterfall continued past LimaData's 404
    assert doc['_treg']['outcome'] == 'hit'
    tried = {t['endpoint_id']: t for t in doc['_treg']['tried']}
    # LimaData should be recorded as miss, not error
    limadata_attempts = [t for t in doc['_treg']['tried'] if t['provider'] == 'limadata']
    assert limadata_attempts, doc['_treg']['tried']
    for attempt in limadata_attempts:
        assert attempt['outcome'] == 'miss', f"LimaData 404 should be miss, not {attempt['outcome']}"


# ---- miss.when: one status, two meanings (prospeo 400 NO_MATCH vs INVALID_DATAPOINTS) ----


def test_declared_miss_honours_when_predicate_and_never_crashes():
    from treg.application.call import route as call_route
    ep = {"id": "x", "miss": {"status": 400, "when": "error_code == 'NO_MATCH'", "means": "no match"}}
    assert call_route._declared_miss(ep, 400, b'{"error":true,"error_code":"NO_MATCH"}')
    assert not call_route._declared_miss(ep, 400, b'{"error":true,"error_code":"INVALID_DATAPOINTS"}')
    assert not call_route._declared_miss(ep, 404, b'{"error":true,"error_code":"NO_MATCH"}')
    assert not call_route._declared_miss(ep, 400, b'["NO_MATCH"]')      # array body: no crash, not a miss
    assert not call_route._declared_miss(ep, 400, b'not json')
    plain = {"id": "y", "miss": {"status": 404, "means": "gone"}}
    assert call_route._declared_miss(plain, 404, b'Not Found')
    assert not call_route._declared_miss(plain, 400, b'')


def test_prospeo_and_limadata_person_finders_declare_their_miss():
    cat = catalog_store.load()
    from treg.domain.catalog.routing.contracts import declared_miss
    for eid in ("prospeo.people.email.find", "prospeo.people.phone.find", "prospeo.people.enrich"):
        ep = cat.by_id[eid]
        assert ep["miss"]["status"] == 400, eid
        # evaluate the predicate, not just its spelling: a misspelt path would silently never match
        assert declared_miss(ep, 400, {"error": True, "error_code": "NO_MATCH"}), eid
        assert not declared_miss(ep, 400, {"error": True, "error_code": "INVALID_DATAPOINTS"}), eid
        assert "when" not in catalog_store.endpoint_view(ep, "Prospeo", cat)["miss"], "internal predicate leaks to agents"
    for eid in ("limadata.people.email.find.name", "limadata.people.email.find.linkedin", "limadata.people.phone.find"):
        assert cat.by_id[eid]["miss"]["status"] == 404, eid


async def test_prospeo_invalid_datapoints_400_stays_a_vendor_fault(
    clients: AsyncClient, enrichment_with_miss_declarers_on, monkeypatch,
):
    """The same 400 with a non-NO_MATCH body is a rejected request: recorded as an error, the
    waterfall goes on to free-on-failure providers, and the outcome is never a clean miss."""
    seen = []
    monkeypatch.setattr(call_service, 'relay', _relay_by_provider({
        '*': [(200, {'data': None})] * 10,
        'prospeo': [(400, {'error': True, 'error_code': 'INVALID_DATAPOINTS'})],
    }, seen))
    r = await clients.post(f'/call/{ROUTED}', json={'full_name': 'Example Person', 'domain': 'example.com'},
                           headers={'X-Treg-Route-Prefer': 'prospeo'})  # ask it first; otherwise tomba's hit ends the waterfall before it runs
    assert r.status_code == 502, r.text
    prospeo = [t for t in r.json()['detail']['tried'] if t['provider'] == 'prospeo']
    assert prospeo and all(t['outcome'] == 'error' for t in prospeo)


def test_linkedin_url_is_normalised_once_for_every_adapter():
    from treg.domain.catalog.routing import paths as P
    from treg.domain.catalog.routing.contracts import canonical_identity
    assert P.linkedin_url("linkedin.com/in/patrickcollison") == "https://linkedin.com/in/patrickcollison"
    assert P.linkedin_url("www.linkedin.com/in/patrickcollison/") == "https://www.linkedin.com/in/patrickcollison/"
    assert P.linkedin_url("https://www.linkedin.com/in/patrickcollison") == "https://www.linkedin.com/in/patrickcollison"
    assert P.linkedin_url("patrickcollison") == "https://www.linkedin.com/in/patrickcollison"
    contract = catalog_store.load().contracts["people.email.find"]
    ident, variant = canonical_identity(contract, {"linkedin_url": "linkedin.com/in/patrickcollison"})
    assert variant == ("linkedin_url",)
    assert ident["linkedin_url"] == "https://linkedin.com/in/patrickcollison"
    assert ident["linkedin_handle"] == "patrickcollison"


def test_linkedin_url_only_trusts_a_linkedin_host():
    from treg.domain.catalog.routing import paths as P
    # a path that merely mentions linkedin.com is a handle-shaped string, never promoted to that host
    assert P.linkedin_url("evil.example/?linkedin.com/in/x") == "https://www.linkedin.com/in/evil.example/?linkedin.com/in/x"
    assert P.linkedin_url("uk.linkedin.com/in/x") == "https://uk.linkedin.com/in/x"


def test_arena_and_router_read_the_miss_block_the_same_way():
    from treg.domain import arena
    cat = catalog_store.load()
    ep = cat.by_id["prospeo.people.email.find"]; ad = cat.adapters[ep["id"]]; contract = cat.contracts["people.email.find"]
    assert arena.classify(contract, ad, ep, 400, {"error": True, "error_code": "NO_MATCH"})[0] == "miss"
    assert arena.classify(contract, ad, ep, 400, {"error": True, "error_code": "INVALID_DATAPOINTS"})[0] == "error"


def test_linkedin_url_lowercases_the_host_so_the_handle_derives():
    from treg.domain.catalog.routing import paths as P
    assert P.linkedin_url("LinkedIn.com/in/Patrick") == "https://linkedin.com/in/Patrick"
    assert P.linkedin_handle(P.linkedin_url("WWW.LinkedIn.com/in/Patrick")) == "Patrick"


async def test_company_blind_search_provider_is_dropped_not_billed(clients: AsyncClient, platform_on, monkeypatch):
    """`people.search` declares `scoping: [company_domain]`. A title-only provider asked for
    `{company_domain, title}` answers with the same strangers for every company and bills them as
    a hit, so it leaves the plan instead of ranking last — and still serves a title-only search."""
    for p in ("LUSHA", "COMPANYENRICH"):
        monkeypatch.setenv(f"TREG_PLATFORM_KEY_{p}", f"PLATFORM-{p}-KEY")
    monkeypatch.setenv("TREG_PLATFORM_PROVIDERS", "lusha,companyenrich")
    get_settings.cache_clear()
    cat = catalog_store.load()
    for eid in cat.by_id["treg.people.search"]["routed_children"]:
        if eid not in ("lusha.people.search", "companyenrich.people.search"):
            monkeypatch.delitem(cat.adapters, eid, raising=False)
    seen = []
    monkeypatch.setattr(call_service, "relay", _relay_by_provider({
        "lusha": [(200, {"results": [{"name": "Some CEO"}], "pagination": {"total": 1}})],
        "companyenrich": [(200, {"items": [], "totalItems": 0})]}, seen))
    r = await clients.post("/call/treg.people.search", json={"company_domain": "example.com", "title": "CEO"})
    assert [s[0] for s in seen] == ["companyenrich"], seen
    assert r.status_code == 200 and r.json()["_treg"]["outcome"] == "miss", r.text
    assert any(d["endpoint_id"] == "lusha.people.search" and "company_domain" in d["why"]
               for d in r.json()["_treg"]["dropped"]), r.json()["_treg"]
    seen.clear()
    r = await clients.post("/call/treg.people.search", json={"title": "CEO"})
    assert [s[0] for s in seen] == ["lusha"] and r.json()["_treg"]["served_by"] == "lusha.people.search", r.text
    get_settings.cache_clear()