Circuit-Breaker Traceability

The problem

Complex software systems can spiral into endless loops of errors when a single failure triggers a chain reaction of nested problems. It becomes difficult to stop these failures before they overwhelm the entire system.

What it does

It monitors the path of errors and automatically shuts down a process if a specific sequence of failures occurs. It acts like a safety switch that cuts the power when it detects a dangerous pattern.

Why it matters

It prevents a single error from cascading into a larger system failure by stopping the process at the right moment.

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 circuit_breaker_traceability_v2.py
File "/work/circuit_breaker.py", line 56
    print(f"Function failed: {e")
                               ^
SyntaxError: f-string: expecting '}'

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 — 77 lines, one file, standard library only.

# Modified CircuitBreaker with Failure Score metric
import time
from functools import wraps

class CircuitBreakerTripped(Exception): pass

class CircuitBreaker:
    def __init__(self, threshold=3, reset_timeout=10):
        self.threshold = threshold
        self.reset_timeout = reset_timeout
        self.tripped_until = 0
        self.trace = []
        self.current_path = []
        self.failure_scores = []  # New attribute for failure scoring

    def __call__(self, func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            if time.time() < self.tripped_until:
                raise CircuitBreakerTripped("Circuit tripped due to excessive nested failures")

            self.current_path.append(func.__name__)
            try:
                result = func(*args, **kwargs)
                return result
            except Exception as e:
                self.trace.append((func.__name__, str(e)))
                
                # Calculate failure score based on recursion depth
                depth = len(self.current_path)
                score = 2 ** depth  # Exponential weighting by recursion depth
                self.failure_scores.append({
                    'func': func.__name__,
                    'error': str(e),
                    'depth': depth,
                    'score': score
                })

                if len(self.current_path) >= self.threshold:
                    self.tripped_until = time.time() + self.reset_timeout
                    raise CircuitBreakerTripped(
                        f"Circuit tripped due to nested failures in path: {self.current_path[-self.threshold:]}")
                raise
            finally:
                if self.current_path and self.current_path[-1] == func.__name__:
                    self.current_path.pop()
        return wrapper

# Example usage
if __name__ == "__main__":
    cb = CircuitBreaker(threshold=2)
    
    @cb
    def function_a():
        print("Executing function_a")
        raise Exception("Failure in A")

    @cb
    def function_b():
        print("Executing function_b")
        function_a()  # This will trigger nested failure
        raise Exception("Failure in B")

    try:
        function_b()
    except CircuitBreakerTripped as e:
        print(f"Circuit tripped: {e}")
    except Exception as e:
        print(f"Function failed: {e}")
    finally:
        print("\nTrace log:")
        for func, error in cb.trace:
            print(f"{func}: {error}")
        
        print("\nFailure Scores (depth-weighted):")
        for entry in cb.failure_scores:
            print(f"{entry['func']} (Depth: {entry['depth']}, Score: {entry['score']}): {entry['error']}")