diff --git a/ir/__init__.py b/ir/__init__.py index 4753953..64e1592 100644 --- a/ir/__init__.py +++ b/ir/__init__.py @@ -37,6 +37,13 @@ sentence_window_policy, ) from .formulate import Formulator, make_llm_formulator +from .graph import ( + CorpusGraph, + EdgeExtractor, + GraphStore, + canonical_node_id, + default_edge_extractor, +) from .index import Corpus, build, open_corpus from .registry import retriever_for, retrievers from .retrieve import Retriever, as_retriever, fuse_hits, records_for_artifact @@ -76,6 +83,11 @@ "retriever_for", "fuse_hits", "records_for_artifact", + "GraphStore", + "CorpusGraph", + "EdgeExtractor", + "default_edge_extractor", + "canonical_node_id", "expand", "Passage", "NeighborhoodPolicy", diff --git a/ir/graph.py b/ir/graph.py new file mode 100644 index 0000000..6dd1bf4 --- /dev/null +++ b/ir/graph.py @@ -0,0 +1,211 @@ +"""The semantic link graph — typed edges between artifacts (report 12). + +ir indexes artifacts that *refer to each other*: a package depends on other +packages, a skill belongs to a parent, a chunk follows the previous one, a +report cites another. This module models those references as **typed, directed +edges** between nodes identified by ``(source, artifact_id)`` (ir's canonical +node identity since 0.1.16), queryable at retrieval time — the substrate the +traversal operator (``traverse``, follow-up) walks. + +Two pieces: + +- the :data:`GraphStore` **protocol** — a minimal ``__getitem__`` + + ``neighbors`` structural contract, so a traversal operator binds to *any* + conforming store, not only ir's; +- :class:`CorpusGraph` — the corpus-backed implementation. Node payloads are + the artifact's stored records (via :func:`ir.retrieve.records_for_artifact`); + neighbors come from the ``links`` view on :class:`~ir.store.CorpusStore`. + +Edges are **regenerable derived state** (persisted in the ``links`` view, like +``calibration``) — never part of build identity. They are ingested at build +time by an injectable :data:`EdgeExtractor`; the shipped +:func:`default_edge_extractor` turns the edge data already latent in the index +(``Package`` ``deps`` → ``REF``, ``Skill`` ``parent`` → ``PARENT``) into edges. + +Vocabulary (documented, **not** enforced): :data:`NEXT` / :data:`PREV` / +:data:`PARENT` / :data:`CHILD` / :data:`REF`. ``NEXT`` / ``PREV`` are *not* +materialized — they are derivable from the ledger's surface order (#44); the +view stores only what isn't derivable. + +**Naming guard** (ADR #43): this is the **semantic link graph** — directed, +**possibly cyclic** (citations, cross-references), traversed at query time. It +is *not* ``ef.artifact_graph``, which is an acyclic build-time +derivation/lineage DAG. The id conventions are kept compatible so a later +unification stays open, but the two are distinct substrates. +""" + +from __future__ import annotations + +import re +from collections.abc import Iterable, Mapping +from typing import Any, Callable, Protocol, runtime_checkable + +from .base import Record +from .retrieve import records_for_artifact + +# ---- edge vocabulary (documented, not enforced) --------------------------- # + +#: Next sibling surface in sequence (derivable from the ledger — not stored). +NEXT = "NEXT" +#: Previous sibling surface in sequence (derivable from the ledger — not stored). +PREV = "PREV" +#: Containment up: a child points at its parent (e.g. a skill → its package). +PARENT = "PARENT" +#: Containment down: a parent points at a child. +CHILD = "CHILD" +#: A generic reference / dependency / citation (e.g. a package → its deps). +REF = "REF" + +#: A node's identity: ``artifact_id`` within one corpus, or ``(source, +#: artifact_id)`` across corpora. The visited-set of a traversal keys on this. +NodeId = "str | tuple[str, str]" + + +@runtime_checkable +class GraphStore(Protocol): + """Structural contract a traversal operator binds to — node + neighbors. + + Deliberately minimal (two methods) so it is satisfied by ir's + :class:`CorpusGraph` *and* by any external graph: ``__getitem__`` resolves + a node id to its scorable payload, ``neighbors`` lists adjacent node ids + (optionally of one edge type). Granularity-agnostic on purpose — an + artifact graph and a surface-level tree are both ``GraphStore``s. + + ``runtime_checkable`` makes ``isinstance(x, GraphStore)`` a *structural* + check on attribute **names** only (not signatures) — enough to tell a + conforming adapter from an arbitrary object, but it cannot validate that + ``neighbors`` takes the right arguments; treat it as a smoke check. + """ + + def __getitem__(self, node_id: Any) -> Any: ... + + def neighbors(self, node_id: Any, *, edge_type: str | None = None) -> Iterable: ... + + +class CorpusGraph: + """A :data:`GraphStore` over one corpus — artifact nodes, ``links`` edges. + + ``node_id`` is an ``artifact_id``. ``graph[aid]`` is the artifact's stored + records (its scorable surfaces, in plan order); ``graph.neighbors(aid, + edge_type=...)`` reads the corpus store's ``links`` view. Single-corpus, so + it resolves **intra-corpus** targets; cross-corpus ``[source, artifact_id]`` + targets are returned by :meth:`neighbors` verbatim but ``__getitem__`` only + dereferences ids in *this* corpus (federated traversal is a follow-up). + """ + + def __init__(self, store_or_corpus: Any): + self.store = getattr(store_or_corpus, "store", store_or_corpus) + #: The corpus name, when known — the ``source`` half of this graph's + #: node identities (``None`` for a bare store). + self.source = getattr(store_or_corpus, "name", None) + + def __getitem__(self, node_id: str) -> list[Record]: + """The artifact's records, plan-ordered (``NoLedgerEntry`` if unknown).""" + return records_for_artifact(self.store, node_id) + + def neighbors(self, node_id: str, *, edge_type: str | None = None) -> list: + """Outgoing neighbor ids of *node_id*, optionally of one *edge_type*. + + Returns target ids in stored form (a bare ``artifact_id``, whose source + is *this* graph's :attr:`source`; or a ``[source, artifact_id]`` list + for a cross-corpus edge), de-duplicated with first-seen order + preserved. An artifact with no edges — or a store without a links view — + yields ``[]``. Pass a canonical ``(source, artifact_id)`` to + :func:`canonical_node_id` for a traversal's visited-set. + + *node_id* is an intra-corpus ``artifact_id`` (a ``str``); a cross-corpus + target fetched from another graph is out of contract here (it has no + edges *in this corpus*). + """ + edges = self.store.get_links(node_id) + if edge_type is not None: + targets = list(edges.get(edge_type, [])) + else: + targets = [t for ts in edges.values() for t in ts] + seen: set = set() + out: list = [] + for t in targets: + key = tuple(t) if isinstance(t, list) else t + if key not in seen: + seen.add(key) + out.append(t) + return out + + def edge_types(self, node_id: str) -> list[str]: + """The edge types present on *node_id* (``[]`` if none).""" + return list(self.store.get_links(node_id)) + + +def canonical_node_id(target: Any, *, source: str | None) -> tuple[str | None, str]: + """Canonicalize a neighbor *target* to a ``(source, artifact_id)`` node id. + + The repo's node identity is ``(source, artifact_id)`` — the key a traversal + visited-set must use so the same id in two corpora stays two nodes. + :meth:`CorpusGraph.neighbors` returns targets in stored form: a bare + ``artifact_id`` (implicitly in *source*, the graph it came from) or a + ``[source, artifact_id]`` cross-corpus pair. This resolves either to the + canonical tuple. + + >>> canonical_node_id("dol", source="packages") + ('packages', 'dol') + >>> canonical_node_id(["skills", "deploy"], source="packages") + ('skills', 'deploy') + """ + if isinstance(target, (list, tuple)) and len(target) == 2: + return (target[0], target[1]) + return (source, target) + + +# ---- edge ingest ---------------------------------------------------------- # + +#: An edge extractor: ``(artifact_id, filter_fields) -> {edge_type: [target]}``. +#: An injectable build-time seam (like :data:`~ir.formulate.Formulator`) that +#: turns an artifact's filter fields into outgoing edges. ``build(..., +#: edge_extractor=...)`` runs it; :func:`default_edge_extractor` is shipped. +EdgeExtractor = Callable[[str, Mapping[str, Any]], "Mapping[str, list]"] + +#: Split a PEP 508 dependency spec at the first version/extra/marker/url +#: delimiter, leaving the bare distribution name (``"numpy[fast]>=1.2; +#: python>='3.9'"`` → ``"numpy"``; the direct-reference ``"dol@git+https://…"`` +#: → ``"dol"``). ``@`` is included so PEP 508 URL deps yield the bare name. +_DEP_DELIM = re.compile(r"[<>=!~;@\[\s(]") + + +def _dep_name(dep: str) -> str: + """The bare distribution name of a dependency spec (lower-cased, stripped).""" + return _DEP_DELIM.split(dep.strip(), 1)[0].strip().lower() + + +def default_edge_extractor( + artifact_id: str, filter_fields: Mapping[str, Any] +) -> dict[str, list]: + """Edges latent in the standard filter fields: ``deps`` → REF, ``parent`` → PARENT. + + - ``deps`` (Package) → :data:`REF` edges to each dependency's bare name + (version specifiers / extras / markers stripped). Self-edges and blanks + are dropped. + - ``parent`` (Skill) → a single :data:`PARENT` edge. + + A package whose ``deps`` name other packages in the same corpus gets + intra-corpus REF edges; third-party deps become REF edges to ids not in the + corpus (harmless — :meth:`CorpusGraph.neighbors` lists them, and a + traversal simply finds no node to expand). + + Self-edges are dropped case-insensitively (``_dep_name`` lower-cases, so a + package ``"AA"`` depending on ``"aa"`` is recognized as a self-reference). + """ + edges: dict[str, list] = {} + self_id = artifact_id.lower() + deps = filter_fields.get("deps") + if isinstance(deps, (list, tuple)): + refs = [] + for dep in deps: + name = _dep_name(str(dep)) + if name and name != self_id: + refs.append(name) + if refs: + edges[REF] = list(dict.fromkeys(refs)) # de-dup, keep order + parent = filter_fields.get("parent") + if isinstance(parent, str) and parent and parent.lower() != self_id: + edges[PARENT] = [parent] + return edges diff --git a/ir/index.py b/ir/index.py index 57963a3..c495b90 100644 --- a/ir/index.py +++ b/ir/index.py @@ -92,6 +92,7 @@ def build( embedder: Any = None, full: bool = True, batch_size: int = 256, + edge_extractor: Callable | None = None, ) -> Corpus: """Build or incrementally update *source* into a :class:`Corpus`. @@ -101,6 +102,18 @@ def build( embedder : override the source's embedder spec. full : when True (default), prune artifacts no longer in the source. batch_size : embedding batch size. + edge_extractor : an optional :data:`~ir.graph.EdgeExtractor` + (``(artifact_id, filter_fields) -> {edge_type: [target]}``) that + populates the corpus's semantic ``links`` graph (see :mod:`ir.graph`; + pass :func:`ir.default_edge_extractor` for the latent deps/parent + edges). Ingest is **eager** — edges are (re)written for *every* + in-scope artifact, a decompose-only pass with no embedding, so the + graph never goes partially stale — while embedding stays fully + incremental. Edges are derived state, **not** part of build identity. + A rebuild *without* an extractor leaves existing edges untouched + (they are only refreshed by re-running with one, and only cleared per + artifact by the ``full`` prune below) — so dropping ``edge_extractor`` + does not wipe a graph. """ store = CorpusStore.local(source.name) if store is None else store spec = embedder if embedder is not None else source.embedder @@ -115,16 +128,23 @@ def build( version = source.change_signal(artifact_id, raw) key = ledger_key(artifact_id) prev = store.get_ledger_entry(key) - if ( + unchanged = ( prev and prev.get("version") == version and prev.get("embedder_id") == emb_id and prev.get("strategy_id") == strat_id - ): - continue # unchanged content + same model + same strategy + ) + # Unchanged + no edges to refresh → nothing to do for this artifact. + if unchanged and edge_extractor is None: + continue meta_extra = source.metadata_of(artifact_id, raw) if source.metadata_of else {} plan = source.indexing_strategy.decompose(artifact_id, raw, meta_extra) - changed[artifact_id] = (key, version, prev, plan) + if not unchanged: + changed[artifact_id] = (key, version, prev, plan) + if edge_extractor is not None: + store.set_links( + artifact_id, edge_extractor(artifact_id, plan.filter_fields) + ) # Embed all surfaces of changed artifacts together (cache makes this cheap). flat = [ @@ -178,6 +198,7 @@ def build( for rid in entry.get("record_ids", []): store.delete_record(rid) store.delete_ledger_entry(key) + store.delete_links(entry.get("artifact_id", "")) store.set_config( { diff --git a/ir/store.py b/ir/store.py index 41f1c01..e7a72d5 100644 --- a/ir/store.py +++ b/ir/store.py @@ -13,8 +13,13 @@ Kept apart from ``config`` on purpose — a calibration is regenerable, derived from an eval run, and not part of the corpus's build identity, so it must never clobber (or be clobbered by) the build settings. +- ``links`` : ``artifact_id -> {edge_type: [target, ...]}`` (the semantic link + graph — typed directed edges between artifacts; see :mod:`ir.graph`). Like + ``calibration`` it is regenerable derived state, kept out of build identity; a + target is a bare ``artifact_id`` (intra-corpus) or a ``[source, artifact_id]`` + pair (cross-corpus). Optional — an absent view is simply "no edges". -The default factory :meth:`CorpusStore.local` roots all five under +The default factory :meth:`CorpusStore.local` roots all six under ``~/.local/share/ir/corpora/`` via ``dol`` file stores; :meth:`CorpusStore.memory` gives a dependency-free in-memory store for tests. Brute-force search reads vectors into a single normalized matrix @@ -72,14 +77,17 @@ def __init__( ledger: MutableMapping[str, Any], config: MutableMapping[str, Any], calibration: MutableMapping[str, Any] | None = None, + links: MutableMapping[str, Any] | None = None, ): self.meta = meta self.vectors = vectors self.ledger = ledger self.config = config - # Optional 5th view (defaults to an in-memory dict so older call sites and - # tests that construct a store with four views keep working unchanged). + # Optional 5th/6th views (default to in-memory dicts so older call sites + # and tests that construct a store with fewer views keep working — and an + # absent links view simply means "no edges"). self.calibration = {} if calibration is None else calibration + self.links = {} if links is None else links self._matrix_cache: tuple | None = None # ----- factories ------------------------------------------------------ # @@ -96,6 +104,7 @@ def local(cls, name: str) -> "CorpusStore": ledger=_json_store(root / "ledger"), config=_json_store(root / "config"), calibration=_json_store(root / "calibration"), + links=_json_store(root / "links"), ) @classmethod @@ -205,6 +214,37 @@ def calibration_modes(self) -> list[str]: """The ranking modes that currently have a stored calibration.""" return list(self.calibration) + # ----- links (semantic edge graph) ------------------------------------ # + + def get_links(self, artifact_id: str) -> dict: + """The outgoing edges of *artifact_id* — ``{edge_type: [target, ...]}``. + + Empty dict when the artifact has no stored edges (or no links view). + A copy, so a caller cannot mutate the persisted adjacency in place. + """ + return copy.deepcopy(self.links.get(artifact_id, {})) + + def set_links(self, artifact_id: str, edges: Mapping[str, Any]) -> None: + """Persist *artifact_id*'s outgoing *edges* (``{edge_type: [target]}``). + + Empty edge-type lists are dropped; an empty result deletes the entry + (no empty adjacency rows linger). Targets are stored verbatim — a bare + ``artifact_id`` or a ``[source, artifact_id]`` pair. + """ + cleaned = {et: list(ts) for et, ts in edges.items() if ts} + if cleaned: + self.links[artifact_id] = cleaned + else: + self.links.pop(artifact_id, None) + + def delete_links(self, artifact_id: str) -> None: + """Remove an artifact's edges; a missing entry is tolerated.""" + self.links.pop(artifact_id, None) + + def link_items(self) -> Iterator[tuple[str, dict]]: + """Iterate ``(artifact_id, {edge_type: [target]})`` adjacency pairs.""" + return iter(list(self.links.items())) + # ----- search matrix -------------------------------------------------- # def matrix(self) -> tuple[list[str], np.ndarray, list[dict]]: diff --git a/tests/test_graph.py b/tests/test_graph.py new file mode 100644 index 0000000..42ee772 --- /dev/null +++ b/tests/test_graph.py @@ -0,0 +1,350 @@ +"""Tests for the semantic link graph (#46) — links view, GraphStore, ingest. + +Pins the #46 acceptance: the ``links`` view round-trips (memory + file), edge +ingest turns Package ``deps`` / Skill ``parent`` into REF / PARENT neighbors, +the corpus-backed adapter satisfies the ``GraphStore`` protocol, edges are +derived state outside build identity, and an absent view degrades to "no +edges". Hermetic: light embedder + in-memory / tmp stores. +""" + +import json + +import pytest + +import ir +from ir.base import ledger_key +from ir.graph import ( + CHILD, + NEXT, + PARENT, + PREV, + REF, + CorpusGraph, + GraphStore, + _dep_name, + canonical_node_id, + default_edge_extractor, +) +from ir.retrieve import NoLedgerEntry +from ir.store import CorpusStore + +PKG = { + "aa": { + "name": "aa", + "description": "package a", + "readme": "", + "owner": "ours", + "deps": ["dol>=1.0", "numpy[fast]>=1.2 ; python_version>='3.9'", "aa"], + }, + "bb": {"name": "bb", "description": "package b", "readme": "", "deps": ["aa"]}, + "cc": {"name": "cc", "description": "package c", "readme": "", "deps": []}, +} + +SKILLS = { + "deploy": {"name": "deploy", "description": "push apps", "parent": "tw"}, + "build": {"name": "build", "description": "compile", "parent": "tw"}, +} + + +def _packages(name="pkgs", store=None, edges=True): + # NB: explicit `is None` — an empty CorpusStore is falsy (len 0), so + # `store or CorpusStore.memory()` would discard a freshly-passed store. + src = ir.CorpusSource.from_mapping(PKG, name=name, strategy=ir.Package()) + return ir.build( + src, + store=CorpusStore.memory() if store is None else store, + embedder="light", + edge_extractor=default_edge_extractor if edges else None, + ) + + +def _skills(store=None): + src = ir.CorpusSource.from_mapping(SKILLS, name="sk", strategy=ir.Skill()) + return ir.build( + src, + store=CorpusStore.memory() if store is None else store, + embedder="light", + edge_extractor=default_edge_extractor, + ) + + +# --------------------------------------------------------------------------- # +# links view — round-trip, hygiene +# --------------------------------------------------------------------------- # + + +def test_links_view_round_trips_in_memory(): + s = CorpusStore.memory() + s.set_links("aa", {"REF": ["dol", "i2"]}) + assert s.get_links("aa") == {"REF": ["dol", "i2"]} + assert list(s.link_items()) == [("aa", {"REF": ["dol", "i2"]})] + + +def test_links_view_round_trips_file_backed(tmp_path, monkeypatch): + monkeypatch.setenv("IR_DATA_DIR", str(tmp_path / "data")) + monkeypatch.setenv("IR_CONFIG_DIR", str(tmp_path / "config")) + monkeypatch.setenv("IR_CACHE_DIR", str(tmp_path / "cache")) + s = CorpusStore.local("g") + s.set_links("aa", {"REF": ["dol"], "PARENT": [["packages", "tw"]]}) + reopened = CorpusStore.local("g") + got = reopened.get_links("aa") + assert got == {"REF": ["dol"], "PARENT": [["packages", "tw"]]} + json.dumps(got) # JSON-clean + + +def test_set_links_drops_empty_edge_lists_and_rows(): + s = CorpusStore.memory() + s.set_links("aa", {"REF": ["dol"], "PARENT": []}) + assert s.get_links("aa") == {"REF": ["dol"]} + s.set_links("aa", {"REF": []}) # all-empty → row removed + assert s.get_links("aa") == {} + assert list(s.link_items()) == [] + + +def test_get_links_returns_a_copy(): + s = CorpusStore.memory() + s.set_links("aa", {"REF": ["dol"]}) + got = s.get_links("aa") + got["REF"].append("MUTATED") + assert s.get_links("aa") == {"REF": ["dol"]} # store untouched + + +def test_absent_links_view_degrades_to_no_edges(): + s = CorpusStore.memory() # 4-arg construction, links defaults to {} + assert s.get_links("anything") == {} + assert CorpusGraph(s).neighbors("anything") == [] + + +# --------------------------------------------------------------------------- # +# default_edge_extractor + _dep_name +# --------------------------------------------------------------------------- # + + +@pytest.mark.parametrize( + "spec,expected", + [ + ("dol", "dol"), + ("dol>=1.0", "dol"), + ("numpy[fast]>=1.2", "numpy"), + ("Requests >= 2 ; python_version>='3.9'", "requests"), + ("pkg (>=1.0)", "pkg"), + ("dol@git+https://github.com/i2mint/dol.git", "dol"), # PEP 508 URL dep + ("dol @ git+https://github.com/i2mint/dol.git@main", "dol"), # spaced + ], +) +def test_dep_name_strips_specifiers(spec, expected): + assert _dep_name(spec) == expected + + +def test_default_extractor_self_edge_drop_is_case_insensitive(): + # _dep_name lower-cases, so an "AA" package depending on "aa" (or "AA") + # is recognized as a self-reference and dropped. + assert default_edge_extractor("AA", {"deps": ["AA", "dol"]}) == {"REF": ["dol"]} + + +def test_default_extractor_dedups_duplicate_deps(): + edges = default_edge_extractor("x", {"deps": ["dol", "dol>=1.0", "i2"]}) + assert edges == {"REF": ["dol", "i2"]} + + +def test_default_extractor_handles_url_dep(): + edges = default_edge_extractor("x", {"deps": ["dol@git+https://h/i2mint/dol"]}) + assert edges == {"REF": ["dol"]} + + +@pytest.mark.parametrize( + "target,source,expected", + [ + ("dol", "packages", ("packages", "dol")), + (["skills", "deploy"], "packages", ("skills", "deploy")), + ("x", None, (None, "x")), + ], +) +def test_canonical_node_id(target, source, expected): + assert canonical_node_id(target, source=source) == expected + + +def test_default_extractor_deps_to_ref_and_parent(): + edges = default_edge_extractor("aa", PKG["aa"]) + # version specifiers / extras stripped; self-edge "aa" dropped; de-duped + assert edges == {"REF": ["dol", "numpy"]} + assert default_edge_extractor("deploy", SKILLS["deploy"]) == {"PARENT": ["tw"]} + assert default_edge_extractor("cc", PKG["cc"]) == {} # empty deps → no edges + + +# --------------------------------------------------------------------------- # +# CorpusGraph — neighbors, node payload, protocol +# --------------------------------------------------------------------------- # + + +def test_neighbors_returns_package_deps(): + g = CorpusGraph(_packages()) + assert g.neighbors("aa", edge_type=REF) == ["dol", "numpy"] + assert g.neighbors("bb", edge_type=REF) == ["aa"] + assert g.neighbors("cc", edge_type=REF) == [] # empty deps + + +def test_neighbors_all_edge_types_when_unfiltered(): + s = CorpusStore.memory() + s.set_links("aa", {"REF": ["dol", "i2"], "PARENT": ["grp"]}) + assert CorpusGraph(s).neighbors("aa") == ["dol", "i2", "grp"] + + +def test_neighbors_dedups_preserving_order(): + s = CorpusStore.memory() + s.set_links("aa", {"REF": ["dol", "i2"], "PARENT": ["dol"]}) + assert CorpusGraph(s).neighbors("aa") == ["dol", "i2"] + + +def test_neighbors_preserves_cross_corpus_pair_targets(): + s = CorpusStore.memory() + s.set_links("deploy", {"PARENT": [["packages", "tw"]]}) + assert CorpusGraph(s).neighbors("deploy", edge_type=PARENT) == [["packages", "tw"]] + + +def test_neighbors_dedups_cross_corpus_pairs_unfiltered(): + s = CorpusStore.memory() + s.set_links("a", {"REF": [["c", "x"], "y"], "PARENT": [["c", "x"]]}) + # The [c, x] pair appears under two edge types — deduped once (lists are + # unhashable, so dedup keys on the tuple form). + assert CorpusGraph(s).neighbors("a") == [["c", "x"], "y"] + + +def test_neighbors_round_trips_through_a_reopened_file_store(tmp_path, monkeypatch): + monkeypatch.setenv("IR_DATA_DIR", str(tmp_path / "data")) + monkeypatch.setenv("IR_CONFIG_DIR", str(tmp_path / "config")) + monkeypatch.setenv("IR_CACHE_DIR", str(tmp_path / "cache")) + src = ir.CorpusSource.from_mapping(PKG, name="fpkgs", strategy=ir.Package()) + ir.build(src, embedder="light", edge_extractor=default_edge_extractor) + reopened = ir.open_corpus("fpkgs") + assert CorpusGraph(reopened).neighbors("aa", edge_type=REF) == ["dol", "numpy"] + + +def test_skill_collision_ids_keep_correct_parent_edges(): + # Same-named skills from different parents get collision-safe ids + # ("name@parent"); the PARENT edge must still point at each one's parent, + # and the id ("deploy@ops") must not be mistaken for a self-edge. + skills = { + "deploy": {"name": "deploy", "description": "d", "parent": "tw"}, + "deploy@ops": {"name": "deploy", "description": "d2", "parent": "ops"}, + } + src = ir.CorpusSource.from_mapping(skills, name="skc", strategy=ir.Skill()) + g = CorpusGraph( + ir.build( + src, + store=CorpusStore.memory(), + embedder="light", + edge_extractor=default_edge_extractor, + ) + ) + assert g.neighbors("deploy", edge_type=PARENT) == ["tw"] + assert g.neighbors("deploy@ops", edge_type=PARENT) == ["ops"] + + +def test_skill_parent_becomes_parent_edge(): + g = CorpusGraph(_skills()) + assert g.neighbors("deploy", edge_type=PARENT) == ["tw"] + + +def test_getitem_returns_node_records(): + corpus = _packages() + g = CorpusGraph(corpus) + recs = g["aa"] + assert recs and all(r.artifact_id == "aa" for r in recs) + assert any(r.surface_kind == "description" for r in recs) + + +def test_getitem_unknown_artifact_raises(): + g = CorpusGraph(_packages()) + with pytest.raises(NoLedgerEntry): + g["nope"] + + +def test_corpus_graph_satisfies_graphstore_protocol(): + g = CorpusGraph(_packages()) + assert isinstance(g, GraphStore) + # A bare object missing neighbors is NOT a GraphStore. + assert not isinstance(object(), GraphStore) + + +def test_corpus_graph_accepts_store_or_corpus(): + corpus = _packages() + via_corpus = CorpusGraph(corpus) + via_store = CorpusGraph(corpus.store) + assert via_corpus.neighbors("aa", edge_type=REF) == via_store.neighbors( + "aa", edge_type=REF + ) + assert via_corpus.source == "pkgs" and via_store.source is None + + +def test_edge_vocabulary_constants(): + assert (NEXT, PREV, PARENT, CHILD, REF) == ( + "NEXT", + "PREV", + "PARENT", + "CHILD", + "REF", + ) + + +# --------------------------------------------------------------------------- # +# Edges are derived state, outside build identity +# --------------------------------------------------------------------------- # + + +def test_edges_do_not_change_build_identity(): + with_edges = _packages(name="x", edges=True) + without = _packages(name="x", edges=False) + # Same records, same ledger entries (edge ingest touches only the links view). + assert set(with_edges.store.record_ids()) == set(without.store.record_ids()) + for aid in PKG: + e1 = with_edges.store.get_ledger_entry(ledger_key(aid)) + e0 = without.store.get_ledger_entry(ledger_key(aid)) + assert e1 == e0 # ledger carries no edge data + # ... yet only the edge-ingested build has edges. + assert CorpusGraph(with_edges).neighbors("aa", edge_type=REF) == ["dol", "numpy"] + assert CorpusGraph(without).neighbors("aa", edge_type=REF) == [] + + +def test_edge_ingest_is_eager_on_rebuild(): + # Build WITHOUT edges, then rebuild the same store WITH an extractor: + # every artifact gains edges (not just content-changed ones), because + # ingest decomposes the whole scope, not only the changed set. + store = CorpusStore.memory() + _packages(name="x", store=store, edges=False) + assert CorpusGraph(store).neighbors("aa", edge_type=REF) == [] + _packages(name="x", store=store, edges=True) # nothing changed but edges + assert CorpusGraph(store).neighbors("aa", edge_type=REF) == ["dol", "numpy"] + assert CorpusGraph(store).neighbors("bb", edge_type=REF) == ["aa"] + + +def test_prune_deletes_links(): + store = CorpusStore.memory() + src_full = ir.CorpusSource.from_mapping(PKG, name="x", strategy=ir.Package()) + ir.build( + src_full, store=store, embedder="light", edge_extractor=default_edge_extractor + ) + assert CorpusGraph(store).neighbors("bb", edge_type=REF) == ["aa"] + smaller = {k: v for k, v in PKG.items() if k != "bb"} + src_small = ir.CorpusSource.from_mapping(smaller, name="x", strategy=ir.Package()) + ir.build( + src_small, store=store, embedder="light", edge_extractor=default_edge_extractor + ) + assert store.get_links("bb") == {} # pruned artifact's edges removed + assert CorpusGraph(store).neighbors("bb") == [] + + +def test_custom_edge_extractor_is_injectable(): + def only_first_dep(aid, fields): + deps = fields.get("deps") or [] + return {"REF": [str(deps[0]).split(">")[0]]} if deps else {} + + g = CorpusGraph(_packages_custom(only_first_dep)) + assert g.neighbors("aa", edge_type=REF) == ["dol"] + + +def _packages_custom(extractor): + src = ir.CorpusSource.from_mapping(PKG, name="pc", strategy=ir.Package()) + return ir.build( + src, store=CorpusStore.memory(), embedder="light", edge_extractor=extractor + )