Merge pull request #741 from semantica-agi/fix/735-provenance-lineage-derived-from

Fix ProvenanceManager.get_lineage not linking entities via derived_from

CHANGELOG.md

@@ -22,6 +22,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **`ProvenanceManager.get_lineage` does not link entities that share a source URL** (#735) by @KaifAhmad1
+  - `track_entity()`'s only auto-linking logic looked up `source` as if it were an existing entity's `entity_id`, so passing the same real URL/DOI as `source` for two conceptually linked entities (e.g. a document and a decision derived from it) never produced a parent link, leaving `get_lineage()` returning a chain of length 1
+  - `metadata["derived_from"]` was preserved and echoed back in the output JSON but was never consulted by any linking or traversal code, so the caller's explicit relationship was silently inert
+  - `track_entity()` now treats `metadata["derived_from"]` as an explicit parent link (unless `parent_entity_id` was already passed directly), so `InMemoryStorage.trace_lineage()`'s existing BFS over `parent_entity_id` picks it up for free
+  - `metadata["derived_from"]` is now recognized on any `collections.abc.Mapping`, not just a concrete `dict`, so e.g. `types.MappingProxyType` metadata still creates the parent link
+  - `get_lineage()`'s metadata aggregation now applies the queried entity's own metadata last so it wins over ancestor metadata on conflicting keys, matching the documented "most recent entry's metadata takes precedence" behavior — previously `trace_lineage()`'s BFS order caused ancestor metadata (now reachable via `derived_from` chains) to silently overwrite the queried entity's own values
+  - Added 9 regression/edge-case tests in `tests/provenance/test_manager.py` covering the happy path, explicit `parent_entity_id` precedence over `derived_from`, precedence over the `source`-as-known-entity-id fallback, a `derived_from` pointing at a never-tracked entity, non-string/empty-string `derived_from` values being ignored, a self-referencing `derived_from` not hanging traversal, multi-hop `derived_from` chains, metadata precedence between a queried entity and its ancestors, and non-`dict` `Mapping` metadata, closing #735
+
 - **`Reasoner.add_rule` had no deduplication, doubling rules and silently emptying `forward_chain()` on rerun** (#732) by @KaifAhmad1
   - `add_rule()` unconditionally appended to `self.rules`, so re-running the same setup code on an existing `Reasoner` instance (e.g. re-executing a Jupyter cell) duplicated every rule; since `forward_chain()` only records a conclusion if it isn't already in `self.facts`, the second run's duplicated rules matched but produced no new results, with no error or warning
   - `add_rule()` now compares an incoming rule's `rule_type`, `conditions`, and `conclusion` against existing rules and returns the existing `Rule` instead of appending a duplicate, keeping repeated `add_rule()` calls with the same definition idempotent

semantica/provenance/manager.py

@@ -25,6 +25,7 @@ License: MIT
 """
 
 from typing import Optional, List, Dict, Any
+from collections.abc import Mapping
 from datetime import datetime
 
 from .schemas import ProvenanceEntry, SourceReference, PropertySource
@@ -115,7 +116,14 @@ class ProvenanceManager:
         # Check if entity already exists
         existing = self.storage.retrieve(entity_id)
         parent_id = kwargs.get("parent_entity_id")
-        
+
+        # If caller declared an explicit parent via metadata, honor it
+        # (unless parent_entity_id was already passed directly)
+        if not parent_id and metadata and isinstance(metadata, Mapping):
+            derived_from = metadata.get("derived_from")
+            if derived_from and isinstance(derived_from, str):
+                parent_id = derived_from
+
         # If source is a known entity, link it as parent (unless parent already set)
         if not parent_id and source and isinstance(source, str):
             try:
@@ -453,10 +461,14 @@ class ProvenanceManager:
         if not lineage_entries:
             return {}
         
-        # Aggregate metadata from all lineage entries
-        # Most recent entry's metadata takes precedence
+        # Aggregate metadata from all lineage entries.
+        # trace_lineage() is a BFS starting at entity_id, so lineage_entries[0]
+        # is always the queried entity itself, followed by its ancestors
+        # (parent, grandparent, ...). Apply ancestors first and the queried
+        # entity last so its own keys win on conflict, matching the intent
+        # that the "most recent"/current entity's metadata takes precedence.
         aggregated_metadata = {}
-        for entry in lineage_entries:
+        for entry in reversed(lineage_entries):
             if entry.metadata:
                 meta = entry.metadata
                 if isinstance(meta, str):

tests/provenance/test_manager.py

@@ -104,7 +104,192 @@ class TestProvenanceManager:
         assert lineage is not None
         assert "lineage_chain" in lineage
         assert len(lineage["lineage_chain"]) > 0
-    
+
+    def test_get_lineage_via_derived_from_metadata(self):
+        """metadata['derived_from'] should link entities into the lineage chain
+        even when they share a source URL rather than one being a known entity_id."""
+        prov_mgr = ProvenanceManager()
+
+        prov_mgr.track_entity(
+            entity_id="doc:X",
+            source="https://example.com/api",
+            metadata={"content_type": "drug_label"},
+        )
+        prov_mgr.track_entity(
+            entity_id="decision:Y",
+            source="https://example.com/api",
+            metadata={"derived_from": "doc:X"},
+        )
+
+        lineage = prov_mgr.get_lineage("decision:Y")
+
+        assert lineage["entity_count"] == 2
+        entity_ids = [e["entity_id"] for e in lineage["lineage_chain"]]
+        assert "doc:X" in entity_ids
+        assert "decision:Y" in entity_ids
+
+    def test_derived_from_does_not_override_explicit_parent(self):
+        """An explicit parent_entity_id kwarg should win over metadata['derived_from']."""
+        prov_mgr = ProvenanceManager()
+
+        prov_mgr.track_entity(entity_id="explicit_parent", source="doc_1")
+        prov_mgr.track_entity(entity_id="ignored_parent", source="doc_1")
+        entry = prov_mgr.track_entity(
+            entity_id="child",
+            source="doc_1",
+            metadata={"derived_from": "ignored_parent"},
+            parent_entity_id="explicit_parent",
+        )
+
+        assert entry.parent_entity_id == "explicit_parent"
+
+    def test_derived_from_takes_precedence_over_source_as_entity_id(self):
+        """If `source` happens to also be a known entity_id, an explicit
+        metadata['derived_from'] should still win over that fallback linking."""
+        prov_mgr = ProvenanceManager()
+
+        prov_mgr.track_entity(entity_id="source_as_entity", source="doc_0")
+        prov_mgr.track_entity(entity_id="real_parent", source="doc_0")
+        entry = prov_mgr.track_entity(
+            entity_id="child",
+            source="source_as_entity",  # resolvable as an entity_id
+            metadata={"derived_from": "real_parent"},
+        )
+
+        assert entry.parent_entity_id == "real_parent"
+
+    def test_derived_from_nonexistent_entity_does_not_crash(self):
+        """derived_from pointing at an entity that was never tracked should be
+        stored as the parent link without raising, and lineage traversal should
+        stop gracefully instead of erroring."""
+        prov_mgr = ProvenanceManager()
+
+        entry = prov_mgr.track_entity(
+            entity_id="orphan_child",
+            source="doc_1",
+            metadata={"derived_from": "never_tracked"},
+        )
+
+        assert entry.parent_entity_id == "never_tracked"
+
+        lineage = prov_mgr.get_lineage("orphan_child")
+        entity_ids = [e["entity_id"] for e in lineage["lineage_chain"]]
+        assert entity_ids == ["orphan_child"]
+
+    def test_derived_from_non_string_is_ignored(self):
+        """A non-string derived_from value (e.g. accidentally passing an int or
+        list) should be ignored rather than raising or being used as a parent id."""
+        prov_mgr = ProvenanceManager()
+
+        entry = prov_mgr.track_entity(
+            entity_id="entity_bad_derived_from",
+            source="doc_1",
+            metadata={"derived_from": 12345},
+        )
+
+        assert entry.parent_entity_id is None
+
+    def test_derived_from_empty_string_is_ignored(self):
+        """An empty-string derived_from is falsy and should not be treated as a parent link."""
+        prov_mgr = ProvenanceManager()
+
+        entry = prov_mgr.track_entity(
+            entity_id="entity_empty_derived_from",
+            source="doc_1",
+            metadata={"derived_from": ""},
+        )
+
+        assert entry.parent_entity_id is None
+
+    def test_derived_from_self_reference_does_not_infinite_loop(self):
+        """An entity that (incorrectly) declares itself as its own derived_from
+        parent should not cause get_lineage to hang or infinitely recurse."""
+        prov_mgr = ProvenanceManager()
+
+        prov_mgr.track_entity(
+            entity_id="self_ref",
+            source="doc_1",
+            metadata={"derived_from": "self_ref"},
+        )
+
+        lineage = prov_mgr.get_lineage("self_ref")
+        entity_ids = [e["entity_id"] for e in lineage["lineage_chain"]]
+        assert entity_ids == ["self_ref"]
+
+    def test_derived_from_multi_hop_chain(self):
+        """derived_from links should chain transitively: A <- B <- C should
+        all appear when tracing lineage from C."""
+        prov_mgr = ProvenanceManager()
+
+        prov_mgr.track_entity(entity_id="grandparent", source="doc_1")
+        prov_mgr.track_entity(
+            entity_id="parent",
+            source="doc_1",
+            metadata={"derived_from": "grandparent"},
+        )
+        prov_mgr.track_entity(
+            entity_id="child",
+            source="doc_1",
+            metadata={"derived_from": "parent"},
+        )
+
+        lineage = prov_mgr.get_lineage("child")
+
+        assert lineage["entity_count"] == 3
+        entity_ids = {e["entity_id"] for e in lineage["lineage_chain"]}
+        assert entity_ids == {"grandparent", "parent", "child"}
+
+    def test_derived_from_without_metadata_dict_does_not_crash(self):
+        """track_entity called with no metadata at all should behave as before
+        (no parent link derived), exercising the `metadata and isinstance(...)` guard."""
+        prov_mgr = ProvenanceManager()
+
+        entry = prov_mgr.track_entity(entity_id="no_metadata_entity", source="doc_1")
+
+        assert entry.parent_entity_id is None
+
+    def test_get_lineage_metadata_prefers_queried_entity_over_ancestors(self):
+        """Aggregated lineage metadata should let the queried entity's own
+        values win over ancestor values on conflicting keys, matching the
+        documented "most recent entry's metadata takes precedence" intent."""
+        prov_mgr = ProvenanceManager()
+
+        prov_mgr.track_entity(
+            entity_id="ancestor",
+            source="doc_1",
+            metadata={"status": "draft", "shared_only_on_ancestor": True},
+        )
+        prov_mgr.track_entity(
+            entity_id="descendant",
+            source="doc_1",
+            metadata={"status": "final", "derived_from": "ancestor"},
+        )
+
+        lineage = prov_mgr.get_lineage("descendant")
+
+        assert lineage["metadata"]["status"] == "final"
+        assert lineage["metadata"]["shared_only_on_ancestor"] is True
+
+    def test_derived_from_accepts_non_dict_mapping(self):
+        """metadata['derived_from'] should be honored for any Mapping
+        implementation, not just a concrete dict (e.g. types.MappingProxyType
+        or a custom collections.abc.Mapping)."""
+        from types import MappingProxyType
+
+        prov_mgr = ProvenanceManager()
+
+        prov_mgr.track_entity(entity_id="mapping_parent", source="doc_1")
+        entry = prov_mgr.track_entity(
+            entity_id="mapping_child",
+            source="doc_1",
+            metadata=MappingProxyType({"derived_from": "mapping_parent"}),
+        )
+
+        assert entry.parent_entity_id == "mapping_parent"
+
+        lineage = prov_mgr.get_lineage("mapping_child")
+        assert lineage["entity_count"] == 2
+
     def test_batch_entity_tracking(self):
         """Test batch entity tracking."""
         prov_mgr = ProvenanceManager()