diff --git a/docs/aeron-has-type-links-spec.md b/docs/aeron-has-type-links-spec.md new file mode 100644 index 0000000..c4c34fe --- /dev/null +++ b/docs/aeron-has-type-links-spec.md @@ -0,0 +1,169 @@ +# LineageLens HAS_TYPE Peer Links Spec + +> **Status:** Decisions recorded. Query behavior follows [§7](#7-decisions). +> **Behavior:** `impact_of`, `explain`, and `find_paths` report a shared hub when the edges support it. The indexer and the contract adapters are unchanged. Search stays a name lookup. + +## Objective + +On a Rust repository, LineageLens stored only `HAS_TYPE` from each Aeron send site and each Aeron receive site onto the same command enum. No messaging adapter ran. `search`, `explain`, `impact_of`, and `find_paths` did not surface the matching peer. + +Go channels fail the same way. Kafka and HTTP fail the same way when no adapter matches. When a contract node does exist, `search`, `explain`, and `find_paths` still miss the peer, and `impact_of` is one-sided. + +The two gaps below are implemented as [§7](#7-decisions) decides them. An entry states only what the edges record. `verification` is present only when those edges do not give a direction. + +--- + +## 1. What the Rust run stored + +The index did not use a messaging adapter, an Aeron detector, or an `EXPOSES` / `CONSUMES` contract. Those adapters exist for Kafka, Celery, and SQS. They did not fire on this repository. + +The only relationship stored for the send site and the receive site was `HAS_TYPE` onto the same command enum: + +```text +sender fn --HAS_TYPE--> Command enum <--HAS_TYPE-- receiver fn +``` + +`RefKind.TYPE` becomes that edge in `src/lineagelens/resolve/resolver.py`. Rust emits the reference from any `type_identifier` in `src/lineagelens/spec/rust/refs.scm`. The shared enum name is why both sites land on one type node. + +Search and blast radius still did not surface the receiver from the sender. + +--- + +## 2. Why the tools miss the peer + +Putting `HAS_TYPE` on an allowlist is not sufficient. For blast radius it is already there. + +| Tool | What it does today | Why the receiver stays hidden | +|---|---|---| +| `impact_of` | Walks `HAS_TYPE` through `DEPENDENT_EDGES` in `src/lineagelens/query/analysis.py` | The walk is backward: who depends on the symbol you named. From the sender, the enum is an outgoing type, so the receiver is never reached. Blast radius of the enum lists both sites as dependents of the type, which is a different question. | +| `search` | BM25 over nodes only, in `src/lineagelens/query/api.py` | It never reads edges. | +| `explain` | Returns only a direct `src → dst` edge, in `src/lineagelens/query/analysis.py` | There is no sender-to-receiver edge. | +| `find_paths` | Forward-only, in `src/lineagelens/query/traverse.py`, defaulting to call-chain kinds (`CALL_CHAIN_EDGES` in `src/lineagelens/core/kinds.py`) | Sender → enum stops. The receiver’s `HAS_TYPE` points at the enum, so a forward walk never leaves the type toward the other site. | + +The missing behavior is a sibling hop over the graph that was actually built: from the sender, follow `HAS_TYPE` to the enum, then to every other site that `HAS_TYPE` that same enum. + +```text +sender --HAS_TYPE--> Command <--HAS_TYPE-- receiver + ^ + | + other co-user +``` + +--- + +## 3. Go channels + +Go channels are the same miss. Nothing in `src/lineagelens/spec/go/refs.scm` captures `ch <- v` or `<-ch`, and `src/lineagelens/contracts/definitions/messaging.yaml` has no Go adapter. Two functions that share a `chan T` are not paired. At best each mentions the element type `T` via `HAS_TYPE`. + +There is no channel-identity edge and no contract hub to walk. + +--- + +## 4. Kafka and HTTP + +Kafka and HTTP split in two, depending on whether an adapter already joined the two sides. + +### 4.1 No adapter matches + +Unmatched clients are the same miss as the Rust run. + +- There is no Go or Rust Kafka adapter (sarama, kafka-go, or a wrapped producer). +- HTTP route adapters cover FastAPI, Flask, Django, Spring, Express, Nest, ASP.NET, and Go mux only (`src/lineagelens/contracts/definitions/http.yaml`). gin, chi, and `http.Client` do not join. + +Two sides that only share a request struct or a message struct are `HAS_TYPE` again. A sibling hop can see that shared type. It does not know that a topic or a route is the real key, because no contract node was stored. + +### 4.2 An adapter matches + +When an adapter does match — Java `@KafkaListener` / `send`, a Python producer or consumer, or a normalised HTTP route — the graph gets a contract node. Both `EXPOSES` and `CONSUMES` point into that hub (`CONTRACT_EDGES` in `src/lineagelens/core/kinds.py`). Those edges are outside `DEPENDENT_EDGES`. + +| Tool | What a reviewer sees | +|---|---| +| `search` | Still misses the peer. Search does not read the hub. | +| `explain` | Still misses the peer. There is no direct symbol-to-symbol edge. | +| `find_paths` | Still misses the peer. The walk is forward-only, and the contract node has no outgoing edge to the other side. | +| `impact_of` | One-sided. `_contract_consumers` in `src/lineagelens/query/analysis.py` adds consumers only when the named symbol `EXPOSES` the contract, and it skips a consumer whose `service_id` equals the target’s `service_id`. | +| `contract_map` | Lists both sides. This is the tool that already shows the join. | + +`js.kafkajs` in `src/lineagelens/contracts/definitions/messaging.yaml` marks both `producer.send` and `consumer.subscribe` as `exposes`. The subscriber never joins the topic as a consumer, so that consumer stays off the hub. + +A `HAS_TYPE` sibling hop covers the Aeron enum, a Go channel element type, and a shared DTO. It does not know that a Kafka topic or an HTTP route is the real key. + +--- + +## 5. Proposed gaps + +One hop does not cover a shared type and a matched contract. Both are the same graph shape: two sites point at one hub. The implementation is that shape, not a detector for one framework. The choices in [§7](#7-decisions) are what keep a hub from being reported as a flow the index did not store. + +### Gap 1 — `HAS_TYPE` sibling hop + +For a shared command enum, a Go `chan T` element type, and a shared request or message struct. + +From a site, follow `HAS_TYPE` to the type node, then to every other site that `HAS_TYPE` that same type. On `impact_of`, this is a new relation after the normal dependent walk. It is not another kind added to `DEPENDENT_EDGES`. + +What this uses: + +- The edges the Rust index already stored. No reindex, and no Aeron adapter. +- Same-language pairs that share one command type, so they can show up in blast radius and explain once a tool grows the hop. +- The tools that failed on this repository. + +What this does not mean: + +- The hop is “shares a type.” It is not “sends to that receiver.” Every other user of that enum is a hit. +- Unfiltered, the same hop also ties together co-users of `String`, `Option`, `Vec`, and `Result`. The filter that keeps that noise out is [decision 1](#7-decisions): a declaration file plus a fan-in cap, not a denylist of those names. +- It fails when the other side does not reuse the type name. That case was not in this Rust run. It needs a graph fact other than `HAS_TYPE`, which this index did not produce. +- It does not cross languages. Resolution is language-scoped, so a Rust enum and a Java class never become one node. +- It misses sites that never put the type in type position: a Rust variant path, a Java `UnsafeBuffer` offer, or encoder and decoder class names that differ. + +Which tools grow this hop is [decision 2](#7-decisions). + +### Gap 2 — contract-hub walk + +For Kafka and HTTP once an adapter has already joined the two sides. + +`search`, `explain`, and `find_paths` never cross the hub. `impact_of` lists consumers only from the exposer and drops same-service peers. `contract_map` already lists both sides. Go channels and unmatched Kafka or HTTP clients have no hub, so this walk does not apply to them. + +The walk stays on the contract kinds the adapter already stored. Same-service peers are included, and they are not labeled cross-service. Both are [§7](#7-decisions). + +### What this run did not use + +An Aeron ground contract — channel plus stream id, stored as `EXPOSES` / `CONSUMES` — is not the graph this run built. Building one would be new detection and a reindex. Once those edges exist, the contract-hub walk in `peers` reports them, including a same-service peer. `_contract_consumers` still fills `cross_service` only for a consumer in another service. That detector is a later spec if the product wants pub/sub semantics. It does not explain this Rust result. + +--- + +## 6. Language matrix + +The sibling hop succeeds only when both sides resolve to the same type node. + +| Case | Shared type node | +|---|---| +| Rust, Java, Go, C# | Yes, when the shared name resolves to a type. | +| TypeScript | Yes. `@ref.type` is broad. | +| Python | Yes, from annotations only. | +| JavaScript | No. There is no `@ref.type`. | +| Kotlin, Ruby, Bash, C | No. The hop does not fire. | +| Cross-language | No. Resolution is language-scoped. A Rust enum and a Java class stay two nodes. | +| Type not in type position | No edge to the shared node. Examples: a Rust variant path, a Java `UnsafeBuffer` offer, encoder vs decoder class names. | + +--- + +## 7. Decisions + +Each choice is the one that lets an answer state a stored fact, and withholds a claim the edges do not support. `verification` is how an ambiguous flow is handed to an agent: the field is present only then, and it says to read the code. It is absent when the direction was recorded. + +1. **Sibling-hop filter.** Any declared type, not enums only and not a denylist of `String`, `Option`, `Vec`, or `Result`. The hub must be a type node with a declaration file. A type referenced from more than 16 distinct sites is not a peer list; the envelope records `peer_fan_in` and names no co-user. A type with no declaration file and more than one site records `peer_undeclared_type` and names no peer. The ordinary dependent walk of the type itself is unchanged: that question is who references the type, and the edges answer it. + +2. **Which tools grow the hop.** `impact_of` adds a `peers` list after the dependent walk. `explain` reports the inward hub path when no direct edge exists. `find_paths` reports a `shared_hub` result only when the forward walk found nothing, so a type share cannot outrank a call chain. `search` stays name lookup. A peer attached to a name hit would read as a match. + +3. **Contract kind.** The walk uses the kind already stored on the contract edge (`topic`, `http_route`, and the other `ContractKind` values). No new kind is added. A kind the adapter did not write is not invented in the answer. + +4. **Same-service contract peers.** They are included in `peers` with `service_relation` `same_service`. They are not placed in `cross_service`. `cross_service` remains what it was: other services that consume a contract the target exposes. A missing `service_id` is `unknown`, not same-service and not cross-service. + +A contract peer whose roles are exactly one `EXPOSES` and one `CONSUMES` states those roles and the weaker evidence tier, and it has no `verification` field. A shared type, two exposes, two consumes, or a site that both exposes and consumes carries `verification`. + +--- + +## 8. Out of scope + +- Indexer, resolver, grammar, or contract-adapter changes. +- An Aeron channel / stream-id detector. +- Turning `search` into a relationship query. diff --git a/src/lineagelens/mcp/server.py b/src/lineagelens/mcp/server.py index 85b4c66..8fb6951 100644 --- a/src/lineagelens/mcp/server.py +++ b/src/lineagelens/mcp/server.py @@ -460,11 +460,20 @@ async def find_paths( Returns ordered hop lists, so you get the route rather than a set of reachable symbols. Use this for "how does X reach Y". + When no forward chain exists, a shared hub may be returned with + ranking "shared_hub". That result is not a call chain. A hop with + "reversed": true is the other site's edge into the hub. An entry + carries "verification" only when the edges do not record a direction; + read the code at both sites before treating it as a flow. A contract + pair with roles EXPOSES and CONSUMES states that recorded direction + and has no verification field. + Args: from_symbol: id, qualified name, or unique search term. to_symbol: same. - kinds: edge kinds to follow. Defaults to call-chain kinds; pass - ["CALLS"] to exclude inheritance and contract hops. + kinds: edge kinds to follow. Defaults to call-chain kinds, and + the shared-hub fallback includes declared types and contract + hubs. Pass ["CALLS"] to keep the walk on calls only. intent: "plan" (default, cheap) or "precise" (adds verbatim source and data edges). max_paths: distinct routes to return. Default 6 (plan) / 12. @@ -561,6 +570,14 @@ async def impact_of( (a wire-compatibility and deploy-ordering problem). Defaults to intent="precise" for a line and "plan" for a symbol. + + `peers` lists symbols that share a declared type or a contract hub + with the target. They are not dependents and they are not callers. + A peer carries `verification` only when the edges do not record a + direction; read the code at both sites before treating that peer as + a sender or a receiver. A peer without `verification` states a + recorded exposes/consumes pair. `cross_service` is unchanged: other + services that consume a contract this symbol exposes. """ return engine().impact_of( target, intent=intent, limit=limit, max_depth=max_depth @@ -700,6 +717,8 @@ async def search( Searches only symbols defined in this project; for external packages (anthropic.Anthropic, openai.OpenAI) explore() reports usage sites. + Search does not return shared-hub peers. A name match is not a flow. + Use impact_of or explain for that. BM25-ranked. kinds filters by node kind: class, interface, enum, struct, trait, function, method, constructor, property, field, @@ -784,6 +803,11 @@ async def explain(from_symbol: str, to_symbol: str, **_: Any) -> dict[str, Any]: component that produced the edge, and where it is written. Use this when an answer looks wrong -- it will say whether the edge is a syntactic fact, a type-resolver result, or a name match. + + When no direct edge exists, a shared type or contract hub is returned + instead. The path is drawn with both edges pointing at the hub. That + entry carries `verification` only when the edges do not record a + direction; read the code at both sites before treating it as a flow. """ return engine().explain(from_symbol, to_symbol).as_dict() diff --git a/src/lineagelens/ontology.py b/src/lineagelens/ontology.py index bfcd276..dfa53eb 100644 --- a/src/lineagelens/ontology.py +++ b/src/lineagelens/ontology.py @@ -347,6 +347,11 @@ def ontology_instructions(project: Path | None = None) -> str: " Pass intent='plan' when surveying and intent='precise' when " "debugging. Precise adds verbatim source and data flow; plan is far " "cheaper and is the default for traversal tools.", + " A peer, explain, or find_paths entry carries `verification` only " + "when the graph cannot say how data moves between the two sites. " + "Read the code at those sites before treating that pair as a flow. " + "An entry without `verification` states a role or an edge the index " + "recorded.", "", "Unresolved is not the same as absent: a reference the resolver could " "not pin down is recorded with its candidates rather than guessed at, " diff --git a/src/lineagelens/query/analysis.py b/src/lineagelens/query/analysis.py index 5899c20..4fc4091 100644 --- a/src/lineagelens/query/analysis.py +++ b/src/lineagelens/query/analysis.py @@ -64,6 +64,10 @@ class ImpactReport: cross_service: list[dict[str, Any]] = field(default_factory=list) data: list[dict[str, Any]] = field(default_factory=list) entry_points: list[str] = field(default_factory=list) + #: Symbols that share a declared type or a contract hub with the target. + #: Not dependents. ``verification`` is present only when the edges do not + #: record a direction. + peers: list[dict[str, Any]] = field(default_factory=list) source: str = "" def as_dict(self) -> dict[str, Any]: @@ -79,6 +83,7 @@ def as_dict(self) -> dict[str, Any]: ("cross_service", self.cross_service), ("data", self.data), ("entry_points", self.entry_points), + ("peers", self.peers), ): if value: out[name] = value @@ -212,6 +217,10 @@ def _fill_impact( report.entry_points.append(node.qualified_name) report.cross_service.extend(self._contract_consumers(target, own_service)) + report.peers.extend( + link.as_dict() + for link in self._shared_hubs(target, budget=budget, envelope=envelope) + ) if intent is Intent.PRECISE: report.data = self._data_impact(target, budget=budget, envelope=envelope) @@ -220,6 +229,18 @@ def _fill_impact( # cheaper, and the shape of the dependency is what a survey needs. envelope.dataflow = {"mode": "skipped", "reason": "intent=plan"} + def _shared_hubs(self, target: Node, *, budget: Budget, envelope: Envelope): + """Type and contract peers of ``target``. + + Kept off ``DEPENDENT_EDGES``. A co-user of a type is not something that + depends on the target, and a contract peer is not a caller. + """ + from .peers import shared_hubs + + return shared_hubs( + self.store, target.id, limit=budget.limit, envelope=envelope + ) + def _contract_consumers( self, target: Node, own_service: str | None ) -> list[dict[str, Any]]: @@ -583,8 +604,16 @@ def _match_file(self, path: str) -> list[str]: # ---- §10.2 explain ---------------------------------------------------- - def explain(self, src: str, dst: str) -> list[dict[str, Any]]: - """Evidence and provenance for the edges between two nodes.""" + def explain( + self, src: str, dst: str, *, envelope: Envelope | None = None + ) -> list[dict[str, Any]]: + """Evidence for a direct edge, or a shared hub when there is none. + + A direct edge is returned as stored. A shared type or an ambiguous + contract role is returned only in that case, and only then may the + entry carry ``verification``. A recorded exposes/consumes pair does + not. + """ out: list[dict[str, Any]] = [] for edge in self.store.edges_from(src): if edge.dst != dst: @@ -602,7 +631,17 @@ def explain(self, src: str, dst: str) -> list[dict[str, Any]]: ), "metadata": edge.metadata or None, }) - return out + if out: + return out + + from .peers import shared_hubs + + return [ + link.as_explain() + for link in shared_hubs( + self.store, src, peer_id=dst, limit=8, envelope=envelope + ) + ] # --------------------------------------------------------------------------- diff --git a/src/lineagelens/query/api.py b/src/lineagelens/query/api.py index 0bc4e70..801c531 100644 --- a/src/lineagelens/query/api.py +++ b/src/lineagelens/query/api.py @@ -321,9 +321,34 @@ def find_paths( sources = self._sources( [h.node for p in paths for h in p.hops], resolved, budget ) + items = [p.as_dict(sources) for p in paths] + ranking = "shortest_path_first" + # A shared hub is not a shorter route. It is reported only when the + # forward walk found no path, so it cannot outrank a real chain. + if not items: + from .peers import shared_hubs + + kind_values = {str(k) for k in kinds} if kinds is not None else None + links = shared_hubs( + self.store, + src.id, + peer_id=dst.id, + include_types=kind_values is None or "HAS_TYPE" in kind_values, + include_contracts=( + kind_values is None + or bool(kind_values & {"EXPOSES", "CONSUMES"}) + ), + limit=budget.max_paths, + envelope=envelope, + ) + hub_nodes = [node for link in links for node in (link.site, link.hub, link.peer)] + sources = self._sources(hub_nodes, resolved, budget) + items = [link.as_path(sources) for link in links] + if items: + ranking = "shared_hub" return QueryResult.of( - "find_paths", [p.as_dict(sources) for p in paths], - budget=budget, envelope=envelope, ranking="shortest_path_first", + "find_paths", items, + budget=budget, envelope=envelope, ranking=ranking, from_symbol=src.qualified_name, to_symbol=dst.qualified_name, ) @@ -571,7 +596,7 @@ def explain(self, src: str, dst: str) -> QueryResult: return QueryResult(kind="explain", envelope=envelope, extra={"error": "unknown symbol"}) return QueryResult.of( - "explain", self.analyser.explain(a.id, b.id), + "explain", self.analyser.explain(a.id, b.id, envelope=envelope), budget=budget, envelope=envelope, from_symbol=a.qualified_name, to_symbol=b.qualified_name, ) diff --git a/src/lineagelens/query/peers.py b/src/lineagelens/query/peers.py new file mode 100644 index 0000000..786df95 --- /dev/null +++ b/src/lineagelens/query/peers.py @@ -0,0 +1,397 @@ +"""Peers that share a hub, and nothing those edges do not say. + +A site and its peer both point *at* a hub. There is no edge from one site to +the other, so a forward walk stops on the hub. The two hubs this knows are a +declared type (``HAS_TYPE``) and a contract node (``EXPOSES`` / ``CONSUMES``). + +What gets said: + +* A shared declared type is not a call and not a send. The entry names the + type and carries ``verification``, which tells the reader to check the code. +* A contract whose roles are exactly one ``EXPOSES`` and one ``CONSUMES`` has + a direction the index recorded. The entry states those roles and the weaker + evidence tier. It does not carry ``verification``. +* Any other contract role pair has no single direction. The entry states the + roles that were stored and carries ``verification``. + +``verification`` is omitted when the direction is recorded. It is not attached +to every peer. + +A type referenced from more than :data:`MAX_SHARED_TYPE_SITES` sites is not a +peer list. Naming every co-user would read as a flow. The envelope records +``peer_fan_in`` and no peer is emitted for that type. Asking for the type +itself still uses the ordinary dependent walk; that question is "who references +this type", which the edges do answer. +""" + +from __future__ import annotations + +from collections import defaultdict +from dataclasses import dataclass + +from ..core import Edge, EdgeKind, EvidenceTier, Node, NodeKind +from ..store import GraphStore +from .budget import Envelope + +#: Above this many sites, a type is too widely shared to name co-users as +#: peers. The cap is a count of distinct sites, not a denylist of type names, +#: so it applies to whatever the index actually resolved. +MAX_SHARED_TYPE_SITES = 16 + +_CONTRACT_KINDS = frozenset({EdgeKind.EXPOSES, EdgeKind.CONSUMES}) +_TIER_RANK = { + EvidenceTier.FACT: 2, + EvidenceTier.HEURISTIC: 1, + EvidenceTier.PROBABILISTIC: 0, +} + +TYPE_VERIFICATION = ( + "Both sites reference this type. The index has no edge that says one " + "calls or sends to the other. Read the code at both sites before " + "treating this as a flow." +) + +CONTRACT_VERIFICATION = ( + "The index does not record a single direction between these sites. " + "Read the code before treating one as the sender or the receiver." +) + + +@dataclass(frozen=True, slots=True) +class PeerLink: + """One other site that points at the same hub.""" + + relation: str + site: Node + peer: Node + hub: Node + site_kinds: frozenset[EdgeKind] + peer_kinds: frozenset[EdgeKind] + site_edges: tuple[Edge, ...] + peer_edges: tuple[Edge, ...] + service_relation: str + + @property + def flow_ambiguous(self) -> bool: + """True when the edges do not record a single direction. + + A shared type never does. A contract does only when one side exposes + and the other consumes, and neither side does both. + """ + if self.relation == "shares_type": + return True + one_each = ( + (frozenset({EdgeKind.EXPOSES}), frozenset({EdgeKind.CONSUMES})), + (frozenset({EdgeKind.CONSUMES}), frozenset({EdgeKind.EXPOSES})), + ) + return (self.site_kinds, self.peer_kinds) not in one_each + + @property + def evidence_tier(self) -> str: + """The weaker tier on the edges that form this link. + + A fact must not be reported when the other edge is a heuristic. + """ + edges = (*self.site_edges, *self.peer_edges) + weakest = min(edges, key=lambda edge: _TIER_RANK[edge.evidence.tier]) + return weakest.evidence.tier.value + + def as_dict(self) -> dict[str, object]: + """The impact / search shape. ``verification`` only when ambiguous.""" + out: dict[str, object] = { + "relation": self.relation, + "node": self.peer.qualified_name, + "at": _at(self.peer), + "hub": self.hub.qualified_name, + "hub_key": self.hub.signature or self.hub.name, + "hub_kind": _hub_kind(self.hub, self.site_edges), + "self_role": _role_names(self.site_kinds), + "peer_role": _role_names(self.peer_kinds), + "evidence_tier": self.evidence_tier, + "service_relation": self.service_relation, + "distance": 2, + } + _attach_verification(out, self) + return out + + def as_explain(self) -> dict[str, object]: + """A path through the hub, drawn with both edges pointing inward.""" + site_role = ",".join(_role_names(self.site_kinds)) + peer_role = ",".join(_role_names(self.peer_kinds)) + out: dict[str, object] = { + "kind": self.relation, + "evidence_tier": self.evidence_tier, + "evidence": sorted({ + edge.evidence.label for edge in (*self.site_edges, *self.peer_edges) + }), + "provenance": sorted({ + edge.provenance for edge in (*self.site_edges, *self.peer_edges) + }), + "hub": self.hub.qualified_name, + "hub_key": self.hub.signature or self.hub.name, + "hub_kind": _hub_kind(self.hub, self.site_edges), + "self_role": _role_names(self.site_kinds), + "peer_role": _role_names(self.peer_kinds), + "service_relation": self.service_relation, + "path": ( + f"{self.site.name} -{site_role}-> {self.hub.name} " + f"<-{peer_role}- {self.peer.name}" + ), + "written_at": [ + _written(edge) for edge in (*self.site_edges, *self.peer_edges) + ], + } + _attach_verification(out, self) + return out + + def as_path(self, sources: dict[str, str] | None = None) -> dict[str, object]: + """A find_paths result that is a hub, not a forward chain. + + The last hop is the peer's own edge into the hub, so ``reversed`` is + set. The kind string stays the stored edge kind. + """ + sources = sources or {} + site_edge = self.site_edges[0] + peer_edge = self.peer_edges[0] + out: dict[str, object] = { + "direction": "shared_hub", + "relation": self.relation, + "length": 2, + "chain": f"{self.site.name} -> {self.hub.name} -> {self.peer.name}", + "hops": [ + _hop(self.site, None, 0, sources), + _hop(self.hub, site_edge, 1, sources, reversed_=False), + _hop(self.peer, peer_edge, 2, sources, reversed_=True), + ], + "hub": self.hub.qualified_name, + "hub_kind": _hub_kind(self.hub, self.site_edges), + "self_role": _role_names(self.site_kinds), + "peer_role": _role_names(self.peer_kinds), + "evidence_tier": self.evidence_tier, + "service_relation": self.service_relation, + } + _attach_verification(out, self) + return out + + +def shared_hubs( + store: GraphStore, + site_id: str, + *, + peer_id: str | None = None, + include_types: bool = True, + include_contracts: bool = True, + limit: int, + envelope: Envelope | None = None, +) -> list[PeerLink]: + """Other sites that share a type or a contract hub with ``site_id``. + + ``peer_id`` restricts the result to one other site, for explain and for + find_paths. Results are ordered by relation, hub, then peer, so two indexes + of one repository return the same list. + """ + site = store.get_node(site_id) + if site is None or limit <= 0: + return [] + + links: list[PeerLink] = [] + if include_types: + links.extend(_type_links(store, site, envelope)) + if include_contracts: + links.extend(_contract_links(store, site)) + + if peer_id is not None: + links = [link for link in links if link.peer.id == peer_id] + + links.sort(key=lambda link: ( + link.relation, + link.hub.qualified_name, + link.peer.qualified_name, + tuple(sorted(kind.value for kind in link.peer_kinds)), + )) + if len(links) > limit: + if envelope is not None: + envelope.add_boundary( + "peer_limit", + node_id=site_id, + detail=( + f"{len(links)} shared-hub peers; returning {limit}" + ), + ) + links = links[:limit] + return links + + +def _type_links( + store: GraphStore, site: Node, envelope: Envelope | None +) -> list[PeerLink]: + by_hub: dict[str, list[Edge]] = defaultdict(list) + for edge in store.edges_from(site.id, [EdgeKind.HAS_TYPE]): + by_hub[edge.dst].append(edge) + + links: list[PeerLink] = [] + for hub_id, site_edges in by_hub.items(): + hub = store.get_node(hub_id) + if hub is None or not hub.is_type: + continue + incoming = store.edges_to(hub.id, [EdgeKind.HAS_TYPE]) + others = {edge.src for edge in incoming if edge.src != site.id} + if not others: + continue + if not hub.file_path: + if envelope is not None: + envelope.add_boundary( + "peer_undeclared_type", + node_id=hub.id, + detail=( + f"{hub.qualified_name} has no declaration file; " + "no peer is named" + ), + ) + continue + sites = others | {site.id} + if len(sites) > MAX_SHARED_TYPE_SITES: + if envelope is not None: + envelope.add_boundary( + "peer_fan_in", + node_id=hub.id, + detail=( + f"{hub.qualified_name} is referenced by {len(sites)} " + "sites; no peer is named" + ), + ) + continue + grouped = _group_by_src(incoming, exclude=site.id) + for peer_id, peer_edges in grouped.items(): + peer = store.get_node(peer_id) + if peer is None: + continue + links.append(_link( + "shares_type", site, peer, hub, site_edges, peer_edges + )) + return links + + +def _contract_links(store: GraphStore, site: Node) -> list[PeerLink]: + by_hub: dict[str, list[Edge]] = defaultdict(list) + for edge in store.edges_from(site.id, _CONTRACT_KINDS): + by_hub[edge.dst].append(edge) + + links: list[PeerLink] = [] + for hub_id, site_edges in by_hub.items(): + hub = store.get_node(hub_id) + if hub is None or hub.kind is not NodeKind.CONTRACT: + continue + incoming = store.edges_to(hub.id, _CONTRACT_KINDS) + grouped = _group_by_src(incoming, exclude=site.id) + for peer_id, peer_edges in grouped.items(): + peer = store.get_node(peer_id) + if peer is None: + continue + links.append(_link( + "shares_contract", site, peer, hub, site_edges, peer_edges + )) + return links + + +def _link( + relation: str, + site: Node, + peer: Node, + hub: Node, + site_edges: list[Edge], + peer_edges: list[Edge], +) -> PeerLink: + site_edges = sorted(site_edges, key=lambda edge: edge.identity) + peer_edges = sorted(peer_edges, key=lambda edge: edge.identity) + return PeerLink( + relation=relation, + site=site, + peer=peer, + hub=hub, + site_kinds=frozenset(edge.kind for edge in site_edges), + peer_kinds=frozenset(edge.kind for edge in peer_edges), + site_edges=tuple(site_edges), + peer_edges=tuple(peer_edges), + service_relation=_service_relation(site, peer), + ) + + +def _group_by_src( + edges: list[Edge], *, exclude: str +) -> dict[str, list[Edge]]: + grouped: dict[str, list[Edge]] = defaultdict(list) + for edge in edges: + if edge.src == exclude: + continue + grouped[edge.src].append(edge) + return grouped + + +def _service_relation(site: Node, peer: Node) -> str: + if not site.service_id or not peer.service_id: + return "unknown" + if site.service_id == peer.service_id: + return "same_service" + return "cross_service" + + +def _hub_kind(hub: Node, site_edges: list[Edge] | tuple[Edge, ...]) -> str: + """The stored contract kind, or the type node's own kind. + + Contract kinds stay the ones the adapter wrote (``topic``, ``http_route``, + and the rest). No extra kind is introduced for the hop. + """ + for edge in site_edges: + kind = edge.metadata.get("contract_kind") + if kind: + return str(kind) + return hub.kind.value + + +def _role_names(kinds: frozenset[EdgeKind]) -> list[str]: + return sorted(kind.value for kind in kinds) + + +def _attach_verification(out: dict[str, object], link: PeerLink) -> None: + if not link.flow_ambiguous: + return + out["verification"] = ( + TYPE_VERIFICATION if link.relation == "shares_type" else CONTRACT_VERIFICATION + ) + + +def _at(node: Node) -> str: + return f"{node.file_path}:{node.span.start_line}" + + +def _written(edge: Edge) -> str: + if edge.span is not None and edge.file_path: + return f"{edge.file_path}:{edge.span.start_line}" + return edge.file_path or "" + + +def _hop( + node: Node, + edge: Edge | None, + depth: int, + sources: dict[str, str], + reversed_: bool = False, +) -> dict[str, object]: + payload: dict[str, object] = { + "depth": depth, + "node": node.qualified_name, + "kind": node.kind.value, + "at": _at(node), + } + if edge is not None: + payload["via"] = edge.kind.value + payload["evidence"] = edge.evidence.tier.value + if reversed_: + payload["reversed"] = True + if edge.span is not None and edge.file_path: + payload["written_at"] = f"{edge.file_path}:{edge.span.start_line}" + source = sources.get(node.id) + if source is not None: + payload["source"] = source + return payload diff --git a/tests/test_peers.py b/tests/test_peers.py new file mode 100644 index 0000000..2b46a38 --- /dev/null +++ b/tests/test_peers.py @@ -0,0 +1,387 @@ +"""Shared-hub peers. + +The hop is the graph shape "two sites point at one hub". It is not a detector +for one framework. A result states only the edges that were stored, and it +carries ``verification`` only when those edges do not give a direction. +""" + +from __future__ import annotations + +import pytest + +from lineagelens.core import ( + Edge, + EdgeKind, + Evidence, + EvidenceTier, + Node, + NodeKind, + Resolution, + Service, + Span, +) +from lineagelens.query import QueryEngine +from lineagelens.query.peers import ( + CONTRACT_VERIFICATION, + MAX_SHARED_TYPE_SITES, + TYPE_VERIFICATION, +) +from lineagelens.store import GraphStore + +SVC_A = "svc-a" +SVC_B = "svc-b" + + +def _node( + name: str, + kind: NodeKind, + *, + service: str | None = SVC_A, + file_path: str = "src/mod.py", + line: int = 1, + signature: str | None = None, +) -> Node: + return Node( + id=name, + kind=kind, + name=name.rsplit("/", 1)[-1], + qualified_name=name, + lang="python", + span=Span.at_line(line), + file_path=file_path, + service_id=service, + signature=signature, + ) + + +def _edge( + src: Node, + dst: Node, + kind: EdgeKind, + *, + line: int, + tier: EvidenceTier = EvidenceTier.FACT, + label: str = "test", + metadata: dict | None = None, +) -> Edge: + return Edge( + src=src.id, + dst=dst.id, + kind=kind, + evidence=Evidence(tier=tier, label=label), + resolution=Resolution.EXACT, + provenance=label, + span=Span.at_line(line), + file_path=src.file_path or "src/mod.py", + metadata=metadata or {}, + ) + + +def _engine(tmp_path, nodes: list[Node], edges: list[Edge]) -> QueryEngine: + tmp_path.mkdir(parents=True, exist_ok=True) + store = GraphStore.create(tmp_path / "graph.sqlite", project_root=str(tmp_path)) + store.write_services([ + Service(id=SVC_A, name="a", root=".", kind="pypi"), + Service(id=SVC_B, name="b", root="web", kind="npm"), + ]) + store.write_nodes(nodes) + store.write_edges(edges) + return QueryEngine(store, str(tmp_path)) + + +def _peer(report: dict, name: str) -> dict: + matches = [entry for entry in report.get("peers", []) if entry["node"] == name] + assert matches, report.get("peers") + return matches[0] + + +class TestSharedType: + def test_a_declared_type_names_the_other_site_and_asks_for_code(self, tmp_path): + """Enum, struct, and class are the same hop. The entry is not a send.""" + for kind in (NodeKind.ENUM, NodeKind.STRUCT, NodeKind.CLASS): + command = _node("app/Command", kind, line=1, file_path="src/command.py") + send = _node("app/send", NodeKind.FUNCTION, line=10) + recv = _node("app/recv", NodeKind.FUNCTION, line=20) + engine = _engine(tmp_path / kind.value, [command, send, recv], [ + _edge(send, command, EdgeKind.HAS_TYPE, line=11), + _edge(recv, command, EdgeKind.HAS_TYPE, line=21), + ]) + + report = engine.impact_of(send.id).results[0].as_dict() + peer = _peer(report, recv.qualified_name) + assert peer["relation"] == "shares_type" + assert peer["hub_kind"] == kind.value + assert peer["self_role"] == ["HAS_TYPE"] + assert peer["peer_role"] == ["HAS_TYPE"] + assert peer["service_relation"] == "same_service" + assert peer["verification"] == TYPE_VERIFICATION + assert recv.qualified_name not in { + entry["node"] for entry in report.get("in_process", []) + } + assert "via_contract" not in peer + + def test_fan_in_names_no_peer(self, tmp_path): + """A widely shared type is a boundary, not a list of flows.""" + shared = _node("app/Wide", NodeKind.CLASS, line=1, file_path="src/wide.py") + sites = [ + _node(f"app/use{i}", NodeKind.FUNCTION, line=10 + i) + for i in range(MAX_SHARED_TYPE_SITES + 1) + ] + rare = _node("app/Rare", NodeKind.STRUCT, line=2, file_path="src/rare.py") + other = _node("app/other", NodeKind.FUNCTION, line=80) + edges = [ + _edge(site, shared, EdgeKind.HAS_TYPE, line=100 + i) + for i, site in enumerate(sites) + ] + edges.append(_edge(sites[0], rare, EdgeKind.HAS_TYPE, line=200)) + edges.append(_edge(other, rare, EdgeKind.HAS_TYPE, line=201)) + engine = _engine(tmp_path, [shared, rare, other, *sites], edges) + + result = engine.impact_of(sites[0].id) + report = result.results[0].as_dict() + names = {entry["node"] for entry in report.get("peers", [])} + assert other.qualified_name in names + assert all(not name.startswith("app/use") for name in names) + assert "peer_fan_in" in result.envelope.boundaries + detail = " ".join( + item["detail"] for item in result.envelope.boundary_detail + ) + assert "no peer is named" in detail + + def test_a_type_with_no_declaration_file_names_no_peer(self, tmp_path): + phantom = _node("app/Phantom", NodeKind.CLASS, line=1, file_path="") + send = _node("app/send", NodeKind.FUNCTION, line=4) + recv = _node("app/recv", NodeKind.FUNCTION, line=8) + engine = _engine(tmp_path, [phantom, send, recv], [ + _edge(send, phantom, EdgeKind.HAS_TYPE, line=5), + _edge(recv, phantom, EdgeKind.HAS_TYPE, line=9), + ]) + result = engine.impact_of(send.id) + assert result.results[0].as_dict().get("peers", []) == [] + assert "peer_undeclared_type" in result.envelope.boundaries + + def test_impact_of_the_type_still_lists_its_references(self, tmp_path): + """The fan-in cap applies to the sibling hop, not to 'who uses this type'.""" + shared = _node("app/Wide", NodeKind.CLASS, line=1, file_path="src/wide.py") + sites = [ + _node(f"app/use{i}", NodeKind.FUNCTION, line=10 + i) + for i in range(MAX_SHARED_TYPE_SITES + 1) + ] + edges = [ + _edge(site, shared, EdgeKind.HAS_TYPE, line=100 + i) + for i, site in enumerate(sites) + ] + engine = _engine(tmp_path, [shared, *sites], edges) + report = engine.impact_of(shared.id, limit=50).results[0].as_dict() + reached = {entry["node"] for entry in report.get("in_process", [])} + assert {site.qualified_name for site in sites} <= reached + + +class TestSharedContract: + @pytest.mark.parametrize("contract_kind", ["topic", "http_route"]) + def test_complementary_roles_state_the_direction(self, tmp_path, contract_kind): + hub = _node( + f"contract/{contract_kind}/orders", + NodeKind.CONTRACT, + service=None, + file_path="", + signature="orders", + ) + exposer = _node("app/publish", NodeKind.FUNCTION, service=SVC_A, line=3) + consumer = _node( + "web/subscribe", NodeKind.FUNCTION, service=SVC_B, line=9, + file_path="web/client.ts", + ) + meta = {"contract_kind": contract_kind, "contract_key": "orders"} + engine = _engine(tmp_path / contract_kind, [hub, exposer, consumer], [ + _edge( + exposer, hub, EdgeKind.EXPOSES, line=4, + tier=EvidenceTier.HEURISTIC, label="adapter", metadata=meta, + ), + _edge( + consumer, hub, EdgeKind.CONSUMES, line=10, + tier=EvidenceTier.FACT, label="registry", metadata=meta, + ), + ]) + + report = engine.impact_of(exposer.id).results[0].as_dict() + peer = _peer(report, consumer.qualified_name) + assert peer["relation"] == "shares_contract" + assert peer["hub_kind"] == contract_kind + assert peer["self_role"] == ["EXPOSES"] + assert peer["peer_role"] == ["CONSUMES"] + assert peer["evidence_tier"] == "heuristic" + assert peer["service_relation"] == "cross_service" + assert "verification" not in peer + assert any( + entry["node"] == consumer.qualified_name and "via_contract" in entry + for entry in report["cross_service"] + ) + + from_consumer = engine.impact_of(consumer.id).results[0].as_dict() + reverse = _peer(from_consumer, exposer.qualified_name) + assert reverse["peer_role"] == ["EXPOSES"] + assert "verification" not in reverse + assert exposer.qualified_name not in { + entry["node"] for entry in from_consumer.get("cross_service", []) + } + + def test_same_service_is_not_called_cross_service(self, tmp_path): + hub = _node( + "contract/topic/orders", NodeKind.CONTRACT, + service=None, file_path="", signature="orders", + ) + exposer = _node("app/publish", NodeKind.FUNCTION, service=SVC_A, line=3) + consumer = _node("app/handle", NodeKind.FUNCTION, service=SVC_A, line=12) + meta = {"contract_kind": "topic", "contract_key": "orders"} + engine = _engine(tmp_path, [hub, exposer, consumer], [ + _edge(exposer, hub, EdgeKind.EXPOSES, line=4, metadata=meta), + _edge(consumer, hub, EdgeKind.CONSUMES, line=13, metadata=meta), + ]) + report = engine.impact_of(exposer.id).results[0].as_dict() + peer = _peer(report, consumer.qualified_name) + assert peer["service_relation"] == "same_service" + assert "verification" not in peer + assert consumer.qualified_name not in { + entry["node"] for entry in report.get("cross_service", []) + } + + def test_the_same_role_on_both_sides_asks_for_code(self, tmp_path): + hub = _node( + "contract/topic/orders", NodeKind.CONTRACT, + service=None, file_path="", signature="orders", + ) + left = _node("app/publish", NodeKind.FUNCTION, service=SVC_A, line=3) + right = _node("web/also", NodeKind.FUNCTION, service=SVC_B, line=8) + meta = {"contract_kind": "topic"} + engine = _engine(tmp_path, [hub, left, right], [ + _edge(left, hub, EdgeKind.EXPOSES, line=4, metadata=meta), + _edge(right, hub, EdgeKind.EXPOSES, line=9, metadata=meta), + ]) + report = engine.impact_of(left.id).results[0].as_dict() + peer = _peer(report, right.qualified_name) + assert peer["self_role"] == ["EXPOSES"] + assert peer["peer_role"] == ["EXPOSES"] + assert peer["verification"] == CONTRACT_VERIFICATION + assert right.qualified_name not in { + entry["node"] for entry in report.get("cross_service", []) + } + + def test_a_missing_service_is_unknown(self, tmp_path): + hub = _node( + "contract/topic/orders", NodeKind.CONTRACT, + service=None, file_path="", signature="orders", + ) + exposer = _node("app/publish", NodeKind.FUNCTION, service=None, line=3) + consumer = _node("app/handle", NodeKind.FUNCTION, service=SVC_A, line=8) + meta = {"contract_kind": "topic"} + engine = _engine(tmp_path, [hub, exposer, consumer], [ + _edge(exposer, hub, EdgeKind.EXPOSES, line=4, metadata=meta), + _edge(consumer, hub, EdgeKind.CONSUMES, line=9, metadata=meta), + ]) + peer = _peer( + engine.impact_of(exposer.id).results[0].as_dict(), + consumer.qualified_name, + ) + assert peer["service_relation"] == "unknown" + assert "verification" not in peer + + +class TestExplainAndPaths: + def test_explain_draws_the_type_inward_and_asks_for_code(self, tmp_path): + command = _node("app/Command", NodeKind.ENUM, line=1, file_path="src/command.py") + send = _node("app/send", NodeKind.FUNCTION, line=10) + recv = _node("app/recv", NodeKind.FUNCTION, line=20) + engine = _engine(tmp_path, [command, send, recv], [ + _edge(send, command, EdgeKind.HAS_TYPE, line=11), + _edge(recv, command, EdgeKind.HAS_TYPE, line=21), + ]) + result = engine.explain(send.id, recv.id) + assert result.results + entry = result.results[0] + assert entry["kind"] == "shares_type" + assert entry["path"] == "send -HAS_TYPE-> Command <-HAS_TYPE- recv" + assert entry["verification"] == TYPE_VERIFICATION + + def test_a_direct_edge_is_the_explanation(self, tmp_path): + send = _node("app/send", NodeKind.FUNCTION, line=10) + recv = _node("app/recv", NodeKind.FUNCTION, line=20) + command = _node("app/Command", NodeKind.ENUM, line=1, file_path="src/command.py") + engine = _engine(tmp_path, [send, recv, command], [ + _edge(send, recv, EdgeKind.CALLS, line=12, label="call"), + _edge(send, command, EdgeKind.HAS_TYPE, line=11), + _edge(recv, command, EdgeKind.HAS_TYPE, line=21), + ]) + result = engine.explain(send.id, recv.id) + assert [entry["kind"] for entry in result.results] == ["CALLS"] + assert "verification" not in result.results[0] + + def test_find_paths_keeps_a_call_chain_ahead_of_a_hub(self, tmp_path): + send = _node("app/send", NodeKind.FUNCTION, line=10) + recv = _node("app/recv", NodeKind.FUNCTION, line=20) + command = _node("app/Command", NodeKind.ENUM, line=1, file_path="src/command.py") + engine = _engine(tmp_path, [send, recv, command], [ + _edge(send, recv, EdgeKind.CALLS, line=12), + _edge(send, command, EdgeKind.HAS_TYPE, line=11), + _edge(recv, command, EdgeKind.HAS_TYPE, line=21), + ]) + result = engine.find_paths(send.id, recv.id) + assert result.ranking == "shortest_path_first" + assert result.results[0]["direction"] == "forward" + assert all(item.get("relation") != "shares_type" for item in result.results) + + def test_find_paths_reports_a_type_hub_only_when_nothing_else_connects( + self, tmp_path + ): + command = _node("app/Command", NodeKind.ENUM, line=1, file_path="src/command.py") + send = _node("app/send", NodeKind.FUNCTION, line=10) + recv = _node("app/recv", NodeKind.FUNCTION, line=20) + engine = _engine(tmp_path, [command, send, recv], [ + _edge(send, command, EdgeKind.HAS_TYPE, line=11), + _edge(recv, command, EdgeKind.HAS_TYPE, line=21), + ]) + result = engine.find_paths(send.id, recv.id) + assert result.ranking == "shared_hub" + entry = result.results[0] + assert entry["relation"] == "shares_type" + assert entry["hops"][2]["reversed"] is True + assert entry["hops"][2]["via"] == "HAS_TYPE" + assert entry["verification"] == TYPE_VERIFICATION + + calls_only = engine.find_paths(send.id, recv.id, kinds=["CALLS"]) + assert calls_only.results == [] + assert calls_only.ranking == "shortest_path_first" + + def test_find_paths_states_a_recorded_contract_direction(self, tmp_path): + hub = _node( + "contract/http_route/orders", NodeKind.CONTRACT, + service=None, file_path="", signature="/orders", + ) + exposer = _node("app/publish", NodeKind.FUNCTION, service=SVC_A, line=3) + consumer = _node("web/subscribe", NodeKind.FUNCTION, service=SVC_B, line=9) + meta = {"contract_kind": "http_route"} + engine = _engine(tmp_path, [hub, exposer, consumer], [ + _edge(exposer, hub, EdgeKind.EXPOSES, line=4, metadata=meta), + _edge(consumer, hub, EdgeKind.CONSUMES, line=10, metadata=meta), + ]) + result = engine.find_paths(exposer.id, consumer.id) + assert result.ranking == "shared_hub" + entry = result.results[0] + assert entry["relation"] == "shares_contract" + assert entry["hub_kind"] == "http_route" + assert entry["self_role"] == ["EXPOSES"] + assert entry["peer_role"] == ["CONSUMES"] + assert "verification" not in entry + + def test_search_does_not_attach_peers(self, tmp_path): + command = _node("app/Command", NodeKind.ENUM, line=1, file_path="src/command.py") + send = _node("app/send", NodeKind.FUNCTION, line=10) + recv = _node("app/recv", NodeKind.FUNCTION, line=20) + engine = _engine(tmp_path, [command, send, recv], [ + _edge(send, command, EdgeKind.HAS_TYPE, line=11), + _edge(recv, command, EdgeKind.HAS_TYPE, line=21), + ]) + result = engine.search("send") + assert result.results + assert "peers" not in result.results[0] + assert "verification" not in result.results[0]