fix(coordination): make manual controls authoritative

CHANGELOG.md

@@ -6,6 +6,14 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
 
+### Fixed
+
+- Local manual controls now take precedence over shared state for coordinated
+  breakers. `force_open()` rejects locally, while `disable()` and
+  `metrics_only()` admit locally without consuming a shared `HALF_OPEN` probe.
+  `reset()` clears only local control and metrics, then resumes the cached
+  shared state rather than resetting the cluster.
+
 ## [2.1.2] - 2026-07-14
 
 ### Fixed

docs/guides/states.md

@@ -66,7 +66,7 @@ Three special states are set manually and stay until you `reset()`:
 | `breaker.force_open()` | `FORCED_OPEN` | Reject all traffic regardless of metrics. |
 | `breaker.disable()` | `DISABLED` | Admit all traffic, record nothing — the breaker is a no-op. |
 | `breaker.metrics_only()` | `METRICS_ONLY` | Admit all traffic, record metrics, but never trip. |
-| `breaker.reset()` | `CLOSED` | Return to closed with a fresh, empty window. |
+| `breaker.reset()` | `CLOSED` | Return to closed with a fresh, empty window. In coordinated mode, resume the cached shared state instead. |
 
 ```python
 breaker.metrics_only()  # observe in production without enforcing
@@ -90,6 +90,12 @@ admission everywhere, and the HALF_OPEN probe budget is shared globally.
 governs admission, the local one otherwise (including while the storage is
 unreachable).
 
+Local operator overrides always take precedence over a healthy shared view:
+`force_open()` rejects locally, while `disable()` and `metrics_only()` admit
+locally without claiming a shared HALF_OPEN probe. `reset()` clears that local
+override and freshens local metrics; it does not change the cluster. The
+instance immediately resumes the cached shared `OPEN` or `HALF_OPEN` state.
+
 ## Observing transitions
 
 Every transition (and reset) is delivered to the breaker's

docs/integrations/redis.md

@@ -101,6 +101,15 @@ of a round, the deciding instance applies the same thresholds as the local
 state machine and writes the transition guarded by a version check, so a
 delayed decision can never overwrite a newer state.
 
+## Manual controls
+
+Manual controls are local to one process and take precedence over Redis while
+they are active. `force_open()` rejects every local call; `disable()` and
+`metrics_only()` admit local calls without consuming a Redis HALF_OPEN probe.
+`reset()` clears the local override and local metrics, but does not reset Redis
+for the fleet: the breaker immediately resumes its cached shared `OPEN` or
+`HALF_OPEN` state.
+
 ## Degradation: Redis down ≠ breaker down
 
 A storage error never reaches your calls. On the first failure the breaker

interlock/_coordination.py

@@ -12,12 +12,11 @@
   decision) are fire-and-forget: they run on a background *lane* (one daemon
   thread for a sync storage, one asyncio task for an async one) that doubles as
   the poller refreshing the cached view every ``poll_interval``.
-- Storage failures never reach the protected path (T3.1): any storage error
+- Storage failures never reach the protected path: any storage error
   flips the coordinator into degraded mode — the breaker runs on local state,
   writes are dropped, and the lane keeps retrying after ``retry_backoff``
   seconds. Degradation and recovery surface through the engine's listener
-  callbacks (T3.2); on recovery the shared view becomes authoritative again
-  (T3.3).
+  callbacks; on recovery the shared view becomes authoritative again.
 
 Tuning knobs are read from optional attributes on the storage object
 (``state_ttl``, ``poll_interval``, ``retry_backoff``) with conservative

interlock/_engine.py

@@ -42,6 +42,7 @@
 
 # Shared states that override local admission (a shared CLOSED defers to local).
 _SHARED_AUTHORITATIVE = frozenset({State.OPEN, State.HALF_OPEN})
+_MANUAL_STATES = frozenset({State.FORCED_OPEN, State.DISABLED, State.METRICS_ONLY})
 
 
 class _NoopListener:
@@ -146,9 +147,9 @@ def __init__(
     def state(self) -> State:
         """The breaker's effective lifecycle state.
 
-        In coordinated mode a shared OPEN/HALF_OPEN overrides the local state
-        (it governs admission); otherwise — including while the storage is
-        degraded — the local state machine's state is reported.
+        A local manual override always governs admission. Otherwise, in
+        coordinated mode a shared OPEN/HALF_OPEN overrides the local state;
+        while storage is degraded the local state is reported.
         """
         with self._lock:
             return self._effective_state_locked()
@@ -254,7 +255,7 @@ def exit_block(
         self._settle(result=None, exception=exception, start=start, admission=admission)
 
     def reset(self) -> None:
-        """Return to ``CLOSED`` with a fresh window, discarding past metrics."""
+        """Clear local control and metrics, then resume the shared state if present."""
         with self._lock:
             effective_before = self._effective_state_locked()
             before = self._machine.state
@@ -339,6 +340,8 @@ def _shared_gate(self) -> State | None:
         admitted there, and local state must not overrule a coordinated trip.
         """
         with self._lock:
+            if self._machine.state in _MANUAL_STATES:
+                return None
             view = self._shared_view
             if view is None or self._storage_degraded:
                 return None
@@ -430,6 +433,9 @@ def _settle(
 
     def _effective_state_locked(self) -> State:
         """The state that governs admission; caller must hold ``self._lock``."""
+        if self._machine.state in _MANUAL_STATES:
+            return self._machine.state
+
         view = self._shared_view
         if view is not None and not self._storage_degraded and view.state in _SHARED_AUTHORITATIVE:
             return view.state

interlock/breaker.py

@@ -93,7 +93,7 @@ def snapshot(self) -> WindowSnapshot:
         return self._engine.snapshot()
 
     def reset(self) -> None:
-        """Return the breaker to ``CLOSED`` with a fresh window."""
+        """Clear local control and metrics, then resume the shared state if present."""
         self._engine.reset()
 
     def force_open(self) -> None:

tests/inmemory_storage.py

@@ -3,7 +3,8 @@
 Not shipped: a deterministic double that mirrors the atomic semantics the Redis
 backend will provide via Lua, so the contract suite (and later engine-coordination
 tests) run without a live server. Time comes from an injected ``Clock``; ``ttl``
-is accepted but inert (key expiry is a Redis concern, covered by T2.4).
+is accepted but inert (key expiry is a Redis concern covered by Redis integration
+tests).
 """
 
 import dataclasses

tests/test_coordination.py

@@ -9,6 +9,7 @@
 import asyncio
 import gc
 import weakref
+from typing import cast
 
 import pytest
 
@@ -23,6 +24,7 @@
     _sync_lane_tick,
 )
 from interlock.errors import InterlockError
+from interlock.protocols import AsyncStorage, EventListener, Storage
 from interlock.shared import ProbeLease, SharedState
 
 NAME = 'svc'
@@ -112,11 +114,15 @@ def close(self, *, name: str, ttl: float, expected_version: int | None = None) -
 def _breaker(
     config: Config,
     fake_clock: FakeClock,
-    storage: object,
+    storage: Storage | AsyncStorage,
     listener: RecordingListener | None = None,
 ) -> CircuitBreaker:
     return CircuitBreaker(
-        name=NAME, config=config, clock=fake_clock, storage=storage, listener=listener
+        name=NAME,
+        config=config,
+        clock=fake_clock,
+        storage=storage,
+        listener=cast('EventListener | None', listener),
     )
 
 
@@ -189,6 +195,88 @@ def test__half_open__probe_budget_is_global(
         b.call(lambda: 'probe-c')  # budget exhausted across instances
 
 
+def _adopt_shared_state(
+    breaker: CircuitBreaker, storage: InMemoryStorage, fake_clock: FakeClock, state: State
+) -> None:
+    storage.trip_open(name=NAME, ttl=60.0)
+    coordinator = _coordinator(breaker)
+    coordinator.poll_once()
+    if state is State.HALF_OPEN:
+        fake_clock.advance(WAIT)
+        coordinator.poll_once()
+
+
+@pytest.mark.parametrize('shared_state', [State.OPEN, State.HALF_OPEN])
+@pytest.mark.parametrize(
+    'manual_control',
+    [
+        ('force_open', State.FORCED_OPEN, False),
+        ('disable', State.DISABLED, True),
+        ('metrics_only', State.METRICS_ONLY, True),
+    ],
+)
+def test__manual_control__shared_state__governs_local_admission_without_leasing_probe(
+    config: Config,
+    fake_clock: FakeClock,
+    storage: InMemoryStorage,
+    shared_state: State,
+    manual_control: tuple[str, State, bool],
+) -> None:
+    control, expected_state, admitted = manual_control
+    listener = RecordingListener()
+    breaker = _breaker(config, fake_clock, storage, listener)
+    _adopt_shared_state(breaker, storage, fake_clock, shared_state)
+    listener.state_changes.clear()
+
+    getattr(breaker, control)()
+
+    assert breaker.state is expected_state
+    assert listener.state_changes == [(shared_state, expected_state)]
+    if admitted:
+        assert breaker.call(lambda: 'local') == 'local'
+    else:
+        with pytest.raises(CircuitOpenError):
+            breaker.call(lambda: 'must not run')
+
+    shared = storage.read(NAME)
+    assert shared is not None
+    expected_remaining = (
+        config.permitted_calls_in_half_open if shared_state is State.HALF_OPEN else 0
+    )
+    assert shared.probes_remaining == expected_remaining
+
+
+@pytest.mark.parametrize('shared_state', [State.OPEN, State.HALF_OPEN])
+def test__reset__shared_state__resumes_shared_admission(
+    config: Config,
+    fake_clock: FakeClock,
+    storage: InMemoryStorage,
+    shared_state: State,
+) -> None:
+    listener = RecordingListener()
+    breaker = _breaker(config, fake_clock, storage, listener)
+    _adopt_shared_state(breaker, storage, fake_clock, shared_state)
+    breaker.force_open()
+    listener.state_changes.clear()
+
+    breaker.reset()
+
+    assert breaker.state is shared_state
+    assert listener.state_changes == [(State.FORCED_OPEN, shared_state)]
+    assert listener.resets == 1
+    if shared_state is State.OPEN:
+        with pytest.raises(CircuitOpenError):
+            breaker.call(lambda: 'must not run')
+        return
+
+    assert breaker.call(lambda: 'probe') == 'probe'
+    _coordinator(breaker).wait_idle()
+    shared = storage.read(NAME)
+    assert shared is not None
+    assert shared.probes_remaining == config.permitted_calls_in_half_open - 1
+    assert shared.probes_completed == 1
+
+
 def test__successful_probe_round__closes_all_instances(
     config: Config, fake_clock: FakeClock, storage: InMemoryStorage
 ) -> None:
@@ -251,9 +339,6 @@ def test__registry__passes_storage_through(
     assert shared.state is State.OPEN
 
 
-# --- T3.1/T3.2/T3.3: degradation, observability, recovery ---
-
-
 def test__storage_failure__degrades_to_local_state(
     config: Config, fake_clock: FakeClock, storage: InMemoryStorage
 ) -> None:
@@ -264,7 +349,8 @@ def test__storage_failure__degrades_to_local_state(
     _trip(tripper)
     _coordinator(tripper).wait_idle()
     _coordinator(follower).poll_once()
-    assert follower.state is State.OPEN  # shared OPEN adopted
+    state = follower.state
+    assert state is State.OPEN  # shared OPEN adopted
 
     flaky.fail = True
     fake_clock.advance(flaky.retry_backoff)
@@ -340,7 +426,7 @@ def test__recovery__emits_event_and_shared_becomes_authoritative(
     _coordinator(breaker).poll_once()
 
     assert listener.recovered == 1
-    assert breaker.state is State.OPEN  # shared OPEN authoritative again (T3.3)
+    assert breaker.state is State.OPEN
     with pytest.raises(CircuitOpenError):
         breaker.call(lambda: 'nope')
 
@@ -440,11 +526,13 @@ async def ok() -> str:
     shared = await astorage.read(NAME)
     assert shared is not None
     assert shared.state is State.OPEN
-    assert breaker.state is State.OPEN
+    state = breaker.state
+    assert state is State.OPEN
 
     fake_clock.advance(WAIT)
     await coordinator.poll_once()
-    assert breaker.state is State.HALF_OPEN
+    state = breaker.state
+    assert state is State.HALF_OPEN
 
     assert await breaker.call(ok) == 'ok'
     await coordinator.wait_idle()
@@ -476,6 +564,93 @@ async def test__async__lease_rejection_when_budget_exhausted(
             pass
 
 
+async def _adopt_async_shared_state(
+    breaker: CircuitBreaker,
+    storage: AsyncInMemoryStorage,
+    fake_clock: FakeClock,
+    state: State,
+) -> None:
+    await storage.trip_open(name=NAME, ttl=60.0)
+    coordinator = _async_coordinator(breaker)
+    await coordinator.poll_once()
+    if state is State.HALF_OPEN:
+        fake_clock.advance(WAIT)
+        await coordinator.poll_once()
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('shared_state', [State.OPEN, State.HALF_OPEN])
+@pytest.mark.parametrize(
+    'manual_control',
+    [
+        ('force_open', State.FORCED_OPEN, False),
+        ('disable', State.DISABLED, True),
+        ('metrics_only', State.METRICS_ONLY, True),
+    ],
+)
+async def test__async__manual_control__shared_state__governs_local_admission_without_leasing_probe(
+    config: Config,
+    fake_clock: FakeClock,
+    shared_state: State,
+    manual_control: tuple[str, State, bool],
+) -> None:
+    control, expected_state, admitted = manual_control
+    storage = AsyncInMemoryStorage(clock=fake_clock)
+    storage.poll_interval = 3600.0
+    listener = RecordingListener()
+    breaker = _breaker(config, fake_clock, storage, listener)
+    await _adopt_async_shared_state(breaker, storage, fake_clock, shared_state)
+    listener.state_changes.clear()
+
+    getattr(breaker, control)()
+
+    assert breaker.state is expected_state
+    assert listener.state_changes == [(shared_state, expected_state)]
+    if admitted:
+        assert await breaker.call(_async_ok) == 'ok'
+    else:
+        with pytest.raises(CircuitOpenError):
+            await breaker.call(_async_ok)
+
+    shared = await storage.read(NAME)
+    assert shared is not None
+    expected_remaining = (
+        config.permitted_calls_in_half_open if shared_state is State.HALF_OPEN else 0
+    )
+    assert shared.probes_remaining == expected_remaining
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('shared_state', [State.OPEN, State.HALF_OPEN])
+async def test__async__reset__shared_state__resumes_shared_admission(
+    config: Config, fake_clock: FakeClock, shared_state: State
+) -> None:
+    storage = AsyncInMemoryStorage(clock=fake_clock)
+    storage.poll_interval = 3600.0
+    listener = RecordingListener()
+    breaker = _breaker(config, fake_clock, storage, listener)
+    await _adopt_async_shared_state(breaker, storage, fake_clock, shared_state)
+    breaker.force_open()
+    listener.state_changes.clear()
+
+    breaker.reset()
+
+    assert breaker.state is shared_state
+    assert listener.state_changes == [(State.FORCED_OPEN, shared_state)]
+    assert listener.resets == 1
+    if shared_state is State.OPEN:
+        with pytest.raises(CircuitOpenError):
+            await breaker.call(_async_ok)
+        return
+
+    assert await breaker.call(_async_ok) == 'ok'
+    await _async_coordinator(breaker).wait_idle()
+    shared = await storage.read(NAME)
+    assert shared is not None
+    assert shared.probes_remaining == config.permitted_calls_in_half_open - 1
+    assert shared.probes_completed == 1
+
+
 class AsyncFlakyStorage:
     """Async in-memory storage whose reads can be made to raise."""