fix(tui): fix thread-safety race in reference_parser catalog cache
CI / benchmark-publish (pull_request) Has been skipped
CI / push-validation (pull_request) Successful in 34s
CI / helm (pull_request) Successful in 42s
CI / build (pull_request) Successful in 1m8s
CI / lint (pull_request) Failing after 1m32s
CI / quality (pull_request) Successful in 1m43s
CI / typecheck (pull_request) Successful in 1m52s
CI / security (pull_request) Successful in 1m53s
CI / integration_tests (pull_request) Successful in 4m5s
CI / e2e_tests (pull_request) Successful in 5m24s
CI / unit_tests (pull_request) Successful in 7m34s
CI / coverage (pull_request) Has been skipped
CI / docker (pull_request) Has been skipped
CI / status-check (pull_request) Failing after 3s

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
This commit is contained in:
2026-04-30 13:26:03 +00:00
parent 9888c2f6e6
commit cbb85ec595
5 changed files with 327 additions and 51 deletions
+13
View File
@@ -47,6 +47,19 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
execution and confirms proper cleanup behavior. execution and confirms proper cleanup behavior.
### Fixed ### 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.
- **Atomic `server_connect` config writes** (#993): Fixed `server_connect` in - **Atomic `server_connect` config writes** (#993): Fixed `server_connect` in
`cli/commands/server.py` to write all three config values (`server.url`, `cli/commands/server.py` to write all three config values (`server.url`,
+1
View File
@@ -25,3 +25,4 @@ Below are some of the specific details of various contributions.
* HAL 9000 has contributed the file edit encoding parameter fix (PR #8258 / issue #7559). * HAL 9000 has contributed the file edit encoding parameter fix (PR #8258 / issue #7559).
* HAL 9000 has contributed the architecture-pool-supervisor milestone assignment feature (PR #8188 / issue #7521): added `forgejo_update_pull_request` permission and documented the PR workflow for major spec changes, enabling automatic milestone assignment for specification PRs. * HAL 9000 has contributed the architecture-pool-supervisor milestone assignment feature (PR #8188 / issue #7521): added `forgejo_update_pull_request` permission and documented the PR workflow for major spec changes, enabling automatic milestone assignment for specification PRs.
* HAL 9000 has contributed the git worktree TOCTOU race condition fix (PR #8178 / issue #7507): replaced the unsafe mkdtemp() + rmdir() pattern with a parent-directory approach to eliminate the race window in concurrent git worktree operations. * HAL 9000 has contributed the git worktree TOCTOU race condition fix (PR #8178 / issue #7507): replaced the unsafe mkdtemp() + rmdir() pattern with a parent-directory approach to eliminate the race window in concurrent git worktree operations.
* 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.
@@ -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}"
)
@@ -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
+56 -51
View File
@@ -3,9 +3,11 @@
from __future__ import annotations from __future__ import annotations
import os import os
import threading
import time import time
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from threading import Lock
from cleveragents.tui.search.fuzzy import rank_candidates 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_LIMIT = 400
_CATALOG_CACHE_TTL_SECONDS = 5.0 _CATALOG_CACHE_TTL_SECONDS = 5.0
_catalog_cache: dict[str, object] = {"cwd": None, "created_at": 0.0, "catalog": None} _catalog_cache: dict[str, object] = {"cwd": None, "created_at": 0.0, "catalog": None}
_catalog_lock: Lock = threading.Lock()
@dataclass(slots=True, frozen=True) @dataclass(slots=True, frozen=True)
@@ -35,61 +38,63 @@ class ReferenceParseResult:
def _catalog() -> dict[str, list[str]]: def _catalog() -> dict[str, list[str]]:
cwd = Path.cwd() with _catalog_lock:
now = time.time() cwd = Path.cwd()
cached_cwd = _catalog_cache.get("cwd") now = time.time()
cached_time = _catalog_cache.get("created_at") cached_cwd = _catalog_cache.get("cwd")
cached_catalog = _catalog_cache.get("catalog") cached_time = _catalog_cache.get("created_at")
if ( cached_catalog = _catalog_cache.get("catalog")
isinstance(cached_cwd, Path) if (
and cached_cwd == cwd isinstance(cached_cwd, Path)
and isinstance(cached_time, float) and cached_cwd == cwd
and now - cached_time < _CATALOG_CACHE_TTL_SECONDS and isinstance(cached_time, float)
and isinstance(cached_catalog, dict) and now - cached_time < _CATALOG_CACHE_TTL_SECONDS
): and isinstance(cached_catalog, dict)
return cached_catalog ):
return cached_catalog
files = [] files = []
for root, dirs, filenames in os.walk(cwd, followlinks=False): for root, dirs, filenames in os.walk(cwd, followlinks=False):
dirs[:] = [name for name in dirs if name not in _IGNORED_DIRS] dirs[:] = [name for name in dirs if name not in _IGNORED_DIRS]
root_path = Path(root) root_path = Path(root)
for filename in filenames: for filename in filenames:
try: try:
files.append((root_path / filename).relative_to(cwd).as_posix()) files.append((root_path / filename).relative_to(cwd).as_posix())
except ValueError: except ValueError:
files.append((root_path / filename).as_posix()) files.append((root_path / filename).as_posix())
if len(files) >= _CATALOG_LIMIT:
break
if len(files) >= _CATALOG_LIMIT: if len(files) >= _CATALOG_LIMIT:
break break
if len(files) >= _CATALOG_LIMIT:
break
catalog = { catalog = {
"resource": sorted(files), "resource": sorted(files),
"project": [value for value in [cwd.name] if value], "project": [value for value in [cwd.name] if value],
"plan": [], "plan": [],
"actor": sorted( "actor": sorted(
f"local/{file.stem}" f"local/{file.stem}"
for file in (cwd / "examples" / "actors").glob("*.y*ml") for file in (cwd / "examples" / "actors").glob("*.y*ml")
) )
if (cwd / "examples" / "actors").is_dir() if (cwd / "examples" / "actors").is_dir()
else [], else [],
"tool": sorted( "tool": sorted(
f"local/{file.stem}" for file in (cwd / "examples" / "tools").glob("*.y*ml") f"local/{file.stem}"
) for file in (cwd / "examples" / "tools").glob("*.y*ml")
if (cwd / "examples" / "tools").is_dir() )
else [], if (cwd / "examples" / "tools").is_dir()
"skill": sorted( else [],
f"local/{entry.name}" "skill": sorted(
for entry in (cwd / "examples" / "skills").iterdir() f"local/{entry.name}"
if entry.is_dir() for entry in (cwd / "examples" / "skills").iterdir()
) if entry.is_dir()
if (cwd / "examples" / "skills").is_dir() )
else [], if (cwd / "examples" / "skills").is_dir()
} else [],
_catalog_cache["cwd"] = cwd }
_catalog_cache["created_at"] = now _catalog_cache["cwd"] = cwd
_catalog_cache["catalog"] = catalog _catalog_cache["created_at"] = now
return catalog _catalog_cache["catalog"] = catalog
return catalog
def _resolve(category: str, query: str, items: dict[str, list[str]]) -> str | None: def _resolve(category: str, query: str, items: dict[str, list[str]]) -> str | None: