diff --git a/api/models.py b/api/models.py index 6e6a9530..65434ce8 100644 --- a/api/models.py +++ b/api/models.py @@ -49,7 +49,6 @@ _CLI_SESSIONS_CACHE_INFLIGHT: "dict[tuple, threading.Event]" = {} _CLI_SESSIONS_CACHE_INVALIDATION_VERSION = 0 _CLI_SESSIONS_CACHE = {} # Event waits that keep stale rows visible while a rebuild is in flight. -_CLI_SESSIONS_CACHE_WAIT_SECONDS = 0.25 _CLI_SESSIONS_CACHE_STALE_WAIT_SECONDS = 0.10 # Per-file parse cache for Claude Code JSONL transcripts (#4718/#4662 phase 4). @@ -4213,7 +4212,6 @@ def clear_cli_sessions_cache() -> None: global _CLI_SESSIONS_CACHE_INVALIDATION_VERSION _CLI_SESSIONS_CACHE_INVALIDATION_VERSION += 1 _CLI_SESSIONS_CACHE.clear() - _CLI_SESSIONS_CACHE_INFLIGHT.clear() # The sidecar-metadata projection cache is stat-keyed (self-invalidating on # any file change), but clear it alongside the CLI cache so an explicit # reset — a mutating sidebar action or test isolation — starts fully cold. @@ -4249,6 +4247,61 @@ def _cli_sessions_cache_done(cache_key: tuple, event: threading.Event | None) -> event.set() +def _reload_cli_sessions_after_inflight( + *, + cache_key: tuple, + ttl: float, + stale_sessions, + stale_stamp, + load_sessions, + all_profiles: bool, + db_path: str, +) -> list: + event, is_owner = _cli_sessions_cache_claim_rebuild(cache_key) + if not is_owner: + try: + event.wait() + except Exception: + pass + with _CLI_SESSIONS_CACHE_LOCK: + cached_entry = _CLI_SESSIONS_CACHE.get(cache_key) + if cached_entry is not None: + if len(cached_entry) == 3: + cached_expires_at, cached_stamp, cached_sessions = cached_entry + else: + cached_expires_at, cached_sessions = cached_entry + cached_stamp = _CLI_SESSIONS_CACHE_INVALIDATION_VERSION + if ( + cached_stamp == _CLI_SESSIONS_CACHE_INVALIDATION_VERSION + and cached_expires_at > time.monotonic() + ): + return _copy_cli_sessions(cached_sessions) + if stale_sessions is not None and stale_stamp == _cli_sessions_cache_invalidation_stamp(): + return stale_sessions + return [] + try: + try: + sessions = load_sessions() + except Exception as _cli_err: + logger.warning( + "get_cli_sessions() failed — check state.db schema or path (%s): %s", + "all profiles" if all_profiles else db_path, _cli_err, + ) + if stale_sessions is not None and stale_stamp == _cli_sessions_cache_invalidation_stamp(): + return stale_sessions + return [] + if _cli_sessions_cache_invalidation_stamp() == fallback_invalidation_stamp: + with _CLI_SESSIONS_CACHE_LOCK: + _CLI_SESSIONS_CACHE[cache_key] = ( + time.monotonic() + ttl, + _CLI_SESSIONS_CACHE_INVALIDATION_VERSION, + _copy_cli_sessions(sessions), + ) + return _copy_cli_sessions(sessions) + finally: + _cli_sessions_cache_done(cache_key, event) + + def _cli_sessions_cache_ttl_seconds() -> float: # #4842: widen the freshness window while a turn is streaming so the fixed # ~5s streaming poll cadence doesn't force a rebuild on every poll. Paired @@ -4876,12 +4929,11 @@ def get_cli_sessions(source_filter=None, *, all_profiles: bool = False) -> list: finally: _cli_sessions_cache_done(cache_key, event) try: - timeout = ( + event.wait( _CLI_SESSIONS_CACHE_STALE_WAIT_SECONDS if stale_sessions is not None - else _CLI_SESSIONS_CACHE_WAIT_SECONDS + else None ) - event.wait(timeout) except Exception: pass with _CLI_SESSIONS_CACHE_LOCK: @@ -4899,6 +4951,15 @@ def get_cli_sessions(source_filter=None, *, all_profiles: bool = False) -> list: return _copy_cli_sessions(cached_sessions) if stale_sessions is not None and stale_stamp == _cli_sessions_cache_invalidation_stamp(): return stale_sessions + return _reload_cli_sessions_after_inflight( + cache_key=cache_key, + ttl=ttl, + stale_sessions=stale_sessions, + stale_stamp=stale_stamp, + load_sessions=_load_sessions, + all_profiles=all_profiles, + db_path=db_path, + ) fallback_invalidation_stamp = _cli_sessions_cache_invalidation_stamp() try: sessions = _load_sessions() diff --git a/tests/test_issue4842_cli_cache_streaming.py b/tests/test_issue4842_cli_cache_streaming.py index 6352a345..54a0a47c 100644 --- a/tests/test_issue4842_cli_cache_streaming.py +++ b/tests/test_issue4842_cli_cache_streaming.py @@ -100,6 +100,59 @@ def test_get_cli_sessions_follower_reuses_stale_rows_during_slow_rebuild(monkeyp assert results.get("owner") == [{"session_id": "fresh", "title": "fresh-row"}] +def test_get_cli_sessions_cold_followers_join_single_rebuild(monkeypatch, tmp_path): + hermes_home = tmp_path / "hermes" + hermes_home.mkdir() + monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: str(hermes_home)) + monkeypatch.setattr(profiles, "get_active_profile_name", lambda: "default") + models.clear_cli_sessions_cache() + monkeypatch.setattr(models, "_CLI_SESSIONS_CACHE_TTL_SECONDS", 60.0, raising=False) + + owner_started = threading.Event() + owner_block = threading.Event() + load_count = {"value": 0} + load_count_lock = threading.Lock() + results = {} + + def _blocking_loader(*_args, **_kwargs): + with load_count_lock: + load_count["value"] += 1 + call_number = load_count["value"] + if call_number == 1: + owner_started.set() + owner_block.wait() + return [{"session_id": "fresh", "title": "fresh-row"}] + + monkeypatch.setattr(models, "_load_cli_sessions_uncached", _blocking_loader) + + def _owner(): + results["owner"] = models.get_cli_sessions() + + def _follower(): + results["follower"] = models.get_cli_sessions() + + owner = threading.Thread(target=_owner, daemon=True) + follower = threading.Thread(target=_follower, daemon=True) + + owner.start() + assert owner_started.wait(1.0), "owner did not start" + follower.start() + time.sleep(0.2) + + assert load_count["value"] == 1 + assert follower.is_alive() + + owner_block.set() + owner.join(2.0) + follower.join(2.0) + + assert not owner.is_alive() + assert not follower.is_alive() + assert load_count["value"] == 1 + assert results.get("owner") == [{"session_id": "fresh", "title": "fresh-row"}] + assert results.get("follower") == [{"session_id": "fresh", "title": "fresh-row"}] + + def test_get_cli_sessions_clear_during_rebuild_does_not_restore_stale_rows(monkeypatch, tmp_path): hermes_home = tmp_path / "hermes" hermes_home.mkdir() @@ -142,3 +195,74 @@ def test_get_cli_sessions_clear_during_rebuild_does_not_restore_stale_rows(monke assert recovered == [{"session_id": "recovered", "title": "recovered-row"}] with models._CLI_SESSIONS_CACHE_LOCK: assert cache_key in models._CLI_SESSIONS_CACHE + + +def test_get_cli_sessions_clear_during_rebuild_preserves_joiners(monkeypatch, tmp_path): + hermes_home = tmp_path / "hermes" + hermes_home.mkdir() + monkeypatch.setattr(profiles, "get_active_hermes_home", lambda: str(hermes_home)) + monkeypatch.setattr(profiles, "get_active_profile_name", lambda: "default") + models.clear_cli_sessions_cache() + monkeypatch.setattr(models, "_CLI_SESSIONS_CACHE_TTL_SECONDS", 60.0, raising=False) + + owner_started = threading.Event() + owner_block = threading.Event() + second_started = threading.Event() + second_block = threading.Event() + load_count = {"value": 0} + load_count_lock = threading.Lock() + results = {} + + def _blocking_loader(*_args, **_kwargs): + with load_count_lock: + load_count["value"] += 1 + call_number = load_count["value"] + if call_number == 1: + owner_started.set() + owner_block.wait() + return [{"session_id": "owner", "title": "owner-row"}] + second_started.set() + second_block.wait() + return [{"session_id": "fresh", "title": "fresh-row"}] + + monkeypatch.setattr(models, "_load_cli_sessions_uncached", _blocking_loader) + + def _owner(): + results["owner"] = models.get_cli_sessions() + + def _follower(name): + results[name] = models.get_cli_sessions() + + owner = threading.Thread(target=_owner, daemon=True) + follower_a = threading.Thread(target=lambda: _follower("follower_a"), daemon=True) + follower_b = threading.Thread(target=lambda: _follower("follower_b"), daemon=True) + + owner.start() + assert owner_started.wait(1.0), "owner did not enter rebuild" + follower_a.start() + models.clear_cli_sessions_cache() + follower_b.start() + time.sleep(0.2) + + assert load_count["value"] == 1 + assert follower_a.is_alive() + assert follower_b.is_alive() + + owner_block.set() + assert second_started.wait(1.0), "second rebuild did not start" + time.sleep(0.2) + + assert load_count["value"] == 2 + second_block.set() + + owner.join(2.0) + follower_a.join(2.0) + follower_b.join(2.0) + + assert not owner.is_alive() + assert not follower_a.is_alive() + assert not follower_b.is_alive() + assert load_count["value"] == 2 + assert results.get("owner") == [{"session_id": "owner", "title": "owner-row"}] + assert results.get("follower_a") == [{"session_id": "fresh", "title": "fresh-row"}] + assert results.get("follower_b") == [{"session_id": "fresh", "title": "fresh-row"}]