Contextual State Circuit Breaker

The problem

Systems often keep trying to perform a task even when it is failing, wasting resources and causing repeated errors. Standard tools can't always tell if a failure is a one-time glitch or a persistent problem.

What it does

It monitors how a system is performing and automatically stops a process if it detects a pattern of repeated failures. It tracks the history of these errors to decide when to cut off the process.

Why it matters

It prevents a system from repeatedly attempting a broken task by understanding the context of the failures.

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.py
Traceback (most recent call last):
  File "/work/contextual_circuit_breaker.py", line 70, in <module>
    wrapped_operation = cb(flaky_operation)
                        ^^^^^^^^^^^^^^^^^^^
TypeError: CircuitBreaker.__call__() takes 1 positional argument but 2 were given

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

import sys
import time
from poly_blemish_breaker import ContextualCircuitBreaker, CircuitBreakerError


def test_failure_in_one_group_does_not_trip_another():
    cb = ContextualCircuitBreaker(threshold=3, cooldown=600)
    cb.reset()

    def failer(name):
        raise RuntimeError(f"{name} failure")

    def succeeder(name):
        return f"{name} OK"

    # Trip group alpha by forcing 3 failures
    for _ in range(3):
        try:
            cb.execute(failer, "alpha", context="alpha")
        except RuntimeError:
            pass

    alpha_stream = cb.stream_state("alpha")
    assert alpha_stream["state"] == "OPEN", (
        f"alpha should be OPEN after 3 failures, got {alpha_stream['state']}"
    )

    # Group bravo should still be CLOSED and operational
    bravo_stream = cb.stream_state("bravo")
    assert bravo_stream["state"] == "CLOSED", (
        f"bravo should still be CLOSED, got {bravo_stream['state']}"
    )

    result = cb.execute(succeeder, "bravo", context="bravo")
    assert result == "bravo OK", f"bravo should succeed, got {result}"

    # Verify alpha rejects
    try:
        cb.execute(succeeder, "alpha", context="alpha")
        assert False, "alpha should reject — circuit should be OPEN"
    except CircuitBreakerError:
        pass

    print("PASS: test_failure_in_one_group_does_not_trip_another")


def test_multiple_independent_failure_streams_trip_correctly():
    cb = ContextualCircuitBreaker(threshold=2, cooldown=600)
    cb.reset()

    def failer(name):
        raise RuntimeError(f"{name} failure")

    contexts = ["x", "y", "z"]
    for ctx in contexts:
        for _ in range(2):
            try:
                cb.execute(failer, ctx, context=ctx)
            except RuntimeError:
                pass

    for ctx in contexts:
        stream = cb.stream_state(ctx)
        assert stream["state"] == "OPEN", (
            f"{ctx} should be OPEN after 2 failures, got {stream['state']}"
        )

    print("PASS: multiple independent failure streams trip correctly")


def test_different_thresholds_independent():
    cb = ContextualCircuitBreaker(threshold=4, cooldown=600)
    cb.reset()

    def failer():
        raise RuntimeError("fail")

    def succeeder():
        return "ok"

    # Give group-A 3 failures (below threshold 4) and group-B 4 failures (trips)
    for _ in range(3):
        try:
            cb.execute(failer, context="A")
        except RuntimeError:
            pass

    for _ in range(4):
        try:
            cb.execute(failer, context="B")
        except RuntimeError:
            pass

    a_stream = cb.stream_state("A")
    b_stream = cb.stream_state("B")

    assert a_stream["state"] == "CLOSED", (
        f"A should be CLOSED (3/4 failures), got {a_stream['state']}"
    )
    assert b_stream["state"] == "OPEN", (
        f"B should be OPEN (4/4 failures), got {b_stream['state']}"
    )

    # A should still accept requests
    result = cb.execute(succeeder, context="A")
    assert result == "ok", f"A should return ok, got {result}"

    # B should reject
    try:
        cb.execute(succeeder, context="B")
        assert False, "B should be OPEN and reject"
    except CircuitBreakerError:
        pass

    print("PASS: different thresholds per group (A=3 open, B=4 open)")


if __name__ == "__main__":
    tests = [
        test_failure_in_one_group_does_not_trip_another,
        test_multiple_independent_failure_streams_trip_correctly,
        test_different_thresholds_independent,
    ]
    failed = 0
    for test in tests:
        try:
            test()
        except AssertionError as e:
            print(f"FAIL: {test.__name__}: {e}")
            failed += 1
        except Exception as e:
            print(f"FAIL: {test.__name__}: {e}")
            failed += 1

    if failed:
        sys.exit(1)
    print("\nAll tests passed.")