test(provenance): assert stored records in KG provenance suites (#946) (#1132)

* fix(provenance): use timezone-aware UTC and assert stored records (#946)

Replace datetime.utcnow() in ProvenanceManager, ProvenanceEntry,
BridgeAxiom, and GraphBuilderWithProvenance with
datetime.now(timezone.utc), matching PipelineWithProvenance.

KG workflow and integration tests now read provenance back through
get_provenance() and assert algorithm metadata instead of generated
IDs, and call tracker methods that actually persist records.

* fix(provenance): compare provenance timestamps as instants, not strings

query_recorded_between() and audit_log() filtered and sorted on raw ISO
strings. With the timezone-aware change, a store can hold both pre-existing
naive stamps and offset-bearing ones, and the two are not string-comparable:
"...500000+00:00" sorts above "...500000", so a record at the identical
instant as a naive bound falls outside the range that should contain it.

Both now parse through _parse_timestamp() before comparing, reading naive
values as UTC. This mirrors ProvenanceTracker._parse_dt() in kg/, the class
ProvenanceManager replaces, so both sides of the migration answer a range
query the same way. Unparseable stored timestamps are skipped and logged
rather than silently dropped; unparseable bounds raise ValueError.

---------

Co-authored-by: Pravit Ampapathini <pravit.amp@gmail.com>

CHANGELOG.md

@@ -119,6 +119,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- **KG provenance tests asserted on generated ID strings instead of stored records, and `kg_provenance.py` was missed by the `utcnow` sweep** (closes #946) by @pravit-amp
+  - The KG workflow and integration suites checked that a tracker call returned an ID matching a prefix (`assert cent_id.startswith("centrality_")`) without ever reading the record back, so an ID generator that returned a well-formed string and wrote nothing would have passed. Worse, some of those calls named tracker methods that do not exist anywhere in `semantica/` (`track_layer_analysis`, `track_centrality_score`), so the assertions were satisfied with no real interaction behind them
+  - Those tests now read provenance back through `get_provenance()` and assert on algorithm metadata, and call the methods that actually persist records. Verified by mutation rather than by a green run alone: neutering the manager's storage write (`self.storage.store(...)` → no-op) fails 10 tests
+  - `GraphBuilderWithProvenance` in `semantica/kg/kg_provenance.py` still stamped `activity_started_at_time`/`activity_ended_at_time` with the deprecated `datetime.utcnow()`; it was outside the `export/`+`provenance/` scope of the #1114 sweep below and now uses the same `utc_now_iso()` helper. `docs/guides/provenance.md` and `docs/reference/provenance.md` were still documenting `utcnow()` and a naive timestamp example, and now show the helper and the offset-bearing form
+  - 16 tests across the affected suites ended in `return <value>` instead of asserting, which pytest reports as `PytestReturnNotNoneWarning`; now zero
+
 - **The temporal-evolution `stability` metric was a hardcoded placeholder, not a duration**
   - `TemporalGraphQuery.analyze_evolution()` documents `stability` as a "relationship duration/stability measure", but the implementation appended a constant `1` for every relationship with both `valid_from` and `valid_until` set (`durations.append(1)  # Placeholder`). The reported stability was therefore always `1.0` when any bounded relationship existed and `0` otherwise — it never reflected how long relationships actually stayed valid, so it could not distinguish a graph of decade-long relationships from one of one-second relationships
   - `stability` now computes the mean valid-time duration in seconds (`(valid_until - valid_from).total_seconds()`) across relationships that have both bounds set. Relationships with a missing or open `valid_from`/`valid_until` are skipped (their duration is unbounded), and non-positive intervals are clamped to `0`; an empty set still reports `0`

docs/guides/provenance.md

@@ -639,7 +639,7 @@ Every `ProvenanceEntry` maps directly to W3C PROV-O terms. If your compliance te
 | — | `previous_version_id` | This entry corrects/replaces a prior version of the *same* fact |
 | `prov:wasDerivedFrom` | `derived_from_id` | This entry was derived from a *different* source entity |
 | `prov:used` | `used_entities` | Entity IDs consumed to produce this one |
-| `prov:generatedAtTime` | `timestamp` | ISO datetime, auto-set to `datetime.utcnow()` at write time |
+| `prov:generatedAtTime` | `timestamp` | ISO datetime, auto-set to `utc_now_iso()` at write time |
 | `prov:qualifiedInvalidation` | `invalidated`, `invalidated_at_time`, `invalidated_by`, `invalidation_reason` | A retraction/correction recorded as a tombstone via `ProvenanceManager.invalidate()`, never a hard delete |
 | `prov:startedAtTime` / `prov:endedAtTime` | `activity_started_at_time`, `activity_ended_at_time` | Typed Activity timing — pass an `ActivityRecord` via the `activity=` kwarg to set these together with `activity_id` |
 | `prov:qualifiedGeneration`/`Generation`, `qualifiedUsage`/`Usage`, `qualifiedDerivation`/`Derivation` | (derived from the fields above) | Additive qualified forms of `wasGeneratedBy`/`used`/`wasDerivedFrom`, emitted automatically alongside the plain triples |

docs/reference/provenance.md

@@ -250,7 +250,7 @@ entry = ProvenanceEntry(
     source_document="report.pdf",     # str: default ""
     source_location="Page 4",         # Optional[str]: default None
     source_quote="Relevant text...",  # Optional[str]: default None
-    timestamp="2024-01-01T12:00:00",  # str: auto-set to utcnow()
+    timestamp="2024-01-01T12:00:00+00:00",  # str: auto-set to utc_now_iso()
     first_seen=None,                  # Optional[str]: ISO timestamp
     last_updated=None,                # Optional[str]: ISO timestamp
     confidence=0.9,                   # float: default 1.0

semantica/kg/kg_provenance.py

@@ -54,10 +54,11 @@ Version: 1.0.0
 """
 
 from typing import Any, Dict, List, Optional
-from datetime import datetime
 import uuid
 import time
 
+from ..utils.helpers import utc_now_iso
+
 
 class GraphBuilderWithProvenance:
     """
@@ -103,7 +104,7 @@ class GraphBuilderWithProvenance:
     
     def build(self, sources, **kwargs):
         """Build graph with provenance tracking."""
-        activity_started_at_time = datetime.utcnow().isoformat()
+        activity_started_at_time = utc_now_iso()
         # Track the build operation (recorded before the build runs, so it
         # has no end time yet — this is the "in progress" marker).
         if self.provenance and self._prov_manager:
@@ -124,7 +125,7 @@ class GraphBuilderWithProvenance:
             )
 
         result = self._builder.build(sources, **kwargs)
-        activity_ended_at_time = datetime.utcnow().isoformat()
+        activity_ended_at_time = utc_now_iso()
 
         # Track individual entities and relationships if available
         if self.provenance and self._prov_manager and hasattr(result, 'get'):
@@ -180,7 +181,7 @@ class GraphBuilderWithProvenance:
 
     def build_single_source(self, kg_data, **kwargs):
         """Build graph from single source with provenance tracking."""
-        activity_started_at_time = datetime.utcnow().isoformat()
+        activity_started_at_time = utc_now_iso()
         # Track the build operation (recorded before the build runs, so it
         # has no end time yet — this is the "in progress" marker).
         if self.provenance and self._prov_manager:
@@ -200,7 +201,7 @@ class GraphBuilderWithProvenance:
             )
 
         result = self._builder.build_single_source(kg_data, **kwargs)
-        activity_ended_at_time = datetime.utcnow().isoformat()
+        activity_ended_at_time = utc_now_iso()
 
         # Track entities and relationships if available
         if self.provenance and self._prov_manager and isinstance(result, dict):

tests/kg/test_integration_comprehensive.py

@@ -7,6 +7,7 @@ Tests integration between all KG components and algorithms.
 import pytest
 import networkx as nx
 import numpy as np
+from datetime import datetime, timedelta
 from typing import Dict, List, Any, Tuple
 import time
 import json
@@ -14,7 +15,6 @@ import json
 from semantica.kg import (
     GraphBuilderWithProvenance,
     AlgorithmTrackerWithProvenance,
-    NodeEmbedder,
     SimilarityCalculator,
     PathFinder,
     LinkPredictor,
@@ -24,6 +24,40 @@ from semantica.kg import (
 )
 
 
+def _stored(owner, entity_id):
+    """Read a provenance record back. Fails if tracking only minted an ID."""
+    record = owner._prov_manager.get_provenance(entity_id)
+    assert record is not None, (
+        f"no stored provenance for {entity_id!r} — an ID was generated without a write"
+    )
+    assert record.get("entity_id") == entity_id
+    return record
+
+
+def _assert_utc_iso(value, field="timestamp"):
+    assert value, f"missing {field}"
+    parsed = datetime.fromisoformat(value)
+    assert parsed.tzinfo is not None, (
+        f"{field}={value!r} is naive (datetime.utcnow leftover)"
+    )
+    assert parsed.utcoffset() == timedelta(0), f"{field}={value!r} is not UTC"
+    return parsed
+
+
+def _assert_tracked(owner, entity_id, source, **metadata):
+    record = _stored(owner, entity_id)
+    assert record["source_document"] == source
+    actual = record.get("metadata") or {}
+    for key, expected in metadata.items():
+        assert actual.get(key) == expected, (
+            f"{entity_id} metadata[{key!r}]={actual.get(key)!r}, expected {expected!r}"
+        )
+    _assert_utc_iso(record["timestamp"], "timestamp")
+    if record.get("last_updated"):
+        _assert_utc_iso(record["last_updated"], "last_updated")
+    return record
+
+
 class TestComprehensiveIntegration:
     """Comprehensive integration tests for KG module."""
     
@@ -157,8 +191,6 @@ class TestComprehensiveIntegration:
         # Initialize all components
         builder = GraphBuilderWithProvenance(provenance=True)
         tracker = AlgorithmTrackerWithProvenance(provenance=True)
-        embedder = NodeEmbedder()
-        embedder.enable_provenance = True
         sim_calc = SimilarityCalculator()
         path_finder = PathFinder()
         link_predictor = LinkPredictor()
@@ -351,14 +383,34 @@ class TestComprehensiveIntegration:
             source='comprehensive_integration_test'
         )
         
-        # Verify all phases completed successfully
+        expected = {
+            'construction': 'graph_construction',
+            'centrality': 'centrality_calculation',
+            'connectivity': 'connectivity_analysis',
+            'community': 'community_detection',
+            'similarity': 'similarity_calculation',
+            'link_prediction': 'link_prediction',
+            'path_analysis': 'path_analysis',
+            'cross_layer': 'cross_layer_analysis',
+        }
         assert len(execution_ids) == 8
         for phase, exec_id in execution_ids.items():
-            assert exec_id is not None
-            assert len(exec_id) > 10
-        
-        print(f"Full pipeline integration completed: {pipeline_id}")
-        return pipeline_id
+            _assert_tracked(
+                tracker,
+                exec_id,
+                source=pipeline_id,
+                entity_type=expected[phase],
+            )
+        summary_record = _assert_tracked(
+            tracker,
+            summary_id,
+            source='comprehensive_integration_test',
+            entity_type='pipeline_summary',
+            pipeline_id=pipeline_id,
+            phases_count=8,
+        )
+        assert summary_id.startswith('pipeline_summary_')
+        assert summary_record['metadata']['input_data_size'] == len(complex_graph_data)
     
     def test_multi_layer_network_analysis(self, multi_layer_network, realistic_embeddings):
         """Test multi-layer network analysis."""
@@ -382,39 +434,39 @@ class TestComprehensiveIntegration:
             
             # Centrality analysis
             if graph.number_of_nodes() > 0:
-                try:
-                    degree_cent = centrality_calc.calculate_degree_centrality(graph_dict)
-                    layer_results[f"{layer_name}_centrality"] = degree_cent
-                    
-                    # Track with provenance
-                    cent_id = tracker.track_layer_analysis(
-                        layer_name=layer_name,
-                        graph=graph,
-                        analysis_type='centrality',
-                        results=degree_cent,
-                        source=multi_layer_id
-                    )
-                    
-                except Exception as e:
-                    print(f"Centrality analysis failed for {layer_name}: {e}")
+                degree_cent = centrality_calc.calculate_degree_centrality(graph_dict)
+                layer_results[f"{layer_name}_centrality"] = degree_cent
+                cent_id = tracker.track_centrality_calculation(
+                    graph=graph,
+                    centrality_scores=degree_cent['centrality'],
+                    method='degree',
+                    source=multi_layer_id
+                )
+                _assert_tracked(
+                    tracker,
+                    cent_id,
+                    source=multi_layer_id,
+                    method='degree',
+                    scores_count=len(degree_cent['centrality']),
+                )
             
             # Community detection
             if graph.number_of_edges() > 0:
-                try:
-                    communities = community_detector.detect_communities(graph_dict, method='label_propagation')
-                    layer_results[f"{layer_name}_communities"] = communities
-                    
-                    # Track with provenance
-                    comm_id = tracker.track_layer_analysis(
-                        layer_name=layer_name,
-                        graph=graph,
-                        analysis_type='communities',
-                        results=communities,
-                        source=multi_layer_id
-                    )
-                    
-                except Exception as e:
-                    print(f"Community detection failed for {layer_name}: {e}")
+                communities = community_detector.detect_communities(graph_dict, method='label_propagation')
+                layer_results[f"{layer_name}_communities"] = communities
+                comm_id = tracker.track_community_detection(
+                    graph=graph,
+                    communities=communities['communities'],
+                    method='label_propagation',
+                    source=multi_layer_id
+                )
+                _assert_tracked(
+                    tracker,
+                    comm_id,
+                    source=multi_layer_id,
+                    method='label_propagation',
+                    communities_count=len(communities['communities']),
+                )
         
         # Cross-layer similarity analysis
         print("Cross-layer similarity analysis")
@@ -433,39 +485,43 @@ class TestComprehensiveIntegration:
                         similarity_score = len(common_nodes) / max(len(graph1.nodes()), len(graph2.nodes()))
                         layer_similarities[f"{layer1_name}_{layer2_name}"] = similarity_score
         
-        # Track cross-layer analysis
         cross_layer_id = tracker.track_cross_layer_analysis(
-            multi_layer_network=multi_layer_network,
-            layer_similarities=layer_similarities,
+            graph_data=multi_layer_network,
+            cross_layer_results=layer_similarities,
             source='multi_layer_test'
         )
+        _assert_tracked(
+            tracker,
+            cross_layer_id,
+            source='multi_layer_test',
+            entity_type='cross_layer_analysis',
+            layers_count=len(layer_similarities),
+        )
         
         # Embedding-based entity similarity
-        print("Embedding-based entity similarity")
-        
         entity_similarities = sim_calc.pairwise_similarity(realistic_embeddings)
         
-        # Track embedding analysis
         embed_id = tracker.track_embedding_analysis(
             embeddings=realistic_embeddings,
-            similarities=entity_similarities,
+            analysis_results=entity_similarities,
             source='multi_layer_test'
         )
+        _assert_tracked(
+            tracker,
+            embed_id,
+            source='multi_layer_test',
+            entity_type='embedding_analysis',
+            embeddings_count=len(realistic_embeddings),
+        )
         
-        # Verify results
         assert len(layer_results) > 0
         assert len(layer_similarities) > 0
         assert len(entity_similarities) > 0
     
-        print(f"Multi-layer analysis completed")
-        print(f"Layers analyzed: {list(multi_layer_network.keys())}")
-        print(f"Layer similarities: {list(layer_similarities.keys())}")
-        
-        return multi_layer_id
-    
     def test_error_handling_and_recovery(self):
         """Test error handling and recovery mechanisms."""
         tracker = AlgorithmTrackerWithProvenance(provenance=True)
+        centrality_calc = CentralityCalculator()
         
         # Test with invalid graph data
         invalid_graph = {
@@ -473,46 +529,24 @@ class TestComprehensiveIntegration:
             'edges': []
         }
         
-        # Should handle gracefully
-        try:
-            centrality_calc = CentralityCalculator()
-            result = centrality_calc.calculate_degree_centrality(invalid_graph)
-            # Should return empty result or handle gracefully
-            assert isinstance(result, dict)
-        except Exception as e:
-            # Should be a controlled exception
-            assert isinstance(e, (ValueError, RuntimeError))
+        result = centrality_calc.calculate_degree_centrality(invalid_graph)
+        assert isinstance(result, dict)
         
-        # Test with invalid embeddings
-        invalid_embeddings = {
-            'node1': [1, 2],  # Different dimensions
-            'node2': [1, 2, 3, 4]  # Different dimensions
-        }
-        
-        try:
-            sim_calc = SimilarityCalculator()
-            result = sim_calc.batch_similarity(
-                embeddings=invalid_embeddings,
-                query_embedding=[1, 2, 3, 4],
-                method='cosine'
-            )
-            # Should handle dimension mismatch
-        except Exception as e:
-            # Should handle gracefully
-            assert isinstance(e, ValueError)
-        
-        # Test provenance tracking with invalid data
-        try:
-            result = tracker.track_embedding_computation(
-                graph=None,  # Invalid graph
-                algorithm='test',
-                embeddings={},
-                parameters={}
-            )
-            # Should either return None or handle gracefully
-        except Exception as e:
-            # Should be a controlled exception
-            assert isinstance(e, (ValueError, TypeError))
+        # Provenance tracking with a None graph still writes a record.
+        result = tracker.track_embedding_computation(
+            graph=None,
+            algorithm='test',
+            embeddings={},
+            parameters={},
+            source='integration_error_recovery'
+        )
+        _assert_tracked(
+            tracker,
+            result,
+            source='integration_error_recovery',
+            algorithm='test',
+            input_data_type='NoneType',
+        )
         
         # Test graceful degradation when provenance is disabled
         tracker_no_prov = AlgorithmTrackerWithProvenance(provenance=False)
@@ -526,8 +560,7 @@ class TestComprehensiveIntegration:
         
         # Should return None when provenance is disabled
         assert result is None
-        
-        print("Error handling and recovery test completed")
+        assert tracker_no_prov._prov_manager is None
     
     def test_performance_benchmarks(self, realistic_embeddings):
         """Test performance benchmarks with realistic data."""

tests/kg/test_provenance_workflows.py

@@ -7,13 +7,13 @@ Tests complete provenance tracking workflows across multiple algorithms.
 import pytest
 import networkx as nx
 import time
+from datetime import datetime, timedelta
 from typing import Dict, List, Any
 import uuid
 
 from semantica.kg import (
     GraphBuilderWithProvenance,
     AlgorithmTrackerWithProvenance,
-    NodeEmbedder,
     SimilarityCalculator,
     LinkPredictor,
     CentralityCalculator,
@@ -21,6 +21,40 @@ from semantica.kg import (
 )
 
 
+def _stored(owner, entity_id):
+    """Read a provenance record back. Fails if tracking only minted an ID."""
+    record = owner._prov_manager.get_provenance(entity_id)
+    assert record is not None, (
+        f"no stored provenance for {entity_id!r} — an ID was generated without a write"
+    )
+    assert record.get("entity_id") == entity_id
+    return record
+
+
+def _assert_utc_iso(value, field="timestamp"):
+    assert value, f"missing {field}"
+    parsed = datetime.fromisoformat(value)
+    assert parsed.tzinfo is not None, (
+        f"{field}={value!r} is naive (datetime.utcnow leftover)"
+    )
+    assert parsed.utcoffset() == timedelta(0), f"{field}={value!r} is not UTC"
+    return parsed
+
+
+def _assert_tracked(owner, entity_id, source, **metadata):
+    record = _stored(owner, entity_id)
+    assert record["source_document"] == source
+    actual = record.get("metadata") or {}
+    for key, expected in metadata.items():
+        assert actual.get(key) == expected, (
+            f"{entity_id} metadata[{key!r}]={actual.get(key)!r}, expected {expected!r}"
+        )
+    _assert_utc_iso(record["timestamp"], "timestamp")
+    if record.get("last_updated"):
+        _assert_utc_iso(record["last_updated"], "last_updated")
+    return record
+
+
 class TestProvenanceWorkflows:
     """Test provenance tracking workflows."""
     
@@ -116,10 +150,25 @@ class TestProvenanceWorkflows:
             source=workflow_id
         )
         
-        assert construction_id is not None
+        _assert_tracked(
+            tracker,
+            construction_id,
+            source=workflow_id,
+            entity_type="graph_construction",
+            entities_count=8,
+            relationships_count=12,
+        )
         assert construction_id.startswith('graph_construction_')
 
+        for entity in graph_result['entities']:
+            built = _stored(builder, entity['id'])
+            assert built['metadata']['operation'] == 'build_entity'
+            assert built['metadata']['entity_type'] == entity['type']
+            _assert_utc_iso(built['activity_started_at_time'], 'activity_started_at_time')
+            _assert_utc_iso(built['activity_ended_at_time'], 'activity_ended_at_time')
+        
         # Step 3: Track entity processing
+        processed = []
         for entity in graph_result['entities']:
             entity_id = tracker.track_entity_processing(
                 entity_id=entity['id'],
@@ -127,9 +176,19 @@ class TestProvenanceWorkflows:
                 entity_data=entity,
                 source=workflow_id
             )
-            assert entity_id is not None
+            _assert_tracked(
+                tracker,
+                entity_id,
+                source=workflow_id,
+                processed_entity_id=entity['id'],
+                processed_entity_type=entity['type'],
+            )
+            processed.append(entity_id)
+        assert len(processed) == 8
+        assert len(set(processed)) == 8
         
         # Step 4: Track relationship processing
+        rel_ids = []
         for relationship in graph_result['relationships']:
             rel_id = tracker.track_relationship_processing(
                 relationship_id=f"{relationship['source']}-{relationship['target']}",
@@ -137,10 +196,15 @@ class TestProvenanceWorkflows:
                 relationship_data=relationship,
                 source=workflow_id
             )
-            assert rel_id is not None
-        
-        print(f"Graph construction workflow completed: {workflow_id}")
-        return workflow_id
+            _assert_tracked(
+                tracker,
+                rel_id,
+                source=workflow_id,
+                processed_relationship_type=relationship['type'],
+            )
+            rel_ids.append(rel_id)
+        assert len(rel_ids) == 12
+        assert len(set(rel_ids)) == 12
     
     def test_embedding_workflow(self, workflow_graph, workflow_embeddings):
         """Test complete embedding workflow with provenance."""
@@ -176,8 +240,20 @@ class TestProvenanceWorkflows:
             source=workflow_id
         )
         
-        assert embed_id is not None
+        _assert_tracked(
+            tracker,
+            embed_id,
+            source=workflow_id,
+            algorithm='node2vec',
+            node_count=6,
+            embedding_dimension=4,
+        )
         assert embed_id.startswith('embedding_')
+        for node_id, vector in computed_embeddings.items():
+            node_record = _stored(tracker, f"embedding_{node_id}")
+            assert node_record['metadata']['node_id'] == node_id
+            assert node_record['metadata']['execution_id'] == embed_id
+            assert node_record['metadata']['embedding_dimension'] == len(vector)
         
         # Step 2: Track embedding quality metrics
         quality_metrics = {
@@ -196,10 +272,13 @@ class TestProvenanceWorkflows:
             source=workflow_id
         )
         
-        assert quality_id is not None
-        
-        print(f"Embedding workflow completed: {workflow_id}")
-        return workflow_id
+        _assert_tracked(
+            tracker,
+            quality_id,
+            source=workflow_id,
+            algorithm='node2vec_quality_check',
+        )
+        assert quality_id != embed_id
     
     def test_similarity_analysis_workflow(self, workflow_embeddings):
         """Test complete similarity analysis workflow with provenance."""
@@ -229,10 +308,18 @@ class TestProvenanceWorkflows:
             source=workflow_id
         )
         
-        assert sim_id is not None
+        _assert_tracked(
+            tracker,
+            sim_id,
+            source=workflow_id,
+            method='cosine',
+            similarities_count=len(similarities),
+        )
         assert sim_id.startswith('similarity_')
+        assert len(similarities) == 3
         
         # Step 2: Track individual similarity results
+        result_ids = []
         for node_id, similarity_score in similarities.items():
             result_id = tracker.track_similarity_result(
                 node_id=node_id,
@@ -241,7 +328,17 @@ class TestProvenanceWorkflows:
                 execution_id=sim_id,
                 source=workflow_id
             )
-            assert result_id is not None
+            record = _assert_tracked(
+                tracker,
+                result_id,
+                source=workflow_id,
+                node_id=node_id,
+                method='cosine',
+                execution_id=sim_id,
+            )
+            assert record['metadata']['similarity_score'] == similarity_score
+            result_ids.append(result_id)
+        assert len(result_ids) == len(similarities)
         
         # Step 3: Track similarity threshold analysis
         threshold = 0.7
@@ -254,10 +351,14 @@ class TestProvenanceWorkflows:
             source=workflow_id
         )
         
-        assert threshold_id is not None
-        
-        print(f"Similarity analysis workflow completed: {workflow_id}")
-        return workflow_id
+        _assert_tracked(
+            tracker,
+            threshold_id,
+            source=workflow_id,
+            threshold=threshold,
+            execution_id=sim_id,
+        )
+        assert _stored(tracker, threshold_id)['metadata']['high_similarity_count'] == len(high_similarity)
     
     def test_link_prediction_workflow(self, workflow_graph):
         """Test complete link prediction workflow with provenance."""
@@ -287,7 +388,13 @@ class TestProvenanceWorkflows:
                 source=workflow_id
             )
             
-            assert pred_id is not None
+            _assert_tracked(
+                tracker,
+                pred_id,
+                source=workflow_id,
+                method=method,
+                predictions_count=len(predictions),
+            )
             assert pred_id.startswith('link_prediction_')
             
             # Step 2: Track individual predictions
@@ -300,10 +407,17 @@ class TestProvenanceWorkflows:
                     execution_id=pred_id,
                     source=workflow_id
                 )
-                assert result_id is not None
-        
-        print(f"Link prediction workflow completed: {workflow_id}")
-        return workflow_id
+                record = _assert_tracked(
+                    tracker,
+                    result_id,
+                    source=workflow_id,
+                    source_node=source,
+                    target_node=target,
+                    method=method,
+                    execution_id=pred_id,
+                )
+                assert record['metadata']['prediction_score'] == score
+        assert len(methods) == 3
     
     def test_centrality_analysis_workflow(self, workflow_graph):
         """Test complete centrality analysis workflow with provenance."""
@@ -326,54 +440,42 @@ class TestProvenanceWorkflows:
             ('eigenvector', centrality_calc.calculate_eigenvector_centrality)
         ]
         
+        tracked_methods = []
         for method_name, method_func in centrality_methods:
             try:
-                start_time = time.time()
                 result = method_func(graph_dict)
-                calculation_time = time.time() - start_time
+            except Exception:
+                # Algorithm failure is allowed for optional methods, not for degree.
+                if method_name == 'degree':
+                    raise
+                continue
 
-                cent_id = tracker.track_centrality_calculation(
-                    graph=workflow_graph,
-                    centrality_scores=result['centrality'],
-                    method=method_name,
-                    parameters={},
-                    calculation_time=calculation_time,
-                    source=workflow_id
-                )
+            calculation_time = 0.0
+            cent_id = tracker.track_centrality_calculation(
+                graph=workflow_graph,
+                centrality_scores=result['centrality'],
+                method=method_name,
+                parameters={},
+                calculation_time=calculation_time,
+                source=workflow_id
+            )
+            _assert_tracked(
+                tracker,
+                cent_id,
+                source=workflow_id,
+                method=method_name,
+                scores_count=len(result['centrality']),
+            )
+            assert cent_id.startswith('centrality_')
+            for node_id, score in result['centrality'].items():
+                score_record = _stored(tracker, f"centrality_{node_id}_{cent_id}")
+                assert score_record['metadata']['node_id'] == node_id
+                assert score_record['metadata']['method'] == method_name
+                assert score_record['metadata']['centrality_score'] == score
+            tracked_methods.append(method_name)
 
-                assert cent_id is not None
-                assert cent_id.startswith('centrality_')
-                
-                # Step 2: Track individual centrality scores
-                for node_id, score in result['centrality'].items():
-                    score_id = tracker.track_centrality_score(
-                        node_id=node_id,
-                        centrality_score=score,
-                        method=method_name,
-                        execution_id=cent_id,
-                        source=workflow_id
-                    )
-                    assert score_id is not None
-                
-                # Step 3: Track centrality ranking analysis
-                rankings = result['rankings']
-                top_nodes = rankings[:3]  # Top 3 nodes
-                
-                ranking_id = tracker.track_centrality_ranking(
-                    execution_id=cent_id,
-                    rankings=rankings,
-                    top_nodes=top_nodes,
-                    method=method_name,
-                    source=workflow_id
-                )
-                
-                assert ranking_id is not None
-                
-            except Exception as e:
-                print(f"Warning: {method_name} centrality failed: {e}")
-        
-        print(f"Centrality analysis workflow completed: {workflow_id}")
-        return workflow_id
+        assert 'degree' in tracked_methods
+        assert len(tracked_methods) >= 1
     
     def test_community_detection_workflow(self, workflow_graph):
         """Test complete community detection workflow with provenance."""
@@ -391,56 +493,39 @@ class TestProvenanceWorkflows:
         # Step 1: Track community detection
         methods = ['label_propagation', 'louvain']
         
+        tracked_methods = []
         for method in methods:
             try:
-                start_time = time.time()
                 result = community_detector.detect_communities(graph_dict, method=method)
-                detection_time = time.time() - start_time
+            except Exception:
+                if method == 'label_propagation':
+                    raise
+                continue
 
-                comm_id = tracker.track_community_detection(
-                    graph=workflow_graph,
-                    communities=result['communities'],
-                    method=method,
-                    parameters={},
-                    detection_time=detection_time,
-                    source=workflow_id
-                )
+            comm_id = tracker.track_community_detection(
+                graph=workflow_graph,
+                communities=result['communities'],
+                method=method,
+                parameters={},
+                detection_time=0.0,
+                source=workflow_id
+            )
+            _assert_tracked(
+                tracker,
+                comm_id,
+                source=workflow_id,
+                method=method,
+                communities_count=len(result['communities']),
+            )
+            assert comm_id.startswith('community_')
+            for i, community in enumerate(result['communities']):
+                comm_record = _stored(tracker, f"community_{comm_id}_{i}")
+                assert comm_record['metadata']['community_id'] == i
+                assert comm_record['metadata']['method'] == method
+                assert comm_record['metadata']['nodes'] == community
+            tracked_methods.append(method)
 
-                assert comm_id is not None
-                assert comm_id.startswith('community_')
-                
-                # Step 2: Track individual communities
-                for i, community in enumerate(result['communities']):
-                    comm_result_id = tracker.track_community_result(
-                        community_id=i,
-                        nodes=community,
-                        method=method,
-                        execution_id=comm_id,
-                        source=workflow_id
-                    )
-                    assert comm_result_id is not None
-                
-                # Step 3: Track community quality metrics
-                quality_metrics = {
-                    'modularity': 0.3,
-                    'num_communities': len(result['communities']),
-                    'avg_community_size': len(result['communities']) / len(result['communities']) if result['communities'] else 0
-                }
-                
-                quality_id = tracker.track_community_quality(
-                    execution_id=comm_id,
-                    metrics=quality_metrics,
-                    method=method,
-                    source=workflow_id
-                )
-                
-                assert quality_id is not None
-                
-            except Exception as e:
-                print(f"Warning: {method} community detection failed: {e}")
-        
-        print(f"Community detection workflow completed: {workflow_id}")
-        return workflow_id
+        assert 'label_propagation' in tracked_methods
     
     def test_comprehensive_provenance_workflow(self, workflow_data, workflow_graph, workflow_embeddings):
         """Test comprehensive provenance workflow combining all algorithms."""
@@ -555,30 +640,38 @@ class TestProvenanceWorkflows:
             source='comprehensive_test'
         )
         
-        # Verify all execution IDs
+        # Verify all execution IDs were stored with the expected payload
         assert len(execution_ids) == 6
+        expected_types = {
+            'construction': ('graph_construction_', 'graph_construction'),
+            'embedding': ('embedding_', 'embedding_computation'),
+            'similarity': ('similarity_', 'similarity_calculation'),
+            'link_prediction': ('link_prediction_', 'link_prediction'),
+            'centrality': ('centrality_', 'centrality_calculation'),
+            'community_detection': ('community_', 'community_detection'),
+        }
         for phase, exec_id in execution_ids.items():
-            assert exec_id is not None
-            assert len(exec_id) > 10
+            prefix, entity_type = expected_types[phase]
+            assert exec_id.startswith(prefix)
+            _assert_tracked(
+                tracker,
+                exec_id,
+                source=master_workflow_id,
+                entity_type=entity_type,
+            )
+
+        summary_record = _assert_tracked(
+            tracker,
+            summary_id,
+            source='comprehensive_test',
+            entity_type='workflow_summary',
+        )
+        assert summary_id.startswith('workflow_summary_')
+        assert summary_record['metadata']['master_workflow_id'] == master_workflow_id
 
-        # Verify all IDs are unique
         all_ids = list(execution_ids.values()) + [summary_id]
         assert len(set(all_ids)) == len(all_ids)
     
-        # Verify ID prefixes
-        assert execution_ids['construction'].startswith('graph_construction_')
-        assert execution_ids['embedding'].startswith('embedding_')
-        assert execution_ids['similarity'].startswith('similarity_')
-        assert execution_ids['link_prediction'].startswith('link_prediction_')
-        assert execution_ids['centrality'].startswith('centrality_')
-        assert execution_ids['community_detection'].startswith('community_')
-        assert summary_id.startswith('workflow_summary_')
-        
-        print(f"Comprehensive provenance workflow completed: {master_workflow_id}")
-        print(f"Execution IDs: {list(execution_ids.keys())}")
-        
-        return master_workflow_id
-    
     def test_provenance_data_integrity(self, workflow_graph, workflow_embeddings):
         """Test provenance data integrity and consistency."""
         tracker = AlgorithmTrackerWithProvenance(provenance=True)
@@ -618,24 +711,24 @@ class TestProvenanceWorkflows:
         )
         operations.append(('link_prediction', link_id))
         
-        # Verify data integrity
+        expected = {
+            'embedding': ('embedding_', 'embedding_computation', embed_id),
+            'similarity': ('similarity_', 'similarity_calculation', sim_id),
+            'link_prediction': ('link_prediction_', 'link_prediction', link_id),
+        }
         for op_type, op_id in operations:
-            assert op_id is not None
-            assert len(op_id) > 10
+            prefix, entity_type, expected_id = expected[op_type]
+            assert op_id == expected_id
+            assert op_id.startswith(prefix)
+            _assert_tracked(
+                tracker,
+                op_id,
+                source=workflow_id,
+                entity_type=entity_type,
+            )
 
-            # Verify ID format consistency
-            if op_type == 'embedding':
-                assert op_id.startswith('embedding_')
-            elif op_type == 'similarity':
-                assert op_id.startswith('similarity_')
-            elif op_type == 'link_prediction':
-                assert op_id.startswith('link_prediction_')
-        
-        # Verify workflow consistency
         workflow_ids = [op_id for _, op_id in operations]
-        assert len(set(workflow_ids)) == len(workflow_ids)  # All unique
-        
-        print(f"Provenance data integrity test completed: {workflow_id}")
+        assert len(set(workflow_ids)) == len(workflow_ids)
     
     def test_provenance_error_recovery(self):
         """Test provenance system error recovery."""
@@ -650,24 +743,27 @@ class TestProvenanceWorkflows:
         )
         
         assert result is None  # Should return None when provenance is disabled
+        assert tracker_no_prov._prov_manager is None
         
-        # Test error handling during tracking
+        # Tracking still records an execution when the graph is None — that is
+        # current production behavior, so assert the write rather than
+        # "either returns or raises".
         tracker = AlgorithmTrackerWithProvenance(provenance=True)
-        
-        # This should not raise exceptions even with invalid data
-        try:
-            result = tracker.track_embedding_computation(
-                graph=None,  # Invalid graph
-                algorithm='test',
-                embeddings={},
-                parameters={}
-            )
-            # Should either return None or handle gracefully
-        except Exception as e:
-            # If it raises, it should be a controlled exception
-            assert isinstance(e, (ValueError, TypeError))
-        
-        print("Provenance error recovery test completed")
+        result = tracker.track_embedding_computation(
+            graph=None,
+            algorithm='test',
+            embeddings={},
+            parameters={},
+            source='error_recovery'
+        )
+        _assert_tracked(
+            tracker,
+            result,
+            source='error_recovery',
+            algorithm='test',
+            input_data_type='NoneType',
+            node_count=0,
+        )
 
 
 if __name__ == '__main__':

tests/kg/test_provenance_workflows_simple.py

@@ -7,13 +7,13 @@ Tests provenance tracking workflows using only available methods.
 import pytest
 import networkx as nx
 import time
+from datetime import datetime, timedelta
 from typing import Dict, List, Any
 import uuid
 
 from semantica.kg import (
     GraphBuilderWithProvenance,
     AlgorithmTrackerWithProvenance,
-    NodeEmbedder,
     SimilarityCalculator,
     LinkPredictor,
     CentralityCalculator,
@@ -21,6 +21,40 @@ from semantica.kg import (
 )
 
 
+def _stored(owner, entity_id):
+    """Read a provenance record back. Fails if tracking only minted an ID."""
+    record = owner._prov_manager.get_provenance(entity_id)
+    assert record is not None, (
+        f"no stored provenance for {entity_id!r} — an ID was generated without a write"
+    )
+    assert record.get("entity_id") == entity_id
+    return record
+
+
+def _assert_utc_iso(value, field="timestamp"):
+    assert value, f"missing {field}"
+    parsed = datetime.fromisoformat(value)
+    assert parsed.tzinfo is not None, (
+        f"{field}={value!r} is naive (datetime.utcnow leftover)"
+    )
+    assert parsed.utcoffset() == timedelta(0), f"{field}={value!r} is not UTC"
+    return parsed
+
+
+def _assert_tracked(owner, entity_id, source, **metadata):
+    record = _stored(owner, entity_id)
+    assert record["source_document"] == source
+    actual = record.get("metadata") or {}
+    for key, expected in metadata.items():
+        assert actual.get(key) == expected, (
+            f"{entity_id} metadata[{key!r}]={actual.get(key)!r}, expected {expected!r}"
+        )
+    _assert_utc_iso(record["timestamp"], "timestamp")
+    if record.get("last_updated"):
+        _assert_utc_iso(record["last_updated"], "last_updated")
+    return record
+
+
 class TestProvenanceWorkflowsSimple:
     """Test provenance tracking workflows with available methods."""
     
@@ -93,11 +127,19 @@ class TestProvenanceWorkflowsSimple:
             source=workflow_id
         )
         
-        assert embed_id is not None
+        _assert_tracked(
+            tracker,
+            embed_id,
+            source=workflow_id,
+            algorithm='node2vec',
+            node_count=6,
+            embedding_dimension=4,
+        )
         assert embed_id.startswith('embedding_')
-        
-        print(f"Simple embedding workflow completed: {workflow_id}")
-        return workflow_id
+        for node_id in workflow_embeddings:
+            node_record = _stored(tracker, f"embedding_{node_id}")
+            assert node_record['metadata']['execution_id'] == embed_id
+            assert node_record['metadata']['node_id'] == node_id
     
     def test_similarity_workflow_simple(self, workflow_embeddings):
         """Test simple similarity workflow with provenance."""
@@ -124,11 +166,15 @@ class TestProvenanceWorkflowsSimple:
             source=workflow_id
         )
         
-        assert sim_id is not None
+        _assert_tracked(
+            tracker,
+            sim_id,
+            source=workflow_id,
+            method='cosine',
+            similarities_count=len(similarities),
+        )
         assert sim_id.startswith('similarity_')
-        
-        print(f"Simple similarity workflow completed: {workflow_id}")
-        return workflow_id
+        assert len(similarities) == 3
     
     def test_link_prediction_workflow_simple(self, workflow_graph):
         """Test simple link prediction workflow with provenance."""
@@ -153,11 +199,15 @@ class TestProvenanceWorkflowsSimple:
             source=workflow_id
         )
         
-        assert link_id is not None
+        _assert_tracked(
+            tracker,
+            link_id,
+            source=workflow_id,
+            method='preferential_attachment',
+            predictions_count=len(predictions),
+        )
         assert link_id.startswith('link_prediction_')
-        
-        print(f"Simple link prediction workflow completed: {workflow_id}")
-        return workflow_id
+        assert len(predictions) >= 1
     
     def test_centrality_workflow_simple(self, workflow_graph):
         """Test simple centrality workflow with provenance."""
@@ -184,11 +234,15 @@ class TestProvenanceWorkflowsSimple:
             source=workflow_id
         )
         
-        assert cent_id is not None
+        _assert_tracked(
+            tracker,
+            cent_id,
+            source=workflow_id,
+            method='degree',
+            scores_count=len(degree_cent['centrality']),
+        )
         assert cent_id.startswith('centrality_')
-        
-        print(f"Simple centrality workflow completed: {workflow_id}")
-        return workflow_id
+        assert set(degree_cent['centrality']) == set(workflow_graph.nodes())
     
     def test_community_detection_workflow_simple(self, workflow_graph):
         """Test simple community detection workflow with provenance."""
@@ -215,11 +269,15 @@ class TestProvenanceWorkflowsSimple:
             source=workflow_id
         )
         
-        assert comm_id is not None
+        _assert_tracked(
+            tracker,
+            comm_id,
+            source=workflow_id,
+            method='label_propagation',
+            communities_count=len(communities['communities']),
+        )
         assert comm_id.startswith('community_')
-        
-        print(f"Simple community detection workflow completed: {workflow_id}")
-        return workflow_id
+        assert len(communities['communities']) >= 1
     
     def test_graph_construction_workflow_simple(self, workflow_data):
         """Test simple graph construction workflow with provenance."""
@@ -237,23 +295,27 @@ class TestProvenanceWorkflowsSimple:
         assert len(graph_result['entities']) == 4
         assert len(graph_result['relationships']) == 3
         
-        # Track graph construction using embedding computation method as proxy
-        construction_id = tracker.track_embedding_computation(
-            graph={'nodes': list(graph_result['entities']), 'edges': list(graph_result['relationships'])},
-            algorithm='graph_construction',
-            embeddings={'graph_size': len(graph_result['entities'])},
-            parameters={
-                'entities_count': len(graph_result['entities']),
-                'relationships_count': len(graph_result['relationships'])
-            },
+        construction_id = tracker.track_graph_construction(
+            input_data=workflow_data,
+            output_graph=graph_result,
+            entities_count=len(graph_result['entities']),
+            relationships_count=len(graph_result['relationships']),
+            construction_time=0.0,
             source=workflow_id
         )
-        
-        assert construction_id is not None
-        assert construction_id.startswith('embedding_')  # Using embedding method as proxy
-        
-        print(f"Simple graph construction workflow completed: {workflow_id}")
-        return workflow_id
+        _assert_tracked(
+            tracker,
+            construction_id,
+            source=workflow_id,
+            entity_type='graph_construction',
+            entities_count=4,
+            relationships_count=3,
+        )
+        assert construction_id.startswith('graph_construction_')
+        for entity in graph_result['entities']:
+            built = _stored(builder, entity['id'])
+            assert built['metadata']['operation'] == 'build_entity'
+            _assert_utc_iso(built['activity_started_at_time'], 'activity_started_at_time')
     
     def test_comprehensive_workflow_simple(self, workflow_data, workflow_graph, workflow_embeddings):
         """Test comprehensive workflow with all available methods."""
@@ -271,11 +333,12 @@ class TestProvenanceWorkflowsSimple:
         # Phase 1: Graph Construction
         graph_result = builder.build_single_source(workflow_data)
         
-        construction_id = tracker.track_embedding_computation(
-            graph={'nodes': list(graph_result['entities']), 'edges': list(graph_result['relationships'])},
-            algorithm='graph_construction',
-            embeddings={'graph_size': len(graph_result['entities'])},
-            parameters={'entities_count': len(graph_result['entities'])},
+        construction_id = tracker.track_graph_construction(
+            input_data=workflow_data,
+            output_graph=graph_result,
+            entities_count=len(graph_result['entities']),
+            relationships_count=len(graph_result['relationships']),
+            construction_time=0.0,
             source=master_workflow_id
         )
         execution_ids['construction'] = construction_id
@@ -351,27 +414,25 @@ class TestProvenanceWorkflowsSimple:
         )
         execution_ids['community_detection'] = comm_id
         
-        # Verify all execution IDs
+        expected = {
+            'construction': ('graph_construction_', 'graph_construction'),
+            'embedding': ('embedding_', 'embedding_computation'),
+            'similarity': ('similarity_', 'similarity_calculation'),
+            'link_prediction': ('link_prediction_', 'link_prediction'),
+            'centrality': ('centrality_', 'centrality_calculation'),
+            'community_detection': ('community_', 'community_detection'),
+        }
         assert len(execution_ids) == 6
         for phase, exec_id in execution_ids.items():
-            assert exec_id is not None
-            assert len(exec_id) > 10
-        
-        # Verify all IDs are unique
-        all_ids = list(execution_ids.values())
-        assert len(set(all_ids)) == len(all_ids)
-        
-        # Verify ID prefixes
-        assert execution_ids['embedding'].startswith('embedding_')
-        assert execution_ids['similarity'].startswith('similarity_')
-        assert execution_ids['link_prediction'].startswith('link_prediction_')
-        assert execution_ids['centrality'].startswith('centrality_')
-        assert execution_ids['community_detection'].startswith('community_')
-        
-        print(f"Comprehensive workflow completed: {master_workflow_id}")
-        print(f"Execution IDs: {list(execution_ids.keys())}")
-        
-        return master_workflow_id
+            prefix, entity_type = expected[phase]
+            assert exec_id.startswith(prefix)
+            _assert_tracked(
+                tracker,
+                exec_id,
+                source=master_workflow_id,
+                entity_type=entity_type,
+            )
+        assert len(set(execution_ids.values())) == len(execution_ids)
     
     def test_provenance_data_integrity_simple(self, workflow_graph, workflow_embeddings):
         """Test provenance data integrity with available methods."""
@@ -412,24 +473,19 @@ class TestProvenanceWorkflowsSimple:
         )
         operations.append(('link_prediction', link_id))
         
-        # Verify data integrity
+        expected = {
+            'embedding': 'embedding_computation',
+            'similarity': 'similarity_calculation',
+            'link_prediction': 'link_prediction',
+        }
         for op_type, op_id in operations:
-            assert op_id is not None
-            assert len(op_id) > 10
-            
-            # Verify ID format consistency
-            if op_type == 'embedding':
-                assert op_id.startswith('embedding_')
-            elif op_type == 'similarity':
-                assert op_id.startswith('similarity_')
-            elif op_type == 'link_prediction':
-                assert op_id.startswith('link_prediction_')
-        
-        # Verify workflow consistency
-        workflow_ids = [op_id for _, op_id in operations]
-        assert len(set(workflow_ids)) == len(workflow_ids)  # All unique
-        
-        print(f"Provenance data integrity test completed: {workflow_id}")
+            _assert_tracked(
+                tracker,
+                op_id,
+                source=workflow_id,
+                entity_type=expected[op_type],
+            )
+        assert len(set(op_id for _, op_id in operations)) == 3
     
     def test_provenance_error_recovery_simple(self):
         """Test provenance system error recovery."""
@@ -443,25 +499,25 @@ class TestProvenanceWorkflowsSimple:
             parameters={}
         )
         
-        assert result is None  # Should return None when provenance is disabled
+        assert result is None
+        assert tracker_no_prov._prov_manager is None
 
-        # Test error handling during tracking
         tracker = AlgorithmTrackerWithProvenance(provenance=True)
-        
-        # This should not raise exceptions even with invalid data
-        try:
-            result = tracker.track_embedding_computation(
-                graph=None,  # Invalid graph
-                algorithm='test',
-                embeddings={},
-                parameters={}
-            )
-            # Should either return None or handle gracefully
-        except Exception as e:
-            # If it raises, it should be a controlled exception
-            assert isinstance(e, (ValueError, TypeError))
-        
-        print("Provenance error recovery test completed")
+        result = tracker.track_embedding_computation(
+            graph=None,
+            algorithm='test',
+            embeddings={},
+            parameters={},
+            source='error_recovery'
+        )
+        _assert_tracked(
+            tracker,
+            result,
+            source='error_recovery',
+            algorithm='test',
+            input_data_type='NoneType',
+            node_count=0,
+        )
 
 
 if __name__ == '__main__':

tests/provenance/test_manager.py

@@ -6,8 +6,9 @@ chunk tracking, source tracking, and lineage tracing.
 """
 
 import pytest
+import warnings
 from unittest.mock import patch
-from datetime import datetime
+from datetime import datetime, timedelta, timezone
 from semantica.provenance import ProvenanceManager, SourceReference, ProvenanceEntry
 from semantica.provenance.storage import InMemoryStorage, SQLiteStorage
 
@@ -733,8 +734,8 @@ class TestProvenanceManager:
             entity_type="entity",
             activity_id="test",
             source_document="doc_1",
-            first_seen=datetime.utcnow().isoformat(),
-            last_updated=datetime.utcnow().isoformat(),
+            first_seen=datetime.now(timezone.utc).isoformat(),
+            last_updated=datetime.now(timezone.utc).isoformat(),
         )
         with patch.object(prov_mgr.storage, "store", side_effect=RuntimeError("storage error")), \
              patch.object(prov_mgr.logger, "error") as mock_log_error:
@@ -1672,3 +1673,195 @@ class TestActivityTimingAcrossWrappers:
         assert entry["activity_ended_at_time"] is not None
 
 
+class TestTimezoneAwareUtcTimestamps:
+    """Issue #946 — timezone-aware UTC stamps without datetime.utcnow()."""
+
+    def test_track_entity_stamps_timezone_aware_utc(self):
+        before = datetime.now(timezone.utc) - timedelta(seconds=1)
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter("always", DeprecationWarning)
+            prov_mgr = ProvenanceManager()
+            entry = prov_mgr.track_entity("utc_entity", source="doc_1")
+        after = datetime.now(timezone.utc) + timedelta(seconds=1)
+
+        utcnow_warnings = [
+            w
+            for w in caught
+            if issubclass(w.category, DeprecationWarning)
+            and "utcnow" in str(w.message).lower()
+        ]
+        assert utcnow_warnings == []
+
+        for field in ("timestamp", "first_seen", "last_updated"):
+            value = getattr(entry, field)
+            parsed = datetime.fromisoformat(value)
+            assert parsed.tzinfo is not None, f"{field}={value!r} is naive"
+            assert parsed.utcoffset() == timedelta(0), f"{field}={value!r} is not UTC"
+            assert before <= parsed <= after
+
+        stored = prov_mgr.get_provenance("utc_entity")
+        assert stored is not None
+        assert stored["source_document"] == "doc_1"
+        assert stored["last_updated"] == entry.last_updated
+
+    def test_invalidate_stamps_timezone_aware_utc(self):
+        prov_mgr = ProvenanceManager()
+        prov_mgr.track_entity("utc_invalidate", source="doc_1")
+        result = prov_mgr.invalidate(
+            "utc_invalidate", agent_id="reviewer", reason="test"
+        )
+        parsed = datetime.fromisoformat(result.invalidated_at_time)
+        assert parsed.tzinfo is not None
+        assert parsed.utcoffset() == timedelta(0)
+
+    def test_graph_builder_activity_times_are_timezone_aware_utc(self):
+        from semantica.kg.kg_provenance import GraphBuilderWithProvenance
+
+        builder = GraphBuilderWithProvenance(provenance=True, agent_id="builder_svc")
+        result = builder.build_single_source({
+            "entities": [{"id": "person1", "type": "Person", "name": "Ada"}],
+            "relationships": [],
+        })
+        assert result["entities"][0]["id"] == "person1"
+
+        person = builder._prov_manager.get_provenance("person1")
+        assert person is not None
+        assert person["metadata"]["operation"] == "build_entity"
+        for field in ("activity_started_at_time", "activity_ended_at_time"):
+            parsed = datetime.fromisoformat(person[field])
+            assert parsed.tzinfo is not None, (
+                f"{field}={person[field]!r} is naive"
+            )
+            assert parsed.utcoffset() == timedelta(0)
+        started = datetime.fromisoformat(person["activity_started_at_time"])
+        ended = datetime.fromisoformat(person["activity_ended_at_time"])
+        assert started <= ended
+
+class TestMixedFormatTimestampComparisons:
+    """Issue #946 review — naive and offset-bearing stamps must compare as instants.
+
+    Records written before the timezone-aware change carry a naive UTC stamp
+    (``2026-08-19T22:35:38.501697``); records written after carry ``+00:00``.
+    Comparing the two as raw strings puts the offset-bearing form above a naive
+    bound at the identical instant, so the record falls outside a range that
+    should contain it. These tests pin the boundary in both directions.
+    """
+
+    NAIVE = "2026-08-19T12:00:00.500000"
+    AWARE = "2026-08-19T12:00:00.500000+00:00"
+
+    def _manager_with(self, *timestamps):
+        prov_mgr = ProvenanceManager()
+        for index, stamp in enumerate(timestamps):
+            prov_mgr.storage.store(
+                ProvenanceEntry(
+                    entity_id=f"entity_{index}",
+                    entity_type="entity",
+                    activity_id="test",
+                    source_document="doc_1",
+                    timestamp=stamp,
+                )
+            )
+        return prov_mgr
+
+    def test_raw_string_compare_would_exclude_the_boundary_record(self):
+        """Guards the premise: the two forms are not string-comparable."""
+        assert not (self.AWARE <= self.NAIVE)
+        assert datetime.fromisoformat(self.AWARE) == datetime.fromisoformat(
+            self.NAIVE
+        ).replace(tzinfo=timezone.utc)
+
+    def test_query_range_with_naive_bounds_includes_aware_record(self):
+        prov_mgr = self._manager_with(self.AWARE)
+        results = prov_mgr.query_recorded_between(
+            "2026-08-19T12:00:00.500000", "2026-08-19T12:00:00.500000"
+        )
+        assert [r["entity_id"] for r in results] == ["entity_0"]
+
+    def test_query_range_with_aware_bounds_includes_naive_record(self):
+        prov_mgr = self._manager_with(self.NAIVE)
+        results = prov_mgr.query_recorded_between(
+            "2026-08-19T12:00:00.500000+00:00", "2026-08-19T12:00:00.500000+00:00"
+        )
+        assert [r["entity_id"] for r in results] == ["entity_0"]
+
+    def test_query_range_returns_both_formats_from_a_mixed_store(self):
+        prov_mgr = self._manager_with(self.NAIVE, self.AWARE)
+        results = prov_mgr.query_recorded_between(
+            "2026-08-19T11:00:00", "2026-08-19T13:00:00+00:00"
+        )
+        assert {r["entity_id"] for r in results} == {"entity_0", "entity_1"}
+
+    def test_query_range_sorts_mixed_formats_chronologically(self):
+        prov_mgr = self._manager_with(
+            "2026-08-19T12:00:02+00:00",  # entity_0, latest
+            "2026-08-19T12:00:00",        # entity_1, earliest
+            "2026-08-19T12:00:01+00:00",  # entity_2, middle
+        )
+        results = prov_mgr.query_recorded_between(
+            "2026-08-19T11:00:00", "2026-08-19T13:00:00"
+        )
+        assert [r["entity_id"] for r in results] == [
+            "entity_1",
+            "entity_2",
+            "entity_0",
+        ]
+
+    def test_query_range_accepts_trailing_z(self):
+        prov_mgr = self._manager_with(self.NAIVE)
+        results = prov_mgr.query_recorded_between(
+            "2026-08-19T11:00:00Z", "2026-08-19T13:00:00Z"
+        )
+        assert len(results) == 1
+
+    def test_query_range_skips_unparseable_timestamp(self):
+        prov_mgr = self._manager_with("not-a-timestamp", self.AWARE)
+        results = prov_mgr.query_recorded_between(
+            "2026-08-19T11:00:00", "2026-08-19T13:00:00"
+        )
+        assert [r["entity_id"] for r in results] == ["entity_1"]
+
+    def test_audit_log_since_naive_bound_includes_aware_record(self):
+        prov_mgr = self._manager_with(self.AWARE)
+        entries = prov_mgr.audit_log(
+            since="2026-08-19T12:00:00.500000", format="json"
+        )
+        assert [e["entity_id"] for e in entries] == ["entity_0"]
+
+    def test_audit_log_since_aware_bound_includes_naive_record(self):
+        prov_mgr = self._manager_with(self.NAIVE)
+        entries = prov_mgr.audit_log(
+            since="2026-08-19T12:00:00.500000+00:00", format="json"
+        )
+        assert [e["entity_id"] for e in entries] == ["entity_0"]
+
+    def test_audit_log_excludes_records_before_since_across_formats(self):
+        prov_mgr = self._manager_with(
+            "2026-08-19T11:59:59",        # entity_0, before
+            "2026-08-19T12:00:01+00:00",  # entity_1, after
+        )
+        entries = prov_mgr.audit_log(since="2026-08-19T12:00:00+00:00", format="json")
+        assert [e["entity_id"] for e in entries] == ["entity_1"]
+
+    def test_audit_log_sorts_mixed_formats_chronologically(self):
+        prov_mgr = self._manager_with(
+            "2026-08-19T12:00:02+00:00",
+            "2026-08-19T12:00:00",
+            "2026-08-19T12:00:01+00:00",
+        )
+        entries = prov_mgr.audit_log(format="json")
+        assert [e["entity_id"] for e in entries] == [
+            "entity_1",
+            "entity_2",
+            "entity_0",
+        ]
+
+    def test_audit_log_unparseable_timestamp_sorts_first(self):
+        prov_mgr = self._manager_with("not-a-timestamp", "2026-08-19T12:00:00+00:00")
+        entries = prov_mgr.audit_log(format="json")
+        assert [e["entity_id"] for e in entries] == ["entity_0", "entity_1"]
+
+    def test_audit_log_since_excludes_unparseable_timestamp(self):
+        prov_mgr = self._manager_with("not-a-timestamp", "2026-08-19T12:00:00+00:00")
+        entries = prov_mgr.audit_log(since="2026-08-19T11:00:00", format="json")
+        assert [e["entity_id"] for e in entries] == ["entity_1"]