From 2dd06aadcd93a012af49079ac188f42ff80ac783 Mon Sep 17 00:00:00 2001 From: Sameer6305 Date: Sun, 26 Jul 2026 16:01:50 +0530 Subject: [PATCH 1/6] feat(provenance): implement Global Default Storage pattern, CLI methods, and orchestrator config wiring - Add _default_storage_path, set_default_storage_path(), and test-isolation context manager default_storage_path() in ProvenanceManager - Accept config kwarg in ProvenanceManager.__init__ to fix CLI initialization bug - Implement audit_log(), lineage(), export_prov(), and check() on ProvenanceManager matching cli.py expectations - Wire provenance.storage_path in Semantica.__init__ before pipeline stages execute - Add comprehensive unit tests in tests/provenance/test_manager.py for CLI methods and test isolation --- semantica/cli.py | 3 + semantica/core/config_manager.py | 4 +- semantica/core/orchestrator.py | 16 +-- semantica/provenance/__init__.py | 3 +- semantica/provenance/manager.py | 196 ++++++++++++++++++++++++++++++- tests/provenance/test_manager.py | 123 ++++++++++++++++++- 6 files changed, 328 insertions(+), 17 deletions(-) diff --git a/semantica/cli.py b/semantica/cli.py index c9c362e6..c66d383f 100644 --- a/semantica/cli.py +++ b/semantica/cli.py @@ -2535,6 +2535,9 @@ def provenance_audit(cli_ctx: CLIContext, since: Optional[str], fmt: str, cli_ctx = _require_ctx(cli_ctx) def _action() -> None: + if _is_dry(cli_ctx, False): + _dry(cli_ctx, "export audit log", since=since, format=fmt, output=output) + return try: from .provenance import ProvenanceManager pm = ProvenanceManager(config=cli_ctx.config.to_dict()) diff --git a/semantica/core/config_manager.py b/semantica/core/config_manager.py index 219bd157..1446c0bf 100644 --- a/semantica/core/config_manager.py +++ b/semantica/core/config_manager.py @@ -51,7 +51,6 @@ Author: Semantica Contributors License: MIT """ -import json import os from pathlib import Path from typing import Any, Dict, List, Optional, Union @@ -67,7 +66,6 @@ from ..utils.helpers import ( set_nested_value, ) from ..utils.progress_tracker import get_progress_tracker -from ..utils.validators import validate_config class Config: @@ -157,6 +155,7 @@ class Config: self.security = config_data.get( "security", DEFAULT_CONFIG.get("security", {}) ) + self.provenance = config_data.get("provenance", {}) self.custom = config_data.get("custom", {}) def _load_from_env(self, config_dict: Dict[str, Any]) -> None: @@ -346,6 +345,7 @@ class Config: "logging": self.logging, "quality": self.quality, "security": self.security, + "provenance": self.provenance, "custom": self.custom, } diff --git a/semantica/core/orchestrator.py b/semantica/core/orchestrator.py index e9d9a138..f0c8cfb2 100644 --- a/semantica/core/orchestrator.py +++ b/semantica/core/orchestrator.py @@ -27,11 +27,11 @@ License: MIT from pathlib import Path from typing import Any, Dict, List, Optional, Union -from ..utils.exceptions import ConfigurationError, ProcessingError +from ..utils.exceptions import ProcessingError from ..utils.logging import get_logger, log_execution_time from ..utils.progress_tracker import get_progress_tracker from .config_manager import Config, ConfigManager -from .lifecycle import LifecycleManager, SystemState +from .lifecycle import LifecycleManager from .plugin_registry import PluginRegistry @@ -92,7 +92,9 @@ class Semantica: # Configure global provenance storage path if specified try: - prov_storage_path = self.config.get("provenance", {}).get("storage_path") + prov_storage_path = self.config.get("provenance.storage_path") + if not prov_storage_path and isinstance(self.config.get("provenance"), dict): + prov_storage_path = self.config.get("provenance", {}).get("storage_path") if prov_storage_path: from ..provenance import ProvenanceManager ProvenanceManager.set_default_storage_path(prov_storage_path) @@ -640,10 +642,10 @@ class Semantica: try: # Import key modules to verify they're available # These imports don't create instances, just verify module availability - from ..ingest import FileIngestor - from ..kg import GraphBuilder - from ..parse import DocumentParser - from ..pipeline import PipelineBuilder + from ..ingest import FileIngestor # noqa: F401 + from ..kg import GraphBuilder # noqa: F401 + from ..parse import DocumentParser # noqa: F401 + from ..pipeline import PipelineBuilder # noqa: F401 self.logger.debug("Framework modules verified and available") except (ImportError, OSError) as e: diff --git a/semantica/provenance/__init__.py b/semantica/provenance/__init__.py index 706f804a..c6faaea4 100644 --- a/semantica/provenance/__init__.py +++ b/semantica/provenance/__init__.py @@ -43,7 +43,7 @@ License: MIT from .schemas import ProvenanceEntry, SourceReference from .storage import ProvenanceStorage, InMemoryStorage, SQLiteStorage -from .manager import ProvenanceManager +from .manager import ProvenanceManager, default_storage_path from .integrity import compute_checksum, verify_checksum __all__ = [ @@ -58,6 +58,7 @@ __all__ = [ # Manager "ProvenanceManager", + "default_storage_path", # Utilities "compute_checksum", diff --git a/semantica/provenance/manager.py b/semantica/provenance/manager.py index 85b516ce..04414c02 100644 --- a/semantica/provenance/manager.py +++ b/semantica/provenance/manager.py @@ -24,15 +24,30 @@ Author: Semantica Contributors License: MIT """ -from typing import Optional, List, Dict, Any +from typing import Optional, List, Dict, Any, Union from collections.abc import Mapping from datetime import datetime +from contextlib import contextmanager -from .schemas import ProvenanceEntry, SourceReference, PropertySource +from .schemas import ProvenanceEntry, SourceReference from .storage import ProvenanceStorage, InMemoryStorage, SQLiteStorage from .integrity import compute_checksum +@contextmanager +def default_storage_path(path: Optional[str]): + """ + Context manager for temporarily setting ProvenanceManager._default_storage_path. + Guarantees restoration to the previous value on exit (safe for tests). + """ + original_path = ProvenanceManager._default_storage_path + ProvenanceManager._default_storage_path = path + try: + yield + finally: + ProvenanceManager._default_storage_path = original_path + + class ProvenanceManager: """ Unified provenance tracking manager. @@ -57,15 +72,29 @@ class ProvenanceManager: _default_storage_path: Optional[str] = None @classmethod - def set_default_storage_path(cls, path: str) -> None: + def set_default_storage_path(cls, path: Optional[str]) -> None: """Set a global default storage path for all new instances.""" cls._default_storage_path = path + + @classmethod + @contextmanager + def default_storage_path(cls, path: Optional[str]): + """ + Context manager for temporarily setting the global default storage path. + Guarantees restoration to the previous value on exit (safe for tests). + """ + original_path = cls._default_storage_path + cls._default_storage_path = path + try: + yield + finally: + cls._default_storage_path = original_path def __init__( self, storage: Optional[ProvenanceStorage] = None, storage_path: Optional[str] = None, - config: Optional[Dict[str, Any]] = None, + config: Optional[Any] = None, **kwargs ): """ @@ -74,14 +103,19 @@ class ProvenanceManager: Args: storage: Custom storage backend (optional) storage_path: Path to SQLite database (optional, uses in-memory if None) - config: Configuration dictionary (optional) + config: Configuration dictionary or mapping (optional) """ if storage: self.storage = storage return if not storage_path and config: - storage_path = config.get("provenance", {}).get("storage_path") + if isinstance(config, Mapping) or isinstance(config, dict): + prov_config = config.get("provenance", {}) + if (isinstance(prov_config, Mapping) or isinstance(prov_config, dict)) and "storage_path" in prov_config: + storage_path = prov_config.get("storage_path") + elif "storage_path" in config: + storage_path = config.get("storage_path") if not storage_path and self._default_storage_path: storage_path = self._default_storage_path @@ -626,3 +660,153 @@ class ProvenanceManager: if e.source_document )) } + + # === CLI Integration Methods === + + def lineage(self, entity_id: str, depth: int = 3) -> Dict[str, Any]: + """ + Get lineage information formatted for CLI display. + + Args: + entity_id: ID of the entity to trace lineage for + depth: Maximum traversal depth (default: 3) + + Returns: + Dict containing entity_id, depth, count, lineage entries, and sources + """ + base_lineage = self.get_lineage(entity_id) + entries = base_lineage.get("entries", []) + if len(entries) > depth: + entries = entries[:depth] + sources = self.get_all_sources(entity_id) + return { + "entity_id": entity_id, + "depth": depth, + "chain_length": len(entries), + "source_documents": base_lineage.get("source_documents", []), + "lineage": entries, + "entries": entries, + "sources": sources, + "metadata": base_lineage.get("metadata", {}), + } + + def audit_log( + self, since: Optional[str] = None, format: str = "table" + ) -> Union[str, List[Dict[str, Any]]]: + """ + Export audit log of provenance entries. + + Args: + since: Optional ISO 8601 date string filter + format: Output format ('table', 'csv', 'json') + + Returns: + Formatted audit log as string or list of dicts + """ + entries = self.storage.retrieve_all() + if since: + entries = [e for e in entries if getattr(e, "timestamp", "") >= since] + entries.sort(key=lambda e: getattr(e, "timestamp", "")) + + if format == "json": + return [ + e.to_dict() if hasattr(e, "to_dict") else getattr(e, "__dict__", {}) + for e in entries + ] + elif format == "csv": + lines = ["entity_id,entity_type,activity_id,agent_id,timestamp"] + for e in entries: + lines.append( + f"{e.entity_id},{e.entity_type},{e.activity_id},{e.agent_id},{getattr(e, 'timestamp', '')}" + ) + return "\n".join(lines) + else: + lines = [ + f"{'ENTITY_ID':<20} {'TYPE':<15} {'ACTIVITY':<15} {'TIMESTAMP':<25}" + ] + lines.append("-" * 75) + for e in entries: + lines.append( + f"{str(e.entity_id):<20} {str(e.entity_type):<15} {str(e.activity_id):<15} {str(getattr(e, 'timestamp', '')):<25}" + ) + return "\n".join(lines) + + def export_prov(self, format: str = "turtle") -> str: + """ + Export provenance as W3C PROV-O RDF. + + Args: + format: RDF format ('turtle', 'ntriples', 'jsonld') + + Returns: + Serialized RDF string + """ + from rdflib import Graph, Literal, Namespace, URIRef + from rdflib.namespace import RDF, XSD + + PROV = Namespace("http://www.w3.org/ns/prov#") + EX = Namespace("http://example.org/ns/") + + g = Graph() + g.bind("prov", PROV) + g.bind("ex", EX) + + for e in self.storage.retrieve_all(): + ent_uri = URIRef(EX[str(e.entity_id)]) + g.add((ent_uri, RDF.type, PROV.Entity)) + + if getattr(e, "timestamp", None): + g.add( + ( + ent_uri, + PROV.generatedAtTime, + Literal(e.timestamp, datatype=XSD.dateTime), + ) + ) + + if getattr(e, "agent_id", None) and e.agent_id != "unknown": + ag_uri = URIRef(EX[str(e.agent_id)]) + g.add((ag_uri, RDF.type, PROV.Agent)) + g.add((ent_uri, PROV.wasAttributedTo, ag_uri)) + + if getattr(e, "activity_id", None) and e.activity_id != "unknown": + act_uri = URIRef(EX[str(e.activity_id)]) + g.add((act_uri, RDF.type, PROV.Activity)) + g.add((ent_uri, PROV.wasGeneratedBy, act_uri)) + + for parent_id in getattr(e, "derived_from", []): + p_uri = URIRef(EX[str(parent_id)]) + g.add((ent_uri, PROV.wasDerivedFrom, p_uri)) + + rdf_format = "json-ld" if format == "jsonld" else format + return g.serialize(format=rdf_format) + + def check(self, strict: bool = False) -> Dict[str, Any]: + """ + Validate provenance integrity. + + Args: + strict: Whether to perform strict validation + + Returns: + Dictionary with validation results + """ + entries = self.storage.retrieve_all() + all_ids = {e.entity_id for e in entries} + + missing_refs = [] + for e in entries: + for p in getattr(e, "derived_from", []): + if p not in all_ids: + missing_refs.append(f"{e.entity_id} -> {p}") + + valid = len(missing_refs) == 0 if strict else True + + return { + "valid": valid, + "total_entries": len(entries), + "missing_references": missing_refs, + "strict": strict, + "errors": len(missing_refs) if strict else 0, + } + diff --git a/tests/provenance/test_manager.py b/tests/provenance/test_manager.py index 6d704d8f..872277e5 100644 --- a/tests/provenance/test_manager.py +++ b/tests/provenance/test_manager.py @@ -11,7 +11,89 @@ from semantica.provenance import ProvenanceManager, SourceReference class TestProvenanceManager: """Test ProvenanceManager functionality.""" - + + @pytest.fixture(autouse=True) + def reset_default_storage_path(self): + original = ProvenanceManager._default_storage_path + try: + yield + finally: + ProvenanceManager._default_storage_path = original + + def test_default_storage_path_pattern(self, tmp_path): + """Test global default storage pattern, config kwarg, and test-isolation context manager.""" + from semantica.provenance import default_storage_path, InMemoryStorage, SQLiteStorage + + # 1. Default should fallback to InMemoryStorage when no path/config + prov_mgr = ProvenanceManager() + assert isinstance(prov_mgr.storage, InMemoryStorage) + + # 2. Config kwarg should extract storage_path gracefully (CLI bug fix) + cfg_path = str(tmp_path / "cfg_test.db") + cfg = {"provenance": {"storage_path": cfg_path}} + prov_mgr = ProvenanceManager(config=cfg) + assert isinstance(prov_mgr.storage, SQLiteStorage) + + # 3. Explicit storage_path arg overrides config and default + explicit_path = str(tmp_path / "explicit.db") + prov_mgr = ProvenanceManager(storage_path=explicit_path, config=cfg) + assert isinstance(prov_mgr.storage, SQLiteStorage) + + # 4. default_storage_path context manager should temporarily set default and restore on exit + ctx_path = str(tmp_path / "ctx_test.db") + with default_storage_path(ctx_path): + assert ProvenanceManager._default_storage_path == ctx_path + prov_mgr = ProvenanceManager() + assert isinstance(prov_mgr.storage, SQLiteStorage) + + # Should be restored after context exit + assert ProvenanceManager._default_storage_path is None + prov_mgr = ProvenanceManager() + assert isinstance(prov_mgr.storage, InMemoryStorage) + + # 5. Guaranteed restoration even on exception + exc_path = str(tmp_path / "exception_test.db") + try: + with ProvenanceManager.default_storage_path(exc_path): + assert ProvenanceManager._default_storage_path == exc_path + raise RuntimeError("Test exception inside context") + except RuntimeError: + pass + assert ProvenanceManager._default_storage_path is None + + def test_isolation_part_1_with_context_manager(self, tmp_path): + """Part 1: Prove context manager sets default storage path for ProvenanceManager created inside context.""" + from semantica.provenance import default_storage_path, SQLiteStorage + iso_db = str(tmp_path / "iso_part1.db") + with default_storage_path(iso_db): + prov_mgr = ProvenanceManager() + assert isinstance(prov_mgr.storage, SQLiteStorage) + assert prov_mgr.storage.db_path == iso_db + + def test_isolation_part_2_no_leakage_after_context(self): + """Part 2: Prove that in a SEPARATE test function after context exit, ProvenanceManager falls back to InMemoryStorage without leakage.""" + from semantica.provenance import InMemoryStorage + assert ProvenanceManager._default_storage_path is None + prov_mgr = ProvenanceManager() + assert isinstance(prov_mgr.storage, InMemoryStorage) + + def test_orchestrator_config_wiring(self, tmp_path): + """Test Semantica orchestrator configures ProvenanceManager._default_storage_path.""" + from semantica.core import Semantica + from semantica.provenance import ProvenanceManager, InMemoryStorage, SQLiteStorage + + # 1. No path set in config means InMemoryStorage fallback + _ = Semantica() + prov_mgr = ProvenanceManager() + assert isinstance(prov_mgr.storage, InMemoryStorage) + + # 2. Config with provenance.storage_path sets default storage path + test_db = str(tmp_path / "orch_test.db") + _ = Semantica(config={"provenance": {"storage_path": test_db}}) + assert ProvenanceManager._default_storage_path == test_db + prov_mgr = ProvenanceManager() + assert isinstance(prov_mgr.storage, SQLiteStorage) + def test_initialization(self): """Test manager initialization.""" prov_mgr = ProvenanceManager() @@ -407,3 +489,42 @@ class TestProvenanceManager: assert e2.used_entities[0] in entity_ids, ( "Archived history entry should be reachable via used_entities in lineage" ) + + def test_cli_lineage(self): + """Test CLI lineage wrapper method on ProvenanceManager.""" + prov_mgr = ProvenanceManager() + prov_mgr.track_entity("e_cli", source="doc_cli") + res = prov_mgr.lineage("e_cli", depth=2) + assert res["entity_id"] == "e_cli" + assert res["depth"] == 2 + assert isinstance(res["lineage"], list) + assert isinstance(res["sources"], list) + + def test_cli_audit_log(self): + """Test CLI audit_log wrapper method on ProvenanceManager.""" + prov_mgr = ProvenanceManager() + prov_mgr.track_entity("e_audit", source="doc_audit") + res_table = prov_mgr.audit_log(format="table") + assert isinstance(res_table, str) + assert "ENTITY_ID" in res_table + res_csv = prov_mgr.audit_log(format="csv") + assert "entity_id,entity_type,activity_id,agent_id,timestamp" in res_csv + res_json = prov_mgr.audit_log(format="json") + assert isinstance(res_json, list) + + def test_cli_export_prov(self): + """Test CLI export_prov method on ProvenanceManager.""" + prov_mgr = ProvenanceManager() + prov_mgr.track_entity("e_rdf", source="doc_rdf") + ttl = prov_mgr.export_prov(format="turtle") + assert "prov:Entity" in ttl or "http://www.w3.org/ns/prov#Entity" in ttl + + def test_cli_check(self): + """Test CLI check integrity method on ProvenanceManager.""" + prov_mgr = ProvenanceManager() + prov_mgr.track_entity("e_valid", source="doc_1") + check_res = prov_mgr.check(strict=True) + assert check_res["valid"] is True + assert check_res["total_entries"] >= 1 + assert check_res["errors"] == 0 + From 6cc2e67b925996ab371f53ffc1a6e110f23f182a Mon Sep 17 00:00:00 2001 From: Sameer6305 Date: Sun, 26 Jul 2026 16:10:26 +0530 Subject: [PATCH 2/6] fix(provenance): populate entries alias in get_lineage and read lineage_chain in lineage() - Add entries alias in get_lineage() return dictionary so CLI and programmatic callers can access lineage entries via either key - Update lineage() wrapper method to fallback to lineage_chain when entries is missing - Add assertions in test_cli_lineage confirming lineage and entries lists are non-empty --- semantica/provenance/manager.py | 6 ++++-- tests/provenance/test_manager.py | 3 +++ 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/semantica/provenance/manager.py b/semantica/provenance/manager.py index 04414c02..2cd1cca7 100644 --- a/semantica/provenance/manager.py +++ b/semantica/provenance/manager.py @@ -557,9 +557,11 @@ class ProvenanceManager: if isinstance(meta, dict): aggregated_metadata.update(meta) + chain_dicts = [entry.to_dict() for entry in lineage_entries] return { "entity_id": entity_id, - "lineage_chain": [entry.to_dict() for entry in lineage_entries], + "lineage_chain": chain_dicts, + "entries": chain_dicts, "source_documents": list(set( e.source_document for e in lineage_entries if e.source_document @@ -675,7 +677,7 @@ class ProvenanceManager: Dict containing entity_id, depth, count, lineage entries, and sources """ base_lineage = self.get_lineage(entity_id) - entries = base_lineage.get("entries", []) + entries = base_lineage.get("lineage_chain") or base_lineage.get("entries", []) if len(entries) > depth: entries = entries[:depth] sources = self.get_all_sources(entity_id) diff --git a/tests/provenance/test_manager.py b/tests/provenance/test_manager.py index 872277e5..6206284f 100644 --- a/tests/provenance/test_manager.py +++ b/tests/provenance/test_manager.py @@ -498,6 +498,9 @@ class TestProvenanceManager: assert res["entity_id"] == "e_cli" assert res["depth"] == 2 assert isinstance(res["lineage"], list) + assert len(res["lineage"]) > 0 + assert len(res["entries"]) > 0 + assert res["lineage"][0]["entity_id"] == "e_cli" assert isinstance(res["sources"], list) def test_cli_audit_log(self): From fd84be8b66608a83cad72a059f5ba3b96bffd60a Mon Sep 17 00:00:00 2001 From: Sameer6305 Date: Sun, 26 Jul 2026 16:26:47 +0530 Subject: [PATCH 3/6] fixed qodo reviews --- semantica/cli.py | 4 ++++ semantica/provenance/manager.py | 28 ++++++++++++++++++++-------- tests/provenance/test_manager.py | 32 +++++++++++++++++++++++++++++--- 3 files changed, 53 insertions(+), 11 deletions(-) diff --git a/semantica/cli.py b/semantica/cli.py index c66d383f..2523e2e2 100644 --- a/semantica/cli.py +++ b/semantica/cli.py @@ -2608,6 +2608,10 @@ def provenance_check(cli_ctx: CLIContext, strict: bool, local_json: bool) -> Non _jecho(result if isinstance(result, dict) else {"valid": bool(result)}) else: _ok(cli_ctx, f"Provenance check: {result}") + if strict and isinstance(result, dict) and not result.get("valid", True): + raise click.ClickException( + f"Provenance integrity check failed: {result.get('errors')} error(s)" + ) _run_with_error_handling(_action) diff --git a/semantica/provenance/manager.py b/semantica/provenance/manager.py index 2cd1cca7..3ec7c6cb 100644 --- a/semantica/provenance/manager.py +++ b/semantica/provenance/manager.py @@ -235,7 +235,8 @@ class ProvenanceManager: metadata=metadata or {}, first_seen=existing.first_seen if existing else datetime.utcnow().isoformat(), last_updated=datetime.utcnow().isoformat(), - parent_entity_id=parent_id # Link to history or explicit parent + parent_entity_id=parent_id, # Link to history or explicit parent + used_entities=kwargs.get("used_entities", []), ) # Make the archived history entry discoverable via trace_lineage()'s @@ -776,10 +777,17 @@ class ProvenanceManager: g.add((act_uri, RDF.type, PROV.Activity)) g.add((ent_uri, PROV.wasGeneratedBy, act_uri)) - for parent_id in getattr(e, "derived_from", []): - p_uri = URIRef(EX[str(parent_id)]) + if getattr(e, "parent_entity_id", None): + p_uri = URIRef(EX[str(e.parent_entity_id)]) g.add((ent_uri, PROV.wasDerivedFrom, p_uri)) + for u_id in getattr(e, "used_entities", []): + u_uri = URIRef(EX[str(u_id)]) + g.add((ent_uri, PROV.wasDerivedFrom, u_uri)) + if getattr(e, "activity_id", None) and e.activity_id != "unknown": + act_uri = URIRef(EX[str(e.activity_id)]) + g.add((act_uri, PROV.used, u_uri)) + rdf_format = "json-ld" if format == "jsonld" else format return g.serialize(format=rdf_format) @@ -798,17 +806,21 @@ class ProvenanceManager: missing_refs = [] for e in entries: - for p in getattr(e, "derived_from", []): - if p not in all_ids: - missing_refs.append(f"{e.entity_id} -> {p}") + if getattr(e, "parent_entity_id", None): + if e.parent_entity_id not in all_ids: + missing_refs.append(f"{e.entity_id} -> {e.parent_entity_id}") + for u_id in getattr(e, "used_entities", []): + if u_id not in all_ids: + missing_refs.append(f"{e.entity_id} -> {u_id}") - valid = len(missing_refs) == 0 if strict else True + valid = len(missing_refs) == 0 + errors = len(missing_refs) return { "valid": valid, "total_entries": len(entries), "missing_references": missing_refs, "strict": strict, - "errors": len(missing_refs) if strict else 0, + "errors": errors, } diff --git a/tests/provenance/test_manager.py b/tests/provenance/test_manager.py index 6206284f..d74521d3 100644 --- a/tests/provenance/test_manager.py +++ b/tests/provenance/test_manager.py @@ -518,16 +518,42 @@ class TestProvenanceManager: def test_cli_export_prov(self): """Test CLI export_prov method on ProvenanceManager.""" prov_mgr = ProvenanceManager() - prov_mgr.track_entity("e_rdf", source="doc_rdf") + prov_mgr.track_entity("e_parent", source="doc_rdf") + prov_mgr.track_entity( + "e_child", + source="doc_rdf", + parent_entity_id="e_parent", + used_entities=["e_parent"], + activity_id="act_transform", + ) ttl = prov_mgr.export_prov(format="turtle") assert "prov:Entity" in ttl or "http://www.w3.org/ns/prov#Entity" in ttl + assert "wasDerivedFrom" in ttl + assert "used" in ttl def test_cli_check(self): """Test CLI check integrity method on ProvenanceManager.""" prov_mgr = ProvenanceManager() - prov_mgr.track_entity("e_valid", source="doc_1") + prov_mgr.track_entity("e_valid_parent", source="doc_1") + prov_mgr.track_entity( + "e_valid_child", + source="doc_1", + parent_entity_id="e_valid_parent", + used_entities=["e_valid_parent"], + ) check_res = prov_mgr.check(strict=True) assert check_res["valid"] is True - assert check_res["total_entries"] >= 1 + assert check_res["total_entries"] >= 2 assert check_res["errors"] == 0 + # Confirm non-strict mode still reports valid=False and errors > 0 when references are missing + prov_mgr.track_entity( + "e_broken_child", + source="doc_1", + parent_entity_id="non_existent_parent", + ) + check_broken = prov_mgr.check(strict=False) + assert check_broken["valid"] is False + assert check_broken["errors"] >= 1 + assert "e_broken_child -> non_existent_parent" in check_broken["missing_references"] + From a16bd9600bc01a65157fceee1209e2c774544c8d Mon Sep 17 00:00:00 2001 From: Sameer6305 Date: Sun, 26 Jul 2026 16:46:50 +0530 Subject: [PATCH 4/6] fixes reviews from qodo free for open source --- semantica/provenance/manager.py | 29 ++++++++++++---------- tests/provenance/test_manager.py | 41 ++++++++++++++++++++++++++++++++ 2 files changed, 58 insertions(+), 12 deletions(-) diff --git a/semantica/provenance/manager.py b/semantica/provenance/manager.py index 3ec7c6cb..c4f6434c 100644 --- a/semantica/provenance/manager.py +++ b/semantica/provenance/manager.py @@ -28,6 +28,7 @@ from typing import Optional, List, Dict, Any, Union from collections.abc import Mapping from datetime import datetime from contextlib import contextmanager +import threading from .schemas import ProvenanceEntry, SourceReference from .storage import ProvenanceStorage, InMemoryStorage, SQLiteStorage @@ -40,12 +41,8 @@ def default_storage_path(path: Optional[str]): Context manager for temporarily setting ProvenanceManager._default_storage_path. Guarantees restoration to the previous value on exit (safe for tests). """ - original_path = ProvenanceManager._default_storage_path - ProvenanceManager._default_storage_path = path - try: + with ProvenanceManager.default_storage_path(path): yield - finally: - ProvenanceManager._default_storage_path = original_path class ProvenanceManager: @@ -70,11 +67,14 @@ class ProvenanceManager: """ _default_storage_path: Optional[str] = None + _lock = threading.RLock() + _path_stack: List[Optional[str]] = [] @classmethod def set_default_storage_path(cls, path: Optional[str]) -> None: """Set a global default storage path for all new instances.""" - cls._default_storage_path = path + with cls._lock: + cls._default_storage_path = path @classmethod @contextmanager @@ -83,12 +83,16 @@ class ProvenanceManager: Context manager for temporarily setting the global default storage path. Guarantees restoration to the previous value on exit (safe for tests). """ - original_path = cls._default_storage_path - cls._default_storage_path = path + cls._lock.acquire() try: + cls._path_stack.append(cls._default_storage_path) + cls._default_storage_path = path yield finally: - cls._default_storage_path = original_path + cls._default_storage_path = ( + cls._path_stack.pop() if cls._path_stack else None + ) + cls._lock.release() def __init__( self, @@ -117,8 +121,9 @@ class ProvenanceManager: elif "storage_path" in config: storage_path = config.get("storage_path") - if not storage_path and self._default_storage_path: - storage_path = self._default_storage_path + if not storage_path: + with self._lock: + storage_path = self._default_storage_path if storage_path: self.storage = SQLiteStorage(storage_path) @@ -678,7 +683,7 @@ class ProvenanceManager: Dict containing entity_id, depth, count, lineage entries, and sources """ base_lineage = self.get_lineage(entity_id) - entries = base_lineage.get("lineage_chain") or base_lineage.get("entries", []) + entries = base_lineage.get("lineage_chain", []) if len(entries) > depth: entries = entries[:depth] sources = self.get_all_sources(entity_id) diff --git a/tests/provenance/test_manager.py b/tests/provenance/test_manager.py index d74521d3..157e8300 100644 --- a/tests/provenance/test_manager.py +++ b/tests/provenance/test_manager.py @@ -61,6 +61,47 @@ class TestProvenanceManager: pass assert ProvenanceManager._default_storage_path is None + # 6. Nested contexts should restore correctly from stack + outer_path = str(tmp_path / "outer.db") + inner_path = str(tmp_path / "inner.db") + with default_storage_path(outer_path): + assert ProvenanceManager._default_storage_path == outer_path + with default_storage_path(inner_path): + assert ProvenanceManager._default_storage_path == inner_path + assert ProvenanceManager._default_storage_path == outer_path + assert ProvenanceManager._default_storage_path is None + + def test_default_storage_path_concurrency_safety(self, tmp_path): + """Test that default_storage_path context manager is thread-safe under concurrent use.""" + import threading + import time + from semantica.provenance import default_storage_path, SQLiteStorage + + errors = [] + + def _worker(path, delay): + try: + with default_storage_path(path): + time.sleep(delay) + prov_mgr = ProvenanceManager() + assert isinstance(prov_mgr.storage, SQLiteStorage) + assert prov_mgr.storage.db_path == path + except Exception as e: + errors.append(e) + + path_a = str(tmp_path / "thread_a.db") + path_b = str(tmp_path / "thread_b.db") + + t1 = threading.Thread(target=_worker, args=(path_a, 0.05)) + t2 = threading.Thread(target=_worker, args=(path_b, 0.01)) + t1.start() + t2.start() + t1.join() + t2.join() + + assert not errors, f"Errors in concurrent context worker: {errors}" + assert ProvenanceManager._default_storage_path is None + def test_isolation_part_1_with_context_manager(self, tmp_path): """Part 1: Prove context manager sets default storage path for ProvenanceManager created inside context.""" from semantica.provenance import default_storage_path, SQLiteStorage From ac69c27b86a80d5b8e0854bd18627299b8f2319e Mon Sep 17 00:00:00 2001 From: Sameer6305 Date: Sun, 26 Jul 2026 16:56:06 +0530 Subject: [PATCH 5/6] fixes reviews from qodo free for open source --- semantica/provenance/manager.py | 26 +++++++++++++++----------- 1 file changed, 15 insertions(+), 11 deletions(-) diff --git a/semantica/provenance/manager.py b/semantica/provenance/manager.py index c4f6434c..1f99123b 100644 --- a/semantica/provenance/manager.py +++ b/semantica/provenance/manager.py @@ -28,6 +28,7 @@ from typing import Optional, List, Dict, Any, Union from collections.abc import Mapping from datetime import datetime from contextlib import contextmanager +import copy import threading from .schemas import ProvenanceEntry, SourceReference @@ -116,7 +117,11 @@ class ProvenanceManager: if not storage_path and config: if isinstance(config, Mapping) or isinstance(config, dict): prov_config = config.get("provenance", {}) - if (isinstance(prov_config, Mapping) or isinstance(prov_config, dict)) and "storage_path" in prov_config: + prov_has_path = ( + isinstance(prov_config, (Mapping, dict)) + and "storage_path" in prov_config + ) + if prov_has_path: storage_path = prov_config.get("storage_path") elif "storage_path" in config: storage_path = config.get("storage_path") @@ -164,12 +169,6 @@ class ProvenanceManager: if not isinstance(entity_id, str): raise TypeError(f"entity_id must be a string, got {type(entity_id).__name__}") - if not isinstance(entity_id, str): - raise TypeError(f"entity_id must be a string, got {type(entity_id).__name__}") - - if not isinstance(entity_id, str): - raise TypeError(f"entity_id must be a string, got {type(entity_id).__name__}") - # Check if entity already exists existing = self.storage.retrieve(entity_id) parent_id = kwargs.get("parent_entity_id") @@ -205,7 +204,6 @@ class ProvenanceManager: if existing: # Create a history entry for the previous state # Use timestamp or counter for uniqueness - import copy history_entry = copy.deepcopy(existing) history_id = f"{entity_id}:v:{existing.last_updated}" @@ -724,8 +722,9 @@ class ProvenanceManager: elif format == "csv": lines = ["entity_id,entity_type,activity_id,agent_id,timestamp"] for e in entries: + ts = getattr(e, "timestamp", "") lines.append( - f"{e.entity_id},{e.entity_type},{e.activity_id},{e.agent_id},{getattr(e, 'timestamp', '')}" + f"{e.entity_id},{e.entity_type},{e.activity_id},{e.agent_id},{ts}" ) return "\n".join(lines) else: @@ -734,8 +733,10 @@ class ProvenanceManager: ] lines.append("-" * 75) for e in entries: + ts = str(getattr(e, "timestamp", "")) lines.append( - f"{str(e.entity_id):<20} {str(e.entity_type):<15} {str(e.activity_id):<15} {str(getattr(e, 'timestamp', '')):<25}" + f"{str(e.entity_id):<20} {str(e.entity_type):<15}" + f" {str(e.activity_id):<15} {ts:<25}" ) return "\n".join(lines) @@ -788,7 +789,10 @@ class ProvenanceManager: for u_id in getattr(e, "used_entities", []): u_uri = URIRef(EX[str(u_id)]) - g.add((ent_uri, PROV.wasDerivedFrom, u_uri)) + # Emit wasDerivedFrom only when this used entity is not the same + # as parent_entity_id — which already carries that triple above. + if u_id != getattr(e, "parent_entity_id", None): + g.add((ent_uri, PROV.wasDerivedFrom, u_uri)) if getattr(e, "activity_id", None) and e.activity_id != "unknown": act_uri = URIRef(EX[str(e.activity_id)]) g.add((act_uri, PROV.used, u_uri)) From 7045d7b94e743546b66602e4336b9c79b0456edf Mon Sep 17 00:00:00 2001 From: KaifAhmad1 Date: Mon, 27 Jul 2026 14:19:51 +0530 Subject: [PATCH 6/6] fix(provenance): address review nits and add CHANGELOG entry - track_entity() no longer aliases a caller-supplied used_entities list (it stored the reference directly and later mutated it via .append()) - Remove dead fallback branches in orchestrator.py/manager.py that duplicated what Config.get()'s dotted-path resolution already does - Add local --dry-run to `provenance audit` for parity with `provenance export` - `provenance check --strict` now warns instead of printing a success checkmark before raising on a failed check --- CHANGELOG.md | 8 ++++++++ semantica/cli.py | 13 +++++++++---- semantica/core/orchestrator.py | 5 +++-- semantica/provenance/manager.py | 6 +++--- 4 files changed, 23 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1a7464b7..2a25095a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- **Global default persistent storage for `ProvenanceManager`, plus a working `provenance` CLI** (#795, #802) by @Sameer6305 and @KaifAhmad1 + - Every ingestion/processing module (`kg_provenance.py`, `pipeline_provenance.py`, and 20+ other call sites) instantiated its own `ProvenanceManager()` with no `storage_path`, so all of them silently fell back to `InMemoryStorage` and the SQLite audit trail was never actually written. `ProvenanceManager.set_default_storage_path(path)` now sets a class-level default that every no-arg instantiation picks up, and `Semantica.__init__` wires `config.provenance.storage_path` into it automatically during orchestrator init + - Added the thread-safe `default_storage_path(path)` context manager (`semantica.provenance.default_storage_path`) for test isolation — it stacks nested overrides and guarantees restoration of the previous default on exit, even on exception, so tests can't leak global state into each other + - Fixed `ProvenanceManager.__init__` raising `TypeError` on the CLI's `config=` kwarg, and implemented the four methods the CLI already called but that didn't exist on the class: `lineage()`, `audit_log()`, `export_prov()` (W3C PROV-O turtle/ntriples/jsonld via `rdflib`), and `check()` — unblocking `semantica provenance lineage|audit|export|check` end-to-end + - Follow-up review fixes: `track_entity` no longer aliases a caller-supplied `used_entities` list (it copied the reference and later mutated it in place via `.append()`, which could corrupt a list the caller still held); removed dead fallback branches in `orchestrator.py`/`manager.py` left over from not realizing `Config.get()` already resolves dotted paths; added a `--dry-run` option to `provenance audit` to match `provenance export` (previously only the global `--dry-run` flag worked, not a local one); and `provenance check --strict` no longer prints a green "✓" success line immediately before failing — a failing check now renders as a warning before the `ClickException` is raised + ### Fixed - **`react-hooks/set-state-in-effect` cascading renders across 12 Explorer workspace files** (#769, #796) by @Sameer6305 and @KaifAhmad1 diff --git a/semantica/cli.py b/semantica/cli.py index 2523e2e2..56d8e9a0 100644 --- a/semantica/cli.py +++ b/semantica/cli.py @@ -2523,9 +2523,11 @@ def provenance_lineage(cli_ctx: CLIContext, entity_id: str, depth: int, local_js default="table", show_default=True) @click.option("--output", default=None, type=click.Path()) @click.option("--json", "local_json", is_flag=True, default=False) +@click.option("--dry-run", "local_dry", is_flag=True, default=False) @click.pass_obj def provenance_audit(cli_ctx: CLIContext, since: Optional[str], fmt: str, - output: Optional[str], local_json: bool) -> None: + output: Optional[str], local_json: bool, + local_dry: bool) -> None: """Export the audit log. \b @@ -2535,7 +2537,7 @@ def provenance_audit(cli_ctx: CLIContext, since: Optional[str], fmt: str, cli_ctx = _require_ctx(cli_ctx) def _action() -> None: - if _is_dry(cli_ctx, False): + if _is_dry(cli_ctx, local_dry): _dry(cli_ctx, "export audit log", since=since, format=fmt, output=output) return try: @@ -2604,11 +2606,14 @@ def provenance_check(cli_ctx: CLIContext, strict: bool, local_json: bool) -> Non result = pm.check(strict=strict) except ImportError as exc: raise click.ClickException(f"Provenance module not available: {exc}") from exc + is_valid = not isinstance(result, dict) or result.get("valid", True) if _is_json(cli_ctx, local_json): _jecho(result if isinstance(result, dict) else {"valid": bool(result)}) - else: + elif is_valid: _ok(cli_ctx, f"Provenance check: {result}") - if strict and isinstance(result, dict) and not result.get("valid", True): + else: + _warn(cli_ctx, f"Provenance check: {result}") + if strict and not is_valid: raise click.ClickException( f"Provenance integrity check failed: {result.get('errors')} error(s)" ) diff --git a/semantica/core/orchestrator.py b/semantica/core/orchestrator.py index f0c8cfb2..2d1ead68 100644 --- a/semantica/core/orchestrator.py +++ b/semantica/core/orchestrator.py @@ -92,9 +92,10 @@ class Semantica: # Configure global provenance storage path if specified try: + # Config.get() already resolves dotted paths against nested dicts + # (see get_nested_value), so this single lookup covers both + # top-level and config.provenance={"storage_path": ...} shapes. prov_storage_path = self.config.get("provenance.storage_path") - if not prov_storage_path and isinstance(self.config.get("provenance"), dict): - prov_storage_path = self.config.get("provenance", {}).get("storage_path") if prov_storage_path: from ..provenance import ProvenanceManager ProvenanceManager.set_default_storage_path(prov_storage_path) diff --git a/semantica/provenance/manager.py b/semantica/provenance/manager.py index 1f99123b..ab0abb26 100644 --- a/semantica/provenance/manager.py +++ b/semantica/provenance/manager.py @@ -115,10 +115,10 @@ class ProvenanceManager: return if not storage_path and config: - if isinstance(config, Mapping) or isinstance(config, dict): + if isinstance(config, Mapping): prov_config = config.get("provenance", {}) prov_has_path = ( - isinstance(prov_config, (Mapping, dict)) + isinstance(prov_config, Mapping) and "storage_path" in prov_config ) if prov_has_path: @@ -239,7 +239,7 @@ class ProvenanceManager: first_seen=existing.first_seen if existing else datetime.utcnow().isoformat(), last_updated=datetime.utcnow().isoformat(), parent_entity_id=parent_id, # Link to history or explicit parent - used_entities=kwargs.get("used_entities", []), + used_entities=list(kwargs.get("used_entities", [])), ) # Make the archived history entry discoverable via trace_lineage()'s