Trip the Loop: A Weekend Circuit Breaker for Agent Side Projects

Weekend agent side projects rarely die from a missing API key. They die from a loop that keeps calling the same tool, eating turns, and leaving a terminal full of identical errors at 1 a.m. A small in-process circuit breaker — turn cap, identical-call tripwire, consecutive-error open state — is enough to stop that spiral without a platform rewrite.

The pattern below is a scoped weekend demo. It is not a distributed reliability product. It is a fence around a local agent loop so a side project can fail closed instead of fail forever.

The failure that actually shows up

A typical weekend agent looks innocent. One planner. A handful of tools. A while loop until the model says it is done. That loop has no natural end when the model repeats a broken tool call, when JSON parsing fails, or when a search tool returns empty and the planner retries the same query.

Unit tests on the tools do not catch this. The tools are fine. The composition is not. Green tool tests and a runaway planner can coexist for hours, because the suite never asserted that the loop had to stop.

Three trip conditions cover most of that weekend damage:

• Turn budget. Hard stop after N planner steps.
• Identical call. Open the circuit when the same tool name and payload digest appear K times in a row.
• Error streak. Open the circuit after M consecutive tool exceptions.

A wall-clock deadline is a fourth optional fence. It is cheap. It is also easy to get wrong on a laptop that sleeps, so the demo keeps cooldown as a reject counter the tests can drive without time.sleep.

Scope cut for one weekend

The build log is the product. Scope was cut on purpose.

Shipped

• In-process state machine: closed, open, half_open
• Turn, identical-call, and error-streak trips
• A half-open probe that allows one trial call after cooldown
• A scripted agent driver and a unit test file that treats "opened" as success

Skipped

• Redis or any shared store
• Multi-process leases (liveness belongs to a heartbeat, not this fence)
• Token spend accounting
• Prompt or model lockfiles
• A web dashboard
• Live model HTTP calls inside the test suite

The skip list is the point. A weekend fence that needs a cluster is not a weekend fence.

Decision table

The breaker answers one question before every tool call: may this loop continue?

State	Event	Next state	Action
closed	turn below max, call is new, tool succeeds	closed	execute
closed	identical payload repeated k times	open	reject
closed	consecutive errors reach m	open	reject
closed	turn reaches max_turns	open	reject
open	cooldown not elapsed	open	reject
open	cooldown elapsed	half_open	reject this call, arm one probe
half_open	probe succeeds	closed	execute and reset counters
half_open	probe fails	open	reject and restart cooldown

Cooldown in the demo is measured in rejected attempts, not wall time. That choice keeps tests deterministic. Wall time can replace the counter later. It should not be the first thing built.

Working demo

The module is a single file. Copy it into tripwire.py. Python 3.10+ is enough. No third-party packages.

# tripwire.py
# Weekend demo: in-process circuit breaker for a local agent loop.
# Proposed pattern, not a production reliability library.

from __future__ import annotations

from dataclasses import dataclass, field
from hashlib import sha256
from typing import Any, Callable, Literal

State = Literal["closed", "open", "half_open"]


def payload_digest(tool: str, payload: Any) -> str:
    raw = f"{tool}:{payload!r}".encode("utf-8")
    return sha256(raw).hexdigest()[:16]


@dataclass
class TripConfig:
    max_turns: int = 8
    identical_limit: int = 3
    error_limit: int = 3
    open_cooldown_rejects: int = 4


@dataclass
class Tripwire:
    config: TripConfig = field(default_factory=TripConfig)
    state: State = "closed"
    turns: int = 0
    identical_streak: int = 0
    error_streak: int = 0
    last_digest: str | None = None
    rejects_while_open: int = 0
    probe_armed: bool = False
    last_reason: str = "ok"

    def precheck(self, tool: str, payload: Any) -> None:
        digest = payload_digest(tool, payload)

        if self.state == "open":
            self.rejects_while_open += 1
            if self.rejects_while_open >= self.config.open_cooldown_rejects:
                self.state = "half_open"
                self.probe_armed = True
                self.rejects_while_open = 0
            self.last_reason = "circuit_open"
            raise RuntimeError("circuit_open")

        if self.state == "half_open":
            if not self.probe_armed:
                self.last_reason = "circuit_open"
                raise RuntimeError("circuit_open")
            self.probe_armed = False

        if self.turns >= self.config.max_turns:
            self._open("max_turns")

        if self.last_digest == digest:
            self.identical_streak += 1
        else:
            self.identical_streak = 1
            self.last_digest = digest

        if self.identical_streak >= self.config.identical_limit:
            self._open("identical_call")

    def record_success(self) -> None:
        self.turns += 1
        self.error_streak = 0
        if self.state == "half_open":
            self.state = "closed"
            self.identical_streak = 0
            self.probe_armed = False
        self.last_reason = "ok"

    def record_error(self, exc: BaseException) -> None:
        self.turns += 1
        self.error_streak += 1
        if self.state == "half_open" or self.error_streak >= self.config.error_limit:
            self._open("error_streak")
        self.last_reason = type(exc).__name__

    def _open(self, reason: str) -> None:
        self.state = "open"
        self.rejects_while_open = 0
        self.probe_armed = False
        self.last_reason = reason
        raise RuntimeError(reason)


def run_agent(
    steps: list[tuple[str, Any, bool]],
    breaker: Tripwire,
    tool_fn: Callable[[str, Any], str],
) -> list[str]:
    """Drive a scripted agent. Each step is (tool, payload, should_succeed)."""
    outputs: list[str] = []
    for tool, payload, should_succeed in steps:
        breaker.precheck(tool, payload)
        try:
            if not should_succeed:
                raise ValueError("tool_failed")
            result = tool_fn(tool, payload)
            breaker.record_success()
            outputs.append(result)
        except RuntimeError:
            raise
        except Exception as exc:
            breaker.record_error(exc)
            raise
    return outputs

A tiny driver shows the fence without a model:

# demo_loop.py
from tripwire import TripConfig, Tripwire, run_agent


def echo(tool: str, payload: object) -> str:
    return f"{tool}:{payload}"


def main() -> None:
    breaker = Tripwire(TripConfig(max_turns=5, identical_limit=3, error_limit=3))
    script = [
        ("search", "circuit breaker agent loop", True),
        ("search", "circuit breaker agent loop", True),
        ("search", "circuit breaker agent loop", True),  # should trip
    ]
    try:
        print(run_agent(script, breaker, echo))
    except RuntimeError as err:
        print(f"tripped state={breaker.state} reason={err}")


if __name__ == "__main__":
    main()

Run it:

python demo_loop.py

Expected line:

tripped state=open reason=identical_call

The third identical search never becomes a fourth attempt. That is the whole weekend win. The planner can still be wrong. It just cannot be wrong forever in the same way.

Tests that fail closed

Naive agent tests assert that a tool returned JSON. They do not assert that the loop stopped. The suite below treats "the breaker opened" as the successful outcome.

# test_tripwire.py
import unittest

from tripwire import TripConfig, Tripwire, run_agent


def echo(tool: str, payload: object) -> str:
    return f"{tool}:{payload}"


class TripwireTests(unittest.TestCase):
    def test_identical_call_opens(self) -> None:
        b = Tripwire(TripConfig(identical_limit=3, max_turns=20))
        steps = [("search", "same", True)] * 3
        with self.assertRaisesRegex(RuntimeError, "identical_call"):
            run_agent(steps, b, echo)
        self.assertEqual(b.state, "open")

    def test_error_streak_opens(self) -> None:
        b = Tripwire(TripConfig(error_limit=3, max_turns=20, identical_limit=9))
        steps = [("fetch", i, False) for i in range(3)]
        with self.assertRaisesRegex(RuntimeError, "error_streak"):
            run_agent(steps, b, echo)
        self.assertEqual(b.state, "open")

    def test_max_turns_opens(self) -> None:
        b = Tripwire(TripConfig(max_turns=2, identical_limit=9, error_limit=9))
        steps = [("think", "a", True), ("think", "b", True), ("think", "c", True)]
        with self.assertRaisesRegex(RuntimeError, "max_turns"):
            run_agent(steps, b, echo)

    def test_half_open_probe_closes(self) -> None:
        b = Tripwire(
            TripConfig(
                max_turns=50,
                identical_limit=9,
                error_limit=2,
                open_cooldown_rejects=2,
            )
        )
        with self.assertRaisesRegex(RuntimeError, "error_streak"):
            run_agent([("fetch", 1, False), ("fetch", 2, False)], b, echo)

        for _ in range(2):
            with self.assertRaisesRegex(RuntimeError, "circuit_open"):
                b.precheck("fetch", "probe-wait")

        b.precheck("fetch", "unique-ok")
        b.record_success()
        self.assertEqual(b.state, "closed")


if __name__ == "__main__":
    unittest.main()

python -m unittest test_tripwire.py -v

Four tests. No network. No fixtures that rot when a vendor changes a chat template. The assertions pin control-plane behavior, which is the part a weekend agent actually loses.

Identical-call hashing uses repr of the payload. That is good enough for strings, ints, and small tuples in a demo. It is a poor canonical form for unordered dicts, objects with changing ids, or floating timestamps injected by the planner. If those show up, normalize the payload before payload_digest, or the tripwire will never see a repeat.

Where a hosted planner fits

Local scripted steps are enough to prove the fence. A real planner still needs a model at some point. That is a different layer from the tripwire. Mixing them in the first weekend usually produces an integration test that fails for reasons that have nothing to do with the breaker.

Disclosure: This article was prepared as part of MonkeyCode's product outreach.

When the scripted loop is green and the next step is to point a planner at a live model, MonkeyCode's free model access and free server option are one way to host that planner without standing up a GPU box first. Keep the tripwire on the caller side, next to the tool calls, so an open circuit does not depend on a remote dashboard.

That split is the useful part. The fence is local, deterministic, and tested. The model is optional and replaceable.

What this will not do

The demo does not coordinate two processes. Two agent workers can still double-call a tool. It does not inspect tool side effects, so a breaker that opens after a successful destructive write is already too late. It does not prove the model is truthful. It only proves the loop stopped.

People who should not use this as-is:

• Anyone running multi-tenant or billed agents. In-process state is not an isolation boundary.
• Anyone who needs a hard real-time deadline. The cooldown is a reject counter, not a clock.
• Anyone who already has a workflow engine with retries and sagas. Do not stack a second policy object on top without mapping states.
• Anyone treating "circuit closed" as a safety sign-off for generated code or destructive tools.

A merge gate for generated diffs is a different weekend. A tool allowlist is a different weekend. This fence only bounds the loop.

Build log: what moved and what did not

The first cut tried to hash the entire conversation and trip on a repeated trace. That was too coarse. A planner can legally retry with a small payload change. Hashing the tool name plus payload, and only counting consecutive repeats, matched the failure that actually happened: the same search, three times, no new information.

The second cut wanted wall-clock timeouts. Tests then needed sleep and became flaky on a laptop that throttled. The reject-counter cooldown replaced it. Wall time can come back when the breaker is wired to a real scheduler.

The third cut wanted to log every trip to a JSONL file. Useful later. Not required to prove the state machine. The last_reason field is enough for a weekend demo and for the tests.

Half-open took longer than the closed/open pair. Allowing the cooldown-completing call to run as a probe mixed "I am still rejecting" with "I am testing recovery" in the same request. The demo now rejects the call that arms the probe, then allows exactly one later call. That is easier to test. It also matches how most human operators think about a cooling-off period: the last failure does not get a free retry.

What shipped is a tripwire a side project can paste, run, and keep. What did not ship is a platform. That is the correct ending for a Sunday night.