fix(validator): close crown pin when a reservation expires without a swap

allways/validator/event_watcher.py

@@ -991,6 +991,14 @@ def _emit_reservation_pin_ends(self, block_num: int, miner: str) -> None:
                     rate=0.0,
                 )
 
+    def expire_stale_reservation_pins(self) -> int:
+        """Close crown pins for reservations that lapsed without a swap (no
+        contract event fires on natural expiry), then purge them. End emitted at
+        ``reserved_until + 1`` so crown stops at the last live block. Returns rows purged."""
+        for pin in self.state_store.get_expired_reservation_pins():
+            self._emit_reservation_pin_ends(pin.reserved_until + 1, pin.miner_hotkey)
+        return self.state_store.purge_expired_reservation_pins()
+
     def record_active_transition(self, block_num: int, hotkey: str, active: bool) -> None:
         """Apply an on-chain active-flag transition to both the current-state
         snapshot and the historical event log. A no-op if the flag already

allways/validator/forward.py

@@ -80,7 +80,10 @@ async def forward(self: Validator) -> None:
             f'forward: purged {purged} expired pending_confirms (reservations elapsed without tx confirmation)'
         )
 
-    purged_pins = self.state_store.purge_expired_reservation_pins()
+    # Emits crown pin-end events for expired reservations *before* purging, so a
+    # reservation that lapses without a swap can't keep earning crown at its
+    # pinned rate. Runs before scoring in this same forward pass.
+    purged_pins = self.event_watcher.expire_stale_reservation_pins()
     if purged_pins:
         bt.logging.info(f'forward: purged {purged_pins} expired reservation_pins (reservations elapsed without a swap)')
 

allways/validator/state_store.py

@@ -267,6 +267,22 @@ def update_reservation_pin_reserved_until(self, miner_hotkey: str, reserved_unti
             (reserved_until, miner_hotkey),
         )
 
+    def get_expired_reservation_pins(self) -> List[ReservationPin]:
+        """Pins whose reservation has lapsed as of the current block.
+
+        Read before ``purge_expired_reservation_pins`` so the caller can emit a
+        scoring pin-end event per expired pin — otherwise the crown overlay's
+        'start' outlives the on-chain reservation and keeps earning crown at the
+        pinned rate after expiry.
+        """
+        if self.current_block_fn is None:
+            return []
+        rows = self._fetchall(
+            'SELECT * FROM reservation_pins WHERE reserved_until < ?',
+            (self.current_block_fn(),),
+        )
+        return [self.row_to_reservation_pin(row) for row in rows]
+
     def purge_expired_reservation_pins(self) -> int:
         """Drop pins whose reservation has already expired."""
         if self.current_block_fn is None:

tests/test_event_watcher.py

@@ -567,6 +567,66 @@ def test_swap_timed_out_clears_pin(self, tmp_path: Path):
         assert w.state_store.get_reservation_pin('hk_a') is None
         w.state_store.close()
 
+    def test_expired_reservation_emits_pin_end_and_purges(self, tmp_path: Path):
+        """A reservation that lapses without a swap emits no contract event, so
+        nothing closes the crown pin. ``expire_stale_reservation_pins`` must
+        emit an 'end' (at reserved_until + 1) for every open direction and purge
+        the synchronous pin row — otherwise the miner keeps earning crown at the
+        pinned rate with no live reservation."""
+        w = make_watcher(tmp_path, netuid=2, subtensor=MagicMock())
+        w.state_store.current_block_fn = lambda: 1001  # past reserved_until=1000
+        with patch(
+            'allways.validator.event_watcher.read_miner_commitment',
+            return_value=make_pinned_commitment(),
+        ):
+            w.apply_event(900, 'MinerReserved', {'miner': 'hk_a', 'reserved_until': 1000})
+
+        purged = w.expire_stale_reservation_pins()
+
+        assert purged == 1
+        assert w.state_store.get_reservation_pin('hk_a') is None
+        ends = [ev for ev in w.reservation_pin_events if ev.hotkey == 'hk_a' and ev.kind == 'end']
+        # One 'end' per pinned direction (primary btc→tao + counter tao→btc),
+        # at reserved_until + 1 so crown is credited through the last live block.
+        assert {(ev.from_chain, ev.to_chain) for ev in ends} == {('btc', 'tao'), ('tao', 'btc')}
+        assert all(ev.block_num == 1001 for ev in ends)
+        w.state_store.close()
+
+    def test_live_reservation_not_closed_by_expiry_sweep(self, tmp_path: Path):
+        """A reservation whose TTL is still in the future must be left alone —
+        no pin-end, no purge."""
+        w = make_watcher(tmp_path, netuid=2, subtensor=MagicMock())
+        w.state_store.current_block_fn = lambda: 950  # before reserved_until=1000
+        with patch(
+            'allways.validator.event_watcher.read_miner_commitment',
+            return_value=make_pinned_commitment(),
+        ):
+            w.apply_event(900, 'MinerReserved', {'miner': 'hk_a', 'reserved_until': 1000})
+
+        purged = w.expire_stale_reservation_pins()
+
+        assert purged == 0
+        assert w.state_store.get_reservation_pin('hk_a') is not None
+        assert not [ev for ev in w.reservation_pin_events if ev.hotkey == 'hk_a' and ev.kind == 'end']
+        w.state_store.close()
+
+    def test_expiry_sweep_is_idempotent(self, tmp_path: Path):
+        """Re-running after the row is purged emits no further 'end' events."""
+        w = make_watcher(tmp_path, netuid=2, subtensor=MagicMock())
+        w.state_store.current_block_fn = lambda: 1001
+        with patch(
+            'allways.validator.event_watcher.read_miner_commitment',
+            return_value=make_pinned_commitment(),
+        ):
+            w.apply_event(900, 'MinerReserved', {'miner': 'hk_a', 'reserved_until': 1000})
+
+        assert w.expire_stale_reservation_pins() == 1
+        ends_after_first = [ev for ev in w.reservation_pin_events if ev.kind == 'end']
+        assert w.expire_stale_reservation_pins() == 0
+        ends_after_second = [ev for ev in w.reservation_pin_events if ev.kind == 'end']
+        assert ends_after_second == ends_after_first
+        w.state_store.close()
+
     def test_reservation_extension_finalized_bumps_pin_ttl(self, tmp_path: Path):
         w = make_watcher(tmp_path, netuid=2, subtensor=MagicMock())
         with patch(

tests/test_scoring_v1.py

@@ -26,7 +26,7 @@
     snapshot_current_crown_holders,
     success_rate,
 )
-from allways.validator.state_store import ValidatorStateStore
+from allways.validator.state_store import ReservationPin, ValidatorStateStore
 
 POOL_TAO_BTC = 0.5
 POOL_BTC_TAO = 0.5
@@ -934,6 +934,68 @@ def test_pin_end_lets_live_rate_take_over(self, tmp_path: Path):
         assert crown == {'hk_a': 750.0, 'hk_b': 250.0}
         store.close()
 
+    def test_expired_reservation_pin_stops_earning_crown(self, tmp_path: Path):
+        """End-to-end: a reservation that lapses without a swap must stop
+        earning crown at the pinned rate. ``expire_stale_reservation_pins``
+        emits the missing pin-end at reserved_until + 1, so once the live rate
+        reverts to junk after expiry the miner can no longer crown-squat on the
+        stale pinned rate."""
+        store = ValidatorStateStore(db_path=tmp_path / 'state.db')
+        watcher = make_watcher(store, active={'hk_a', 'hk_b'})
+        conn = store.require_connection()
+        for hk in ('hk_a', 'hk_b'):
+            conn.execute(
+                'INSERT INTO rate_events (hotkey, from_chain, to_chain, rate, block) VALUES (?, ?, ?, ?, ?)',
+                (hk, 'btc', 'tao', 200.0, 0),
+            )
+        # A bumps its live rate to junk at block 700 — only the stale pin would
+        # keep it winning crown at 200 past expiry.
+        conn.execute(
+            'INSERT INTO rate_events (hotkey, from_chain, to_chain, rate, block) VALUES (?, ?, ?, ?, ?)',
+            ('hk_a', 'btc', 'tao', 25000.0, 700),
+        )
+        conn.commit()
+
+        # A is reserved at block 300, TTL through 600, and never swaps — the
+        # synchronous pin row + the scoring 'start' both land, but no terminal
+        # event ever closes the pin.
+        store.upsert_reservation_pin(
+            ReservationPin(
+                miner_hotkey='hk_a',
+                reserve_block=300,
+                from_chain='btc',
+                to_chain='tao',
+                rate_str='200',
+                counter_rate_str='0',
+                miner_from_address='bc1-a',
+                miner_to_address='5a',
+                reserved_until=600,
+                created_at=1.0,
+            )
+        )
+        watcher._record_reservation_pin_event(
+            block_num=300, hotkey='hk_a', from_chain='btc', to_chain='tao', kind='start', rate=200.0
+        )
+
+        # The forward loop's expiry sweep fires once the block passes the TTL.
+        store.current_block_fn = lambda: 601
+        watcher.expire_stale_reservation_pins()
+
+        crown = replay_crown_time_window(
+            store=store,
+            event_watcher=watcher,
+            from_chain='btc',
+            to_chain='tao',
+            window_start=100,
+            window_end=1100,
+            rewardable_hotkeys={'hk_a', 'hk_b'},
+        )
+        # (100, 601]: A pinned at 200, tie with B → 250.5 each.
+        # (601, 700]: pin closed, A live at 200, tie with B → 49.5 each.
+        # (700, 1100]: A live at junk 25000 — wins the rate sort, earns 400.
+        assert crown == {'hk_a': 700.0, 'hk_b': 300.0}
+        store.close()
+
     def test_pin_only_applies_to_pinned_direction(self, tmp_path: Path):
         """A reservation pin is direction-specific — pinning btc→tao must not
         affect crown ranking in the tao→btc direction."""

tests/test_state_store.py

@@ -113,6 +113,26 @@ def test_purge_without_current_block_fn_is_noop(self, tmp_path: Path):
         assert store.get_reservation_pin('miner-1') == PIN_SAMPLE1
         store.close()
 
+    def test_get_expired_returns_only_expired_rows(self, tmp_path: Path):
+        store = ValidatorStateStore(
+            db_path=tmp_path / 'state.db',
+            current_block_fn=lambda: 1001,
+        )
+        store.upsert_reservation_pin(PIN_SAMPLE1)  # reserved_until=1000 → expired at 1001
+        store.upsert_reservation_pin(PIN_SAMPLE2)  # reserved_until=1005 → still live
+
+        expired = store.get_expired_reservation_pins()
+        assert expired == [PIN_SAMPLE1]
+        # Read-only: the row is left for the caller to emit a pin-end before purging.
+        assert store.get_reservation_pin('miner-1') == PIN_SAMPLE1
+        store.close()
+
+    def test_get_expired_without_current_block_fn_is_empty(self, tmp_path: Path):
+        store = ValidatorStateStore(db_path=tmp_path / 'state.db')
+        store.upsert_reservation_pin(PIN_SAMPLE1)
+        assert store.get_expired_reservation_pins() == []
+        store.close()
+
     def test_update_reserved_until_keeps_row_a_purge_would_drop(self, tmp_path: Path):
         """Regression: after the contract extends a reservation, refreshing the
         pin's reserved_until must keep it alive past its original TTL — exactly