Getting ChatGPT to Write Accurate Circuit Breaker Logic Without Flapping

You ask ChatGPT to implement a circuit breaker, and it produces something that looks right at first glance — states, transitions, a failure counter. But in a staging environment with real traffic spikes, the breaker trips and immediately tries to recover, over and over, hammering your already-struggling downstream service. That is flapping, and it makes your circuit breaker worse than useless.

The root cause is almost never your code. It is the prompt. ChatGPT does not know your failure semantics, your concurrency model, or your recovery window unless you tell it. This article shows you exactly what to include.

What You'll Learn

• Why ChatGPT's default circuit breaker output almost always collapses the half-open state
• How to specify threshold, cooldown, and probe-request semantics precisely in your prompt
• How to handle wall-clock time and thread-safety in generated code
• A reusable prompt template that produces production-grade breaker logic
• How to audit the generated code for the three most common flapping bugs

What ChatGPT Gets Wrong by Default

When you type something like "write a circuit breaker in Python", ChatGPT reaches for the most statistically common pattern in its training data. That pattern is usually a two-state toggle: CLOSED (healthy) and OPEN (failing). The half-open state — the controlled recovery probe — is either missing or implemented as an immediate, unconditional flip back to closed after a fixed sleep.

The result is a breaker that opens on failure, waits N seconds, then closes completely regardless of whether the downstream service has recovered. Under sustained load, this creates a sawtooth pattern: open, close, flood, open again. Each cycle dumps a full burst of requests at a service that is still degraded.

ChatGPT also tends to implement failure counting with a plain integer and no time window. That means five failures spread across three hours can trip the breaker the same way five failures in one second would — a counting model that has no relationship to real failure rate.

Why Circuit Breaker Flapping Is Dangerous

A flapping breaker creates a feedback loop. The downstream service is slow because it is overloaded. Your breaker opens, cutting traffic for a few seconds. Then it closes and immediately sends the full traffic volume again. The service never gets breathing room, so it never recovers. Your breaker opens again. Repeat indefinitely.

Beyond the downstream impact, flapping hides signal from your monitoring. Alerts tied to error rate or latency become noisy because the error rate oscillates on a tight cycle. On-call engineers spend time chasing a symptom rather than the root cause. For a concrete example of how noisy alerting erodes confidence, see how ChatGPT-generated Prometheus alerting rules can produce false fires for a parallel failure mode.

The Three-State Model ChatGPT Tends to Collapse

A correct circuit breaker has three distinct states with specific transition rules:

• CLOSED — normal operation. Requests pass through. Failures are counted within a rolling time window. When the failure count (or rate) crosses a threshold, transition to OPEN.
• OPEN — all requests fail fast without touching the downstream service. After a configured cooldown period, transition to HALF_OPEN.
• HALF_OPEN — a limited number of probe requests are allowed through. If they succeed, transition to CLOSED. If any fail, return to OPEN and reset the cooldown timer.

The critical word in that last bullet is limited. In HALF_OPEN, you must block everything except the probe request(s). ChatGPT almost always omits this gate, turning HALF_OPEN into effectively CLOSED with extra steps.

Prompting for Threshold Logic That Actually Holds

The single most important thing you can do is replace vague language with explicit numeric contracts. Compare these two prompt fragments:

Vague: "Trip the breaker when there are too many failures."

Precise: "Trip the breaker when 5 or more failures occur within a 10-second sliding window. A failure is defined as any exception raised by the wrapped call or any response where response.status_code >= 500."

Giving ChatGPT a concrete failure definition also prevents a common bug where the model counts connection timeouts but ignores 502 responses, or vice versa. You know what "failed" means in your system; spell it out.

Here is a prompt fragment for threshold logic that consistently produces accurate output:

Implement threshold tracking with these exact rules:
- Use a deque to store timestamps of recent failures.
- On each call, purge timestamps older than `window_seconds` from the left of the deque.
- A failure is: any exception, or a return value where the caller-supplied `is_failure(result)` predicate returns True.
- Trip (CLOSED -> OPEN) when len(deque) >= failure_threshold after purging.
- Do not use a plain integer counter. The deque IS the counter and the window.

Prompting the Half-Open State Correctly

This is where most generated circuit breaker code falls apart. You need to give ChatGPT an explicit concurrency contract for the HALF_OPEN state. Without it, the model will let every concurrent request through during the probe window, which defeats the purpose entirely.

Add this to your prompt:

HALF_OPEN state rules (non-negotiable):
- Only ONE probe request may pass through at a time.
- Use a threading.Lock (or asyncio.Lock for async) to enforce this.
- Any request that arrives while a probe is in flight must receive an immediate failure (raise CircuitOpenError), NOT wait for the probe to finish.
- If the probe succeeds: acquire the state lock, transition to CLOSED, reset the failure deque, release the lock.
- If the probe fails: acquire the state lock, transition to OPEN, reset the cooldown timer to now(), release the lock.

Specifying the lock behavior explicitly prevents the model from generating a flag-based approach like self.probing = True without any actual synchronization, which breaks under concurrent requests.

This kind of careful concurrency specification mirrors the discipline required when prompting ChatGPT for idempotency key logic — in both cases, the AI skips the hard concurrency details unless you demand them explicitly.

Handling Clock and Concurrency Pitfalls

Wall-clock time, not call count, for cooldown

ChatGPT often implements the cooldown as a time.sleep() call on the thread that triggered the OPEN transition. This blocks the calling thread and does nothing useful for concurrent callers. The correct approach is to record the timestamp when the breaker opened and check elapsed time on each incoming request.

import time
from threading import Lock
from collections import deque

class CircuitBreaker:
    CLOSED = "CLOSED"
    OPEN = "OPEN"
    HALF_OPEN = "HALF_OPEN"

    def __init__(self, failure_threshold, window_seconds, cooldown_seconds, is_failure=None):
        self.failure_threshold = failure_threshold
        self.window_seconds = window_seconds
        self.cooldown_seconds = cooldown_seconds
        self.is_failure = is_failure or (lambda r: False)
        self._state = self.CLOSED
        self._failure_times = deque()
        self._opened_at = None
        self._probe_in_flight = False
        self._lock = Lock()

    @property
    def state(self):
        with self._lock:
            if self._state == self.OPEN:
                if time.monotonic() - self._opened_at >= self.cooldown_seconds:
                    self._state = self.HALF_OPEN
                    self._probe_in_flight = False
            return self._state

    def call(self, fn, *args, **kwargs):
        state = self.state

        if state == self.OPEN:
            raise CircuitOpenError("Circuit is OPEN")

        if state == self.HALF_OPEN:
            with self._lock:
                if self._probe_in_flight:
                    raise CircuitOpenError("Probe already in flight")
                self._probe_in_flight = True

        try:
            result = fn(*args, **kwargs)
            if self.is_failure(result):
                raise CallFailedError("Caller predicate marked result as failure")
            self._on_success()
            return result
        except CircuitOpenError:
            raise
        except Exception:
            self._on_failure()
            raise

    def _on_success(self):
        with self._lock:
            self._state = self.CLOSED
            self._failure_times.clear()
            self._probe_in_flight = False

    def _on_failure(self):
        with self._lock:
            now = time.monotonic()
            self._failure_times.append(now)
            # Purge failures outside the window
            while self._failure_times and now - self._failure_times[0] > self.window_seconds:
                self._failure_times.popleft()
            if len(self._failure_times) >= self.failure_threshold:
                self._state = self.OPEN
                self._opened_at = now
                self._probe_in_flight = False


class CircuitOpenError(Exception):
    pass


class CallFailedError(Exception):
    pass

Use time.monotonic(), not time.time()

ChatGPT frequently uses time.time() for elapsed-time checks. This is wall-clock time and can jump backwards during NTP corrections, causing a cooldown that expires immediately or never. Always specify time.monotonic() in your prompt for any elapsed-time measurement.

A Complete Prompt Template You Can Reuse

Combine everything above into a single prompt block. Copy this, fill in your values, and paste it directly into ChatGPT or your preferred model:

Write a thread-safe Python circuit breaker class. Requirements:

Parameters (all constructor args with defaults):
- failure_threshold: int = 5 (failures needed within window to trip)
- window_seconds: float = 10.0 (sliding window for failure counting)
- cooldown_seconds: float = 30.0 (time OPEN before moving to HALF_OPEN)
- is_failure: callable = None (optional predicate on the return value; if None, only exceptions count)

States: CLOSED, OPEN, HALF_OPEN (use string constants, not an Enum, for readability).

Failure counting:
- Use a collections.deque of monotonic timestamps.
- On each failure, append time.monotonic(), then purge entries older than window_seconds from the left.
- Trip to OPEN when len(deque) >= failure_threshold after purging.
- NEVER use a plain integer counter.

Cooldown:
- Record opened_at = time.monotonic() when transitioning to OPEN.
- Compute elapsed = time.monotonic() - opened_at on each incoming call.
- Transition to HALF_OPEN only when elapsed >= cooldown_seconds.
- NEVER use time.sleep() or time.time().

HALF_OPEN concurrency:
- Exactly one probe request may be in flight at a time.
- Gate this with a boolean _probe_in_flight protected by the same threading.Lock used for state.
- Any request that arrives while _probe_in_flight is True must raise CircuitOpenError immediately.
- Probe success: transition to CLOSED, clear failure deque, set _probe_in_flight = False.
- Probe failure: transition to OPEN, reset opened_at, set _probe_in_flight = False.

Public API:
- call(fn, *args, **kwargs) -> any: wraps a callable.
- state property: returns current state string.

Raise CircuitOpenError (custom exception, subclass of Exception) when the circuit is open or a probe is in flight.
Do not use any external libraries. Include a brief docstring on the class.

This level of specificity might feel like over-engineering a prompt, but it is exactly the same discipline you would apply to a code review checklist. The more precisely you define the contract, the less the model has to infer, and the fewer bugs it introduces. This mirrors a broader principle: when you ask ChatGPT to generate logic with complex failure semantics, ambiguity in the prompt translates directly to bugs in the output — as covered in depth in the guide on avoiding poison-pill loops in AI-generated message queue consumers.

Common Pitfalls When Reviewing AI-Generated Breaker Code

Even with a detailed prompt, you should scan the output for these specific issues before committing:

• State check outside the lock. If the model reads self._state before acquiring the lock and then re-reads it inside, the transition logic can race. All state reads that gate behavior must happen inside a single lock acquisition.
• Swallowed exceptions. Some generated implementations catch the exception, record the failure, then return None instead of re-raising. This hides errors from the caller. The breaker should always re-raise after recording.
• CircuitOpenError caught by the failure handler. If the broad except Exception in call() also catches CircuitOpenError, it will record a failure every time the breaker is already open. Guard against this with an explicit except CircuitOpenError: raise before the broad handler — exactly as shown in the code block above.
• Missing reset on close. When transitioning from HALF_OPEN to CLOSED after a successful probe, the failure deque must be cleared. If it is not, stale failure timestamps from before the outage can re-trip the breaker seconds after recovery.
• No is_failure predicate. If your downstream service returns 200 OK with an error payload (common in legacy RPC systems), exception-only counting will never trip the breaker. Always include the predicate hook and mention it in your prompt.

Reviewing for these issues is a lot like auditing AI-generated rate-limit logic — the code looks correct until traffic hits an edge case. The article on getting ChatGPT to write accurate API rate limit headers walks through a similar review checklist for a related class of problems.

Next Steps

You now have a repeatable process for generating circuit breaker logic that holds up under real conditions. Here is what to do next:

1. Copy the prompt template above and adapt the numeric parameters to match your service's actual SLOs — failure threshold, window size, and cooldown should be derived from your p99 latency and retry budget, not arbitrary defaults.
2. Add the is_failure predicate for your specific downstream API. Pass in a lambda that checks status codes, error fields in the response body, or whatever signal your service emits.
3. Write a pytest fixture that mocks time.monotonic() and drives the breaker through all three state transitions, including the concurrent HALF_OPEN probe-blocking case. Ask ChatGPT to write the tests too — use the same level of specificity in that prompt.
4. Expose the state as a metric. Emit a gauge (0 = CLOSED, 1 = HALF_OPEN, 2 = OPEN) to your metrics system so you can alert on sustained OPEN time rather than guessing from error rates alone.
5. Revisit the cooldown period after your first real incident. Cooldown is not a set-and-forget value; it should reflect the actual recovery time of your slowest downstream dependency.