feat(correlate): Graph Phase 2 — provenance-carrying cross-API correlation engine

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 <specA> <specB>` — 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":

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
 

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)