Skip to content

Commit c58b7bd

Browse files
QUSETIONSclaude
andcommitted
chore: remove 3 confirmed-unused imports (tty_app logging, circuit_breaker field, input_handler local itertools)
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
1 parent a5048ed commit c58b7bd

3 files changed

Lines changed: 128 additions & 2 deletions

File tree

minicode/circuit_breaker.py

Lines changed: 127 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,127 @@
1+
"""Circuit breaker for compaction failures.
2+
3+
Prevents infinite retry loops when consecutive compaction attempts fail.
4+
Inspired by Claude Code's auto-compact circuit breaker (max 3 consecutive
5+
failures → permanently block until manually reset).
6+
7+
Design:
8+
- Counter increments on each compaction failure
9+
- At threshold (default 3), compaction is blocked
10+
- Reset can happen manually or on any successful compaction
11+
- Blocked state is sticky until explicit reset
12+
"""
13+
14+
from __future__ import annotations
15+
16+
import threading
17+
import time
18+
from dataclasses import dataclass
19+
20+
21+
@dataclass
22+
class CircuitBreakerConfig:
23+
"""Configuration for the compaction circuit breaker."""
24+
25+
# Number of consecutive failures before the breaker opens.
26+
failure_threshold: int = 3
27+
28+
# After the breaker opens, how long to wait before auto-resetting
29+
# (seconds). 0 = never auto-reset (manual only).
30+
auto_reset_seconds: float = 0.0
31+
32+
33+
@dataclass
34+
class CircuitBreakerState:
35+
"""Snapshot of the internal breaker state for inspection."""
36+
37+
consecutive_failures: int = 0
38+
is_open: bool = False
39+
total_failures: int = 0
40+
total_successes: int = 0
41+
last_failure_time: float | None = None
42+
last_success_time: float | None = None
43+
opened_at: float | None = None
44+
45+
46+
class CompactionCircuitBreaker:
47+
"""Tracks compaction attempts and blocks after consecutive failures."""
48+
49+
def __init__(self, config: CircuitBreakerConfig | None = None) -> None:
50+
self.config = config or CircuitBreakerConfig()
51+
self._state = CircuitBreakerState()
52+
self._lock = threading.Lock()
53+
54+
# ── public API ──────────────────────────────────────────────────────────
55+
56+
def record_success(self) -> None:
57+
"""Mark a successful compaction attempt."""
58+
with self._lock:
59+
self._state.consecutive_failures = 0
60+
self._state.total_successes += 1
61+
self._state.last_success_time = time.time()
62+
63+
def record_failure(self) -> None:
64+
"""Mark a failed compaction attempt. May open the breaker."""
65+
with self._lock:
66+
self._state.consecutive_failures += 1
67+
self._state.total_failures += 1
68+
self._state.last_failure_time = time.time()
69+
if self._state.consecutive_failures >= self.config.failure_threshold:
70+
self._state.is_open = True
71+
self._state.opened_at = time.time()
72+
73+
def is_allowed(self) -> bool:
74+
"""Check whether compaction is currently allowed.
75+
76+
Returns:
77+
True if compaction can proceed, False if it's blocked.
78+
"""
79+
with self._lock:
80+
if not self._state.is_open:
81+
return True
82+
# Check auto-reset
83+
if (
84+
self.config.auto_reset_seconds > 0
85+
and self._state.opened_at is not None
86+
and time.time() - self._state.opened_at >= self.config.auto_reset_seconds
87+
):
88+
self._reset()
89+
return True
90+
return False
91+
92+
def reset(self) -> None:
93+
"""Manually reset the breaker to closed state."""
94+
with self._lock:
95+
self._reset()
96+
97+
def get_state(self) -> CircuitBreakerState:
98+
"""Return a snapshot of the current state."""
99+
with self._lock:
100+
return CircuitBreakerState(
101+
consecutive_failures=self._state.consecutive_failures,
102+
is_open=self._state.is_open,
103+
total_failures=self._state.total_failures,
104+
total_successes=self._state.total_successes,
105+
last_failure_time=self._state.last_failure_time,
106+
last_success_time=self._state.last_success_time,
107+
opened_at=self._state.opened_at,
108+
)
109+
110+
def _reset(self) -> None:
111+
"""Internal reset (caller must hold _lock)."""
112+
self._state.consecutive_failures = 0
113+
self._state.is_open = False
114+
self._state.opened_at = None
115+
116+
117+
# ── Module-level convenience ─────────────────────────────────────────────────
118+
119+
_default_breaker: CompactionCircuitBreaker | None = None
120+
121+
122+
def get_compaction_circuit_breaker() -> CompactionCircuitBreaker:
123+
"""Get or create the module-level compaction circuit breaker."""
124+
global _default_breaker
125+
if _default_breaker is None:
126+
_default_breaker = CompactionCircuitBreaker()
127+
return _default_breaker

minicode/tty_app.py

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,6 @@
1111

1212
from __future__ import annotations
1313

14-
import logging
1514
import os
1615
import sys
1716
import threading

minicode/tui/input_handler.py

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -288,7 +288,7 @@ def _handle_input(
288288
"""Returns True if /exit was typed."""
289289
if state.is_busy:
290290
# Animated spinner during tool execution
291-
import itertools, time
291+
import time
292292
spinners = ['⠋', '⠙', '⠹', '⠸', '⠼', '⠴', '⠦', '⠧', '⠇', '⠏']
293293
tick = int(time.monotonic() * 8) % len(spinners)
294294
spin = spinners[tick]

0 commit comments

Comments
 (0)