From e54f28d0a10db1c6ec71559d9ada37c092dd0d6b Mon Sep 17 00:00:00 2001 From: HAL9000 Date: Thu, 30 Apr 2026 13:26:03 +0000 Subject: [PATCH 1/9] fix(tui): fix thread-safety race in reference_parser catalog cache Added threading.Lock to prevent race conditions in _catalog_cache dict. Two concurrent threads calling _catalog() while cache was stale could both walk the filesystem and write results simultaneously, causing partial dict updates or RuntimeError on concurrent iteration. Changes: - Added import threading and from threading import Lock - Added _catalog_lock: Lock = threading.Lock() at module level - Wrapped entire _catalog() body in with _catalog_lock: for atomic access - Added 4 BDD scenarios in tdd_reference_parser_catalog_lock.feature - Updated CHANGELOG.md with fix entry under [Unreleased] > Fixed - Updated CONTRIBUTORS.md with contribution entry ISSUES CLOSED: #7590 --- CHANGELOG.md | 13 + CONTRIBUTORS.md | 1 + ...tdd_reference_parser_catalog_lock_steps.py | 226 ++++++++++++++++++ .../tdd_reference_parser_catalog_lock.feature | 31 +++ .../tui/input/reference_parser.py | 107 +++++---- 5 files changed, 327 insertions(+), 51 deletions(-) create mode 100644 features/steps/tdd_reference_parser_catalog_lock_steps.py create mode 100644 features/tdd_reference_parser_catalog_lock.feature diff --git a/CHANGELOG.md b/CHANGELOG.md index 57513a9d3..5a96ae324 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -709,6 +709,19 @@ _ALL_DATA_COLUMNS + ") " "SELECT " + _ALL_DATA_COLUMNS + " FROM v3_plans"`. to recognize database resource categories. ### Fixed +- **Thread-safe catalog cache in TUI reference parser** (#7590): Fixed a + thread-safety race condition in `src/cleveragents/tui/input/reference_parser.py` + where the module-level `_catalog_cache` dict was written to without any locking. + In a multi-tab TUI session, two input threads calling `_catalog()` concurrently + while the cache was stale could both walk the filesystem and then write their + results to the shared dict simultaneously, causing partial dict updates or a + `RuntimeError` if the dict was iterated while being modified. Added + `import threading`, `from threading import Lock`, and a module-level + `_catalog_lock: Lock = threading.Lock()`. Wrapped the entire body of `_catalog()` + in `with _catalog_lock:` so all cache reads and writes are atomic. Added 4 BDD + scenarios in `features/tdd_reference_parser_catalog_lock.feature` verifying lock + existence, concurrent correctness, cache population, and contention simulation. + - **UAT tester docs PR duplicate detection and examples.json conflict fix** (#5768): Fixed two bugs in the `create_documentation_pr()` function of the uat-tester agent that caused diff --git a/CONTRIBUTORS.md b/CONTRIBUTORS.md index ac904da73..b104c3214 100644 --- a/CONTRIBUTORS.md +++ b/CONTRIBUTORS.md @@ -113,6 +113,7 @@ Below are some of the specific details of various contributions. * HAL 9000 has contributed the A2A stdio transport for local mode (#691): implemented ``A2aStdioTransport`` class with JSON-RPC 2.0 message framing over stdin/stdout for subprocess communication, including process lifecycle management (connect/disconnect), request/response serialization, graceful shutdown with timeout-based termination, and type-safe path resolution for Python modules vs. executable scripts. Added full BDD test coverage (18 scenarios) in ``features/a2a_stdio_transport.feature`` with mock subprocess behavior in step definitions. * HAL 9000 has contributed the `plan apply --format json` spec-compliant envelope fix (PR #9817 / issue #9449): replaced raw plan dictionary output with a spec-required JSON envelope containing structured data fields for artifacts, changes, validation, sandbox cleanup, and lifecycle metrics across all output formats. Full BDD test suite in behave + Robot Framework integration tests added. * HAL 9000 has contributed the ContextTierService defaults fix (PR #1485 / issue #1443): corrected spec-aligned default values for `max_tokens_hot` (16000), `max_decisions_warm` (100), and `max_decisions_cold` (500) in ``context_tier_settings.py``. Added comprehensive BDD regression tests verifying all three interface contracts. (Parent Epic: #935) +* HAL 9000 has contributed the thread-safety fix for the TUI reference parser catalog cache (#7590): added `_catalog_lock: Lock = threading.Lock()` and wrapped the entire `_catalog()` function body in `with _catalog_lock:` to eliminate race conditions under concurrent multi-tab TUI access. # Details (PR Contributions) 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..20425399f --- /dev/null +++ b/features/steps/tdd_reference_parser_catalog_lock_steps.py @@ -0,0 +1,226 @@ +"""Step definitions for tdd_reference_parser_catalog_lock.feature. + +Verifies thread-safety of the _catalog_lock in reference_parser.py: +- Lock existence and correct type +- Concurrent calls produce consistent results +- Cache is fully populated after concurrent access +- Lock prevents interleaved writes under simulated contention +""" + +from __future__ import annotations + +import threading +import time +from concurrent.futures import ThreadPoolExecutor, as_completed +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, +) + + +# --------------------------------------------------------------------------- +# Given steps +# --------------------------------------------------------------------------- + + +@given("the reference parser module is loaded") +def step_module_loaded(context: Any) -> None: + """Verify the module is importable and accessible.""" + assert rp_module is not None + + +@given("the catalog cache is cleared") +def step_clear_catalog_cache(context: Any) -> None: + """Reset the module-level catalog cache so _catalog() rebuilds on next call.""" + context.original_cache = { + "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 + + def restore_cache() -> None: + _catalog_cache["cwd"] = context.original_cache["cwd"] + _catalog_cache["created_at"] = context.original_cache["created_at"] + _catalog_cache["catalog"] = context.original_cache["catalog"] + + context.add_cleanup(restore_cache) + + +@given("os.walk is mocked with a slow delay to force thread overlap") +def step_mock_slow_walk(context: Any) -> None: + """Mock os.walk to introduce a delay, forcing threads to overlap.""" + _SLOW_DELAY = 0.001 + + def slow_walk(top: Any, followlinks: bool = False) -> Any: + time.sleep(_SLOW_DELAY) + yield str(top), [], [] + + context.slow_walk_patch = patch.object(rp_module.os, "walk", side_effect=slow_walk) + context.slow_walk_patch.start() + + context.is_dir_patch = patch.object(Path, "is_dir", return_value=False) + context.is_dir_patch.start() + + def cleanup() -> None: + context.slow_walk_patch.stop() + context.is_dir_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: + """Call _catalog() from multiple threads simultaneously.""" + results: list[dict[str, list[str]]] = [] + errors: list[Exception] = [] + lock = threading.Lock() + + def call_catalog() -> None: + try: + result = _catalog() + with lock: + results.append(result) + except Exception as exc: # broad catch intentional for thread safety + with lock: + errors.append(exc) + + with ThreadPoolExecutor(max_workers=count) as executor: + futures = [executor.submit(call_catalog) for _ in range(count)] + for future in as_completed(futures): + future.result() + + assert not errors, f"Threads raised exceptions: {errors}" + context.catalog_results = results + + +@when("{count:d} threads call _catalog concurrently with the slow mock") +def step_concurrent_catalog_calls_slow(context: Any, count: int) -> None: + """Call _catalog() from multiple threads with the slow walk mock active.""" + results: list[dict[str, list[str]]] = [] + errors: list[Exception] = [] + lock = threading.Lock() + + def call_catalog() -> None: + try: + result = _catalog() + with lock: + results.append(result) + except Exception as exc: # broad catch intentional for thread safety + with lock: + errors.append(exc) + + with ThreadPoolExecutor(max_workers=count) as executor: + futures = [executor.submit(call_catalog) for _ in range(count)] + for future in as_completed(futures): + future.result() + + assert not errors, f"Threads raised exceptions: {errors}" + context.catalog_results = results + + +# --------------------------------------------------------------------------- +# Then steps +# --------------------------------------------------------------------------- + + +@then("the module should have a _catalog_lock attribute") +def step_lock_attribute_exists(context: Any) -> None: + """Verify _catalog_lock is present on the module.""" + assert hasattr(rp_module, "_catalog_lock"), ( + "Expected rp_module to have attribute '_catalog_lock'" + ) + + +@then("_catalog_lock should be an instance of threading.Lock") +def step_lock_is_threading_lock(context: Any) -> None: + """Verify _catalog_lock is a threading.Lock (or _thread.lock).""" + lock = rp_module._catalog_lock + # threading.Lock() returns a _thread.lock instance; check via acquire/release + assert hasattr(lock, "acquire") and hasattr(lock, "release"), ( + f"Expected _catalog_lock to be a Lock-like object, got {type(lock)}" + ) + # Verify it is the same type as threading.Lock() + assert isinstance(lock, type(threading.Lock())), ( + f"Expected threading.Lock type, got {type(lock)}" + ) + + +@then("all threads should receive a valid catalog dict") +def step_all_results_valid(context: Any) -> None: + """Verify every thread received a non-None dict result.""" + results: list[dict[str, list[str]]] = context.catalog_results + assert results, "No results collected from threads" + for i, result in enumerate(results): + assert isinstance(result, dict), ( + f"Thread {i} result is not a dict: {type(result)}" + ) + + +@then("all catalog results should have the same keys") +def step_all_results_same_keys(context: Any) -> None: + """Verify all thread results have identical key sets.""" + results: list[dict[str, list[str]]] = context.catalog_results + assert results, "No results to compare" + first_keys = set(results[0].keys()) + for i, result in enumerate(results[1:], start=1): + assert set(result.keys()) == first_keys, ( + f"Thread {i} keys {set(result.keys())} differ from first {first_keys}" + ) + + +@then("the cache cwd entry should be a Path instance") +def step_cache_cwd_is_path(context: Any) -> None: + """Verify the cache cwd entry is a Path after concurrent access.""" + cwd = _catalog_cache.get("cwd") + assert isinstance(cwd, Path), ( + f"Expected cache['cwd'] to be a Path, got {type(cwd)}: {cwd}" + ) + + +@then("the cache created_at entry should be a positive float") +def step_cache_created_at_positive(context: Any) -> None: + """Verify the cache created_at entry is a positive float.""" + created_at = _catalog_cache.get("created_at") + assert isinstance(created_at, float) and created_at > 0, ( + f"Expected cache['created_at'] to be a positive float, got {created_at}" + ) + + +@then("the cache catalog entry should be a dict") +def step_cache_catalog_is_dict(context: Any) -> None: + """Verify the cache catalog entry is a dict after concurrent access.""" + catalog = _catalog_cache.get("catalog") + assert isinstance(catalog, dict), ( + f"Expected cache['catalog'] to be a dict, got {type(catalog)}: {catalog}" + ) + + +@then("all {count:d} threads should receive identical catalog results") +def step_all_results_identical(context: Any, count: int) -> None: + """Verify all threads received the exact same catalog dict contents.""" + results: list[dict[str, list[str]]] = context.catalog_results + assert len(results) == count, ( + f"Expected {count} results, got {len(results)}" + ) + first = results[0] + for i, result in enumerate(results[1:], start=1): + assert result == first, ( + f"Thread {i} result 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..d110a518d --- /dev/null +++ b/features/tdd_reference_parser_catalog_lock.feature @@ -0,0 +1,31 @@ +Feature: Thread-safe catalog cache lock in reference_parser + Verifies that _catalog_lock prevents race conditions in the + module-level _catalog_cache dict under concurrent access. + + @tdd_issue @tdd_issue_7590 + Scenario: Lock attribute exists and is a threading.Lock + Given the reference parser module is loaded + Then the module should have a _catalog_lock attribute + And _catalog_lock should be an instance of threading.Lock + + @tdd_issue @tdd_issue_7590 + Scenario: Concurrent calls to _catalog produce consistent results + Given the catalog cache is cleared + When 10 threads call _catalog concurrently + Then all threads should receive a valid catalog dict + And all catalog results should have the same keys + + @tdd_issue @tdd_issue_7590 + Scenario: Cache is fully populated after concurrent access + Given the catalog cache is cleared + When 4 threads call _catalog concurrently + Then the cache cwd entry should be a Path instance + And the cache created_at entry should be a positive float + And the cache catalog entry should be a dict + + @tdd_issue @tdd_issue_7590 + Scenario: Lock prevents interleaved writes under simulated contention + Given the catalog cache is cleared + And os.walk is mocked with a slow delay to force thread overlap + When 4 threads call _catalog concurrently with the slow mock + Then all 4 threads should receive identical catalog results diff --git a/src/cleveragents/tui/input/reference_parser.py b/src/cleveragents/tui/input/reference_parser.py index 779f32a73..cef88ba31 100644 --- a/src/cleveragents/tui/input/reference_parser.py +++ b/src/cleveragents/tui/input/reference_parser.py @@ -3,9 +3,11 @@ from __future__ import annotations import os +import threading import time from dataclasses import dataclass from pathlib import Path +from threading import Lock from cleveragents.tui.search.fuzzy import rank_candidates @@ -14,6 +16,7 @@ _IGNORED_DIRS = {".git", ".venv", "node_modules", "__pycache__", ".mypy_cache"} _CATALOG_LIMIT = 400 _CATALOG_CACHE_TTL_SECONDS = 5.0 _catalog_cache: dict[str, object] = {"cwd": None, "created_at": 0.0, "catalog": None} +_catalog_lock: Lock = threading.Lock() @dataclass(slots=True, frozen=True) @@ -35,61 +38,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: -- 2.52.0 From d02aa69924cfcf642661d90a5803f5b1b5135edb Mon Sep 17 00:00:00 2001 From: controller-ci-rerun Date: Thu, 11 Jun 2026 03:30:00 -0400 Subject: [PATCH 2/9] chore: re-trigger CI [controller] -- 2.52.0 From c7c7817fcd1758641ec562c74494604089cb597a Mon Sep 17 00:00:00 2001 From: CleverThis Date: Thu, 11 Jun 2026 15:26:22 -0400 Subject: [PATCH 3/9] fix(tui): apply ruff format to tdd_reference_parser_catalog_lock_steps.py --- features/steps/tdd_reference_parser_catalog_lock_steps.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/features/steps/tdd_reference_parser_catalog_lock_steps.py b/features/steps/tdd_reference_parser_catalog_lock_steps.py index 20425399f..a8a0762f6 100644 --- a/features/steps/tdd_reference_parser_catalog_lock_steps.py +++ b/features/steps/tdd_reference_parser_catalog_lock_steps.py @@ -214,9 +214,7 @@ def step_cache_catalog_is_dict(context: Any) -> None: def step_all_results_identical(context: Any, count: int) -> None: """Verify all threads received the exact same catalog dict contents.""" results: list[dict[str, list[str]]] = context.catalog_results - assert len(results) == count, ( - f"Expected {count} results, got {len(results)}" - ) + assert len(results) == count, f"Expected {count} results, got {len(results)}" first = results[0] for i, result in enumerate(results[1:], start=1): assert result == first, ( -- 2.52.0 From fe52983aeff034e2f71b126e78142f4e43bbc0d0 Mon Sep 17 00:00:00 2001 From: controller-ci-rerun Date: Fri, 12 Jun 2026 15:27:48 -0400 Subject: [PATCH 4/9] chore: re-trigger CI [controller] -- 2.52.0 From 703a5aa7256cda0891bc12f33e6af3b82f847ffd Mon Sep 17 00:00:00 2001 From: controller-ci-rerun Date: Wed, 17 Jun 2026 14:32:19 -0400 Subject: [PATCH 5/9] chore: re-trigger CI [controller] -- 2.52.0 From 6d603ee33dc0371ed80519bbe2ae42e6a7b44fe0 Mon Sep 17 00:00:00 2001 From: controller-ci-rerun Date: Thu, 18 Jun 2026 09:13:36 -0400 Subject: [PATCH 6/9] chore: re-trigger CI [controller] -- 2.52.0 From 1737921628a4547b43a189a0b5d18261dcb6d418 Mon Sep 17 00:00:00 2001 From: controller-ci-rerun Date: Thu, 18 Jun 2026 09:32:52 -0400 Subject: [PATCH 7/9] chore: re-trigger CI [controller] -- 2.52.0 From 9c3fd674d09209c50ffd3314b91f6af9654eebf3 Mon Sep 17 00:00:00 2001 From: controller-ci-rerun Date: Thu, 18 Jun 2026 09:43:10 -0400 Subject: [PATCH 8/9] chore: re-trigger CI [controller] -- 2.52.0 From 494da85edd6426d07f7f4ea63240f61426a0490e Mon Sep 17 00:00:00 2001 From: controller-ci-rerun Date: Thu, 18 Jun 2026 10:19:40 -0400 Subject: [PATCH 9/9] chore: re-trigger CI [controller] -- 2.52.0