Complex software systems can get stuck in messy loops or nested errors that spiral out of control. It is difficult for these systems to automatically recognize when a path is becoming too broken to continue.
It monitors errors and uses logic formulas to decide if a process is becoming too messy. It automatically shuts down those specific paths before they cause further issues.
It provides a way to stop software errors from spiraling by identifying and cutting off problematic paths automatically.
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 symbolic_circuit_breaker.py Success: Success Success: Success Success: Success Success: Success Success: Success Success: Success Success: Success Success: Success Success: Success Success: Success
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.
All of it — 81 lines, one file, standard library only.
import time
from datetime import datetime, timedelta
class SymbolicCircuitBreaker:
ERROR_WEIGHTS = {
'RecursionDepthError': 3,
'NetworkTimeoutError': 2,
'Default': 1
}
def __init__(self, max_score=10, timeout=30, error_weights=None):
self.max_score = max_score
self.timeout = timeout
self.score = 0
self.last_failure_time = None
self.state = 'CLOSED'
self.error_weights = error_weights or self.ERROR_WEIGHTS.copy()
def is_viable(self):
if self.state == 'OPEN':
if datetime.now() - self.last_failure_time > timedelta(seconds=self.timeout):
self.state = 'HALF_OPEN'
return False
return True
def check(self):
if self.state == 'HALF_OPEN':
self.state = 'CLOSED'
self.score = 0
return True
if self.score >= self.max_score:
self.state = 'OPEN'
self.last_failure_time = datetime.now()
return False
return True
def record_failure(self, error_type='Default', error_weight=1):
weight = self.error_weights.get(error_type, self.error_weights['Default'])
self.score += weight * error_weight
self.last_failure_time = datetime.now()
if self.score >= self.max_score:
self.state = 'OPEN'
self.last_failure_time = datetime.now()
def execute(self, func, *args, **kwargs):
if not self.check():
raise Exception("Circuit breaker open")
try:
result = func(*args, **kwargs)
if self.state == 'CLOSED':
self.score = 0
return result
except Exception as e:
self.record_failure(error_type=type(e).__name__)
raise e
# Example usage
if __name__ == "__main__":
class RecursionDepthError(Exception):
pass
class NetworkTimeoutError(Exception):
pass
cb = SymbolicCircuitBreaker(max_score=10, timeout=10)
for i in range(15):
try:
if i % 3 == 0:
raise RecursionDepthError("Recursion too deep")
elif i % 3 == 1:
raise NetworkTimeoutError("Network timeout")
else:
print('Success:', cb.execute(lambda: 'Success'))
except Exception as e:
print(f'Iteration {i}: Error ({type(e).__name__}) - Score: {cb.score}')
time.sleep(1)