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

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)
 

features/steps/tdd_reference_parser_catalog_lock_steps.py

@@ -0,0 +1,224 @@
+"""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}"
+        )

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

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: