fix(engine): reap stale warm spare when desired fingerprint changes

When idle TTL is disabled, a ready spare whose fingerprint no longer
matches _desired blocked refill indefinitely. Reap on mismatch before refill.

Author: gildesmarais

app/engine/warm_pool.py

@@ -367,15 +367,30 @@ def _maybe_reap_ttl(self) -> None:
     def _schedule_deferred_refill(self, not_before: float) -> None:
         self._next_refill_at = not_before
 
+    def _maybe_reap_stale_spare(self, desired: DriverFingerprint) -> None:
+        """Requires ``_lock`` held. Drop ready spare when it no longer matches desired."""
+        spare = self._spare
+        if spare is None or spare.fingerprint == desired:
+            return
+        logger.info(
+            "warm_pool_reap reason=stale_fingerprint fingerprint_hash=%s",
+            spare.fingerprint.fingerprint_hash[:12],
+        )
+        self._close_spare_unlocked(spare, reason="stale_fingerprint")
+        self._spare = None
+
     def _maybe_refill(self) -> None:
         with self._lock:
             if self._stop.is_set():
                 return
-            if self._spare is not None or self._building:
+            if self._building:
                 return
             desired = self._desired
             if desired is None:
                 return
+            self._maybe_reap_stale_spare(desired)
+            if self._spare is not None:
+                return
             if self._headless_only and not desired.headless:
                 return
             now = self._clock()

tests/engine/test_warm_pool.py

@@ -153,6 +153,23 @@ def test_take_miss_on_fingerprint_mismatch(self) -> None:
         self.assertTrue(pool.live_spare_dirs())
         pool.shutdown()
 
+    def test_stale_spare_reaped_when_desired_fingerprint_changes(self) -> None:
+        pool = self._pool(idle_ttl_seconds=0)
+        fp_a = _fp(user_agent="A")
+        fp_b = _fp(user_agent="B")
+        pool.notify_scrape_finished(fp_a)
+        self._wait_spare(pool)
+        stale_driver = self.built[0]
+        self.spare_ready.clear()
+
+        pool.notify_scrape_finished(fp_b)
+        self._wait_spare(pool)
+
+        self.assertTrue(stale_driver.closed)
+        self.assertEqual(len(self.built), 2)
+        self.assertIsNotNone(pool.take(fp_b))
+        pool.shutdown()
+
     def test_unhealthy_spare_is_reaped_not_handed_out(self) -> None:
         pool = self._pool(health_check=HealthCheckStub(healthy=False))
         fp = _fp()