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.
### 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
`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 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 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
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: