Causal Path Traceability

The problem

When a system fails or produces an error, it is often difficult to distinguish between unrelated data points and the actual cause. This makes finding the root source of a problem tedious and confusing.

What it does

It traces a result back through a chain of dependencies to pinpoint the exact input variable responsible for a failure. It filters out irrelevant noise to show only the direct path of cause and effect.

Why it matters

It allows you to identify the specific source of a problem without sifting through unrelated data.

Validation

It was run in the sandbox and it failed. run output shows an error/traceback — the artifact does NOT run clean.

$ python3 causal_path_traceability.py
Traceback (most recent call last):
  File "/work/causal_path_traceability.py", line 69, in <module>
    dependency_path = engine.trace_path(violation_state)
                      ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/work/causal_path_traceability.py", line 26, in trace_path
    if not self.dependency_graph[current]:
           ~~~~~~~~~~~~~~~~~~~~~^^^^^^^^^
KeyError: 'processed_data'

No screenshot — there is nothing working to show. This is recorded as an unfinished sketch so the attempt stays visible instead of being quietly dropped.

The code

All of it — 72 lines, one file, standard library only.

import json
from typing import Dict, List, Any

class StatePathTraceabilityEngine:
    def __init__(self, states: Dict[str, Any], transitions: List[Dict[str, Any]]):
        self.states = states
        self.transitions = transitions
        self.dependency_graph = self._build_dependency_graph()

    def _build_dependency_graph(self) -> Dict[str, List[str]]:
        """Build a graph mapping each state to its dependencies"""
        graph = {state: [] for state in self.states}
        for transition in self.transitions:
            for input_var in transition.get('inputs', []):
                graph[transition['target']].append(input_var)
        return graph

    def trace_path(self, violation_state: str) -> List[str]:
        """Trace backward from a violation state to find potential causes"""
        current = violation_state
        path = []
        while current in self.dependency_graph:
            for dependency in self.dependency_graph[current]:
                path.append(dependency)
                current = dependency
            if not self.dependency_graph[current]:
                break
        return path

class CausalDiscovery:
    @staticmethod
    def filter_non_causal(dependency_path: List[str], data: List[Dict[str, Any]]) -> List[str]:
        """Filter out variables that don't have causal relationship"""
        # Simplified causal check: temporal precedence and value change correlation
        causal_vars = []
        for var in dependency_path:
            # Check if variable changes before state violation
            for entry in data:
                if entry.get(var) and entry.get('timestamp') < entry.get('violation_time'):
                    causal_vars.append(var)
                    break
        return causal_vars

# Example usage
if __name__ == "__main__":
    # Sample system description
    states = {
        'initial': {},
        'processing': {},
        'violation': {'error': 'Invalid state'}
    }

    transitions = [
        {'source': 'initial', 'target': 'processing', 'inputs': ['user_input']},
        {'source': 'processing', 'target': 'violation', 'inputs': ['processed_data']}
    ]

    # Sample execution data
    execution_data = [
        {'timestamp': 1, 'user_input': 'malicious'},
        {'timestamp': 2, 'processed_data': 'corrupted'},
        {'timestamp': 3, 'violation_time': 3}
    ]

    engine = StatePathTraceabilityEngine(states, transitions)
    causal_discovery = CausalDiscovery()

    violation_state = 'violation'
    dependency_path = engine.trace_path(violation_state)
    causal_vars = causal_discovery.filter_non_causal(dependency_path, execution_data)

    print(f"Causal variables leading to violation: {causal_vars}")