How ProvenanceManager Tracks W3C PROV-O Lineage Across Entities in Semantica

The ProvenanceManager tracks W3C PROV-O lineage by persisting structured ProvenanceEntry records, traversing parent-child relationships via breadth-first search, and exporting the graph to standards-compliant RDF.

The semantica-agi/semantica repository implements a comprehensive provenance tracking system through the ProvenanceManager class. This component enables W3C PROV-O lineage tracking across entities by mapping internal records to standard provenance concepts and providing bidirectional graph traversal capabilities.

Storing PROV-O Compliant Provenance Records

Every provenance event is persisted as a ProvenanceEntry defined in semantica/provenance/schemas.py. This dataclass maps directly to W3C PROV-O concepts:

• prov:Entity maps to entity_id and entity_type
• prov:Activity maps to activity_id, activity_started_at_time, and activity_ended_at_time
• prov:Agent maps to agent_id, agent_type, is_automated, and role
• prov:generatedAtTime maps to timestamp, first_seen, and last_updated
• prov:wasDerivedFrom and prov:wasAttributedTo map to parent_entity_id, derived_from_id, source_document, and source_location

When you invoke track_entity(), track_relationship(), or track_chunk(), the manager instantiates a ProvenanceEntry and persists it through the configured storage backend. During persistence, the manager computes a hash-chain checksum using compute_checksum() and stores previous_checksum and sequence_id values to enable cryptographic integrity verification.

Querying Upstream Lineage with get_lineage()

The get_lineage(entity_id) method in semantica/provenance/manager.py serves as the primary API for retrieving full upstream ancestry. The implementation delegates graph traversal to the storage layer while performing post-processing on the results:

def get_lineage(self, entity_id: str) -> Dict[str, Any]:
    lineage_entries = self.storage.trace_lineage(entity_id)
    # ... aggregation and verification logic

The storage layer's trace_lineage() method performs a breadth-first search over parent_entity_id and used_entities links to assemble the complete ancestry chain. After retrieval, get_lineage() aggregates metadata from all ancestors (with later entries taking precedence in key collisions) and verifies the hash chain for each entry using verify_checksum().

The returned dictionary contains the ordered lineage_chain, distinct source_documents, timestamps, aggregated metadata, and an integrity_verified boolean flag.

Tracing Downstream Descendants

For reverse provenance queries, ProvenanceManager provides the trace_descendants() method. This walks the graph in the opposite direction, following used_entities and relationship edges to identify all entities derived from or consuming a specific source entity:

def trace_descendants(self, entity_id: str, max_depth: Optional[int] = None) -> List[ProvenanceEntry]:
    return self.storage.trace_descendants(entity_id, max_depth=max_depth)

This downstream tracing enables impact analysis and audit workflows by revealing the full scope of entities influenced by a particular data source or transformation.

Exporting to W3C PROV-O RDF Standards

The export_prov() method serializes the entire provenance graph to a PROV-O RDF document using the rdflib library:

def export_prov(self, format: str = "turtle", base_uri: Optional[str] = None) -> str:
    from rdflib import Graph, Namespace, URIRef, Literal, RDF, XSD
    PROV = Namespace("http://www.w3.org/ns/prov#")
    # ... RDF generation logic

Each ProvenanceEntry becomes a prov:Entity node with associated prov:Activity and prov:Agent triples. Timestamps are emitted as prov:generatedAtTime literals with ISO-8601 formatting. The method supports Turtle, N-Triples, and JSON-LD output formats, ensuring compatibility with external RDF stores and PROV visualization tools.

Storage Layer Architecture

The actual graph traversal logic resides in semantica/provenance/storage.py. Both trace_lineage() and trace_descendants() utilize queue-based BFS algorithms with optional max_depth parameters. The storage abstraction supports multiple backends:

• SQLiteStorage: Persistent disk-based storage for production workloads
• InMemoryStorage: Volatile storage for testing and ephemeral operations

Because the ProvenanceManager depends only on the abstract ProvenanceStorage interface, you can swap implementations without modifying lineage query logic.

Practical Code Examples

Tracking Entities and Their Relationships

from semantica.provenance import ProvenanceManager

# Initialize with default in-memory storage

prov_mgr = ProvenanceManager()

# Track a base entity

prov_mgr.track_entity(
    entity_id="e1",
    source="docA.pdf",
    metadata={"label": "Apple Inc.", "type": "ORG"},
    activity_id="ingest",
    agent_id="pipeline",
)

# Track a derived entity with explicit parent reference

prov_mgr.track_entity(
    entity_id="e2",
    source="docB.pdf",
    metadata={"label": "Apple", "type": "ORG"},
    derived_from_id="e1",  # establishes lineage link

    activity_id="extraction",
    agent_id="nlp_service",
)

Retrieving Complete Lineage

lineage = prov_mgr.get_lineage("e2")
print(f"Sources: {lineage['source_documents']}")
print(f"Metadata: {lineage['metadata']}")
print(f"Entities in chain: {lineage['entity_count']}")

Exporting PROV-O RDF

turtle_output = prov_mgr.export_prov(format="turtle")

# Save to file or load directly into RDF stores

with open("provenance.ttl", "w") as f:
    f.write(turtle_output)

Querying Downstream Impact

descendants = prov_mgr.trace_descendants("e1", max_depth=5)
print(f"Total descendants: {len(descendants)}")

Summary

• W3C PROV-O Compliance: The ProvenanceEntry schema in semantica/provenance/schemas.py maps directly to PROV-O concepts including entities, activities, agents, and derivation relationships.
• Cryptographic Integrity: Every entry includes hash-chain checksums (previous_checksum, sequence_id) that get_lineage() verifies during retrieval.
• Bidirectional Traversal: The storage layer implements both upstream lineage tracing (trace_lineage) and downstream descendant tracking (trace_descendants) using BFS algorithms.
• Standards-Based Export: The export_prov() method generates valid PROV-O RDF serializations compatible with Turtle, N-Triples, and JSON-LD formats.
• Pluggable Storage: Abstract storage interface allows seamless switching between SQLite and in-memory backends without affecting lineage logic.

Frequently Asked Questions

What is W3C PROV-O and why does it matter for lineage tracking?

W3C PROV-O is the OWL2 Web Ontology Language encoding of the PROV Data Model, a standard for representing provenance information on the web. By aligning ProvenanceEntry fields with PROV-O concepts like prov:Entity and prov:wasDerivedFrom, Semantica ensures interoperability with external provenance tools, RDF databases, and compliance frameworks that expect standards-compliant traceability.

How does ProvenanceManager ensure lineage integrity?

The manager computes a cryptographic checksum for each ProvenanceEntry using compute_checksum() and stores it alongside previous_checksum and sequence_id values. When retrieving lineage via get_lineage(), the system invokes verify_checksum() on every entry in the chain. This hash-linking mechanism detects tampering or corruption by ensuring each entry cryptographically references its predecessor.

Can I limit the depth of lineage queries?

Yes. Both trace_lineage() and trace_descendants() in the storage layer accept an optional max_depth parameter. When specified, the breadth-first search terminates after traversing the specified number of generations, preventing excessive resource consumption when dealing with deeply nested derivation chains.

Which storage backends support lineage tracking?

The lineage system works with any implementation of the abstract ProvenanceStorage class defined in semantica/provenance/storage.py. The repository provides SQLiteStorage for persistent production use and InMemoryStorage for testing and development. Custom backends can implement trace_lineage() and trace_descendants() to support distributed or cloud-native storage solutions while maintaining the same ProvenanceManager API.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}