Merge pull request #751 from semantica-agi/deprecate/744-kg-provenance-tracker

Deprecate kg.ProvenanceTracker and remove tests for unimplemented compatibility APIs

CHANGELOG.md

@@ -22,6 +22,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **`kg.ProvenanceTracker` compatibility wrapper out of sync with `ProvenanceManager`, causing 9 pre-existing test failures** (#744, #751) by @Sameer6305 and @KaifAhmad1
+  - `kg.ProvenanceTracker` was a standalone in-memory implementation that never delegated to the unified `ProvenanceManager` backend; its own test suite asserted the existence of `get_lineage`, `track_relationship`, `track_entities_batch`, `get_provenance`, and `_use_unified`, none of which were ever implemented, plus a stale `get_all_sources()` assertion expecting `"timestamp"` instead of the actual `"recorded_at"` key
+  - Rather than completing the abandoned compatibility layer, `kg.ProvenanceTracker` and its remaining supported methods (`track_entity`, `get_all_sources`, `query_recorded_between`, `revision_history`, `export_audit_log`) now emit `DeprecationWarning`s pointing callers to `semantica.provenance.ProvenanceManager`
+  - Removed/rewrote the 9 tests that only exercised the never-implemented compatibility methods to instead verify the observable behavior of the still-supported API, and corrected the stale `get_all_sources()` assertion
+  - Added the previously-missing `docs/migration/kg-provenance-tracker.md` migration guide referenced by every new deprecation warning, with a method-mapping table to `ProvenanceManager` and a before/after example, closing #744
+
 - **`ProvenanceManager.track_entity` silently overrides an explicit `parent_entity_id`/`derived_from` on re-track** (#742) by @Sameer6305
   - `track_entity()` resolved `parent_id` via a documented precedence chain (`parent_entity_id` kwarg > `metadata["derived_from"]` > source-as-known-entity-id fallback), but the history-preservation block that runs afterward unconditionally overwrote that resolved value with an auto-generated `f"{entity_id}:v:{existing.last_updated}"` history pointer whenever the entity was being re-tracked, discarding whatever parent the caller had just explicitly supplied with no warning
   - `track_entity()` now records whether the precedence chain already resolved an explicit parent (`parent_entity_id` kwarg, `metadata["derived_from"]`, or the source-as-known-entity-id fallback) before the history block runs, and only falls back to the auto-generated history pointer when the caller supplied no explicit parent on that call

docs/migration/kg-provenance-tracker.md

@@ -0,0 +1,56 @@
+---
+title: "Migrating from kg.ProvenanceTracker"
+description: "How to move from the deprecated semantica.kg.ProvenanceTracker to the unified semantica.provenance.ProvenanceManager."
+---
+
+## Why migrate
+
+`semantica.kg.ProvenanceTracker` is deprecated and will be removed in a future major version. It was a standalone, in-memory implementation that never delegated to the unified provenance backend — `semantica.provenance.ProvenanceManager` is that backend, and is now the supported way to track entity and relationship provenance across every Semantica module (see the [Provenance & Audit Trails guide](/guides/provenance)).
+
+Every method on `kg.ProvenanceTracker` now emits a `DeprecationWarning` on use, but existing code keeps working unchanged until the class is removed — there is no forced migration deadline yet.
+
+## Method mapping
+
+| `kg.ProvenanceTracker` | `ProvenanceManager` equivalent | Notes |
+| --- | --- | --- |
+| `ProvenanceTracker()` | `ProvenanceManager()` | `ProvenanceManager` also accepts `storage_path=` for SQLite persistence instead of in-memory only. |
+| `track_entity(entity_id, source, metadata)` | `track_entity(entity_id, source, metadata)` | Same call shape. `ProvenanceManager` additionally auto-links each update to its prior version via `parent_entity_id`. |
+| `get_all_sources(entity_id)` | `get_all_sources(entity_id)` | Field name differs: the `kg` tracker returns each record's time under `"recorded_at"`; `ProvenanceManager` returns `"timestamp"`. |
+| `clear(entity_id=None)` | `clear()` | `ProvenanceManager.clear()` clears all provenance data; there is no per-entity clear yet. |
+| `query_recorded_between(start, end)` | *No direct equivalent yet* | Filter the entries returned by `get_lineage()` / `trace_lineage()` client-side in the meantime. |
+| `revision_history(fact_id)` | *No direct equivalent yet* | `get_lineage(fact_id)["lineage_chain"]` returns the full chain of `ProvenanceEntry` records but not in the same versioned shape. |
+| `export_audit_log(fact_ids, format)` | *No direct equivalent yet* | Build the export from `get_lineage()` output, or serialize `get_statistics()` for a summary view. |
+
+Methods with no direct equivalent are not planned to be reimplemented on `kg.ProvenanceTracker` — they will need a small adapter in caller code, or a feature request against `ProvenanceManager` if you rely on them heavily.
+
+## Example
+
+```python
+# Before
+from semantica.kg import ProvenanceTracker
+
+tracker = ProvenanceTracker()
+tracker.track_entity("entity_1", source="doc_1", metadata={"confidence": 0.9})
+sources = tracker.get_all_sources("entity_1")  # [{"source": ..., "recorded_at": ..., "confidence": 0.9}]
+
+# After
+from semantica.provenance import ProvenanceManager
+
+prov = ProvenanceManager()
+prov.track_entity("entity_1", source="doc_1", metadata={"confidence": 0.9})
+sources = prov.get_all_sources("entity_1")  # [{"source": ..., "timestamp": ..., "metadata": {...}, ...}]
+```
+
+## Suppressing the warning during migration
+
+If you need to keep using `kg.ProvenanceTracker` temporarily and want to silence the warning while you plan the switch:
+
+```python
+import warnings
+
+with warnings.catch_warnings():
+    warnings.simplefilter("ignore", DeprecationWarning)
+    tracker = ProvenanceTracker()
+```
+
+This is a stopgap, not a fix — plan to move to `ProvenanceManager` before `kg.ProvenanceTracker` is removed.

semantica/kg/provenance_tracker.py

@@ -7,14 +7,23 @@ Tracks the sources and lineage of entities and relationships.
 import csv
 import io
 import json
+import warnings
 from datetime import datetime, timezone
 from typing import Any, Dict, List, Optional
 
+_MIGRATION_GUIDE_URL = "docs/migration/kg-provenance-tracker.md"
+
 
 class ProvenanceTracker:
     """
     Tracks provenance (source lineage) for knowledge graph entities.
 
+    .. deprecated::
+        ``ProvenanceTracker`` is deprecated in favor of
+        :class:`semantica.provenance.ProvenanceManager` and will be removed
+        in a future major version. See the migration guide at
+        ``docs/migration/kg-provenance-tracker.md``.
+
     Usage:
         tracker = ProvenanceTracker()
         tracker.track_entity("E1", "doc1.txt", metadata={"type": "file"})
@@ -22,6 +31,13 @@ class ProvenanceTracker:
     """
 
     def __init__(self):
+        warnings.warn(
+            "ProvenanceTracker is deprecated and will be removed in a future "
+            "major version. Use semantica.provenance.ProvenanceManager instead. "
+            f"See migration guide: {_MIGRATION_GUIDE_URL}",
+            DeprecationWarning,
+            stacklevel=2,
+        )
         self._records: Dict[str, List[Dict[str, Any]]] = {}
 
     def track_entity(
@@ -31,6 +47,13 @@ class ProvenanceTracker:
         metadata: Optional[Dict[str, Any]] = None,
     ) -> None:
         """Record that entity_id was derived from source."""
+        warnings.warn(
+            "ProvenanceTracker.track_entity() is deprecated; use "
+            "ProvenanceManager.track_entity() instead. "
+            f"See migration guide: {_MIGRATION_GUIDE_URL}",
+            DeprecationWarning,
+            stacklevel=2,
+        )
         if entity_id not in self._records:
             self._records[entity_id] = []
         entry: Dict[str, Any] = {
@@ -43,6 +66,13 @@ class ProvenanceTracker:
 
     def get_all_sources(self, entity_id: str) -> List[Dict[str, Any]]:
         """Return all provenance records for entity_id."""
+        warnings.warn(
+            "ProvenanceTracker.get_all_sources() is deprecated; use "
+            "ProvenanceManager.get_all_sources() instead. "
+            f"See migration guide: {_MIGRATION_GUIDE_URL}",
+            DeprecationWarning,
+            stacklevel=2,
+        )
         return self._records.get(entity_id, [])
 
     def clear(self, entity_id: Optional[str] = None) -> None:
@@ -66,6 +96,13 @@ class ProvenanceTracker:
             Flat list of matching provenance records (each dict includes
             the entity_id under the key "entity_id").
         """
+        warnings.warn(
+            "ProvenanceTracker.query_recorded_between() is deprecated with no "
+            "direct ProvenanceManager equivalent yet; see "
+            f"migration guide: {_MIGRATION_GUIDE_URL}",
+            DeprecationWarning,
+            stacklevel=2,
+        )
         start_dt = self._parse_dt(start)
         end_dt = self._parse_dt(end)
 
@@ -93,6 +130,21 @@ class ProvenanceTracker:
 
         Returns an empty list for a fact with no recorded provenance.
         """
+        warnings.warn(
+            "ProvenanceTracker.revision_history() is deprecated with no "
+            "direct ProvenanceManager equivalent yet; see "
+            f"migration guide: {_MIGRATION_GUIDE_URL}",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+        return self._revision_history_no_warn(fact_id)
+
+    def _revision_history_no_warn(self, fact_id: str) -> List[Dict[str, Any]]:
+        """Internal, warning-free implementation of revision_history().
+
+        Used by other deprecated methods (e.g. export_audit_log()) that need
+        this logic without emitting a second DeprecationWarning per call.
+        """
         records = self._records.get(fact_id, [])
         if not records:
             return []
@@ -133,9 +185,16 @@ class ProvenanceTracker:
         Returns:
             String containing the serialized audit log.
         """
+        warnings.warn(
+            "ProvenanceTracker.export_audit_log() is deprecated with no "
+            "direct ProvenanceManager equivalent yet; see "
+            f"migration guide: {_MIGRATION_GUIDE_URL}",
+            DeprecationWarning,
+            stacklevel=2,
+        )
         rows = []
         for fact_id in fact_ids:
-            for entry in self.revision_history(fact_id):
+            for entry in self._revision_history_no_warn(fact_id):
                 rows.append({"fact_id": fact_id, **entry})
 
         if format == "json":

tests/provenance/test_all_modules.py

@@ -60,21 +60,25 @@ class TestKGModule:
         assert tracker is not None
         
         # Test basic functionality
+        # NOTE: tracker.get_lineage() was never implemented on
+        # kg.ProvenanceTracker; this tested an intended unified-backend
+        # migration that never happened (#744). ProvenanceTracker is now
+        # deprecated in favor of semantica.provenance.ProvenanceManager.
+        # Instead, verify the observable behavior of the still-supported
+        # track_entity()/get_all_sources() pair.
         tracker.track_entity("test_entity", source="test_source")
-        lineage = tracker.get_lineage("test_entity")
+        sources = tracker.get_all_sources("test_entity")
+        assert len(sources) > 0
+        last_entry = sources[-1]
+        assert last_entry["source"] == "test_source"
+        assert "recorded_at" in last_entry
     
-        assert lineage is not None
+    # NOTE: test_kg_uses_unified_backend removed; it only asserted the
-        assert "sources" in lineage
+    # presence of _use_unified/_unified_manager attributes, which were
-    
+    # never implemented on kg.ProvenanceTracker. This tested an intended
-    def test_kg_uses_unified_backend(self):
+    # unified-backend migration that never happened (#744).
-        """Test that kg module uses unified backend."""
+    # ProvenanceTracker is now deprecated in favor of
-        from semantica.kg import ProvenanceTracker
+    # semantica.provenance.ProvenanceManager.
-        
-        tracker = ProvenanceTracker()
-        
-        # Check if using unified backend
-        assert hasattr(tracker, '_use_unified')
-        assert hasattr(tracker, '_unified_manager')
     
     def test_kg_graph_builder_ready(self):
         """Test GraphBuilder is ready for provenance."""
@@ -405,7 +409,14 @@ class TestCrossModuleIntegration:
     """Test provenance tracking across multiple modules."""
     
     def test_kg_and_split_integration(self):
-        """Test provenance tracking between kg and split modules."""
+        """Test provenance tracking between kg and split modules.
+
+        NOTE: this no longer asserts kg.ProvenanceTracker uses a unified
+        backend (kg_tracker.get_lineage() was never implemented; see #744).
+        It instead verifies, independent of unified-backend behavior, that
+        kg tracking actually produced a record via the still-supported
+        track_entity()/get_all_sources() pair.
+        """
         from semantica.kg import ProvenanceTracker as KGTracker
         from semantica.split import ProvenanceTracker as SplitTracker
         from semantica.split.semantic_chunker import Chunk
@@ -414,17 +425,22 @@ class TestCrossModuleIntegration:
         kg_tracker = KGTracker()
         kg_tracker.track_entity("entity_1", source="doc_1")
         
+        # Verify kg tracking produced a record
+        kg_sources = kg_tracker.get_all_sources("entity_1")
+        assert len(kg_sources) > 0
+        last_kg_entry = kg_sources[-1]
+        assert last_kg_entry["source"] == "doc_1"
+        assert "recorded_at" in last_kg_entry
+        
         # Track with split
         split_tracker = SplitTracker()
         chunk = Chunk(text="Test", start_index=0, end_index=4, metadata={})
         chunk.id = "chunk_1"
         split_tracker.track_chunk(chunk, source_document="doc_1")
         
-        # Both should work
+        # split.ProvenanceTracker.get_provenance() is real and still works
-        kg_lineage = kg_tracker.get_lineage("entity_1")
         split_prov = split_tracker.get_provenance("chunk_1")
         
-        assert kg_lineage is not None
         assert split_prov is not None
     
     def test_unified_manager_with_all_modules(self):

tests/provenance/test_backward_compat.py

@@ -15,30 +15,32 @@ class TestKGProvenanceBackwardCompat:
     """Test kg.ProvenanceTracker backward compatibility."""
     
     def test_existing_code_unchanged(self):
-        """Test that existing kg.ProvenanceTracker code works unchanged."""
+        """Test that existing kg.ProvenanceTracker code works unchanged.
+
+        NOTE: this no longer asserts on tracker.get_lineage(), which was
+        never implemented on kg.ProvenanceTracker (see #744). It instead
+        verifies the observable behavior of the still-supported
+        track_entity()/get_all_sources() pair.
+        """
         # Existing code pattern
         tracker = KGProvenanceTracker()
         
         # Track entity (existing API)
         tracker.track_entity("entity_1", source="doc_1", metadata={"confidence": 0.9})
         
-        # Get lineage (existing API)
+        # Verify the entity was actually tracked
-        lineage = tracker.get_lineage("entity_1")
+        sources = tracker.get_all_sources("entity_1")
+        assert len(sources) > 0
+        last_entry = sources[-1]
+        assert last_entry["source"] == "doc_1"
+        assert "recorded_at" in last_entry
+        assert last_entry["confidence"] == 0.9
     
-        # Verify existing return format
+    # NOTE: test_track_relationship_unchanged removed; it only exercised
-        assert "sources" in lineage
+    # tracker.track_relationship(), which was never implemented on
-        assert "first_seen" in lineage
+    # kg.ProvenanceTracker. This tested an intended unified-backend
-        assert "last_updated" in lineage
+    # migration that never happened (#744). ProvenanceTracker is now
-        assert "metadata" in lineage
+    # deprecated in favor of semantica.provenance.ProvenanceManager.
-    
-    def test_track_relationship_unchanged(self):
-        """Test relationship tracking works unchanged."""
-        tracker = KGProvenanceTracker()
-        
-        tracker.track_relationship("rel_1", source="doc_1", metadata={"type": "founded"})
-        
-        lineage = tracker.get_lineage("rel_1")
-        assert lineage is not None
     
     def test_get_all_sources_unchanged(self):
         """Test get_all_sources returns expected format."""
@@ -53,22 +55,14 @@ class TestKGProvenanceBackwardCompat:
         assert len(sources) >= 2
         for source in sources:
             assert "source" in source
-            assert "timestamp" in source
+            assert "recorded_at" in source
     
-    def test_batch_operations_unchanged(self):
+    # NOTE: test_batch_operations_unchanged removed; it only exercised
-        """Test batch operations work unchanged."""
+    # tracker.track_entities_batch() and tracker.get_lineage(), neither of
-        tracker = KGProvenanceTracker()
+    # which was ever implemented on kg.ProvenanceTracker. This tested an
-        
+    # intended unified-backend migration that never happened (#744).
-        entities = [
+    # ProvenanceTracker is now deprecated in favor of
-            {"id": "entity_1", "confidence": 0.9},
+    # semantica.provenance.ProvenanceManager.
-            {"id": "entity_2", "confidence": 0.85}
-        ]
-        
-        count = tracker.track_entities_batch(entities, "doc_1")
-        
-        assert count == 2
-        assert tracker.get_lineage("entity_1") is not None
-        assert tracker.get_lineage("entity_2") is not None
 
 
 class TestSplitProvenanceBackwardCompat:
@@ -196,10 +190,10 @@ class TestGracefulDegradation:
         
         # Should work even if unified backend has issues
         tracker.track_entity("entity_1", source="doc_1")
-        lineage = tracker.get_lineage("entity_1")
+        # NOTE: tracker.get_lineage() was never implemented on
-        
+        # kg.ProvenanceTracker; this tested an intended unified-backend
-        assert lineage is not None
+        # migration that never happened (#744). ProvenanceTracker is now
-        assert "sources" in lineage
+        # deprecated in favor of semantica.provenance.ProvenanceManager.
     
     def test_split_tracker_fallback(self):
         """Test split.ProvenanceTracker falls back to legacy on error."""
@@ -221,18 +215,20 @@ class TestExistingTestsPass:
         tracker = KGProvenanceTracker()
         
         # Test 1: Basic tracking
+        # NOTE: tracker.get_provenance() was never implemented on
+        # kg.ProvenanceTracker; this tested an intended unified-backend
+        # migration that never happened (#744). ProvenanceTracker is now
+        # deprecated in favor of semantica.provenance.ProvenanceManager.
         tracker.track_entity("e1", "src1")
-        assert tracker.get_provenance("e1") is not None
         
         # Test 2: Multiple sources
         tracker.track_entity("e1", "src2")
         sources = tracker.get_all_sources("e1")
         assert len(sources) >= 2
         
-        # Test 3: Metadata
+        # NOTE: Test 3 (metadata via tracker.get_lineage()) removed; that
-        tracker.track_entity("e2", "src1", metadata={"key": "value"})
+        # method was never implemented on kg.ProvenanceTracker. This tested
-        lineage = tracker.get_lineage("e2")
+        # an intended unified-backend migration that never happened (#744).
-        assert "metadata" in lineage
     
     def test_split_provenance_existing_behavior(self):
         """Test existing split.ProvenanceTracker behavior is preserved."""

tests/provenance/test_integration.py

@@ -31,18 +31,25 @@ class TestEndToEndProvenance:
         assert entity_prov is not None
         assert chunk_prov is not None
     
-    def test_kg_to_unified_integration(self):
+    def test_kg_tracker_records_provenance(self):
-        """Test kg.ProvenanceTracker uses unified backend."""
+        """Test kg.ProvenanceTracker records provenance via its supported API.
+
+        NOTE: this no longer asserts kg.ProvenanceTracker uses a unified
+        backend (kg_tracker.get_lineage() was never implemented; see #744).
+        It instead verifies the observable behavior of the still-supported
+        track_entity()/get_all_sources() pair.
+        """
         kg_tracker = KGTracker()
         
         # Track with kg tracker
         kg_tracker.track_entity("kg_entity_1", source="kg_doc_1")
         
         # Verify it was tracked
-        lineage = kg_tracker.get_lineage("kg_entity_1")
+        sources = kg_tracker.get_all_sources("kg_entity_1")
-        
+        assert len(sources) > 0
-        assert lineage is not None
+        last_entry = sources[-1]
-        assert "sources" in lineage
+        assert last_entry["source"] == "kg_doc_1"
+        assert "recorded_at" in last_entry
     
     def test_split_to_unified_integration(self):
         """Test split.ProvenanceTracker uses unified backend."""