diff --git a/CHANGELOG.md b/CHANGELOG.md index 35247aff4..a1fa75036 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,17 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). ## [Unreleased] +### Fixed + +- **Thread-safe catalog cache in TUI reference parser** (#7590): Added a + `threading.Lock` (`_catalog_lock`) to protect all read and write access to + the module-level `_catalog_cache` dict in + `src/cleveragents/tui/input/reference_parser.py`. Previously, concurrent + calls to `_catalog()` from multiple TUI input threads could race on the + shared dict, causing partial writes or a `RuntimeError` if the dict was + iterated while being modified. All cache reads and writes are now performed + atomically inside a `with _catalog_lock:` block. + ### Added - **TDD Issue-Capture Test Activation** (#7025): Replaced 234 bare `@skip` tags diff --git a/features/steps/tdd_reference_parser_catalog_lock_steps.py b/features/steps/tdd_reference_parser_catalog_lock_steps.py new file mode 100644 index 000000000..15af65f06 --- /dev/null +++ b/features/steps/tdd_reference_parser_catalog_lock_steps.py @@ -0,0 +1,223 @@ +"""Step definitions for tdd_reference_parser_catalog_lock.feature. + +These steps verify that _catalog_cache in reference_parser.py is protected +by a threading.Lock, preventing race conditions when multiple TUI input +threads call _catalog() concurrently. + +Issue: #7590 — _catalog_cache global dict has unsynchronized write +""" + +from __future__ import annotations + +import threading +import time +from pathlib import Path +from typing import Any +from unittest.mock import patch + +from behave import given, then, when + +import cleveragents.tui.input.reference_parser as rp_module +from cleveragents.tui.input.reference_parser import ( + _catalog, + _catalog_cache, +) + + +# --------------------------------------------------------------------------- +# Background +# --------------------------------------------------------------------------- +@given("the reference parser module is loaded for concurrency tests") +def step_load_reference_parser(context: Any) -> None: + """Verify the module is importable and exposes the expected symbols.""" + assert _catalog is not None + assert _catalog_cache is not None + assert hasattr(rp_module, "_catalog_lock"), ( + "_catalog_lock is missing from reference_parser module" + ) + + +# --------------------------------------------------------------------------- +# Given steps +# --------------------------------------------------------------------------- +@given("the catalog cache is cleared for concurrency testing") +def step_clear_catalog_cache(context: Any) -> None: + """Reset the module-level catalog cache so _catalog() rebuilds on next call. + + Also installs a fast os.walk mock so the filesystem is never traversed + during concurrent tests — keeping the test suite fast and deterministic. + """ + original: dict[str, object] = { + "cwd": _catalog_cache.get("cwd"), + "created_at": _catalog_cache.get("created_at"), + "catalog": _catalog_cache.get("catalog"), + } + _catalog_cache["cwd"] = None + _catalog_cache["created_at"] = 0.0 + _catalog_cache["catalog"] = None + + # Mock os.walk to return an empty listing so _catalog() completes instantly. + context._walk_patch = patch.object(rp_module.os, "walk", return_value=iter([])) + context._walk_patch.start() + + # Mock Path.is_dir so actor/tool/skill glob paths are skipped. + context._is_dir_patch = patch.object(Path, "is_dir", return_value=False) + context._is_dir_patch.start() + + def _restore() -> None: + context._walk_patch.stop() + context._is_dir_patch.stop() + _catalog_cache["cwd"] = original["cwd"] + _catalog_cache["created_at"] = original["created_at"] + _catalog_cache["catalog"] = original["catalog"] + + context.add_cleanup(_restore) + + +@given("the filesystem walk is mocked to be slow") +def step_mock_slow_walk(context: Any) -> None: + """Replace the fast empty-walk mock with one that introduces a small delay. + + This increases the window for a race condition to manifest if the lock + were absent, making the concurrent test more reliable. The delay is + intentionally tiny (1 ms) so the test suite stays fast. + """ + _SLOW_DELAY = 0.001 # 1 ms — enough to expose races without slowing tests + + # Stop the fast mock installed by step_clear_catalog_cache. + if hasattr(context, "_walk_patch"): + context._walk_patch.stop() + + def _slow_walk(top: Any, followlinks: bool = False) -> Any: + time.sleep(_SLOW_DELAY) + return iter([]) + + context._slow_walk_patch = patch.object( + rp_module.os, "walk", side_effect=_slow_walk + ) + context._slow_walk_patch.start() + + def _cleanup() -> None: + context._slow_walk_patch.stop() + + context.add_cleanup(_cleanup) + + +# --------------------------------------------------------------------------- +# When steps +# --------------------------------------------------------------------------- +@when("{count:d} threads call _catalog() concurrently") +def step_concurrent_catalog_calls(context: Any, count: int) -> None: + """Spawn *count* threads that each call _catalog() and collect results.""" + results: list[dict[str, list[str]] | None] = [None] * count + exceptions: list[BaseException | None] = [None] * count + + def _call(index: int) -> None: + try: + results[index] = _catalog() + except BaseException as exc: + exceptions[index] = exc + + threads = [threading.Thread(target=_call, args=(i,)) for i in range(count)] + for t in threads: + t.start() + for t in threads: + t.join(timeout=5.0) + + context.concurrent_results = results + context.concurrent_exceptions = exceptions + + +@when("{count:d} threads call _catalog() concurrently with the slow walk") +def step_concurrent_catalog_calls_slow(context: Any, count: int) -> None: + """Same as the plain concurrent step but relies on the slow-walk mock.""" + step_concurrent_catalog_calls(context, count) + + +# --------------------------------------------------------------------------- +# Then steps +# --------------------------------------------------------------------------- +@then("the _catalog_lock should be a threading.Lock") +def step_catalog_lock_is_lock(context: Any) -> None: + """Verify that _catalog_lock is a real threading.Lock (or RLock) instance.""" + lock = getattr(rp_module, "_catalog_lock", None) + assert lock is not None, "_catalog_lock attribute is missing" + # threading.Lock() returns a _thread.lock; threading.RLock() returns + # threading.RLock. Both are acceptable — check for the context-manager + # protocol and the acquire/release interface. + assert hasattr(lock, "acquire"), "_catalog_lock must have an acquire() method" + assert hasattr(lock, "release"), "_catalog_lock must have a release() method" + assert hasattr(lock, "__enter__"), ( + "_catalog_lock must support the context-manager protocol" + ) + + +@then("all threads should receive a valid catalog dict") +def step_all_threads_got_catalog(context: Any) -> None: + """Verify every thread received a non-None dict with the expected keys.""" + expected_keys = {"resource", "project", "plan", "actor", "tool", "skill"} + for i, result in enumerate(context.concurrent_results): + assert result is not None, f"Thread {i} received None instead of a catalog" + assert isinstance(result, dict), ( + f"Thread {i} received {type(result).__name__} instead of dict" + ) + assert expected_keys.issubset(result.keys()), ( + f"Thread {i} catalog is missing keys: {expected_keys - result.keys()}" + ) + + +@then("no catalog thread should have raised an exception") +def step_no_catalog_thread_exceptions(context: Any) -> None: + """Verify no thread raised an exception during concurrent _catalog() calls.""" + for i, exc in enumerate(context.concurrent_exceptions): + assert exc is None, f"Thread {i} raised an exception: {exc!r}" + + +@then("the catalog cache should contain a valid cwd entry") +def step_cache_has_cwd(context: Any) -> None: + """Verify _catalog_cache['cwd'] is a Path after concurrent access.""" + cwd_entry = _catalog_cache.get("cwd") + assert isinstance(cwd_entry, Path), ( + f"Expected _catalog_cache['cwd'] to be a Path, got {type(cwd_entry).__name__}" + ) + + +@then("the catalog cache should contain a valid created_at entry") +def step_cache_has_created_at(context: Any) -> None: + """Verify _catalog_cache['created_at'] is a positive float.""" + created_at = _catalog_cache.get("created_at") + assert isinstance(created_at, float), ( + f"Expected _catalog_cache['created_at'] to be float, " + f"got {type(created_at).__name__}" + ) + assert created_at > 0.0, ( + f"Expected _catalog_cache['created_at'] > 0, got {created_at}" + ) + + +@then("the catalog cache should contain a valid catalog entry") +def step_cache_has_catalog(context: Any) -> None: + """Verify _catalog_cache['catalog'] is a dict with the expected keys.""" + catalog_entry = _catalog_cache.get("catalog") + assert isinstance(catalog_entry, dict), ( + f"Expected _catalog_cache['catalog'] to be dict, " + f"got {type(catalog_entry).__name__}" + ) + expected_keys = {"resource", "project", "plan", "actor", "tool", "skill"} + assert expected_keys.issubset(catalog_entry.keys()), ( + f"Catalog entry is missing keys: {expected_keys - catalog_entry.keys()}" + ) + + +@then("all threads should receive identical catalog results") +def step_all_threads_identical_results(context: Any) -> None: + """Verify all threads received the same catalog (no partial-write corruption).""" + results = [r for r in context.concurrent_results if r is not None] + assert len(results) > 0, "No threads returned a result" + first = results[0] + for i, result in enumerate(results[1:], start=1): + assert result == first, ( + f"Thread {i} catalog differs from thread 0:\n" + f" Thread 0: {first}\n" + f" Thread {i}: {result}" + ) diff --git a/features/tdd_reference_parser_catalog_lock.feature b/features/tdd_reference_parser_catalog_lock.feature new file mode 100644 index 000000000..5f7d2f309 --- /dev/null +++ b/features/tdd_reference_parser_catalog_lock.feature @@ -0,0 +1,31 @@ +@tdd_issue @tdd_issue_7590 +Feature: Thread-safe catalog cache in reference_parser + As a developer using the TUI in a multi-tab session + I want the _catalog_cache global dict to be protected by a threading.Lock + So that concurrent calls to _catalog() cannot cause partial writes or RuntimeError + + Background: + Given the reference parser module is loaded for concurrency tests + + Scenario: _catalog_lock is a threading.Lock instance + Then the _catalog_lock should be a threading.Lock + + Scenario: Concurrent calls to _catalog() produce consistent results + Given the catalog cache is cleared for concurrency testing + When 10 threads call _catalog() concurrently + Then all threads should receive a valid catalog dict + And no catalog thread should have raised an exception + + Scenario: _catalog_cache is fully populated after concurrent access + Given the catalog cache is cleared for concurrency testing + When 5 threads call _catalog() concurrently + Then the catalog cache should contain a valid cwd entry + And the catalog cache should contain a valid created_at entry + And the catalog cache should contain a valid catalog entry + + Scenario: Lock prevents interleaved writes under simulated contention + Given the catalog cache is cleared for concurrency testing + And the filesystem walk is mocked to be slow + When 4 threads call _catalog() concurrently with the slow walk + Then all threads should receive identical catalog results + And no catalog thread should have raised an exception diff --git a/src/cleveragents/tui/input/reference_parser.py b/src/cleveragents/tui/input/reference_parser.py index 779f32a73..a2cb8591f 100644 --- a/src/cleveragents/tui/input/reference_parser.py +++ b/src/cleveragents/tui/input/reference_parser.py @@ -3,6 +3,7 @@ from __future__ import annotations import os +import threading import time from dataclasses import dataclass from pathlib import Path @@ -13,6 +14,7 @@ _REFERENCE_TYPES = ("project", "plan", "resource", "actor", "tool", "skill") _IGNORED_DIRS = {".git", ".venv", "node_modules", "__pycache__", ".mypy_cache"} _CATALOG_LIMIT = 400 _CATALOG_CACHE_TTL_SECONDS = 5.0 +_catalog_lock = threading.Lock() _catalog_cache: dict[str, object] = {"cwd": None, "created_at": 0.0, "catalog": None} @@ -35,61 +37,63 @@ class ReferenceParseResult: def _catalog() -> dict[str, list[str]]: - cwd = Path.cwd() - now = time.time() - cached_cwd = _catalog_cache.get("cwd") - cached_time = _catalog_cache.get("created_at") - cached_catalog = _catalog_cache.get("catalog") - if ( - isinstance(cached_cwd, Path) - and cached_cwd == cwd - and isinstance(cached_time, float) - and now - cached_time < _CATALOG_CACHE_TTL_SECONDS - and isinstance(cached_catalog, dict) - ): - return cached_catalog + with _catalog_lock: + cwd = Path.cwd() + now = time.time() + cached_cwd = _catalog_cache.get("cwd") + cached_time = _catalog_cache.get("created_at") + cached_catalog = _catalog_cache.get("catalog") + if ( + isinstance(cached_cwd, Path) + and cached_cwd == cwd + and isinstance(cached_time, float) + and now - cached_time < _CATALOG_CACHE_TTL_SECONDS + and isinstance(cached_catalog, dict) + ): + return cached_catalog - files = [] - for root, dirs, filenames in os.walk(cwd, followlinks=False): - dirs[:] = [name for name in dirs if name not in _IGNORED_DIRS] - root_path = Path(root) - for filename in filenames: - try: - files.append((root_path / filename).relative_to(cwd).as_posix()) - except ValueError: - files.append((root_path / filename).as_posix()) + files = [] + for root, dirs, filenames in os.walk(cwd, followlinks=False): + dirs[:] = [name for name in dirs if name not in _IGNORED_DIRS] + root_path = Path(root) + for filename in filenames: + try: + files.append((root_path / filename).relative_to(cwd).as_posix()) + except ValueError: + files.append((root_path / filename).as_posix()) + if len(files) >= _CATALOG_LIMIT: + break if len(files) >= _CATALOG_LIMIT: break - if len(files) >= _CATALOG_LIMIT: - break - catalog = { - "resource": sorted(files), - "project": [value for value in [cwd.name] if value], - "plan": [], - "actor": sorted( - f"local/{file.stem}" - for file in (cwd / "examples" / "actors").glob("*.y*ml") - ) - if (cwd / "examples" / "actors").is_dir() - else [], - "tool": sorted( - f"local/{file.stem}" for file in (cwd / "examples" / "tools").glob("*.y*ml") - ) - if (cwd / "examples" / "tools").is_dir() - else [], - "skill": sorted( - f"local/{entry.name}" - for entry in (cwd / "examples" / "skills").iterdir() - if entry.is_dir() - ) - if (cwd / "examples" / "skills").is_dir() - else [], - } - _catalog_cache["cwd"] = cwd - _catalog_cache["created_at"] = now - _catalog_cache["catalog"] = catalog - return catalog + catalog = { + "resource": sorted(files), + "project": [value for value in [cwd.name] if value], + "plan": [], + "actor": sorted( + f"local/{file.stem}" + for file in (cwd / "examples" / "actors").glob("*.y*ml") + ) + if (cwd / "examples" / "actors").is_dir() + else [], + "tool": sorted( + f"local/{file.stem}" + for file in (cwd / "examples" / "tools").glob("*.y*ml") + ) + if (cwd / "examples" / "tools").is_dir() + else [], + "skill": sorted( + f"local/{entry.name}" + for entry in (cwd / "examples" / "skills").iterdir() + if entry.is_dir() + ) + if (cwd / "examples" / "skills").is_dir() + else [], + } + _catalog_cache["cwd"] = cwd + _catalog_cache["created_at"] = now + _catalog_cache["catalog"] = catalog + return catalog def _resolve(category: str, query: str, items: dict[str, list[str]]) -> str | None: