Resilient Path Traceability

The problem

It is difficult to pinpoint exactly where a process is failing when multiple steps are connected together. Standard logs often show that an error occurred without showing the specific path of events that led to it.

What it does

It tracks the sequence of actions and assigns a score to errors based on where they happen in that sequence. It identifies which specific path is most prone to breaking.

Why it matters

It helps identify the root cause of failures by showing how deep in a process a problem actually occurs.

Validation

It was run inside an isolated container with no network access. This is the exact command and the real output it produced — captured process output, not written by a model.

$ python3 resilient_path_traceability.py
File "/work/resilient_path_traceability.py", line 34
    failure_weight = len(self.call_stack) + 1
IndentationError: unexpected indent

A screenshot of that run.

A clean run proves this does what is shown above, in a CPU-only sandbox. It is a small research demo — not a production tool, and nothing here was published anywhere.

The code

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

import inspect
class CircuitBreaker:
    def calculate_path_failure_score(self):
        return sum(self.path_scores.values())
    
    def __init__(self, threshold=5):
        self.failure_score = 0
        self.threshold = threshold
        self.is_open = False
        self.call_stack = []
        self.path_scores = {}
    
    def __call__(self, func):
        def wrapper(*args, **kwargs):
            if self.is_open:
                raise RuntimeError("Circuit breaker is open")
            
            self.call_stack.append(func.__name__)
            try:
                result = func(*args, **kwargs)
                self.call_stack.pop()
                self.failure_score = max(0, self.failure_score - 1)
                return result
            except Exception as e:
                self._handle_failure(e)
                raise
        return wrapper
    
    def _handle_failure(self, error):
        path = '.'.join(self.call_stack)
        self.path_scores[path] = self.path_scores.get(path, 0) + len(self.call_stack) + 1
        failure_weight = len(self.call_stack) + 1
        self.failure_score = sum(self.path_scores.values())
        
        if self.failure_score >= self.threshold:
            self.is_open = True
    
    def reset(self):
        self.failure_score = 0
        self.is_open = False
        self.path_scores = {}
    
    def generate_path_traceability_report(self):
        if not self.path_scores:
            return "No failure paths recorded."
        
        max_score = max(self.path_scores.values())
        max_paths = [path for path, score in self.path_scores.items() if score == max_score]
        most_impactful_path = max_paths[0]
        
        return (
            f"Path Traceability Report:\n"
            f"- Most impactful failure path: {most_impactful_path}\n"
            f"- Cumulative impact score: {max_score}\n"
            f"- Total failed paths: {len(self.path_scores)}"
        )
    

# Example Usage
if __name__ == "__main__":
    cb = CircuitBreaker(threshold=10)
    
    @cb
    def level1():
        @cb
        def level2():
            @cb
            def level3():
                raise RuntimeError("Critical failure at deepest level")
            
            try:
                level3()
            except:
                pass
        
        try:
            level2()
        except:
            pass
    
    try:
        level1()
    except:
        pass
    
    print(f"Path-aware failure score: {cb.calculate_path_failure_score()}")
    print(f"Circuit state: {'Open' if cb.is_open else 'Closed'}")
    print(cb.generate_path_traceability_report())