Resilient Flow-Traceback

The problem

Complex systems can get overwhelmed or drained by expensive tasks that waste resources. It is difficult to stop these specific high-cost paths without crashing the entire system.

What it does

It tracks the cost of different inputs and automatically shuts down specific flows that become too expensive. This prevents a single costly task from draining the system's resources.

Why it matters

It allows a system to stay stable by cutting off expensive tasks before they cause a total failure.

Validation

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

$ python3 flow_traceback.py
Traceback (most recent call last):
  File "/work/flow_traceback.py", line 54, in <module>
    @resilient_traceback(cb)
     ^^^^^^^^^^^^^^^^^^^^^^^
TypeError: 'NoneType' object is not callable

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

# Resilient Flow-Traceback Implementation
import time
from functools import wraps
from collections import defaultdict

class CircuitBreaker:
    def __init__(self, cost_threshold, max_attempts=3, tracker=None):
        self.cost_threshold = cost_threshold
        self.max_attempts = max_attempts
        self.attempts = 0
        self.tracker = tracker if tracker else CostTracker()
        self.halted = False

    def check(self, path, cost):
        self.tracker.add_cost(path, cost)
        current_cost = self.tracker.get_cost(path)
        if current_cost > self.cost_threshold:
            self.attempts += 1
            if self.attempts >= self.max_attempts:
                self.halted = True
            return False
        self.attempts = 0
        return True

class CostTracker:
    def __init__(self):
        self.costs = defaultdict(float)

    def add_cost(self, path, cost):
        self.costs[path] += cost

    def get_cost(self, path):
        return self.costs[path]

def resilient_traceback(circuit_breaker):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            path = func.__name__
            start_time = time.time()
            result = func(*args, **kwargs)
            execution_time = time.time() - start_time
            cost = execution_time * 0.1  # Cost factor example

            if not circuit_breaker.check(path, cost):
                raise RuntimeError(f"Path '{path}' exceeded cost threshold")
            return result
        return wrapper

if __name__ == "__main__":
    tracker = CostTracker()
    cb = CircuitBreaker(cost_threshold=1.0, max_attempts=2, tracker=tracker)

    @resilient_traceback(cb)
    def process_input(data):
        time.sleep(0.5)  # Simulate processing time
        print(f"Processed: {data}")
        return len(data)

    # Test simulation
    for i in range(5):
        try:
            process_input(f"Request {i}")
        except RuntimeError as e:
            print(e)
        time.sleep(1)