diff --git a/features/steps/acms_parallel_indexing_steps.py b/features/steps/acms_parallel_indexing_steps.py index 1bb2e38fb..cbd68b7f4 100644 --- a/features/steps/acms_parallel_indexing_steps.py +++ b/features/steps/acms_parallel_indexing_steps.py @@ -22,9 +22,7 @@ from typing import Any from behave import given, then, when from cleveragents.acms.index import ( - ACMSIndex, FileTraversalEngine, - FileType, IndexProgress, ) @@ -79,9 +77,7 @@ def _create_test_directory_with_gitignore( nm = root / "node_modules" nm.mkdir(parents=True, exist_ok=True) for i in range(node_modules_entries): - (nm / f"pkg_{i}.js").write_text( - f"console.log({i});\n", encoding="utf-8" - ) + (nm / f"pkg_{i}.js").write_text(f"console.log({i});\n", encoding="utf-8") gitignore = root / ".gitignore" gitignore.write_text("node_modules\n", encoding="utf-8") @@ -97,8 +93,9 @@ def _create_test_directory_with_acmsignore( (subdir / f"module_{i:03d}.py").write_text( f"# src module {i}\n", encoding="utf-8" ) + (root / "__pycache__").mkdir(parents=True, exist_ok=True) for i in range(pycache_entries): - (root / f"__pycache__" / f"module_{i:02d}.cpython-312.pyc").write_bytes( + (root / "__pycache__" / f"module_{i:02d}.cpython-312.pyc").write_bytes( bytes([i] * 64) ) acmsignore = root / ".acmsignore" @@ -108,6 +105,7 @@ def _create_test_directory_with_acmsignore( def _create_test_directory_with_unreadable_dir(directory: str, py_count: int): """Create Python files plus one subdirectory we cannot read.""" import platform + if platform.system() == "Windows": # On Windows we create a regular dir but the engine handles PermissionError. root = Path(directory) @@ -135,15 +133,18 @@ def _create_test_directory_with_unreadable_dir(directory: str, py_count: int): @given("a test directory with {count:d} Python files") def step_create_py_dir(context: Any, count: int) -> None: - _create_test_directory_with_py_files( - directory=tempfile.mkdtemp(prefix="parallel-idx-"), - count=count, - prefix="file", - ) + d = tempfile.mkdtemp(prefix="parallel-idx-") + _create_test_directory_with_py_files(directory=d, count=count, prefix="file") + context.temp_dir_path = d + context.add_cleanup(shutil.rmtree, d, True) -@given("a test directory with {text_count:d} text files and {binary_count:d} binary PNG files") -def step_create_mixed_file_dir(context: Any, text_count: int, binary_count: int) -> None: +@given( + "a test directory with {text_count:d} text files and {binary_count:d} binary PNG files" +) +def step_create_mixed_file_dir( + context: Any, text_count: int, binary_count: int +) -> None: d = tempfile.mkdtemp(prefix="parallel-idx-mixed-") _create_test_directory_with_binary_files(d, text_count, binary_count) context.temp_dir_path = d @@ -174,7 +175,9 @@ def step_create_named_files(context: Any) -> None: context.add_cleanup(shutil.rmtree, d, True) -@given("a parallel file traversal engine with {workers:d} workers and chunk size {chunk:d}") +@given( + "a parallel file traversal engine with {workers:d} workers and chunk size {chunk:d}" +) def step_create_parallel_engine(context: Any, workers: int, chunk: int) -> None: engine = FileTraversalEngine(chunk_size=chunk, max_workers=workers) context.parallel_engine = engine @@ -192,7 +195,9 @@ def step_create_sequential_engine(context: Any, chunk: int) -> None: context.parallel_engine = engine -@given("a parallel file traversal engine with {workers:d} workers with progress tracking") +@given( + "a parallel file traversal engine with {workers:d} workers with progress tracking" +) def step_create_parallel_engine_with_progress(context: Any, workers: int) -> None: progress = IndexProgress() engine = FileTraversalEngine(chunk_size=25, max_workers=workers, progress=progress) @@ -200,9 +205,15 @@ def step_create_parallel_engine_with_progress(context: Any, workers: int) -> Non context.progress_tracker = progress -@given('a parallel file traversal engine with {workers:d} workers and cache at "{cache_path}"') -def step_create_parallel_engine_with_cache(context: Any, workers: int, cache_path: str) -> None: - engine = FileTraversalEngine(chunk_size=25, max_workers=workers, cache_path=cache_path) +@given( + 'a parallel file traversal engine with {workers:d} workers and cache at "{cache_path}"' +) +def step_create_parallel_engine_with_cache( + context: Any, workers: int, cache_path: str +) -> None: + engine = FileTraversalEngine( + chunk_size=25, max_workers=workers, cache_path=cache_path + ) context.parallel_engine = engine @@ -234,6 +245,8 @@ def step_traverse_parallel(context: Any) -> None: context.index_elapsed = time.monotonic() - start if hasattr(context, "progress_tracker"): context.progress_snapshot = context.progress_tracker.snapshot() + else: + context.progress_snapshot = engine.progress.snapshot() @when("I traverse and index the directory in parallel with exclusions") @@ -259,15 +272,13 @@ def step_traverse_parallel_acms_exclusions(context: Any) -> None: # --------------------------------------------------------------------------- -@then('the parallel index should contain {count:d} entries') +@then("the parallel index should contain {count:d} entries") def step_check_parallel_count(context: Any, count: int) -> None: actual = context.index.get_entry_count() - assert actual == count, ( - f"Expected {count} entries, got {actual}" - ) + assert actual == count, f"Expected {count} entries, got {actual}" -@then('the parallel index should contain approximately {count:d} entries') +@then("the parallel index should contain approximately {count:d} entries") def step_check_parallel_approx_count(context: Any, count: int) -> None: actual = context.index.get_entry_count() assert abs(actual - count) <= count * 0.15, ( @@ -275,7 +286,7 @@ def step_check_parallel_approx_count(context: Any, count: int) -> None: ) -@then('the indexing should complete within {seconds:d} seconds') +@then("the indexing should complete within {seconds:d} seconds") def step_check_indexing_time(context: Any, seconds: int) -> None: elapsed = context.index_elapsed assert elapsed < seconds, ( @@ -283,7 +294,7 @@ def step_check_indexing_time(context: Any, seconds: int) -> None: ) -@then('all indexed entries should have text (not binary) content') +@then("all indexed entries should have text (not binary) content") def step_check_all_text_entries(context: Any) -> None: for entry in context.index.get_all_entries(): assert not str(entry.path).endswith(".png"), ( @@ -291,7 +302,7 @@ def step_check_all_text_entries(context: Any) -> None: ) -@then('the parallel index should not contain any {pattern:s} paths') +@then("the parallel index should not contain any {pattern} paths") def step_check_no_pattern_paths(context: Any, pattern: str) -> None: for entry in context.index.get_all_entries(): assert pattern not in entry.path, ( @@ -299,12 +310,10 @@ def step_check_no_pattern_paths(context: Any, pattern: str) -> None: ) -@then('the sequential index should contain {count:d} entries') +@then("the sequential index should contain {count:d} entries") def step_check_sequential_count(context: Any, count: int) -> None: actual = context.index.get_entry_count() - assert actual == count, ( - f"Expected {count} entries, got {actual}" - ) + assert actual == count, f"Expected {count} entries, got {actual}" @then("the progress snapshot should show all 200 files indexed") @@ -349,9 +358,7 @@ def step_check_cache_valid_json(context: Any) -> None: @then("the parallel index should contain at least {count:d} entries") def step_check_parallel_min_count(context: Any, count: int) -> None: actual = context.index.get_entry_count() - assert actual >= count, ( - f"Expected at least {count} entries, got {actual}" - ) + assert actual >= count, f"Expected at least {count} entries, got {actual}" @then("the progress snapshot should show zero or few errors") diff --git a/src/cleveragents/acms/index.py b/src/cleveragents/acms/index.py index 64aaa3e9a..8ec76a9ad 100644 --- a/src/cleveragents/acms/index.py +++ b/src/cleveragents/acms/index.py @@ -13,19 +13,15 @@ threads while preserving thread safety and deterministic chunk boundaries. from __future__ import annotations import json -import os import threading from collections.abc import Iterator from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime from enum import StrEnum from pathlib import Path -from typing import TYPE_CHECKING +from typing import Any -from pydantic import BaseModel, Field - -if TYPE_CHECKING: - from typing import Any +from pydantic import BaseModel, Field, PrivateAttr class FileType(StrEnum): @@ -122,7 +118,7 @@ class IndexProgress(BaseModel): Thread-safe via a single lock protecting all counters. """ - _lock: threading.Lock = None # type: ignore[assignment] + _lock: threading.Lock = PrivateAttr(default_factory=threading.Lock) files_processed: int = 0 files_skipped: int = 0 files_indexed: int = 0 @@ -130,11 +126,6 @@ class IndexProgress(BaseModel): binary_files_found: int = 0 errors_occurred: int = 0 - def __init__(self) -> None: - """Initialize progress tracker with a reentrant lock.""" - super().__init__() - self._lock = threading.Lock() - def record_processed(self, bytes_read: int = 0) -> None: """Record that a file was processed (stat+read completed).""" with self._lock: @@ -161,6 +152,21 @@ class IndexProgress(BaseModel): with self._lock: self.errors_occurred += 1 + def reset(self) -> None: + """Reset all counters to zero while preserving the existing lock. + + Caller-held references to this object continue to observe the + cleared counters — useful for engine reuse where the test or + production harness keeps a reference to the same progress tracker. + """ + with self._lock: + self.files_processed = 0 + self.files_skipped = 0 + self.files_indexed = 0 + self.bytes_read = 0 + self.binary_files_found = 0 + self.errors_occurred = 0 + @property def progress_percent(self) -> float: """Return completion percentage (0-100).""" @@ -327,9 +333,16 @@ class ACMSIndex: return len(self.entries) def to_json_dict(self) -> dict[str, Any]: - """Serialize the entire index to a JSON-serializable dictionary.""" + """Serialize the entire index to a JSON-serializable dictionary. + + Uses Pydantic's ``mode='json'`` so non-JSON-native field types + (``datetime``, ``set``, ``StrEnum``) are converted to their JSON + wire forms instead of left as Python objects. + """ return { - "entries": [entry.model_dump() for entry in self.entries.values()], + "entries": [ + entry.model_dump(mode="json") for entry in self.entries.values() + ], "count": len(self.entries), } @@ -448,7 +461,7 @@ class FileTraversalEngine: with open(file_path, "rb") as f: chunk = f.read(8192) return b"\x00" in chunk - except (OSError, IOError): + except OSError: # If we can't read it, treat it as binary (skip it). return True @@ -490,7 +503,7 @@ class FileTraversalEngine: if not gitignore.exists(): return [] try: - with open(gitignore, "r", encoding="utf-8") as f: + with open(gitignore, encoding="utf-8") as f: return [ line.strip() for line in f @@ -514,7 +527,7 @@ class FileTraversalEngine: if not acmsignore.exists(): return [] try: - with open(acmsignore, "r", encoding="utf-8") as f: + with open(acmsignore, encoding="utf-8") as f: return [ line.strip() for line in f @@ -538,10 +551,7 @@ class FileTraversalEngine: True if the file should be excluded. """ rel_str = str(file_path) - for pattern in exclude_patterns: - if pattern in rel_str: - return True - return False + return any(pattern in rel_str for pattern in exclude_patterns) def _traverse_directory( self, @@ -645,9 +655,7 @@ class FileTraversalEngine: self.progress.record_error() return None - def _process_chunk_parallel( - self, chunk: list[Path] - ) -> list[IndexEntry]: + def _process_chunk_parallel(self, chunk: list[Path]) -> list[IndexEntry]: """Process a chunk of files using ThreadPoolExecutor. Uses concurrent.futures.ThreadPoolExecutor to distribute file @@ -686,7 +694,7 @@ class FileTraversalEngine: ordered_entries.append(entries_by_index[idx]) self.progress.record_indexed() - for exc in errors_in_chunk.values(): + for _exc in errors_in_chunk.values(): self.progress.record_error() return ordered_entries @@ -707,7 +715,7 @@ class FileTraversalEngine: tmp_path = self.cache_path.with_suffix(".part") tmp_path.write_text(json_content, encoding="utf-8") tmp_path.rename(self.cache_path) - except (OSError, IOError): + except OSError: # Silently ignore cache write failures -- indexing must not fail. pass @@ -737,7 +745,9 @@ class FileTraversalEngine: # Load exclusion patterns from .gitignore / .acmsignore files. gitignore_patterns = self._load_gitignore_patterns(root) acmsignore_patterns = self._load_acmsignore_patterns(root) - exclude_patterns = list(set(exclude_patterns + gitignore_patterns + acmsignore_patterns)) + exclude_patterns = list( + set((exclude_patterns or []) + gitignore_patterns + acmsignore_patterns) + ) # Single-pass collection respecting exclusions. all_files = self._collect_all_files(root, exclude_patterns) @@ -770,11 +780,13 @@ class FileTraversalEngine: return self.index def reset_index(self) -> None: - """Reset the index to empty state.""" + """Reset the index and progress counters to empty state. + + Preserves the identity of ``self.progress`` so that any caller + holding an external reference to it observes the cleared counters. + """ self.index = ACMSIndex() - # Reset progress tracker. - new_progress = IndexProgress() - new_progress._lock = threading.Lock() # type: ignore[assignment] + self.progress.reset() __all__ = [ @@ -782,6 +794,6 @@ __all__ = [ "FileTraversalEngine", "FileType", "IndexEntry", - "TierLevel", "IndexProgress", + "TierLevel", ]