diff --git a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py index b8e130cb77..aba3e2763d 100644 --- a/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py +++ b/backend/packages/harness/deerflow/community/aio_sandbox/aio_sandbox_provider.py @@ -901,7 +901,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): deferred += 1 logger.debug("Deferring container %s during reconciliation: this instance is tearing it down", info.sandbox_id) continue - self._warm_pool[info.sandbox_id] = (info, current_time) + self._warm_pool[info.sandbox_id] = (info, time.monotonic()) self._warm_pool_identity[info.sandbox_id] = None self._unowned_since.pop(info.sandbox_id, None) adopted += 1 @@ -1509,7 +1509,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): logger.warning(f"Error closing sandbox {sandbox_id} after losing its lease: {e}") def _cleanup_idle_sandboxes(self, idle_timeout: float) -> None: - current_time = time.time() + current_time = time.monotonic() active_to_destroy = [] with self._lock: @@ -1535,7 +1535,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): # Already released or destroyed by another path — skip. logger.info(f"Sandbox {sandbox_id} already gone before idle destroy, skipping") return False - if (time.time() - last_activity) < idle_timeout: + if (time.monotonic() - last_activity) < idle_timeout: # Re-acquired (activity updated) since the snapshot — skip. logger.info(f"Sandbox {sandbox_id} was re-acquired before idle destroy, skipping") return False @@ -1556,7 +1556,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): if timeout <= 0: return - now = time.time() + now = time.monotonic() expired: list[tuple[str, SandboxInfo]] = [] with self._lock: for sandbox_id, (entry, timestamp) in self._warm_pool.items(): @@ -1737,7 +1737,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): suffix = " (post-lock check)" if post_lock else "" logger.info(f"Reusing in-process sandbox {existing_id} for user/thread {effective_user_id}/{thread_id}{suffix}") - self._last_activity[existing_id] = time.time() + self._last_activity[existing_id] = time.monotonic() # Fail closed: an OwnershipBackendError propagates rather than handing out # a sandbox we could not publish ownership for. @@ -1905,7 +1905,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): self._sandbox_infos[sandbox_id] = info getattr(self, "_cleanup_retry_ids", set()).discard(sandbox_id) self._active_sandbox_identity[sandbox_id] = key - self._last_activity[sandbox_id] = time.time() + self._last_activity[sandbox_id] = time.monotonic() self._thread_sandboxes[key] = sandbox_id suffix = " (post-lock check)" if post_lock else f" at {info.sandbox_url}" @@ -1985,7 +1985,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): self._sandbox_infos[info.sandbox_id] = info getattr(self, "_cleanup_retry_ids", set()).discard(info.sandbox_id) self._active_sandbox_identity[info.sandbox_id] = key - self._last_activity[info.sandbox_id] = time.time() + self._last_activity[info.sandbox_id] = time.monotonic() self._thread_sandboxes[key] = info.sandbox_id except Exception: try: @@ -2032,7 +2032,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): self._sandbox_infos[sandbox_id] = info getattr(self, "_cleanup_retry_ids", set()).discard(sandbox_id) self._active_sandbox_identity[sandbox_id] = key - self._last_activity[sandbox_id] = time.time() + self._last_activity[sandbox_id] = time.monotonic() if key is not None: self._thread_sandboxes[key] = sandbox_id except Exception: @@ -2707,7 +2707,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): return None sandbox = self._sandboxes.get(sandbox_id) if sandbox is not None: - self._last_activity[sandbox_id] = time.time() + self._last_activity[sandbox_id] = time.monotonic() return sandbox def get_scoped( @@ -2728,7 +2728,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): return None sandbox = self._sandboxes.get(sandbox_id) if sandbox is not None: - self._last_activity[sandbox_id] = time.time() + self._last_activity[sandbox_id] = time.monotonic() return sandbox def release(self, sandbox_id: str) -> None: @@ -2792,7 +2792,7 @@ class AioSandboxProvider(WarmPoolLifecycleMixin[SandboxInfo], SandboxProvider): self._last_activity.pop(sandbox_id, None) # Park in warm pool — container keeps running if info and sandbox_id not in self._warm_pool: - self._warm_pool[sandbox_id] = (info, time.time()) + self._warm_pool[sandbox_id] = (info, time.monotonic()) self._warm_pool_identity[sandbox_id] = thread_keys_to_remove[0] if thread_keys_to_remove else active_identity if sandbox is not None: diff --git a/backend/packages/harness/deerflow/community/boxlite/provider.py b/backend/packages/harness/deerflow/community/boxlite/provider.py index ab35daa06e..57e647c0e7 100644 --- a/backend/packages/harness/deerflow/community/boxlite/provider.py +++ b/backend/packages/harness/deerflow/community/boxlite/provider.py @@ -338,7 +338,6 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider): def _adopt_existing_boxes(self) -> int: runtime_cls = _import_sync_boxlite_runtime() - now = time.time() adopted = 0 list_runtime = runtime_cls.default().start() @@ -372,7 +371,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider): if sandbox_id in self._boxes or sandbox_id in self._warm_pool: box_runtime.stop() continue - self._warm_pool[sandbox_id] = (wrapped, now) + self._warm_pool[sandbox_id] = (wrapped, time.monotonic()) self._warm_pool_identity[sandbox_id] = None adopted += 1 logger.info("Adopted existing BoxLite box %s (%s) into warm pool", sandbox_id, name) @@ -524,7 +523,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider): self._skip_health_check_warm_ids.discard(sandbox_id) self._warm_pool_identity.pop(sandbox_id, None) else: - self._warm_pool[sandbox_id] = (box, time.time()) + self._warm_pool[sandbox_id] = (box, time.monotonic()) self._warm_pool_identity[sandbox_id] = released_keys[0] if released_keys else active_identity self._skip_health_check_warm_ids.add(sandbox_id) @@ -559,7 +558,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider): skip_eligible = sandbox_id in self._skip_health_check_warm_ids skip_seconds = self._config.get("health_check_skip_seconds", 0.0) - if skip_eligible and skip_seconds > 0 and (time.time() - released_at) < skip_seconds: + if skip_eligible and skip_seconds > 0 and (time.monotonic() - released_at) < skip_seconds: # Recently released by this provider — promote directly without a # health-check round trip, but never return an adapter that this # process already knows is closed. @@ -595,7 +594,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider): logger.debug( "Reclaimed warm-pool box %s (skipped health check, age=%.1fs)", sandbox_id, - time.time() - released_at, + time.monotonic() - released_at, ) return sandbox_id @@ -675,7 +674,7 @@ class BoxliteProvider(WarmPoolLifecycleMixin[BoxliteBox], SandboxProvider): reaper and atexit shutdown instead of orphaning them. """ with self._lock: - now = time.time() + now = time.monotonic() for sandbox_id, box in self._boxes.items(): self._warm_pool.setdefault(sandbox_id, (box, now)) self._warm_pool_identity.setdefault(sandbox_id, self._active_box_identity.get(sandbox_id)) diff --git a/backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox_provider.py b/backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox_provider.py index f11add8b81..39adf386ca 100644 --- a/backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox_provider.py +++ b/backend/packages/harness/deerflow/community/e2b_sandbox/e2b_sandbox_provider.py @@ -1662,9 +1662,9 @@ class E2BSandboxProvider(SandboxProvider): idle_timeout = float(self._config["idle_timeout"]) if idle_timeout <= 0: return - now_wall = time.time() + now_monotonic = time.monotonic() with self._lock: - expired = [(sandbox_id, entry) for sandbox_id, entry in self._warm_pool.items() if now_wall - entry[1] >= idle_timeout] + expired = [(sandbox_id, entry) for sandbox_id, entry in self._warm_pool.items() if now_monotonic - entry[1] >= idle_timeout] for sandbox_id, entry in expired: with self._sandbox_lifecycle(sandbox_id): with self._lock: @@ -2994,7 +2994,7 @@ class E2BSandboxProvider(SandboxProvider): with self._lock: should_kill = self._shutdown_called if not should_kill: - self._warm_pool[sandbox_id] = (seed or "", time.time()) + self._warm_pool[sandbox_id] = (seed or "", time.monotonic()) self._warm_pool.move_to_end(sandbox_id) self._end_transition_locked() transition_slot_held = False diff --git a/backend/packages/harness/deerflow/community/opensandbox/provider.py b/backend/packages/harness/deerflow/community/opensandbox/provider.py index 075b166710..aef7422329 100644 --- a/backend/packages/harness/deerflow/community/opensandbox/provider.py +++ b/backend/packages/harness/deerflow/community/opensandbox/provider.py @@ -403,7 +403,7 @@ class OpenSandboxProvider(WarmPoolLifecycleMixin[OpenSandboxSandbox], SandboxPro self._thread_sandboxes.pop(key, None) destroy_after_unlock = self._shutdown_called if not destroy_after_unlock: - self._warm_pool[sandbox_id] = (sandbox, time.time()) + self._warm_pool[sandbox_id] = (sandbox, time.monotonic()) if destroy_after_unlock: self._destroy_quietly(sandbox, context="released during shutdown") @@ -552,7 +552,7 @@ class OpenSandboxProvider(WarmPoolLifecycleMixin[OpenSandboxSandbox], SandboxPro def reset(self) -> None: """Park active clients so the detached provider still owns their cleanup.""" with self._lock: - now = time.time() + now = time.monotonic() for sandbox_id, sandbox in self._sandboxes.items(): self._warm_pool.setdefault(sandbox_id, (sandbox, now)) self._sandboxes.clear() diff --git a/backend/packages/harness/deerflow/community/tenki/provider.py b/backend/packages/harness/deerflow/community/tenki/provider.py index 29dcb25f3b..1a36a1ab21 100644 --- a/backend/packages/harness/deerflow/community/tenki/provider.py +++ b/backend/packages/harness/deerflow/community/tenki/provider.py @@ -387,7 +387,7 @@ class TenkiSandboxProvider(WarmPoolLifecycleMixin[TenkiSandbox], SandboxProvider if self._shutdown_called: close_sandbox = sandbox else: - self._warm_pool[sandbox_id] = (sandbox, time.time()) + self._warm_pool[sandbox_id] = (sandbox, time.monotonic()) if close_sandbox is not None: self._close_quietly(close_sandbox, context="released during shutdown") @@ -443,7 +443,7 @@ class TenkiSandboxProvider(WarmPoolLifecycleMixin[TenkiSandbox], SandboxProvider instance's idle reaper and atexit shutdown instead of orphaning them. """ with self._lock: - now = time.time() + now = time.monotonic() for sandbox_id, sandbox in self._sandboxes.items(): self._warm_pool.setdefault(sandbox_id, (sandbox, now)) self._sandboxes.clear() diff --git a/backend/packages/harness/deerflow/community/warm_pool_lifecycle.py b/backend/packages/harness/deerflow/community/warm_pool_lifecycle.py index 8ba7e392b6..ac48368998 100644 --- a/backend/packages/harness/deerflow/community/warm_pool_lifecycle.py +++ b/backend/packages/harness/deerflow/community/warm_pool_lifecycle.py @@ -72,7 +72,9 @@ class WarmPoolLifecycleMixin[WarmEntryT]: if timeout <= 0: return - now = time.time() + # Idle age is a duration. Wall-clock steps (NTP, VM resume, manual + # correction) must neither expire fresh entries nor preserve old ones. + now = time.monotonic() expired: list[tuple[str, WarmEntryT]] = [] with self._lock: for sandbox_id, (entry, timestamp) in self._warm_pool.items(): diff --git a/backend/tests/test_aio_sandbox_provider.py b/backend/tests/test_aio_sandbox_provider.py index e27378e327..b4e1735254 100644 --- a/backend/tests/test_aio_sandbox_provider.py +++ b/backend/tests/test_aio_sandbox_provider.py @@ -1870,7 +1870,7 @@ def _make_provider_with_active_sandbox(tmp_path, sandbox_id: str): sandbox_id: aio_mod.SandboxInfo(sandbox_id=sandbox_id, sandbox_url="http://sandbox-host"), } provider._thread_sandboxes = {} - provider._last_activity = {sandbox_id: 0.0} + provider._last_activity = {sandbox_id: time.monotonic()} provider._local_teardown = set() provider._acquire_epoch = {} provider._acquire_epoch_counter = 0 @@ -3414,7 +3414,7 @@ def test_warm_reclaim_keeps_failed_stop_fence_for_unsafe_generation(failure_reco with pytest.raises(RuntimeError, match="stop failed"): provider.destroy(old.sandbox_id) candidate = type(old)(old.sandbox_id, old.sandbox_url, container_id=generation) - provider._warm_pool[old.sandbox_id] = (candidate, time.time()) + provider._warm_pool[old.sandbox_id] = (candidate, time.monotonic()) provider._warm_pool_identity[old.sandbox_id] = None provider_mod = importlib.import_module("deerflow.community.aio_sandbox.aio_sandbox_provider") @@ -3444,7 +3444,7 @@ def test_warm_reclaim_does_not_clear_fence_if_entry_changes(failure_recovery_lif def swap_during_publish(sandbox_id): publish(sandbox_id) with provider._lock: - provider._warm_pool[sandbox_id] = (unvalidated, time.time()) + provider._warm_pool[sandbox_id] = (unvalidated, time.monotonic()) monkeypatch.setattr(provider, "_publish_ownership", swap_during_publish) assert provider._reclaim_warm_pool_sandbox("thread-failure-recovery", old.sandbox_id, user_id="alice") is None @@ -3779,7 +3779,7 @@ def test_cleanup_idle_sandboxes_keeps_active_cleanup_and_delegates_warm_expiry(t "active-old": aio_mod.SandboxInfo(sandbox_id="active-old", sandbox_url="http://active-old"), } provider._thread_sandboxes = {("default", "thread-old"): "active-old"} - provider._last_activity = {"active-old": 0.0} + provider._last_activity = {"active-old": time.monotonic() - 10_000} provider._warm_pool = { "warm-old": ( aio_mod.SandboxInfo(sandbox_id="warm-old", sandbox_url="http://warm-old"), diff --git a/backend/tests/test_bench_opensandbox.py b/backend/tests/test_bench_opensandbox.py index ca8c6b57e6..30d4257b26 100644 --- a/backend/tests/test_bench_opensandbox.py +++ b/backend/tests/test_bench_opensandbox.py @@ -4,6 +4,7 @@ from __future__ import annotations import json import threading +import time from datetime import timedelta from types import SimpleNamespace @@ -248,10 +249,12 @@ def test_image_override_and_failed_construction_restore_config(transport): def test_idle_reaping_forces_a_new_remote(transport, monkeypatch, tmp_path): - from deerflow.community import warm_pool_lifecycle - clock = [100.0] - monkeypatch.setattr(warm_pool_lifecycle.time, "time", lambda: clock[0]) + # Warm-pool idle age is written and read on the monotonic clock, so freeze + # that one and let the benchmark's sleep advance it. Freezing `time.time` + # instead leaves both ends of the comparison on the real clock: the entry + # never expires, the reaper removes nothing, and the next turn is a warm hit. + monkeypatch.setattr(time, "monotonic", lambda: clock[0]) monkeypatch.setattr(bench.time, "sleep", lambda seconds: clock.__setitem__(0, clock[0] + seconds)) output = tmp_path / "idle.jsonl" assert ( diff --git a/backend/tests/test_boxlite_provider.py b/backend/tests/test_boxlite_provider.py index 17cf0c077e..9ac1bbc588 100644 --- a/backend/tests/test_boxlite_provider.py +++ b/backend/tests/test_boxlite_provider.py @@ -746,7 +746,7 @@ def test_adopted_warm_pool_box_still_health_checks(monkeypatch): _fake_run, default_env={}, ) - provider._warm_pool["adopted"] = (adopted, time.time()) + provider._warm_pool["adopted"] = (adopted, time.monotonic()) calls = 0 original_execute = adopted.execute_command @@ -1039,8 +1039,8 @@ def test_reset_parked_resources_are_reaped_after_idle_timeout(monkeypatch): provider.reset() - provider._warm_pool[sid_active] = (active_box, time.time() - 9999) - provider._warm_pool[sid_warm] = (warm_box, time.time() - 9999) + provider._warm_pool[sid_active] = (active_box, time.monotonic() - 9999) + provider._warm_pool[sid_warm] = (warm_box, time.monotonic() - 9999) provider._reap_expired_warm(idle_timeout=1) assert provider._warm_pool == {} @@ -1076,7 +1076,7 @@ def test_idle_reaper_destroys_expired_warm_boxes(monkeypatch): # Backdate the warm-pool timestamp so it appears long-expired warm_box = provider._warm_pool[sid][0] - provider._warm_pool[sid] = (warm_box, time.time() - 9999) + provider._warm_pool[sid] = (warm_box, time.monotonic() - 9999) # Wait long enough for the reaper to detect and destroy it time.sleep(0.3) @@ -1118,10 +1118,10 @@ def test_replica_enforcement_evicts_oldest_warm(monkeypatch): # Make sid_a definitely older by backdating its timestamp box_a = provider._warm_pool[sid_a][0] - provider._warm_pool[sid_a] = (box_a, time.time() - 100) + provider._warm_pool[sid_a] = (box_a, time.monotonic() - 100) # Refresh sid_b's timestamp so it's newer box_b = provider._warm_pool[sid_b][0] - provider._warm_pool[sid_b] = (box_b, time.time()) + provider._warm_pool[sid_b] = (box_b, time.monotonic()) # Acquiring a third thread triggers replica enforcement: # warm pool count (2) >= replicas (2) → evict oldest (sid_a) @@ -1507,12 +1507,12 @@ def test_failed_health_check_does_not_remove_swapped_warm_entry(monkeypatch): _fake_run, default_env={}, ) - provider._warm_pool["shared-id"] = (stale, time.time()) + provider._warm_pool["shared-id"] = (stale, time.monotonic()) provider._warm_pool_identity["shared-id"] = ("user-a", "thread-a") def _swap_during_health_check(*args, **kwargs): with provider._lock: - provider._warm_pool["shared-id"] = (replacement, time.time()) + provider._warm_pool["shared-id"] = (replacement, time.monotonic()) provider._warm_pool_identity["shared-id"] = ( "user-b", "thread-b", diff --git a/backend/tests/test_e2b_sandbox_provider.py b/backend/tests/test_e2b_sandbox_provider.py index 1853a63f28..2d00ebdd39 100644 --- a/backend/tests/test_e2b_sandbox_provider.py +++ b/backend/tests/test_e2b_sandbox_provider.py @@ -2457,7 +2457,7 @@ def test_reconcile_honors_wall_clock_budget(monkeypatch): fake_cls.list_return = [_info("sb-never-probed", "u1", "t1")] p._config["reconciliation_max_seconds"] = 0.5 provider_mod = importlib.import_module("deerflow.community.e2b_sandbox.e2b_sandbox_provider") - ticks = iter([0.0, 0.0, 1.0, 1.0]) + ticks = iter([0.0, 0.0, 0.0, 1.0, 1.0]) monkeypatch.setattr(provider_mod.time, "monotonic", lambda: next(ticks)) stats = p._reconcile_remote_sandboxes(now=100.0) @@ -2485,7 +2485,7 @@ def test_reconcile_never_probes_or_adopts_a_warm_pool_sandbox(monkeypatch): p = _make_provider() fake_cls = _install_fake_sdk(monkeypatch, p) fake_cls.list_return = [_info("sb-warm", "u1", "t1")] - p._warm_pool["sb-warm"] = (p._stable_seed("t1", "u1"), time.time()) + p._warm_pool["sb-warm"] = (p._stable_seed("t1", "u1"), time.monotonic()) stats = p._reconcile_remote_sandboxes(now=100.0) @@ -2649,7 +2649,7 @@ def test_reconcile_sweeps_expired_warm_pool_entry(monkeypatch): fake_cls = _install_fake_sdk(monkeypatch, p) store = _install_shared_deployment_capacity(p) fake_cls.list_return = [] - p._warm_pool["sb-old"] = (p._stable_seed("t1", "u1"), time.time() - 601) + p._warm_pool["sb-old"] = (p._stable_seed("t1", "u1"), time.monotonic() - 601) p._owned_sandbox_ids.add("sb-old") p._mount_results["sb-old"] = MagicMock() p._ownership.take("sb-old") @@ -2671,7 +2671,7 @@ def test_reconcile_keeps_unexpired_warm_pool_entry(monkeypatch): p = _make_provider(idle_timeout=600) fake_cls = _install_fake_sdk(monkeypatch, p) fake_cls.list_return = [] - p._warm_pool["sb-fresh"] = (p._stable_seed("t1", "u1"), time.time()) + p._warm_pool["sb-fresh"] = (p._stable_seed("t1", "u1"), time.monotonic()) p._reconcile_remote_sandboxes(now=100.0) @@ -2688,7 +2688,7 @@ def test_reconcile_kills_remote_duplicate_of_a_warm_pool_sandbox(monkeypatch): ] duplicate = FakeClient(sandbox_id="sb-duplicate") fake_cls.connect_factory = lambda sid, **_kw: duplicate if sid == "sb-duplicate" else FakeClient(sandbox_id=sid) - p._warm_pool["sb-warm"] = (p._stable_seed("t1", "u1"), time.time()) + p._warm_pool["sb-warm"] = (p._stable_seed("t1", "u1"), time.monotonic()) p._config["reconciliation_grace_seconds"] = 0.0 stats = p._reconcile_remote_sandboxes(now=100.0) @@ -2711,7 +2711,7 @@ def test_reconcile_adoption_recheck_treats_warm_pool_as_local(monkeypatch): client = FakeClient(sandbox_id="sb-race") def connect_and_park(sid, **_kw): - p._warm_pool[sid] = (p._stable_seed("t1", "u1"), time.time()) + p._warm_pool[sid] = (p._stable_seed("t1", "u1"), time.monotonic()) return client fake_cls.connect_factory = connect_and_park @@ -3483,7 +3483,7 @@ def test_thread_restore_does_not_overwrite_live_sandbox(monkeypatch, tmp_path, r provider._sandboxes[client.sandbox_id] = _make_sandbox(client, sandbox_id=client.sandbox_id) provider._thread_sandboxes[provider._thread_key("t1", "u1")] = client.sandbox_id elif reuse == "warm": - provider._warm_pool[client.sandbox_id] = (provider._stable_seed("t1", "u1"), time.time()) + provider._warm_pool[client.sandbox_id] = (provider._stable_seed("t1", "u1"), time.monotonic()) else: sdk.list_return = [_info(client.sandbox_id, "u1", "t1")] sandbox_id = provider.acquire("t1", user_id="u1") @@ -5077,7 +5077,7 @@ def test_shutdown_during_initial_eviction_reconnect_failure_tracks_vm(monkeypatc """Shutdown must destroy a VM while its first eviction reconnects.""" p = _make_provider() fake_cls = _install_fake_sdk(monkeypatch, p) - p._warm_pool["warm-1"] = ("seed", time.time()) + p._warm_pool["warm-1"] = ("seed", time.monotonic()) reconnect_started = threading.Event() allow_failure = threading.Event() @@ -5117,7 +5117,7 @@ def test_shutdown_during_gone_sandbox_eviction_releases_slot_once(monkeypatch): """ p = _make_provider() fake_cls = _install_fake_sdk(monkeypatch, p) - p._warm_pool["warm-1"] = ("seed", time.time()) + p._warm_pool["warm-1"] = ("seed", time.monotonic()) reconnect_started = threading.Event() allow_failure = threading.Event() @@ -6569,7 +6569,7 @@ def test_warm_sweep_cannot_delete_concurrent_discovery_lease(monkeypatch): provider = _make_provider(idle_timeout=30) sdk = _install_fake_sdk(monkeypatch, provider) sdk.list_return = [_info("sb-race", "u1", "t1")] - provider._warm_pool["sb-race"] = (provider._stable_seed("t1", "u1"), time.time() - 31) + provider._warm_pool["sb-race"] = (provider._stable_seed("t1", "u1"), time.monotonic() - 31) provider._ownership.take("sb-race") provider._owned_sandbox_ids.add("sb-race") cleanup_started, finish_cleanup, acquire_reached = threading.Event(), threading.Event(), threading.Event() @@ -6615,7 +6615,7 @@ def test_warm_sweep_cannot_delete_concurrent_discovery_lease(monkeypatch): def test_warm_sweep_serializes_with_lapsed_lease_renewal(monkeypatch): provider = _make_provider(idle_timeout=30) - provider._warm_pool["sb-race"] = (provider._stable_seed("t1", "u1"), time.time() - 31) + provider._warm_pool["sb-race"] = (provider._stable_seed("t1", "u1"), time.monotonic() - 31) provider._owned_sandbox_ids.add("sb-race") renewal_started, finish_renewal, cleanup_reached = threading.Event(), threading.Event(), threading.Event() renew = provider._ownership.renew @@ -6737,7 +6737,7 @@ def test_reconcile_does_not_readopt_a_release_in_progress(monkeypatch): def test_warm_sweep_leaves_shared_capacity_to_inventory(monkeypatch, inventory_complete): provider = _make_provider(idle_timeout=30) store = _install_shared_deployment_capacity(provider) - provider._warm_pool["sb-old"] = (provider._stable_seed("t1", "u1"), time.time() - 31) + provider._warm_pool["sb-old"] = (provider._stable_seed("t1", "u1"), time.monotonic() - 31) monkeypatch.setattr(provider, "_list_remote_entries", lambda _metadata: ([], False, inventory_complete)) provider._reconcile_remote_sandboxes() @@ -6825,7 +6825,7 @@ def test_warm_sweep_cannot_release_reacquired_ownership(monkeypatch, acquire_pat provider = _make_provider(idle_timeout=30) sdk = _install_fake_sdk(monkeypatch, provider) sdk.list_return = [_info("sb-race", "u1", "t1")] - provider._warm_pool["sb-race"] = (provider._stable_seed("t1", "u1"), time.time() - 31) + provider._warm_pool["sb-race"] = (provider._stable_seed("t1", "u1"), time.monotonic() - 31) provider._owned_sandbox_ids.add("sb-race") provider._ownership.take("sb-race") cleaning = threading.Event() @@ -6876,7 +6876,7 @@ def test_warm_sweep_cannot_release_reacquired_ownership(monkeypatch, acquire_pat def test_warm_sweep_rechecks_snapshot_after_lifecycle_wait(monkeypatch, transition): provider = _make_provider(idle_timeout=30) sid = "sb-stale-snapshot" - old_entry = (provider._stable_seed("t1", "u1"), time.time() - 31) + old_entry = (provider._stable_seed("t1", "u1"), time.monotonic() - 31) provider._warm_pool[sid] = old_entry provider._owned_sandbox_ids.add(sid) provider._ownership.take(sid) @@ -6893,7 +6893,7 @@ def test_warm_sweep_rechecks_snapshot_after_lifecycle_wait(monkeypatch, transiti # The sweep has already captured old_entry. A competing operation # completes before it acquires the lifecycle lock. if transition == "repark": - provider._warm_pool[sid] = (old_entry[0], time.time()) + provider._warm_pool[sid] = (old_entry[0], time.monotonic()) else: provider._publish_ownership(sid) with lifecycle(sandbox_id): @@ -6980,7 +6980,7 @@ def test_warm_sweep_cannot_race_lapsed_lease_reclaim(monkeypatch): provider = _make_provider(idle_timeout=30) sid = "sb-lapsed" - provider._warm_pool[sid] = (provider._stable_seed("t1", "u1"), time.time() - 31) + provider._warm_pool[sid] = (provider._stable_seed("t1", "u1"), time.monotonic() - 31) provider._owned_sandbox_ids.add(sid) renewing = threading.Event() finish_renewal = threading.Event() diff --git a/backend/tests/test_opensandbox_provider.py b/backend/tests/test_opensandbox_provider.py index de85ef32c0..c387b128bd 100644 --- a/backend/tests/test_opensandbox_provider.py +++ b/backend/tests/test_opensandbox_provider.py @@ -762,7 +762,10 @@ def test_output_sync_is_bounded(output_sync_provider, monkeypatch, caplog, limit monkeypatch.setattr(provider, "_MAX_SYNC_LISTING_BYTES", 4) else: monkeypatch.setattr(provider, "_SYNC_DEADLINE_SECONDS", 1) - monkeypatch.setattr("deerflow.community.opensandbox.provider.time", types.SimpleNamespace(monotonic=iter([0, 2]).__next__, time=time.time)) + monkeypatch.setattr( + "deerflow.community.opensandbox.provider.time", + types.SimpleNamespace(monotonic=iter([0, 2, 2]).__next__, time=time.time), + ) provider.release(sandbox_id) if limit in {"files", "bytes"}: assert (outputs / "first.txt").read_bytes() == b"123" @@ -1071,6 +1074,27 @@ def test_reset_parks_active_and_shutdown_destroys_active_and_warm(monkeypatch: p assert provider._sandboxes == {} and provider._warm_pool == {} +def test_reset_parks_with_monotonic_timestamp_and_reaper_expires_it(monkeypatch: pytest.MonkeyPatch) -> None: + provider, sdk = _install(monkeypatch) + sandbox_id = provider.acquire("active", user_id="user") + remote = sdk.remotes[0] + monkeypatch.setattr(time, "monotonic", lambda: 100.0) + monkeypatch.setattr(time, "time", lambda: 1_000_000.0) + + provider.reset() + assert provider._warm_pool[sandbox_id][1] == 100.0 + + # This reset path was omitted by #6996. A backwards wall-clock step must not + # prevent the shared reaper from observing ten elapsed monotonic seconds. + monkeypatch.setattr(time, "monotonic", lambda: 110.0) + monkeypatch.setattr(time, "time", lambda: -1_000_000.0) + provider._reap_expired_warm(idle_timeout=5) + + assert sandbox_id not in provider._warm_pool + assert remote.destroy_calls == 1 + provider.shutdown() + + def test_shutdown_stops_idle_reaper(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(OpenSandboxProvider, "IDLE_CHECK_INTERVAL", 0.01) provider, _ = _install(monkeypatch, config={"idle_timeout": 60}) diff --git a/backend/tests/test_sandbox_orphan_reconciliation.py b/backend/tests/test_sandbox_orphan_reconciliation.py index d2f280b95f..780c07f3ef 100644 --- a/backend/tests/test_sandbox_orphan_reconciliation.py +++ b/backend/tests/test_sandbox_orphan_reconciliation.py @@ -745,12 +745,12 @@ def test_idle_reap_does_not_destroy_peer_owned_warm_entry(): worker_a = _make_provider_for_reconciliation(worker_id="worker-a", store=shared) worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared) worker_b._config["idle_timeout"] = 60 - now = time.time() + now = time.monotonic() info = SandboxInfo( sandbox_id="a99c8444", sandbox_url="http://localhost:8080", container_name="deer-flow-sandbox-a99c8444", - created_at=now - 50, + created_at=time.time() - 50, ) # Simulate the bad old path: B already has it in warm (or adopted wrongly). worker_b._warm_pool["a99c8444"] = (info, now - 61) @@ -791,7 +791,7 @@ def test_multi_worker_release_then_peer_reconcile_cannot_kill(): worker_a._backend = backend worker_a._config["idle_timeout"] = 60 # A released to warm and holds the lease. - worker_a._warm_pool[sid] = (info, time.time()) + worker_a._warm_pool[sid] = (info, time.monotonic()) worker_a._publish_ownership(sid) worker_b = _make_provider_for_reconciliation(worker_id="worker-b", store=shared) @@ -801,7 +801,7 @@ def test_multi_worker_release_then_peer_reconcile_cannot_kill(): assert sid not in worker_b._warm_pool # Even if B somehow had it warm, reap must refuse. - worker_b._warm_pool[sid] = (info, time.time() - 61) + worker_b._warm_pool[sid] = (info, time.monotonic() - 61) worker_b._reap_expired_warm(idle_timeout=60) assert sid not in destroyed assert sid in running @@ -1013,7 +1013,7 @@ def test_renewal_keeps_the_sandbox_when_the_store_cannot_answer(): worker._sandboxes["live02"] = sandbox worker._sandbox_infos["live02"] = info worker._thread_sandboxes[("u1", "t1")] = "live02" - worker._warm_pool["warm02"] = (info, time.time()) + worker._warm_pool["warm02"] = (info, time.monotonic()) worker._renew_owned_leases() @@ -1161,7 +1161,7 @@ def test_renewal_covers_warm_entries_not_just_active(): created_at=time.time(), ) worker._sandboxes["active01"] = MagicMock() - worker._warm_pool["warm01"] = (info, time.time()) + worker._warm_pool["warm01"] = (info, time.monotonic()) worker._publish_ownership("active01") worker._publish_ownership("warm01") @@ -1189,7 +1189,7 @@ def test_renewal_does_not_forget_a_warm_entry_mid_teardown(): container_name="deer-flow-sandbox-warm-stop", created_at=time.time(), ) - worker._warm_pool["warm-stop"] = (info, time.time()) + worker._warm_pool["warm-stop"] = (info, time.monotonic()) worker._publish_ownership("warm-stop") stop_started = threading.Event() @@ -1250,7 +1250,7 @@ def test_lost_lease_drops_sandbox_without_destroying_container(): worker_a._sandboxes["moved01"] = sandbox worker_a._sandbox_infos["moved01"] = info worker_a._thread_sandboxes[("u1", "t1")] = "moved01" - worker_a._last_activity["moved01"] = time.time() + worker_a._last_activity["moved01"] = time.monotonic() worker_a._publish_ownership("moved01") # The thread's next turn routes to B, which takes over ownership. @@ -1583,7 +1583,7 @@ def test_teardown_marker_is_held_for_a_stop_that_outlives_the_lease_ttl(): release_stop.wait(timeout=5) worker_a._backend.destroy = MagicMock(side_effect=slow_destroy) - worker_a._warm_pool["doomed1"] = (info, time.time()) + worker_a._warm_pool["doomed1"] = (info, time.monotonic()) reaper = threading.Thread( target=lambda: worker_a._destroy_warm_entry("doomed1", info, reason="idle_timeout", still_reapable=lambda: True), @@ -1857,7 +1857,7 @@ def test_teardown_release_waits_for_the_heartbeat_to_exit(): assert blocking.heartbeat_blocked.wait(timeout=5), "the heartbeat never issued a refresh claim" worker_a._backend.destroy = MagicMock(side_effect=slow_destroy) - worker_a._warm_pool["defer1"] = (info, time.time()) + worker_a._warm_pool["defer1"] = (info, time.monotonic()) reaper = threading.Thread( target=lambda: worker_a._destroy_warm_entry("defer1", info, reason="idle_timeout", still_reapable=lambda: True), @@ -1900,7 +1900,7 @@ def test_evict_keeps_the_warm_entry_when_the_claim_is_refused(): container_name="deer-flow-sandbox-peer01", created_at=time.time(), ) - worker_a._warm_pool["peer01"] = (info, time.time() - 5) + worker_a._warm_pool["peer01"] = (info, time.monotonic() - 5) # A peer owns it, so A's eviction claim is refused. worker_b._publish_ownership("peer01") @@ -1929,7 +1929,7 @@ def test_reclaim_drops_a_container_a_peer_is_destroying(): container_name="deer-flow-sandbox-dying02", created_at=time.time(), ) - worker_a._warm_pool["dying02"] = (info, time.time()) + worker_a._warm_pool["dying02"] = (info, time.monotonic()) worker_a._check_tracked_sandbox_alive = MagicMock(return_value=True) # B's reaper marks the teardown; its stop is in flight. @@ -1987,9 +1987,9 @@ def test_shutdown_does_not_stop_a_peers_warm_container(): mine = SandboxInfo(sandbox_id="mine01", sandbox_url="http://localhost:8080", container_name="c-mine01", created_at=time.time()) theirs = SandboxInfo(sandbox_id="peer02", sandbox_url="http://localhost:8081", container_name="c-peer02", created_at=time.time()) - worker_a._warm_pool["mine01"] = (mine, time.time()) + worker_a._warm_pool["mine01"] = (mine, time.monotonic()) worker_a._publish_ownership("mine01") - worker_a._warm_pool["peer02"] = (theirs, time.time()) + worker_a._warm_pool["peer02"] = (theirs, time.monotonic()) worker_b._publish_ownership("peer02") # a live peer owns this one worker_a.shutdown() @@ -2018,7 +2018,7 @@ def test_teardown_heartbeat_stops_when_the_stop_returns(): container_name="deer-flow-sandbox-doomed2", created_at=time.time(), ) - worker_a._warm_pool["doomed2"] = (info, time.time()) + worker_a._warm_pool["doomed2"] = (info, time.monotonic()) assert worker_a._destroy_warm_entry("doomed2", info, reason="idle_timeout", still_reapable=lambda: True) is True @@ -2097,7 +2097,7 @@ def test_refused_idle_destroy_keeps_the_warm_entry(): container_name="deer-flow-sandbox-warmpeer", created_at=time.time(), ) - worker_a._warm_pool["warmpeer"] = (info, time.time() - 999) + worker_a._warm_pool["warmpeer"] = (info, time.monotonic() - 999) worker_b._publish_ownership("warmpeer") worker_a._reap_expired_warm(idle_timeout=60) @@ -2247,7 +2247,7 @@ def _active_sandbox(provider, sandbox_id, info, *, thread_key=("u", "t1")): provider._sandboxes[sandbox_id] = AioSandbox(id=sandbox_id, base_url=info.sandbox_url) provider._sandbox_infos[sandbox_id] = info - provider._last_activity[sandbox_id] = time.time() + provider._last_activity[sandbox_id] = time.monotonic() provider._thread_sandboxes[thread_key] = sandbox_id provider._backend.is_alive.return_value = True @@ -2327,7 +2327,7 @@ def test_warm_reaper_does_not_stop_a_container_this_instance_just_reclaimed(reas provider._ownership = gate info = _info("warm1") # Old enough for the idle reaper; the only entry, so also the eviction pick. - provider._warm_pool["warm1"] = (info, time.time() - 10_000) + provider._warm_pool["warm1"] = (info, time.monotonic() - 10_000) provider._ownership._inner.take("warm1") reap = provider._evict_oldest_warm if reap_name == "_evict_oldest_warm" else lambda: provider._reap_expired_warm(600) @@ -2359,7 +2359,7 @@ def test_idle_destroy_does_not_stop_a_container_this_instance_just_reused(): provider._ownership = gate info = _info("idle1") _active_sandbox(provider, "idle1", info) - provider._last_activity["idle1"] = 0.0 # long idle + provider._last_activity["idle1"] = time.monotonic() - 10_000 # long idle provider._ownership._inner.take("idle1") handed_out = _run_reap_vs_promote( @@ -2386,7 +2386,7 @@ def test_reuse_does_not_return_a_sandbox_reserved_during_its_health_check(): provider._ownership = gate info = _info("idlerace") _active_sandbox(provider, "idlerace", info) - provider._last_activity["idlerace"] = 0.0 + provider._last_activity["idlerace"] = time.monotonic() - 10_000 provider._ownership._inner.take("idlerace") health_started, let_health_finish = threading.Event(), threading.Event() @@ -2622,14 +2622,14 @@ def test_idle_destroy_skips_a_sandbox_re_acquired_after_the_idle_snapshot(): provider = _make_provider_for_reconciliation() info = _info("idle2") _active_sandbox(provider, "idle2", info) - provider._last_activity["idle2"] = 0.0 + provider._last_activity["idle2"] = time.monotonic() - 10_000 provider._ownership.take("idle2") real_destroy_tracked = provider._destroy_tracked def turn_lands_first(sandbox_id, **kwargs): # A turn arrives between the idle snapshot and the reservation. - provider._last_activity[sandbox_id] = time.time() + provider._last_activity[sandbox_id] = time.monotonic() return real_destroy_tracked(sandbox_id, **kwargs) provider._destroy_tracked = turn_lands_first @@ -2829,7 +2829,7 @@ def test_teardown_reservation_is_cleared_once_the_stop_returns(): """ provider = _make_provider_for_reconciliation() info = _info("cleared") - provider._warm_pool["cleared"] = (info, time.time() - 10_000) + provider._warm_pool["cleared"] = (info, time.monotonic() - 10_000) provider._ownership.take("cleared") assert provider._destroy_warm_entry("cleared", info, reason="idle_timeout", still_reapable=lambda: True) is True @@ -2849,7 +2849,7 @@ def test_reclaim_does_not_hand_out_a_container_a_reaper_is_stopping(): """ provider = _make_provider_for_reconciliation() info = _info("handout") - provider._warm_pool["handout"] = (info, time.time() - 10_000) + provider._warm_pool["handout"] = (info, time.monotonic() - 10_000) provider._ownership.take("handout") # Park reclaim between its `take()` and its install. @@ -2899,7 +2899,7 @@ def test_warm_entry_is_removed_before_its_teardown_reservation_is_released(): """ provider = _make_provider_for_reconciliation() info = _info("ordered") - provider._warm_pool["ordered"] = (info, time.time() - 10_000) + provider._warm_pool["ordered"] = (info, time.monotonic() - 10_000) provider._ownership.take("ordered") observed = {} @@ -2927,7 +2927,7 @@ def test_reclaim_short_circuits_a_reserved_entry_without_touching_the_backend(): """ provider = _make_provider_for_reconciliation() info = _info("shortcut") - provider._warm_pool["shortcut"] = (info, time.time()) + provider._warm_pool["shortcut"] = (info, time.monotonic()) provider._ownership = MagicMock(wraps=provider._ownership) with provider._lock: provider._local_teardown.add("shortcut") @@ -2977,7 +2977,7 @@ def test_registering_a_sandbox_clears_any_stale_warm_entry(register): """ provider = _make_provider_for_reconciliation() info = _info("stale") - provider._warm_pool["stale"] = (info, time.time()) + provider._warm_pool["stale"] = (info, time.monotonic()) if register == "discovered": provider._register_discovered_sandbox("t1", info, user_id="u") diff --git a/backend/tests/test_tenki_provider.py b/backend/tests/test_tenki_provider.py index 54e96b6d8a..514ff1ab6a 100644 --- a/backend/tests/test_tenki_provider.py +++ b/backend/tests/test_tenki_provider.py @@ -1112,6 +1112,29 @@ def test_reset_parks_running_for_later_cleanup(monkeypatch): assert active_box.is_closed +def test_reset_parks_running_with_monotonic_timestamp_for_idle_cleanup(monkeypatch): + client = _FakeClient() + provider = _install(monkeypatch, client=client) + sid_active = provider.acquire("thread-active", user_id="u1") + active_box = provider._sandboxes[sid_active] + monkeypatch.setattr(time, "monotonic", lambda: 100.0) + monkeypatch.setattr(time, "time", lambda: 1_000_000.0) + + provider.reset() + assert provider._sandboxes == {} + assert provider._warm_pool[sid_active] == (active_box, 100.0) + + # Ten elapsed monotonic seconds exceed the configured timeout even though + # the wall clock moves backwards. This is the reset writer missed by #6996. + monkeypatch.setattr(time, "monotonic", lambda: 110.0) + monkeypatch.setattr(time, "time", lambda: -1_000_000.0) + provider._reap_expired_warm(idle_timeout=5) + + assert sid_active not in provider._warm_pool + assert active_box.is_closed + provider.shutdown() + + def test_idle_reaper_destroys_expired_warm(monkeypatch): client = _FakeClient() monkeypatch.setattr(TenkiSandboxProvider, "IDLE_CHECK_INTERVAL", 0.1) @@ -1119,7 +1142,7 @@ def test_idle_reaper_destroys_expired_warm(monkeypatch): sid = provider.acquire("thread-1", user_id="u1") provider.release(sid) warm_box = provider._warm_pool[sid][0] - provider._warm_pool[sid] = (warm_box, time.time() - 9999) + provider._warm_pool[sid] = (warm_box, time.monotonic() - 9999) time.sleep(0.3) assert sid not in provider._warm_pool assert warm_box.is_closed @@ -1134,8 +1157,8 @@ def test_replica_enforcement_evicts_oldest_warm(monkeypatch): sid_b = provider.acquire("thread-b", user_id="u1") provider.release(sid_b) box_a = provider._warm_pool[sid_a][0] - provider._warm_pool[sid_a] = (box_a, time.time() - 100) - provider._warm_pool[sid_b] = (provider._warm_pool[sid_b][0], time.time()) + provider._warm_pool[sid_a] = (box_a, time.monotonic() - 100) + provider._warm_pool[sid_b] = (provider._warm_pool[sid_b][0], time.monotonic()) provider.acquire("thread-c", user_id="u1") assert sid_a not in provider._warm_pool assert box_a.is_closed diff --git a/backend/tests/test_warm_pool_lifecycle.py b/backend/tests/test_warm_pool_lifecycle.py index 30f9e91d47..e0e2d53b3e 100644 --- a/backend/tests/test_warm_pool_lifecycle.py +++ b/backend/tests/test_warm_pool_lifecycle.py @@ -34,7 +34,7 @@ class _Provider(WarmPoolLifecycleMixin[str]): def test_replica_count_includes_active_and_warm_entries() -> None: provider = _Provider(replicas=2, active_count=1) - provider._warm_pool["warm-1"] = ("entry-1", time.time()) + provider._warm_pool["warm-1"] = ("entry-1", time.monotonic()) assert provider._replica_count() == (2, 2) @@ -61,7 +61,7 @@ def test_evict_oldest_warm_returns_none_when_pool_empty() -> None: def test_reap_expired_warm_destroys_only_expired_entries() -> None: provider = _Provider() - now = time.time() + now = time.monotonic() provider._warm_pool["expired"] = ("entry-expired", now - 100) provider._warm_pool["fresh"] = ("entry-fresh", now) @@ -74,7 +74,7 @@ def test_reap_expired_warm_destroys_only_expired_entries() -> None: def test_reap_expired_warm_noops_when_timeout_disabled() -> None: provider = _Provider(idle_timeout=0) - provider._warm_pool["expired"] = ("entry-expired", time.time() - 100) + provider._warm_pool["expired"] = ("entry-expired", time.monotonic() - 100) provider._reap_expired_warm(idle_timeout=0) @@ -123,7 +123,7 @@ def test_provider_shutdown_resets_guard_when_idle_checker_stop_fails(module_name def test_start_idle_checker_uses_monkeypatchable_interval(monkeypatch) -> None: provider = _Provider(idle_timeout=0.01) monkeypatch.setattr(_Provider, "IDLE_CHECK_INTERVAL", 0.01) - provider._warm_pool["expired"] = ("entry-expired", time.time() - 10) + provider._warm_pool["expired"] = ("entry-expired", time.monotonic() - 10) provider._start_idle_checker() deadline = time.time() + 1 @@ -135,3 +135,31 @@ def test_start_idle_checker_uses_monkeypatchable_interval(monkeypatch) -> None: assert provider.destroyed == [("expired", "entry-expired", "idle_timeout")] assert provider._idle_checker_thread is not None assert not provider._idle_checker_thread.is_alive() + + +@pytest.mark.parametrize("wall_step", [10_000.0, -10_000.0]) +def test_reap_expired_warm_ignores_wall_clock_steps(monkeypatch, wall_step: float) -> None: + """NTP/VM wall-clock jumps cannot change a warm entry's monotonic idle age.""" + provider = _Provider() + parked_at = time.monotonic() - 1 + provider._warm_pool["fresh"] = ("entry-fresh", parked_at) + wall_now = time.time() + monkeypatch.setattr(time, "time", lambda: wall_now + wall_step) + + provider._reap_expired_warm(idle_timeout=600) + + assert provider._warm_pool == {"fresh": ("entry-fresh", parked_at)} + assert provider.destroyed == [] + + +def test_reap_expired_warm_uses_elapsed_monotonic_time_when_wall_clock_moves_backward(monkeypatch) -> None: + """A backward wall step cannot preserve an entry whose real idle age expired.""" + provider = _Provider() + provider._warm_pool["expired"] = ("entry-expired", time.monotonic() - 601) + wall_now = time.time() + monkeypatch.setattr(time, "time", lambda: wall_now - 10_000) + + provider._reap_expired_warm(idle_timeout=600) + + assert provider._warm_pool == {} + assert provider.destroyed == [("expired", "entry-expired", "idle_timeout")]