From 4e4d5f642b5489132de9a684d6f3742a4a79116c Mon Sep 17 00:00:00 2001 From: ernani Date: Sun, 26 Jul 2026 17:21:44 -0300 Subject: [PATCH] feat(catalog): aggregator correlation tools + e2e cross-provider proof (Graph P4) Phase 2b/4: expose cross-provider correlation as agent-facing, hidden-but-callable tools on CatalogMcpSurface, plus the offline $0 falsifiable proof. - correlate_surfaces(a, b): resolve two providers via resolve_provider, delegate to the shared Phase-2 correlate.correlate_surfaces (same confirmed-gate scorer). - find_correlations(entity) / value_domain_index: the cross-provider "what correlates with X" query, reusing Phase-3 vindex. - Both getattr-guarded for duck-typed clients, control-plane clean, NOT enumerated in list_tools (projection stays byte-identical), callable by name. - tests/test_e2e_correlation.py proves the full chain THROUGH the aggregator: no DECLARED hint -> candidate; one customer-confirmed entity -> plan-eligible + grouped + cross_plan recovers first-plan-correct; provider-only -> stays a candidate (confirmed-gate holds to the executable plan). Co-Authored-By: Claude Opus 4.8 --- gecko/catalog_mcp.py | 68 +++++++- tests/test_e2e_correlation.py | 307 ++++++++++++++++++++++++++++++++++ 2 files changed, 372 insertions(+), 3 deletions(-) create mode 100644 tests/test_e2e_correlation.py diff --git a/gecko/catalog_mcp.py b/gecko/catalog_mcp.py index c5eb857..1ac0f9b 100644 --- a/gecko/catalog_mcp.py +++ b/gecko/catalog_mcp.py @@ -139,16 +139,66 @@ def resolve_provider(self, name: str) -> ProviderSurface | None: ] return matches[0] if len(matches) == 1 else None + def _surfaces(self) -> list[Any]: + """Wrap every connected provider client that carries a call graph as a Surface. + + Getattr-guarded (guardrail for duck-typed clients): a registry client with no + ``surface_graph`` — a test fake, a not-yet-comprehended stub — is skipped rather + than raised on, so the aggregator degrades gracefully instead of crashing the + whole cross-provider query for one odd provider.""" + from .surface import Surface + + return [ + Surface.of(ps.client) + for ps in self.registry.providers() + if getattr(ps.client, "surface_graph", None) is not None + ] + # -- the value-domain index over every connected provider (§13 Phase 3.3) ----- def value_domain_index(self) -> Any: """The ``entity -> {producers, consumers}`` index across every connected provider — the cross-provider "what correlates with X" lookup. Deterministic, control-plane only. Returns a :class:`~gecko.vindex.ValueDomainIndex`.""" - from .surface import Surface from .vindex import value_domain_index - surfaces = [Surface.of(ps.client) for ps in self.registry.providers()] - return value_domain_index(surfaces) + return value_domain_index(self._surfaces()) + + # -- the aggregator correlation tools (§13 Phase 2b / Phase 4) ----------------- + def correlate_surfaces(self, a: str, b: str) -> dict[str, Any]: + """Correlate two named/resolved providers in THIS workspace — which of A's + outputs feed B's inputs, each with a provenance-carrying basis and whether it is + plan-eligible now or a quarantined candidate. + + Resolves both names through :meth:`resolve_provider` (deterministic, no fuzzy + guessing) and delegates to the shared Phase-2 ``correlate.correlate_surfaces`` — + the SAME confirmed-gate scorer as everywhere else, so a provider-only declaration + can never reach a plan-eligible cross-API join through this door. Returns the + control-plane-clean ``CorrelationResult.to_dict()`` (structure only, never a + payload). Raises :class:`~gecko.caller.CallError` on an unknown/ambiguous name.""" + from .correlate import correlate_surfaces as _correlate + from .surface import Surface + + ra = self.resolve_provider(a) + rb = self.resolve_provider(b) + unresolved = [n for n, r in ((a, ra), (b, rb)) if r is None] + if unresolved: + raise CallError(f"unknown or ambiguous provider(s): {unresolved!r}") + assert ra is not None and rb is not None # narrowed by the guard above + return _correlate(Surface.of(ra.client), Surface.of(rb.client)).to_dict() + + def find_correlations(self, entity: str) -> dict[str, Any]: + """The cross-provider query — "what correlates with ``entity`` across every + connected provider." Groups the producers/consumers of the DECLARED value-domain + entity and reports each cross-provider join tagged plan-eligible vs candidate via + the SHARED Phase-2 scorer. Control-plane clean: names + entity + provenance only, + never a response value.""" + idx = self.value_domain_index() + group = idx.group(entity) + return { + "entity": entity, + "group": group.to_dict() if group is not None else None, + "cross_joins": [j.to_dict() for j in idx.find_correlations(entity)], + } def call_tool( self, name: str, arguments: dict[str, Any], session_id: str | None = None @@ -161,6 +211,18 @@ def call_tool( return self.search_capabilities(str(args.get("query", ""))) if name == "get_capability": return self.get_capability(str(args.get("name", ""))) + # The cross-provider correlation doors (hidden: callable by name, NOT enumerated + # in list_tools — the projection stays byte-identical). Siblings of get_capability: + # dispatched by name, resolved in the package (correlate/vindex), control-plane + # only, and never reaching an upstream call. + if name == "correlate_surfaces": + return self.correlate_surfaces( + str(args.get("a", "")), str(args.get("b", "")) + ) + if name == "find_correlations": + return self.find_correlations(str(args.get("entity", ""))) + if name == "value_domain_index": + return self.value_domain_index().to_dict() ps = self.registry.by_tool(name) if ps is None: raise CallError(f"unknown tool: {name!r}") diff --git a/tests/test_e2e_correlation.py b/tests/test_e2e_correlation.py new file mode 100644 index 0000000..bdeacae --- /dev/null +++ b/tests/test_e2e_correlation.py @@ -0,0 +1,307 @@ +"""Phase 4 e2e falsifiable proof — the multi-API correlation system, end to end, +THROUGH the aggregator (``CatalogMcpSurface``), offline and $0 (Pattern B). + +The load-bearing chain, on a real 2-API pair sharing a ``solana-token-mint`` +value-domain (the Phase-2 producer/consumer fixtures, standing in for the +Birdeye x daily-news testbed): + +1. **No DECLARED hint** -> the cross-API token join is a **candidate**, NEVER + plan-eligible — even when name AND signature both match across the boundary + (the §13.6 gate), surfaced identically through ``correlate_surfaces`` and + ``find_correlations``. +2. **One customer-confirmed entity on both sides** -> it jumps to **tier 3, + plan-eligible**; ``value_domain_index`` groups BOTH providers under the entity; + and ``cross_plan`` recovers the cross-API chain **first-plan-correct**. +3. **A provider-only declaration** (untrusted x-gecko, no customer confirm) stays a + **candidate** — the confirmed-gate holds THROUGH the aggregator (an untrusted spec + can't mint a cross-system plan basis). + +Falsified if a bare-name/signature match reaches plan-eligible without a +customer-confirmed DECLARED entity, at any of the aggregator's doors. + +The aggregator tools are also asserted ABSENT from ``list_tools`` (the projection +stays byte-identical) yet callable by name. +""" + +from __future__ import annotations + +from typing import Any + +from gecko.caller import CallError +from gecko.catalog_mcp import CatalogMcpSurface +from gecko.client import AgentApiClient +from gecko.compose import Workspace, cross_plan +from gecko.paysh_catalog import CatalogEntry, CatalogRegistry, ProviderSurface +from gecko.surfaces import safe_surface_id + +# base58-ish discriminating pattern so BOTH name and signature match across the +# boundary — the hardest case for the §13.6 cross-API gate to hold. +_MINT_PATTERN = "^[1-9A-HJ-NP-Za-km-z]{32,44}$" +_ENTITY = "solanatokenmint" # normalized form of "solana-token-mint" +_CONFIRMED = {"tokenId": "solana-token-mint"} + + +def _producer_spec(*, declared: bool) -> dict[str, Any]: + """Birdeye-like: resolves a token to its mint id (a ``tokenId`` output field).""" + prop: dict[str, Any] = {"type": "string", "pattern": _MINT_PATTERN} + if declared: + prop["x-gecko-entity"] = "solana-token-mint" + return { + "openapi": "3.0.0", + "info": {"title": "Birdeye Token Resolver", "version": "1"}, + "servers": [{"url": "https://public-api.birdeye.so"}], + "paths": { + "/token/resolve": { + "get": { + "operationId": "resolveToken", + "summary": "Resolve a token to its mint id", + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "tokenId": prop, + "symbol": {"type": "string"}, + }, + } + } + } + } + }, + } + } + }, + } + + +def _consumer_spec(*, declared: bool) -> dict[str, Any]: + """daily-news-like: prices a token by ``tokenId`` (a required query param).""" + schema: dict[str, Any] = {"type": "string", "pattern": _MINT_PATTERN} + param: dict[str, Any] = { + "name": "tokenId", + "in": "query", + "required": True, + "schema": schema, + } + if declared: + param["x-gecko-entity"] = "solana-token-mint" + return { + "openapi": "3.0.0", + "info": {"title": "daily-news Token Prices", "version": "1"}, + "servers": [{"url": "https://prices.example.com"}], + "paths": { + "/price": { + "get": { + "operationId": "getPrice", + "summary": "Get the price of a token", + "parameters": [param], + "responses": { + "200": { + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": {"price": {"type": "number"}}, + } + } + } + } + }, + } + } + }, + } + + +def _entry(fqn: str, host: str) -> CatalogEntry: + return CatalogEntry( + fqn=fqn, + title=fqn.split("/")[-1].title(), + service_url=host, + description="", + use_case="", + category="finance", + endpoint_count=1, + min_price_usd=0.0, + max_price_usd=0.0, + has_free_tier=True, + has_metering=False, + sha=fqn, + ) + + +def _provider( + spec: dict[str, Any], fqn: str, host: str, *, hints: dict[str, str] | None = None +) -> ProviderSurface: + slug = safe_surface_id(fqn) + client = AgentApiClient( + spec, base_url=host, surface_id=slug, declared_hints=hints or {} + ) + return ProviderSurface( + entry=_entry(fqn, host), + client=client, + tool_name=slug, + host=host, + status="pending", + ) + + +def _registry(*providers: ProviderSurface) -> CatalogRegistry: + reg = CatalogRegistry() + for ps in providers: + reg._install(ps) + return reg + + +def _surface( + *, producer_declared: bool, consumer_declared: bool, hints: dict[str, str] | None +) -> CatalogMcpSurface: + """A 2-provider aggregator: a Birdeye-like producer + a daily-news-like consumer.""" + return CatalogMcpSurface( + _registry( + _provider( + _producer_spec(declared=producer_declared), + "birdeye/data", + "https://public-api.birdeye.so", + hints=hints, + ), + _provider( + _consumer_spec(declared=consumer_declared), + "dailynews/prices", + "https://prices.example.com", + hints=hints, + ), + ) + ) + + +def _token_link(result: dict[str, Any]) -> dict[str, Any]: + """The single tokenId->tokenId cross-API link from a correlate_surfaces result.""" + links = [ + lk + for lk in result["links"] + if lk["basis"]["src_field"] == "tokenId" + and lk["basis"]["dst_param"] == "tokenId" + and lk["basis"]["src_surface"] != lk["basis"]["dst_surface"] + ] + assert links, "expected a cross-API tokenId correlation link" + return links[0] + + +def _cross_joins(payload: dict[str, Any]) -> list[dict[str, Any]]: + return [ + j + for j in payload["cross_joins"] + if j["producer"]["surface"] != j["consumer"]["surface"] + ] + + +# --- PROOF 1: no DECLARED hint -> the cross-API join is a quarantined candidate ---- +def test_no_declared_hint_is_candidate_through_aggregator() -> None: + surface = _surface(producer_declared=False, consumer_declared=False, hints=None) + # both aggregator doors agree: name+signature match, but NEVER plan-eligible. + link = _token_link( + surface.call_tool( + "correlate_surfaces", {"a": "birdeye/data", "b": "dailynews/prices"} + ) + ) + assert ( + link["basis"]["tier"] <= 1 and not link["plan_eligible"] and link["candidate"] + ) + + # find_correlations: with NO declared vocab the entity is absent from the index — + # §13.6 forbids a name-only cross-provider bucket, so there is no join at all. + fc = surface.call_tool("find_correlations", {"entity": _ENTITY}) + assert fc["group"] is None and _cross_joins(fc) == [] + + +# --- PROOF 2: one confirmed entity -> tier 3, grouped, and cross_plan recovers ------ +def test_confirmed_entity_reaches_plan_eligible_and_cross_plan_recovers() -> None: + surface = _surface( + producer_declared=False, consumer_declared=False, hints=_CONFIRMED + ) + link = _token_link( + surface.call_tool( + "correlate_surfaces", {"a": "birdeye/data", "b": "dailynews/prices"} + ) + ) + assert link["basis"]["tier"] == 3 and link["plan_eligible"] + assert link["basis"]["provenance"] == "DECLARED" + + # value_domain_index groups BOTH providers under the shared entity. + vdi = surface.call_tool("value_domain_index", {}) + ent = next(e for e in vdi["entities"] if e["entity"] == _ENTITY) + assert {p["surface"] for p in ent["producers"]} == {"birdeye-data"} + assert {c["surface"] for c in ent["consumers"]} == {"dailynews-prices"} + + # find_correlations reports the plan-eligible cross-provider join. + fc = surface.call_tool("find_correlations", {"entity": _ENTITY}) + cross = _cross_joins(fc) + assert cross and all(j["plan_eligible"] for j in cross) + + # and cross_plan recovers the cross-API chain first-plan-correct. + graphs = tuple(ps.client.surface_graph for ps in surface.registry.providers()) + ws = Workspace(graphs=graphs) + plan = cross_plan(ws, "dailynews-prices", "getPrice", set()) + assert plan is not None + assert [s.surface for s in plan.steps] == ["birdeye-data", "dailynews-prices"] + + +# --- PROOF 3: a provider-only declaration stays a candidate (confirmed-gate) -------- +def test_provider_only_declaration_stays_candidate_through_aggregator() -> None: + # provider x-gecko on both sides, but NO customer confirmation (hints=None). + surface = _surface(producer_declared=True, consumer_declared=True, hints=None) + + link = _token_link( + surface.call_tool( + "correlate_surfaces", {"a": "birdeye/data", "b": "dailynews/prices"} + ) + ) + assert link["basis"]["provenance"] == "DECLARED" and link["basis"]["tier"] == 3 + assert not link["plan_eligible"] and link["candidate"] + assert "provider-declared-unconfirmed" in link["basis"]["signals"] + + # the value-domain index shows the join but marks it a candidate, never confirmed. + fc = surface.call_tool("find_correlations", {"entity": _ENTITY}) + cross = _cross_joins(fc) + assert cross, "the provider-declared cross join should be visible as a candidate" + assert all(not j["plan_eligible"] and j["candidate"] for j in cross) + assert all( + not j["producer"]["confirmed"] and not j["consumer"]["confirmed"] for j in cross + ) + + # the confirmed-gate holds all the way to the executable plan: cross_plan refuses it. + graphs = tuple(ps.client.surface_graph for ps in surface.registry.providers()) + ws = Workspace(graphs=graphs) + assert cross_plan(ws, "dailynews-prices", "getPrice", set()) is None + + +# --- the hidden-tool discipline: absent from list_tools, callable by name ----------- +def test_aggregator_tools_are_hidden_but_callable() -> None: + surface = _surface( + producer_declared=False, consumer_declared=False, hints=_CONFIRMED + ) + listed = {t["name"] for t in surface.list_tools()} + assert not ( + {"correlate_surfaces", "find_correlations", "value_domain_index"} & listed + ), ( + "aggregator correlation tools must NOT be enumerated (projection stays byte-identical)" + ) + # yet each is callable by name (no raise). + assert "links" in surface.call_tool( + "correlate_surfaces", {"a": "birdeye/data", "b": "dailynews/prices"} + ) + assert "cross_joins" in surface.call_tool("find_correlations", {"entity": _ENTITY}) + assert "entities" in surface.call_tool("value_domain_index", {}) + + +def test_correlate_surfaces_unknown_provider_raises() -> None: + surface = _surface(producer_declared=False, consumer_declared=False, hints=None) + try: + surface.call_tool("correlate_surfaces", {"a": "birdeye/data", "b": "nope"}) + except CallError as exc: + assert "nope" in str(exc) + else: # pragma: no cover - the raise is the assertion + raise AssertionError("expected a CallError for an unknown provider")