diff --git a/gecko/cli.py b/gecko/cli.py index 0d424b1..225d327 100644 --- a/gecko/cli.py +++ b/gecko/cli.py @@ -53,6 +53,7 @@ "scan-doc", "auth", "graph", + "correlate", "rm", "list", "doctor", @@ -601,6 +602,34 @@ def _cmd_graph(argv: list[str]) -> int: return 0 +def _cmd_correlate(argv: list[str]) -> int: + """`gecko correlate ` — the cross-API correlation report. + + Thin transport, like `gecko graph json`: build two Surfaces, call + ``Surface.correlate``, print the provenance-carrying result as JSON. Cross-API + links are DECLARED-only for plan-eligibility (§13.6); a bare name/signature + match across the boundary is a quarantined candidate a human confirms. This is a + report — non-zero exit is not used. + """ + from .access import public_session + from .surface import Surface + + p = argparse.ArgumentParser( + prog="gecko correlate", + description="Report which of A's outputs correlate to B's inputs, with provenance.", + ) + p.add_argument("spec_a", help="An OpenAPI URL, path, or docs URL (surface A).") + p.add_argument("spec_b", help="An OpenAPI URL, path, or docs URL (surface B).") + p.add_argument("--id-a", default="A", help="Surface id label for A (default: A).") + p.add_argument("--id-b", default="B", help="Surface id label for B (default: B).") + args = p.parse_args(argv) + + a = Surface.from_spec(args.spec_a, session=public_session(), surface_id=args.id_a) + b = Surface.from_spec(args.spec_b, session=public_session(), surface_id=args.id_b) + print(json.dumps(a.correlate(b).to_dict(), indent=2)) + return 0 + + def _cmd_auth(argv: list[str]) -> int: """`gecko auth set|rm|list` — thin transport over ``credentials`` (keychain). @@ -1312,6 +1341,8 @@ def main(argv: list[str] | None = None) -> int: return _cmd_auth(rest) if cmd == "graph": return _cmd_graph(rest) + if cmd == "correlate": + return _cmd_correlate(rest) if cmd == "rm": return _cmd_rm(rest) if cmd == "list": diff --git a/gecko/client.py b/gecko/client.py index 082f93d..bf2742d 100644 --- a/gecko/client.py +++ b/gecko/client.py @@ -291,8 +291,16 @@ def surface_graph(self) -> SurfaceGraph: # §14), with injected customer confirmations winning on conflict — a # customer's local correction outranks the shipped hint (§13.2). declared = {**declared_entity_hints(self.spec), **self._declared_hints} + # ``confirmed`` is the CUSTOMER-vouched subset ONLY — the injected + # confirmations, never the provider's own x-gecko. Carried onto the + # graph so compose/correlate share one trusted provenance source: only + # a confirmed entity may drive an executable cross-surface plan (the + # §13.6 guardrail, now enforced in the planner, not just the report). self._surface_graph = build_graph( - self.operations, surface_id=self.surface_id, declared=declared + self.operations, + surface_id=self.surface_id, + declared=declared, + confirmed=self._declared_hints, ) return self._surface_graph diff --git a/gecko/compose.py b/gecko/compose.py index e1baeae..1f48689 100644 --- a/gecko/compose.py +++ b/gecko/compose.py @@ -66,15 +66,21 @@ def graph(self, surface_id: str) -> SurfaceGraph | None: return None -def _declared_field_producers( +def _confirmed_field_producers( graph: SurfaceGraph, entity: str ) -> list[tuple[str, str]]: """(producer_op_id, field_name) for id-shaped response fields this graph's - DECLARED vocabulary maps to ``entity`` — deterministic order.""" - decl = dict(graph.declared) + CUSTOMER-CONFIRMED vocabulary maps to ``entity`` — deterministic order. + + Reads ``graph.confirmed``, NOT ``graph.declared``: a producing field whose + entity is only provider-x-gecko-declared (present in ``declared`` but not + ``confirmed``) is NOT a plannable cross-API supplier. An untrusted provider + self-declaration must never mint an executable cross-surface chain (§13.6 + guardrail 3/4) — the same honest refusal correlate reports as a candidate.""" + conf = dict(graph.confirmed) out: list[tuple[str, str]] = [] for node in graph.nodes: - if node.kind != "field" or decl.get(_norm(node.name)) != entity: + if node.kind != "field" or conf.get(_norm(node.name)) != entity: continue raw_type = node.sig.split("|", 1)[0] if node.sig else "" if raw_type not in _ID_SIG_TYPES: @@ -129,20 +135,27 @@ def cross_plan( if target.opnode(intent_op_id) not in target._by_id: return None - decl_target = dict(target.declared) + # Gate on the CUSTOMER-CONFIRMED vocab, not the merged (provider∪customer) + # ``declared``: the consuming param's entity must be customer-vouched on the + # target surface, and the producing field's entity customer-vouched on its own + # surface. A provider-only x-gecko declaration (in ``declared`` but not + # ``confirmed``) resolves to no entity here -> None, the same honest refusal as + # an unmatched input. This makes ``plan_eligible`` the sole plan gate EVERYWHERE + # (§13.6 guardrail 3/4): an untrusted provider cannot mint an executable chain. + conf_target = dict(target.confirmed) cross_steps: list[PlanStep] = [] cross_explain: list[ExplainEntry] = [] supplied: set[str] = set() for pn in sorted(_unsatisfied(target, intent_op_id, sat), key=lambda n: n.name): - entity = decl_target.get(_norm(pn.name)) + entity = conf_target.get(_norm(pn.name)) if not entity: - continue # not declared on the consuming side -> not cross-sourceable + continue # not customer-confirmed on the consuming side -> not sourceable resolved = False for other in sorted(workspace.graphs, key=lambda g: g.surface_id): if other.surface_id == surface_id or resolved: continue - for src_op, field_name in _declared_field_producers(other, entity): + for src_op, field_name in _confirmed_field_producers(other, entity): # the supplier must resolve within its OWN surface (its own # ladder, its own genericity stats) — never a dangling step. sub = intra_plan(other, src_op, sat, max_ops=_MAX_SUB_OPS) diff --git a/gecko/correlate.py b/gecko/correlate.py new file mode 100644 index 0000000..527c72e --- /dev/null +++ b/gecko/correlate.py @@ -0,0 +1,487 @@ +"""The correlation engine (§13 Phase 2) — the confidence, done honestly. + +``correlate_surfaces(a, b)`` reports, WITH PROVENANCE, which of A's outputs +correlate to B's inputs and *why*. The "confidence" is the §13.2 **provenance +ladder**, never a learned float (that would re-import the retired corpus-moat +framing). It builds ON the shipped primitives — it re-derives no join logic: + + * ``graph.py`` — ``_entity_of`` (name -> entity), ``_sig_corroborates`` / + ``_sig_of`` (the §13.1 value-domain signature), the genericity floor. + * ``hints.py`` — ``declared_entity_hints`` (provider x-gecko, UNTRUSTED spec) + vs. the client's injected customer-confirmed vocabulary. + +The ladder (all deterministic; cross-API is effectively BINARY — §13.6): + + tier 3 DECLARED both names map to the same entity in the DECLARED vocab. + Cross-API + both sides CUSTOMER-confirmed -> plan-eligible, + high. Cross-API with a PROVIDER-only declaration -> candidate + (guardrail 3/4: an untrusted spec can't mint a cross-system + join). Intra-API -> plan-eligible. + tier 2 SIG a shared discriminating pattern/enum/format corroborates a + name match. **INTRA-API ONLY** — the §13.6 gate forbids a + cross-API signature from carrying a join (zero false-high). + tier 1 NAME same ``_entity_of``, id-shaped, not genericity-demoted. + Intra-API -> plan-eligible. Cross-API -> a **candidate**, + quarantined, NEVER auto-joined. + tier 0 none. **Type is a HARD gate**: a non-id-type or a mismatched + id-type drops to 0 before any scoring. + +Control-plane only (invariant #1, guardrail 2): the scorer's ONLY input is a +:class:`Correland`, whose typed shape carries ``{name, type, value-domain +signature, name/declared entity, provenance}`` — a response value, sampled +overlap, or observed cardinality literally cannot be passed in. +""" + +from __future__ import annotations + +import math +from collections import defaultdict +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, Literal + +from .graph import ( + _GENERIC_FLOOR, + _GENERIC_FRAC, + SurfaceGraph, + _entity_of, + _norm, + _sig_corroborates, +) + +if TYPE_CHECKING: + from .surface import Surface + +# single source of truth for this module's Literal types (CLAUDE.md). +Provenance = Literal["DECLARED", "INFERRED"] +Confidence = Literal["high", "medium", "low"] +Origin = Literal["SPEC", "DOCS"] +DeclaredSource = Literal["customer", "provider"] + +# id-shaped raw schema types — a join key is number/string-shaped (mirrors +# graph._ID_TYPES / compose._ID_SIG_TYPES); a bool/enum can never be a join key. +_ID_SIG_TYPES = ("string", "integer", "number") +_CONF_RANK = {"high": 0, "medium": 1, "low": 2} + + +def _id_type(sig: str) -> str | None: + """The normalized id-type of a signature's leading raw type, or None (non-id). + ``integer`` and ``number`` collapse to ``number`` so a counter and a float join.""" + raw = sig.split("|", 1)[0] if sig else "" + if raw == "string": + return "string" + if raw in ("integer", "number"): + return "number" + return None + + +def _sig_signal(a: str, b: str) -> str | None: + """The named discriminating signal two signatures share (§13.1), or None. A + corroborator only — never a standalone cross-API basis (that's the §13.6 gate).""" + if not _sig_corroborates(a, b): + return None + _, fa, pa, ea = a.split("|") + _, fb, pb, eb = b.split("|") + if pa and pa == pb: + return "pattern-eq" + if ea and ea == eb: + return "enum-eq" + return "format-eq" # a shared discriminating format (uuid/uri/...) + + +# --- the scorer's ONLY input: a control-plane-typed value object (guardrail 2) --- +@dataclass(frozen=True) +class Correland: + """One side of a candidate correlation — a **control-plane-typed** descriptor. + + Its fields are exactly ``{name, id-type, value-domain signature, name entity, + declared entity + source, origin}``. There is deliberately NO field for a + response value, a sampled overlap, or an observed cardinality — so a payload + cannot be fed to the scorer (invariant #1). ``sig`` carries the *identity* of a + schema constraint (type|format|pattern-hash|enum-hash), never any constraint or + value text. + """ + + surface_id: str + op_id: str + name: str + id_type: str | None # "string" | "number" | None (non-id -> hard-gated) + sig: str # value-domain signature; hashes only, no value text + name_entity: str | None # _entity_of(name): the entity the NAME refers to + declared_entity: str | None # entity from the surface's DECLARED vocab + declared_source: DeclaredSource | None # customer-confirmed vs provider x-gecko + origin: Origin = "SPEC" # weakest data origin this side rests on + + +# --- the verdict objects (frozen; no bare dicts as contracts) ------------------- +@dataclass(frozen=True) +class CorrelationBasis: + """The *why*, riding each link — mirrors the graph's per-edge provenance.""" + + entity: str | None # value-domain entity, e.g. "solanatokenmint" + tier: int + provenance: Provenance + src_surface: str + dst_surface: str + src_field: str + dst_param: str + signals: tuple[str, ...] # each rung that fired + weakest_origin: Origin + confidence: Confidence + + def to_dict(self) -> dict[str, Any]: + return { + "entity": self.entity, + "tier": self.tier, + "provenance": self.provenance, + "src_surface": self.src_surface, + "dst_surface": self.dst_surface, + "src_field": self.src_field, + "dst_param": self.dst_param, + "signals": list(self.signals), + "weakest_origin": self.weakest_origin, + "confidence": self.confidence, + } + + +@dataclass(frozen=True) +class CorrelationLink: + """One A.output <-> B.input correlation, with its basis and whether it is usable + NOW (``plan_eligible``) or a quarantined ``candidate`` needing a human confirm.""" + + src_op: str + dst_op: str + basis: CorrelationBasis + plan_eligible: bool + + @property + def candidate(self) -> bool: + """Quarantined — surfaced for a human to confirm, never auto-joined.""" + return not self.plan_eligible + + def to_dict(self) -> dict[str, Any]: + return { + "src_op": self.src_op, + "dst_op": self.dst_op, + "plan_eligible": self.plan_eligible, + "candidate": self.candidate, + "basis": self.basis.to_dict(), + } + + +@dataclass(frozen=True) +class CorrelationResult: + """The ranked links + a summary. ``to_dict`` is JSON-serializable and + control-plane clean (structure only, never a payload).""" + + links: tuple[CorrelationLink, ...] + summary: dict[str, Any] + + def to_dict(self) -> dict[str, Any]: + return { + "summary": self.summary, + "links": [link.to_dict() for link in self.links], + } + + +# --- the pure scorer (control-plane; the §13.2 ladder) -------------------------- +def _basis( + tier: int, + provenance: Provenance, + src: Correland, + dst: Correland, + entity: str | None, + signals: tuple[str, ...], + confidence: Confidence, +) -> CorrelationBasis: + weakest: Origin = "DOCS" if "DOCS" in (src.origin, dst.origin) else "SPEC" + return CorrelationBasis( + entity=entity, + tier=tier, + provenance=provenance, + src_surface=src.surface_id, + dst_surface=dst.surface_id, + src_field=src.name, + dst_param=dst.name, + signals=signals, + weakest_origin=weakest, + confidence=confidence, + ) + + +def _score(src: Correland, dst: Correland) -> tuple[CorrelationBasis, bool] | None: + """The §13.2 ladder over TWO control-plane descriptors. Returns + ``(basis, plan_eligible)`` or None (tier 0). Cross-API is DECLARED-only for + plan-eligibility; a name/signature match across the boundary is at most a + quarantined candidate (§13.6).""" + # tier 0 — TYPE IS A HARD GATE (before any name/signature scoring). + if src.id_type is None or dst.id_type is None: + return None # a non-id-shaped side can never be a join key + if src.id_type != dst.id_type: + return None # a string field cannot feed an integer param + cross = src.surface_id != dst.surface_id + + # tier 3 — DECLARED entity identity (the only cross-API-grade basis, §13.6). + if src.declared_entity and src.declared_entity == dst.declared_entity: + entity = src.declared_entity + signals = [ + f"declared:{entity}", + f"declared-by:{src.surface_id}", + f"declared-by:{dst.surface_id}", + ] + sig = _sig_signal(src.sig, dst.sig) + if sig: + signals.append(sig) + if not cross: + return _basis(3, "DECLARED", src, dst, entity, tuple(signals), "high"), True + both_confirmed = ( + src.declared_source == "customer" and dst.declared_source == "customer" + ) + if both_confirmed: + return ( + _basis( + 3, "DECLARED", src, dst, entity, (*signals, "confirmed"), "high" + ), + True, + ) + # guardrail 3/4: a provider-DECLARED-from-spec entity is untrusted — it is a + # candidate until the workspace owner confirms it. An untrusted provider must + # not unilaterally mint a cross-system plan basis. + return ( + _basis( + 3, + "DECLARED", + src, + dst, + entity, + (*signals, "provider-declared-unconfirmed"), + "medium", + ), + False, + ) + + # tier 2 — signature corroboration, INTRA-API ONLY (the §13.6 gate). + sig = _sig_signal(src.sig, dst.sig) + if sig and not cross: + ent2: str | None = ( + src.name_entity + if src.name_entity and src.name_entity == dst.name_entity + else None + ) + signals2 = [sig] if ent2 is None else [f"name-entity:{ent2}", sig] + return _basis(2, "INFERRED", src, dst, ent2, tuple(signals2), "medium"), True + + # tier 1 — INFERRED name-entity match. + if src.name_entity and src.name_entity == dst.name_entity: + entity = src.name_entity + signals1 = [f"name-entity:{entity}"] + if sig: + signals1.append(sig) # signature corroborates the name (not a cross basis) + if cross: + # §13.6: a name (and even signature) match across the boundary is a + # quarantined candidate — NEVER an auto-join. + return ( + _basis( + 1, + "INFERRED", + src, + dst, + entity, + (*signals1, "cross-api-quarantined"), + "low", + ), + False, + ) + return _basis(1, "INFERRED", src, dst, entity, tuple(signals1), "low"), True + + return None # tier 0 — nothing correlates + + +# --- enumeration: a surface -> its control-plane descriptors -------------------- +def _declared_split( + surface: Surface, +) -> tuple[dict[str, str], set[str]]: + """``(name -> entity merged vocab, customer-confirmed names)`` — both normalized. + + Both come straight off the graph: ``graph.declared`` is the MERGED + (provider∪customer) vocab, ``graph.confirmed`` is the CUSTOMER-vouched subset. + Reading ``confirmed`` here (rather than reconstructing it from + ``client._declared_hints``) makes the graph the ONE provenance source both the + correlation engine and the compose planner share — so guardrail 3/4 can never + disagree between the report and the executable plan.""" + merged = dict(surface.graph.declared) # already normalized name -> entity + customer = {name for name, _ in surface.graph.confirmed} + return merged, {n for n in merged if n in customer} + + +def _generic_names(graph: SurfaceGraph) -> frozenset[str]: + """The genericity-demoted names of a surface — reuses graph.py's exact floor so + an intra tier-1 link on an over-common name is quarantined, never planned.""" + produced: dict[str, set[str]] = defaultdict(set) + consumed: dict[str, set[str]] = defaultdict(set) + n_ops = 0 + for node in graph.nodes: + if node.kind == "operation": + n_ops += 1 + elif node.kind == "field": + produced[_norm(node.name)].add(node.owner) + elif node.kind == "param": + consumed[_norm(node.name)].add(node.owner) + floor = max(_GENERIC_FLOOR, math.ceil(_GENERIC_FRAC * max(1, n_ops))) + return frozenset( + name + for name in set(produced) | set(consumed) + if len(produced[name]) > floor or len(consumed[name]) > floor + ) + + +def _source_of( + name_norm: str, merged: dict[str, str], customer: set[str] +) -> DeclaredSource | None: + if name_norm not in merged: + return None + return "customer" if name_norm in customer else "provider" + + +def _outputs( + surface: Surface, merged: dict[str, str], customer: set[str] +) -> list[Correland]: + """Producer descriptors: id-shaped response FIELDS this surface emits.""" + out: list[Correland] = [] + for node in surface.graph.nodes: + if node.kind != "field": + continue + idt = _id_type(node.sig) + if idt is None: + continue + nn = _norm(node.name) + out.append( + Correland( + surface_id=surface.surface_id, + op_id=node.owner, + name=node.name, + id_type=idt, + sig=node.sig, + name_entity=_entity_of(node.name, node.detail or None), + declared_entity=merged.get(nn), + declared_source=_source_of(nn, merged, customer), + ) + ) + return out + + +def _inputs( + surface: Surface, merged: dict[str, str], customer: set[str] +) -> list[Correland]: + """Consumer descriptors: required id-shaped PARAMS this surface consumes + (path/query/body — never auth headers/cookies, invariant #4).""" + out: list[Correland] = [] + for node in surface.graph.nodes: + if node.kind != "param": + continue + loc, _, flag = node.detail.partition("|") + if flag != "req" or loc not in ("path", "query", "body"): + continue + idt = _id_type(node.sig) + if idt is None: + continue + nn = _norm(node.name) + out.append( + Correland( + surface_id=surface.surface_id, + op_id=node.owner, + name=node.name, + id_type=idt, + sig=node.sig, + name_entity=_entity_of(node.name), + declared_entity=merged.get(nn), + declared_source=_source_of(nn, merged, customer), + ) + ) + return out + + +def _link(src: Correland, dst: Correland) -> CorrelationLink | None: + scored = _score(src, dst) + if scored is None: + return None + basis, plan_eligible = scored + return CorrelationLink( + src_op=src.op_id, dst_op=dst.op_id, basis=basis, plan_eligible=plan_eligible + ) + + +def _demote_generic(link: CorrelationLink, generic: frozenset[str]) -> CorrelationLink: + """An intra tier-1 name link on a genericity-demoted name is quarantined (the + §10 demotion, applied to the report exactly as graph.py applies it to plans).""" + if link.basis.tier != 1 or not link.plan_eligible: + return link + if link.basis.src_surface != link.basis.dst_surface: + return link + names = {_norm(link.basis.src_field), _norm(link.basis.dst_param)} + if not (names & generic): + return link + demoted = CorrelationBasis( + **{ + **link.basis.to_dict(), + "signals": (*link.basis.signals, "generic-demoted"), + "confidence": "low", + } + ) + return CorrelationLink(link.src_op, link.dst_op, demoted, plan_eligible=False) + + +def correlate_surfaces(a: Surface, b: Surface) -> CorrelationResult: + """Which of A's outputs correlate to B's inputs (and, when ``a is not b``, B's + outputs to A's inputs), each with a provenance-carrying basis. Deterministic, + control-plane only. Cross-API links are DECLARED-only for plan-eligibility; a + bare name/signature match across the boundary is a quarantined candidate.""" + a_merged, a_customer = _declared_split(a) + b_merged, b_customer = _declared_split(b) + a_gen = _generic_names(a.graph) + + a_out = _outputs(a, a_merged, a_customer) + a_in = _inputs(a, a_merged, a_customer) + b_out = _outputs(b, b_merged, b_customer) + b_in = _inputs(b, b_merged, b_customer) + + pairs = [(src, dst) for src in a_out for dst in b_in] + if a is not b: + pairs += [(src, dst) for src in b_out for dst in a_in] + + links: list[CorrelationLink] = [] + for src, dst in pairs: + link = _link(src, dst) + if link is None: + continue + links.append(_demote_generic(link, a_gen)) + + links.sort( + key=lambda lk: ( + not lk.plan_eligible, # plan-eligible first + -lk.basis.tier, # then higher tier + _CONF_RANK.get(lk.basis.confidence, 3), + lk.src_op, + lk.dst_op, + lk.basis.src_field, + lk.basis.dst_param, + ) + ) + ranked = tuple(links) + summary: dict[str, Any] = { + "total": len(ranked), + "plan_eligible": sum(1 for lk in ranked if lk.plan_eligible), + "candidates": sum(1 for lk in ranked if lk.candidate), + "by_tier": { + str(t): sum(1 for lk in ranked if lk.basis.tier == t) for t in (3, 2, 1) + }, + } + return CorrelationResult(links=ranked, summary=summary) + + +__all__ = [ + "Correland", + "CorrelationBasis", + "CorrelationLink", + "CorrelationResult", + "correlate_surfaces", +] diff --git a/gecko/graph.py b/gecko/graph.py index ebe2f70..337f6c9 100644 --- a/gecko/graph.py +++ b/gecko/graph.py @@ -335,6 +335,15 @@ class SurfaceGraph: # (normalized_name, entity) pairs — carried on the graph so compose() can # join across surfaces on entity identity without re-reading any spec. declared: tuple[tuple[str, str], ...] = () + # the CUSTOMER-CONFIRMED subset of ``declared`` — the (normalized_name, entity) + # pairs whose source is the workspace confirm store (injected declared_hints), + # NOT provider-authored x-gecko. This is the TRUSTED subset: only a customer + # vouch may drive an executable cross-surface plan (compose.cross_plan) or a + # plan-eligible cross-API correlation (§13.6 guardrail 3/4). ``declared`` stays + # the merged provider∪customer vocab (untrusted); ``confirmed`` is what a human + # actually stood behind. Empty when no customer hints were injected -> compose + # refuses every cross-join (fail-closed: nothing is customer-vouched). + confirmed: tuple[tuple[str, str], ...] = () # indices (not serialized) — kept out of the content hash on purpose. _by_id: dict[str, Node] = field(default_factory=dict, compare=False, repr=False) @@ -347,6 +356,7 @@ def serialize(self) -> bytes: payload = { "surface_id": self.surface_id, "declared": sorted(list(pair) for pair in self.declared), + "confirmed": sorted(list(pair) for pair in self.confirmed), "nodes": sorted( ([n.kind, n.id, n.name, n.owner, n.detail, n.sig] for n in self.nodes), ), @@ -536,6 +546,7 @@ def build_graph( *, surface_id: str = "", declared: Mapping[str, str] | None = None, + confirmed: Mapping[str, str] | None = None, ) -> SurfaceGraph: """Deterministic surface graph: operation/param/field/resource nodes + EXTRACTED produces/consumes/on edges + DECLARED feeds edges (explicit entity hints, the @@ -545,12 +556,25 @@ def build_graph( ``declared`` maps a param/field NAME to an entity (from x-gecko hints or a customer confirmation); two names declared to the same entity join with provenance DECLARED even when their names differ — the only cross-API-grade - basis (§13.6). Hint names/entities are matched on normalized form.""" + basis (§13.6). Hint names/entities are matched on normalized form. + + ``confirmed`` is the CUSTOMER-vouched subset of ``declared`` (the workspace + confirm store / injected ``declared_hints``, NEVER provider x-gecko). It is + carried onto the graph as the trusted provenance source both correlate and + compose read: only a confirmed entity may drive an executable cross-surface + plan. Omitted -> empty -> compose refuses all cross-joins (fail-closed).""" nodes: dict[str, Node] = {} edges: list[Edge] = [] ns = _ns(surface_id) # normalized declared vocabulary: name -> entity (both sides normalized). decl = {_norm(k): _norm(v) for k, v in (declared or {}).items() if k and v} + # the customer-confirmed subset — same normalization; only names that are also + # in ``decl`` (a confirmation without a matching declared name is inert). + conf = { + _norm(k): _norm(v) + for k, v in (confirmed or {}).items() + if k and v and _norm(k) in decl + } def add_node(node: Node) -> None: nodes.setdefault(node.id, node) @@ -730,5 +754,6 @@ def is_generic(name: str) -> bool: edges=tuple(uniq_edges), surface_id=surface_id, declared=tuple(sorted(decl.items())), + confirmed=tuple(sorted(conf.items())), _by_id=by_id, ) diff --git a/gecko/surface.py b/gecko/surface.py index 3f6b7e6..8da9ebf 100644 --- a/gecko/surface.py +++ b/gecko/surface.py @@ -196,6 +196,19 @@ def graph_data(self) -> dict[str, Any]: return graph_data(self.graph) + # -- the cross-API correlation (the provenance-carrying confidence) ---------- + def correlate(self, other: Surface) -> Any: + """Correlate this Surface's outputs to ``other``'s inputs — a ranked, + provenance-carrying set of "A.output <-> B.input" links, each with a basis + (the *why*) and whether it is plan-eligible now or a quarantined candidate. + + Cross-API links are DECLARED-only for plan-eligibility (§13.6); a bare + name/signature match across the API boundary is a candidate a human confirms. + Returns a :class:`~gecko.correlate.CorrelationResult`.""" + from .correlate import correlate_surfaces + + return correlate_surfaces(self, other) + # -- the visual (graphviz for APIs) ------------------------------------------ def render_svg(self, *, title: str | None = None) -> str: """The Surface as an SVG call graph — operations as nodes, feeds as arrows colored diff --git a/tests/test_compose.py b/tests/test_compose.py index 7495046..e8570d1 100644 --- a/tests/test_compose.py +++ b/tests/test_compose.py @@ -81,12 +81,31 @@ def market_spec(*, with_hints: bool = True) -> dict[str, Any]: } +#: Market-side entity vocabulary as a CUSTOMER confirmation (§12) — what +#: `gecko graph confirm market fixture_ref fixture` persists. A cross-API plan now +#: requires BOTH sides CUSTOMER-confirmed: a provider x-gecko marker alone (the +#: `with_hints` arg on market_spec) is DECLARED-but-untrusted and no longer drives +#: an executable chain. Enforcing that only in correlate (the report) while +#: compose.cross_plan read the merged `declared` was the Graph-P2 BLOCKING hole — +#: a malicious provider self-declaration minted an EXECUTABLE cross-API chain. The +#: planner now gates on `graph.confirmed`; see +#: tests/test_correlate.test_cross_plan_refuses_provider_only_declared for the proof. +MARKET_HINTS = {"fixture_ref": "fixture"} + + def _txline_client() -> AgentApiClient: return AgentApiClient(str(TXLINE), surface_id="txline", declared_hints=TXLINE_HINTS) -def _market_client(**kw) -> AgentApiClient: - return AgentApiClient(market_spec(**kw), surface_id="market") +def _market_client(*, confirmed: bool = False, **kw: Any) -> AgentApiClient: + """The market surface. ``confirmed`` injects the customer confirmation that a + cross-API plan now requires — provider x-gecko (``market_spec(with_hints=...)``) + alone is no longer sufficient to drive an executable chain.""" + return AgentApiClient( + market_spec(**kw), + surface_id="market", + declared_hints=MARKET_HINTS if confirmed else {}, + ) def _workspace(tx: AgentApiClient, mk: AgentApiClient) -> Workspace: @@ -106,8 +125,10 @@ def test_workspace_rejects_unnamespaced_and_duplicate_graphs() -> None: # --- the cross plan -------------------------------------------------------------- def test_cross_plan_two_api_intent() -> None: """The §12 Phase 4 shape: TxLINE supplies the fixture id, the market surface - consumes it via a DECLARED synonym join.""" - tx, mk = _txline_client(), _market_client() + consumes it via a DECLARED synonym join. BOTH sides customer-confirmed — that + is now the requirement for an executable cross chain (provider x-gecko alone is + a candidate, not plan-eligible).""" + tx, mk = _txline_client(), _market_client(confirmed=True) p = cross_plan(_workspace(tx, mk), "market", "openMarket", set()) assert p is not None assert [s.surface for s in p.steps] == ["txline", "market"] @@ -129,10 +150,11 @@ def test_cross_plan_is_declared_only() -> None: def test_cross_plan_needs_both_sides_declared() -> None: - """Market declares, but TxLINE has no confirmed vocabulary -> None: a - one-sided declaration is not an entity identity.""" + """Market is customer-confirmed, but TxLINE has no confirmed vocabulary -> None: + a one-sided confirmation is not an entity identity — the SUPPLIER side must be + customer-vouched too (its provider x-gecko / inferred field does not count).""" tx = AgentApiClient(str(TXLINE), surface_id="txline") # no declared_hints - mk = _market_client() + mk = _market_client(confirmed=True) assert cross_plan(_workspace(tx, mk), "market", "openMarket", set()) is None @@ -157,8 +179,9 @@ def test_satisfied_intent_yields_trivial_intent_plan() -> None: def test_two_api_chain_first_plan_correct_offline() -> None: """THE Step-5 gate (§12): the composed two-API plan executes end-to-end in recorded mode — TxLINE's synthesized FixtureId threads into the market op's - fixture_ref and lands kind-correct. First-plan-correct, offline, $0.""" - tx, mk = _txline_client(), _market_client() + fixture_ref and lands kind-correct. First-plan-correct, offline, $0. Both sides + customer-confirmed (the executable-chain requirement).""" + tx, mk = _txline_client(), _market_client(confirmed=True) p = cross_plan(_workspace(tx, mk), "market", "openMarket", set()) assert p is not None result = evaluate_cross_chain({"txline": tx, "market": mk}, p) @@ -170,7 +193,7 @@ def test_two_api_chain_first_plan_correct_offline() -> None: def test_cross_chain_missing_client_is_honest_failure() -> None: - tx, mk = _txline_client(), _market_client() + tx, mk = _txline_client(), _market_client(confirmed=True) p = cross_plan(_workspace(tx, mk), "market", "openMarket", set()) assert p is not None result = evaluate_cross_chain({"market": mk}, p) # txline client absent diff --git a/tests/test_correlate.py b/tests/test_correlate.py new file mode 100644 index 0000000..0108e4a --- /dev/null +++ b/tests/test_correlate.py @@ -0,0 +1,319 @@ +"""Phase 2 correlation-engine tests (§13, roadmap Phase 2). + +The load-bearing pair is the two Pattern-B falsifiable-proof asserts (offline, $0): + +1. NO DECLARED hint -> the cross-API token link is a **candidate** (tier <= 1, + quarantined), NEVER plan-eligible — even when name AND signature both match, the + §13.6 gate forbids an auto-join across an API boundary. +2. ONE DECLARED entity confirmed on both sides -> the link jumps to **tier 3, + plan-eligible**, and ``cross_plan`` recovers the cross-API chain first-plan-correct. + +Falsified if a bare-name or signature match reaches tier 3 cross-API without a +DECLARED hint. Both are single asserts. + +The remaining tests pin the guardrails defi-security will check: intra tier-2 +signature works but cross-API tier-2 does NOT; type is a hard gate; the basis carries +the why; the scorer input is control-plane typed (no traffic field); provider-DECLARED +alone is a candidate; ``.to_dict()`` is JSON-serializable and control-plane clean. +""" + +from __future__ import annotations + +import dataclasses +import json +from typing import Any + +from gecko.client import AgentApiClient +from gecko.compose import Workspace, cross_plan +from gecko.correlate import Correland, CorrelationResult, correlate_surfaces +from gecko.surface import Surface + +# base58-ish discriminating pattern so BOTH name and signature match across the +# boundary — the hardest case for the §13.6 gate to hold. +_MINT_PATTERN = "^[1-9A-HJ-NP-Za-km-z]{32,44}$" + + +def _producer_spec(*, declared: bool) -> dict[str, Any]: + """Surface A: resolves a token id with no required inputs (stands alone). + + Returns a ``tokenId`` field. ``declared`` adds a provider ``x-gecko-entity`` + marker; a customer confirmation is injected separately via ``declared_hints``. + """ + 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": "Token Resolver", "version": "1"}, + "servers": [{"url": "https://resolver.example.com"}], + "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, id_type: str = "string") -> dict[str, Any]: + """Surface B: prices a token by ``tokenId`` (required query param). + + ``id_type`` lets a test force a type mismatch (integer vs the producer's string). + """ + schema: dict[str, Any] = {"type": id_type} + if id_type == "string": + schema["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": "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"}}, + } + } + } + } + }, + } + } + }, + } + + +_CONFIRMED = {"tokenId": "solana-token-mint"} + + +def _surface( + spec: dict[str, Any], sid: str, *, hints: dict[str, str] | None = None +) -> Surface: + return Surface.of(AgentApiClient(spec, surface_id=sid, declared_hints=hints or {})) + + +def _token_link(result: CorrelationResult): + """The single tokenId->tokenId cross-API link (there is exactly one).""" + links = [ + link + for link in result.links + if link.basis.src_field == "tokenId" and link.basis.dst_param == "tokenId" + ] + assert links, "expected a tokenId correlation link" + return links[0] + + +# --- the two falsifiable-proof asserts (Pattern B, offline, $0) ----------------- +def test_no_declared_hint_stays_candidate() -> None: + """PROOF 1 — name AND signature match across the boundary, yet the link is a + quarantined candidate (tier <= 1), NEVER plan-eligible (§13.6).""" + a = _surface(_producer_spec(declared=False), "resolver") + b = _surface(_consumer_spec(declared=False), "prices") + link = _token_link(correlate_surfaces(a, b)) + assert link.basis.tier <= 1 and not link.plan_eligible + + +def test_one_declared_entity_reaches_tier3_and_cross_plan_recovers() -> None: + """PROOF 2 — confirm the entity on both sides and the link jumps to tier 3, + plan-eligible, AND cross_plan recovers the chain first-plan-correct.""" + a = _surface(_producer_spec(declared=False), "resolver", hints=_CONFIRMED) + b = _surface(_consumer_spec(declared=False), "prices", hints=_CONFIRMED) + link = _token_link(correlate_surfaces(a, b)) + assert link.basis.tier == 3 and link.plan_eligible + ws = Workspace(graphs=(a.graph, b.graph)) + plan = cross_plan(ws, "prices", "getPrice", set()) + assert plan is not None and [s.surface for s in plan.steps] == [ + "resolver", + "prices", + ] + + +# --- the §13.6 signature gate: intra works, cross does NOT ----------------------- +def test_intra_signature_tier2_but_cross_signature_not() -> None: + """A shared discriminating signature reaches tier 2 INTRA-API, but the SAME + signature across two surfaces never reaches tier 2 (it falls to a tier-1 + candidate) — the §13.6 cross-API zero-false-high lock.""" + a = _surface(_producer_spec(declared=False), "resolver") + consumer = _surface(_consumer_spec(declared=False), "prices") + cross = _token_link(correlate_surfaces(a, consumer)) + assert cross.basis.tier <= 1 # cross-API signature is NOT tier 2 + + # Build one surface that both PRODUCES and CONSUMES the signed tokenId so an + # intra field->param link exists; its signature corroboration reaches tier 2. + both = _surface(_both_spec(), "one") + intra = correlate_surfaces(both, both) + sig_links = [ + link + for link in intra.links + if link.basis.tier == 2 and "pattern-eq" in link.basis.signals + ] + assert sig_links, "intra signature corroboration should reach tier 2" + + +def _both_spec() -> dict[str, Any]: + """One surface: resolveToken produces a signed tokenId; getPrice consumes one.""" + prod = _producer_spec(declared=False)["paths"]["/token/resolve"] + cons = _consumer_spec(declared=False)["paths"]["/price"] + return { + "openapi": "3.0.0", + "info": {"title": "One", "version": "1"}, + "servers": [{"url": "https://one.example.com"}], + "paths": {"/token/resolve": prod, "/price": cons}, + } + + +# --- type is a hard gate -------------------------------------------------------- +def test_type_mismatch_drops_to_tier0() -> None: + """A string tokenId field cannot feed an integer tokenId param — mismatched + id-type drops to tier 0 (no link), before any name/signature scoring.""" + a = _surface(_producer_spec(declared=False), "resolver", hints=_CONFIRMED) + b = _surface( + _consumer_spec(declared=False, id_type="integer"), "prices", hints=_CONFIRMED + ) + links = [ + link + for link in correlate_surfaces(a, b).links + if link.basis.dst_param == "tokenId" + ] + assert not links + + +# --- the basis carries the why -------------------------------------------------- +def test_basis_carries_signals_and_provenance() -> None: + a = _surface(_producer_spec(declared=False), "resolver", hints=_CONFIRMED) + b = _surface(_consumer_spec(declared=False), "prices", hints=_CONFIRMED) + basis = _token_link(correlate_surfaces(a, b)).basis + assert basis.provenance == "DECLARED" + assert basis.entity == "solanatokenmint" + assert basis.src_surface == "resolver" and basis.dst_surface == "prices" + assert any(s.startswith("declared:") for s in basis.signals) + assert any(s.startswith("declared-by:") for s in basis.signals) + assert basis.confidence == "high" + + +# --- provider-DECLARED-only is a candidate, not plan-eligible -------------------- +def test_provider_declared_only_is_candidate() -> None: + """Guardrail 3/4: a provider x-gecko entity comes from an UNTRUSTED spec. It is + DECLARED but not workspace-confirmed, so the cross-API link is a candidate, not + plan-eligible — an untrusted provider can't unilaterally mint a cross-system join.""" + a = _surface(_producer_spec(declared=True), "resolver") # provider x-gecko only + b = _surface(_consumer_spec(declared=True), "prices") # provider x-gecko only + link = _token_link(correlate_surfaces(a, b)) + assert link.basis.provenance == "DECLARED" and link.basis.tier == 3 + assert not link.plan_eligible + assert "provider-declared-unconfirmed" in link.basis.signals + + +# --- SECURITY (Graph P2 BLOCKING): plan_eligible is the SOLE plan gate ----------- +def test_cross_plan_refuses_provider_only_declared() -> None: + """The blocking finding: correlate gates a provider-only x-gecko declaration to + ``plan_eligible=False`` (a candidate), but ``cross_plan`` read the MERGED + ``graph.declared`` with provenance stripped, so a malicious provider + self-declaration minted an EXECUTABLE cross-API chain. cross_plan must refuse it + — an untrusted provider can't unilaterally mint a cross-system plan. Provider-only + = x-gecko marker present, NO customer confirmation (declared_hints).""" + a = _surface(_producer_spec(declared=True), "resolver") # provider x-gecko only + b = _surface(_consumer_spec(declared=True), "prices") # provider x-gecko only + ws = Workspace(graphs=(a.graph, b.graph)) + assert cross_plan(ws, "prices", "getPrice", set()) is None + + +def test_confirmed_subset_is_customer_only_not_provider() -> None: + """``graph.confirmed`` is the trusted CUSTOMER-confirmed subset — populated ONLY + from injected ``declared_hints``, never from provider x-gecko. It is the ONE + provenance source both correlate and compose read (no per-engine reconstruction).""" + # provider x-gecko present, but NO customer confirm -> declared has it, confirmed empty + provider_only = _surface(_producer_spec(declared=True), "resolver").graph + assert ("tokenid", "solanatokenmint") in provider_only.declared + assert provider_only.confirmed == () + # a customer confirmation -> present in confirmed (the trusted subset) + confirmed = _surface( + _producer_spec(declared=False), "resolver", hints=_CONFIRMED + ).graph + assert ("tokenid", "solanatokenmint") in confirmed.confirmed + + +# --- control-plane: the scorer input cannot carry a payload --------------------- +_FORBIDDEN = ( + "value", + "sample", + "response", + "payload", + "cardinality", + "overlap", + "traffic", + "observed", + "count", + "example", +) + + +def test_correland_is_control_plane_typed() -> None: + """Guardrail 2: the scorer's ONLY input is a Correland whose typed shape carries + name/type/value-domain signature/entity/provenance — a response value, sampled + overlap, or observed cardinality literally cannot be passed in.""" + fields = {f.name for f in dataclasses.fields(Correland)} + assert fields == { + "surface_id", + "op_id", + "name", + "id_type", + "sig", + "name_entity", + "declared_entity", + "declared_source", + "origin", + } + for f in fields: + for bad in _FORBIDDEN: + assert bad not in f, f"Correland.{f} looks like a traffic field" + + +def test_to_dict_is_json_serializable_and_control_plane_clean() -> None: + a = _surface(_producer_spec(declared=False), "resolver", hints=_CONFIRMED) + b = _surface(_consumer_spec(declared=False), "prices", hints=_CONFIRMED) + payload = correlate_surfaces(a, b).to_dict() + text = json.dumps(payload) # raises if not serializable + lowered = text.lower() + for bad in _FORBIDDEN: + assert bad not in lowered, f"to_dict leaked a {bad}-shaped key/value" + assert payload["summary"]["plan_eligible"] >= 1 + + +def test_surface_correlate_facade_matches_free_function() -> None: + a = _surface(_producer_spec(declared=False), "resolver", hints=_CONFIRMED) + b = _surface(_consumer_spec(declared=False), "prices", hints=_CONFIRMED) + assert a.correlate(b).to_dict() == correlate_surfaces(a, b).to_dict()