fix: preserve relation- and origin-distinct graph paths

datamind/capabilities/graph/providers/networkx_store.py

@@ -298,19 +298,29 @@ async def traverse(
         relation_filter: list[str] | None = None,
         max_results: int = 100,
     ) -> list[GraphPath]:
-        if not self._g.has_node(start):
+        """Return edge-distinct simple paths, including parallel relations.
+
+        Traverse breadth-first in stable target/relation/edge-key order and
+        stop after ``max_results`` paths. Sort that bounded selection by score;
+        this is not an exhaustive global top-k search. Each path excludes
+        repeated nodes, while different edge identities may share its nodes.
+        """
+        if max_hops <= 0 or max_results <= 0 or not self._g.has_node(start):
             return []
         allowed = set(relation_filter) if relation_filter else None
 
         # BFS over (node, path_edges) up to max_hops.
         paths: list[GraphPath] = []
-        visited: set[tuple[str, ...]] = set()
         frontier: list[tuple[str, list[Edge], set[str]]] = [(start, [], {start})]
         depth = 0
         while frontier and depth < max_hops:
             next_frontier: list[tuple[str, list[Edge], set[str]]] = []
             for node, edges_so_far, seen in frontier:
-                for u, v, d in self._g.out_edges(node, data=True):
+                outgoing = sorted(
+                    self._g.out_edges(node, keys=True, data=True),
+                    key=lambda e: (e[1], str(e[3].get("relation", "related")), str(e[2])),
+                )
+                for u, v, _edge_key, d in outgoing:
                     rel = d.get("relation", "related")
                     if allowed is not None and rel not in allowed:
                         continue
@@ -319,10 +329,9 @@ async def traverse(
                     edge_obj = self._edge(u, v, d)
                     path_edges = edges_so_far + [edge_obj]
                     nodes = [start] + [e.target for e in path_edges]
-                    key = tuple(nodes)
-                    if key in visited:
-                        continue
-                    visited.add(key)
+                    # Each frontier entry is an exact edge-sequence prefix.
+                    # Extending it once per keyed edge already enumerates
+                    # distinct paths; node-only dedup loses parallel evidence.
                     paths.append(
                         GraphPath(
                             nodes=nodes,

datamind/tests/test_graph_parallel_paths.py

@@ -0,0 +1,94 @@
+"""Parallel edges are distinct evidence paths, including after persistence."""
+import pytest
+
+from datamind.capabilities.graph.providers.networkx_store import NetworkXGraphStore
+from datamind.core.protocols import GraphTriple
+
+
+def triple(src, rel, dst, source=None, confidence=1.0, **props):
+    return GraphTriple(subject=src, relation=rel, object=dst, source=source,
+                       confidence=confidence, properties=props)
+
+
+@pytest.mark.asyncio
+async def test_parallel_relations_extend_into_distinct_multihop_paths(tmp_path):
+    store = NetworkXGraphStore(persist_path=tmp_path / 'g.json')
+    await store.upsert_triples([
+        triple('A', 'authored', 'B', confidence=0.8),
+        triple('A', 'reviewed', 'B', confidence=0.6),
+        triple('B', 'published_in', 'C'),
+    ])
+    paths = await store.traverse('A', max_hops=2)
+    assert {tuple(e.relation for e in p.edges) for p in paths} == {
+        ('authored',), ('reviewed',),
+        ('authored', 'published_in'), ('reviewed', 'published_in'),
+    }
+    scores = {tuple(e.relation for e in p.edges): p.score for p in paths}
+    assert scores[('authored', 'published_in')] == pytest.approx(0.9)
+    assert [p.score for p in paths] == sorted(scores.values(), reverse=True)
+
+
+@pytest.mark.asyncio
+async def test_same_relation_different_origins_survive_reload_and_upsert(tmp_path):
+    path = tmp_path / 'g.json'
+    store = NetworkXGraphStore(persist_path=path)
+    triples = [triple('A', 'cites', 'B', source='doc1'),
+               triple('A', 'cites', 'B', source='doc2'),
+               triple('A', 'cites', 'B', source='doc1', _profile_managed=True)]
+    await store.upsert_triples(triples)
+    await store.upsert_triples(triples)  # Exact edge identities still overwrite.
+    before = await store.traverse('A')
+    assert len(before) == 3
+    assert {(p.edges[0].properties['source'],
+             p.edges[0].properties.get('_profile_managed', False)) for p in before} == {
+        ('doc1', False), ('doc2', False), ('doc1', True),
+    }
+    await store.persist()
+    after = await NetworkXGraphStore(persist_path=path).traverse('A')
+    assert [p.model_dump() for p in before] == [p.model_dump() for p in after]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('cap', [1, 3, 100])
+async def test_insertion_order_does_not_change_results_even_when_capped(tmp_path, cap):
+    triples = [triple('A', 'reviewed', 'B', source='z'),
+               triple('A', 'authored', 'B', source='b'),
+               triple('A', 'authored', 'B', source='a'),
+               triple('B', 'next', 'C')]
+    outputs = []
+    for i, ordered in enumerate([triples, list(reversed(triples))]):
+        store = NetworkXGraphStore(persist_path=tmp_path / f'{i}.json')
+        await store.upsert_triples(ordered)
+        paths = await store.traverse('A', max_results=cap)
+        assert len(paths) <= cap
+        outputs.append([p.model_dump() for p in paths])
+    assert outputs[0] == outputs[1]
+
+
+@pytest.mark.asyncio
+async def test_filters_hops_and_cycles_remain_bounded(tmp_path):
+    store = NetworkXGraphStore(persist_path=tmp_path / 'g.json')
+    await store.upsert_triples([
+        triple('A', 'skip', 'B'), triple('A', 'keep', 'B'),
+        triple('B', 'keep', 'C'), triple('C', 'keep', 'A'),
+        triple('A', 'keep', 'A'),
+    ])
+    paths = await store.traverse('A', max_hops=10, relation_filter=['keep'])
+    assert [p.nodes for p in paths] == [['A', 'B'], ['A', 'B', 'C']]
+    assert all(e.relation == 'keep' for p in paths for e in p.edges)
+    assert len(await store.traverse('A', max_hops=1, relation_filter=['keep'])) == 1
+    assert await store.traverse('A', relation_filter=['missing']) == []
+    assert await store.traverse('missing') == []
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('cap', [0, -1, 1, 7])
+async def test_dense_parallel_graph_obeys_result_budget(tmp_path, cap):
+    store = NetworkXGraphStore(persist_path=tmp_path / 'g.json')
+    await store.upsert_triples([
+        triple(str(i), f'relation{j}', str(i+1))
+        for i in range(8) for j in range(8)
+    ])
+    paths = await store.traverse('0', max_hops=8, max_results=cap)
+    assert len(paths) == max(0, cap)
+    assert await store.traverse('0', max_hops=0) == []

docs/STABLE_API.md

@@ -112,6 +112,21 @@ Receipt status is one of `stored`, `unchanged`, or `failed`. A receipt records
 what DataMind observed and attempted; it is not a two-phase commit with an
 external database or sink.
 
+## NetworkX graph traversal
+
+`NetworkXGraphStore.traverse` returns paths distinguished by their exact stored
+edge sequences, not just their node sequences. Parallel relations and the same
+relation from distinct origins remain separate evidence paths; relation,
+weight, and source properties are preserved. Upserting an existing edge identity
+still replaces that edge rather than creating another path.
+
+Traversal uses breadth-first expansion in deterministic target/relation/edge-key
+order. `relation_filter` applies at every hop, no path repeats a node, and
+`max_hops` and `max_results` bound expansion. Non-positive limits return no paths.
+Each emitted prefix counts toward `max_results`, so reaching the cap stops
+expansion rather than enumerating all combinations of parallel edges. Selected
+paths are sorted by descending score; this is not a global top-k guarantee.
+
 ## HTTP API contract
 
 The bundled FastAPI app is a thin transport over the same facade. Request body