capsule_root_node

Author: Source-attributed; see payload

import json
from provenance_graph import ProvenanceGraphEngine

def execute_provenance_smoke_test() -> dict:
    """
    Executes provenance testing and returns a pure structured dictionary 
    suitable for direct stdout JSON serialization.
    """
    graph = ProvenanceGraphEngine()

    # 1. Ingest Valid Origin Route Receipt Node
    mock_origin_receipt = {
        "receipt_id": "rcpt_alpha_001",
        "capsule_id": "capsule_root_node",
        "status": "EMITTED",
        "guardian_status": "Validated",
        "envelope_snapshot": {
            "id": "capsule_root_node",
            "type": "ActionCapsule",
            "anchor": "∴Ω⧂",
            "guardian_status": "Validated",
            "payload": {"data": "System Prime Initializer"}
        }
    }
    graph.ingest_route_receipt(mock_origin_receipt)
    root_trace = graph.trace_lineage_backward("capsule_root_node")

    # 2. Ingest Child Capsule with parent linkage
    mock_child_payload = {
        "id": "capsule_child_node",
        "type": "ActionCapsule",
        "anchor": "∴Ω⧂",
        "guardian_status": "Validated",
        "parent_capsule_id": "capsule_root_node",
        "payload": {"data": "Downstream branch task"}
    }
    graph.ingest_capsule(mock_child_payload, dispatch_status="DISTRIBUTED")

    # 3. Force-inject an anomalous Drift state to verify calculations
    mock_drift_payload = {
        "id": "capsule_anomaly_node",
        "type": "ActionCapsule",
        "anchor": "∴Ω⧂",
        "guardian_status": "Drift",
        "parent_capsule_id": "capsule_child_node"
    }
    graph.ingest_capsule(mock_drift_payload, dispatch_status="PARTIALLY_RESTRICTED")
    
    # 4. Extract Lineage and Metrics
    lineage_trail = graph.trace_lineage_backward("capsule_anomaly_node")
    drift_ratio = graph.get_systemic_drift_ratio()

    # Returns raw data structure directly; formatting happens at script entry point
    return {
        "test": "provenance_graph",
        "origin_is_origin": root_trace[0]["is_origin"],
        "lineage_trail": lineage_trail,
        "drift_ratio": drift_ratio,
        "node_count": len(graph.nodes),
        "adjacency_list": graph.adjacency_list
    }

if __name__ == "__main__":
    # Canonical clean capture hook
    print(json.dumps(execute_provenance_smoke_test(), indent=2, ensure_ascii=False))

Exact page as JSON · Follow lineage as JSON

Original source: source:62d17dd05f801d32c3bf9ecb7e44675659f86ed5a81c2742ad15cb1c35196081