fix(acms): rebase ACMS index data model onto master to fix unit test failures
CI / coverage (pull_request) Blocked by required conditions
CI / docker (pull_request) Blocked by required conditions
CI / status-check (pull_request) Blocked by required conditions
CI / typecheck (pull_request) Has started running
CI / unit_tests (pull_request) Has started running
CI / integration_tests (pull_request) Has started running
CI / e2e_tests (pull_request) Has started running
CI / benchmark-publish (pull_request) Has been skipped
CI / helm (pull_request) Successful in 41s
CI / lint (pull_request) Successful in 1m4s
CI / quality (pull_request) Successful in 1m23s
CI / build (pull_request) Successful in 1m6s
CI / push-validation (pull_request) Has started running
CI / security (pull_request) Successful in 1m43s

Rebases the ACMS index data model and file traversal engine implementation
onto the current master branch to resolve merge conflicts and fix unit test
failures caused by the PR branch being 321 commits behind master.

All ACMS-specific changes are preserved:
- src/cleveragents/acms/index.py: ACMS index data model with hot/warm/cold/archive tiers
- src/cleveragents/acms/__init__.py: Updated exports
- features/acms/index_data_model_and_traversal.feature: BDD feature file
- features/steps/acms_index_data_model_traversal_steps.py: Step definitions
- features/environment.py: Added temp_dir cleanup hook
- CHANGELOG.md: ACMS entry preserved

ISSUES CLOSED: #9579
This commit is contained in:
2026-04-30 21:03:13 +00:00
parent 1691def164
commit c2bb41e89e
6 changed files with 980 additions and 3 deletions
+8
View File
@@ -166,6 +166,14 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
logged at debug level for observability.
### Added
- **ACMS Index Data Model and File Traversal Engine** (#9579): Implements the
foundational ACMS index data model with structured fields for file metadata
(path, size, last modified, type), tag system, and hot/warm/cold/archive
storage tier assignment. Introduces a timeout-safe large-project file traversal
engine capable of handling 10,000+ files without memory exhaustion through
chunked processing. Provides a complete index entry pipeline for creation,
storage, and retrieval with full queryability by path, tag, type, and recency.
- Wired `StrategyActor` into the real plan execution path: `_get_plan_executor`
in `plan.py` now resolves the strategy actor via `resolve_strategy_actor()`
(reading the `actor.default.strategy` config key) instead of always
@@ -0,0 +1,138 @@
Feature: ACMS Index Data Model and File Traversal Engine
As a developer
I want to index large projects with 10,000+ files
So that I can efficiently query and manage context entries at scale
Background:
Given I have an ACMS index
And I have a file traversal engine with chunk size 100
Scenario: Create an index entry with file metadata
When I create an index entry with:
| path | /project/src/main.py |
| file_type | python |
| size_bytes | 1024 |
Then the index entry should have path "/project/src/main.py"
And the index entry should have file type "python"
And the index entry should have size 1024 bytes
Scenario: Add tags to an index entry
Given I have an index entry with path "/project/src/main.py"
When I add tag "core" to the entry
And I add tag "important" to the entry
Then the entry should have tag "core"
And the entry should have tag "important"
And the entry should have 2 tags
Scenario: Set tier level for an index entry
Given I have an index entry with path "/project/src/main.py"
When I set the tier level to "hot"
Then the entry should have tier level "hot"
Scenario: Add entry to index
Given I have an index entry with path "/project/src/main.py"
When I add the entry to the index
Then the index should contain 1 entry
And I should be able to retrieve the entry by path "/project/src/main.py"
Scenario: Query index by path pattern
Given I have an index with entries:
| path | file_type |
| /project/src/main.py | python |
| /project/src/utils.py | python |
| /project/tests/test_main.py | python |
| /project/docs/readme.md | markdown |
When I query the index by path pattern "src"
Then I should get 2 results
And the results should include "/project/src/main.py"
And the results should include "/project/src/utils.py"
Scenario: Query index by file type
Given I have an index with entries:
| path | file_type |
| /project/src/main.py | python |
| /project/src/utils.py | python |
| /project/src/app.js | javascript |
| /project/docs/readme.md | markdown |
When I query the index by file type "python"
Then I should get 2 results
And all results should have file type "python"
Scenario: Query index by tag
Given I have an index with entries:
| path | tags |
| /project/src/main.py | core,important |
| /project/src/utils.py | supporting |
| /project/tests/test_main.py | test,important |
When I query the index by tag "important"
Then I should get 2 results
And the results should include "/project/src/main.py"
And the results should include "/project/tests/test_main.py"
Scenario: Query index by tier level
Given I have an index with entries:
| path | tier |
| /project/src/main.py | hot |
| /project/src/utils.py | warm |
| /project/tests/test_main.py | cold |
When I query the index by tier level "hot"
Then I should get 1 result
And the result should have path "/project/src/main.py"
Scenario: Query index by recency
Given I have an index with entries from different dates
When I query the index for entries modified after "2026-04-01"
Then I should get entries modified after that date
Scenario: Traverse and index a directory with multiple files
Given I have a test directory with 50 files
When I traverse and index the directory
Then the index should contain 50 entries
And all entries should have valid file paths
Scenario: Handle large project traversal with chunked processing
Given I have a test directory with 1000 files
When I traverse and index the directory with chunk size 100
Then the index should contain 1000 entries
And the traversal should complete without timeout
Scenario: Exclude patterns during traversal
Given I have a test directory with files including:
| path |
| /project/src/main.py |
| /project/.git/config |
| /project/__pycache__/main.cpython-39.pyc |
| /project/src/utils.py |
When I traverse and index the directory excluding ".git" and "__pycache__"
Then the index should contain 2 entries
And the index should not contain ".git" paths
And the index should not contain "__pycache__" paths
Scenario: Get all entries from index
Given I have an index with 5 entries
When I get all entries from the index
Then I should get 5 results
Scenario: Get entry count from index
Given I have an index with 10 entries
When I get the entry count
Then the count should be 10
Scenario: Remove entry from index
Given I have an index with 3 entries
When I remove an entry by path
Then the index should contain 2 entries
Scenario: Combined query with multiple filters
Given I have an index with entries:
| path | file_type | tags | tier |
| /project/src/main.py | python | core,important | hot |
| /project/src/utils.py | python | supporting | warm |
| /project/tests/test_main.py | python | test | cold |
| /project/docs/readme.md | markdown | docs | cold |
When I query the index with filters:
| path_pattern | src |
| file_type | python |
| tier | hot |
Then I should get 1 result
And the result should have path "/project/src/main.py"
+6
View File
@@ -676,6 +676,12 @@ def after_scenario(context, scenario):
pass # Ignore cleanup errors
context.test_dir = None
# Clean up TemporaryDirectory objects created by ACMS index traversal tests
if hasattr(context, "temp_dir") and context.temp_dir is not None:
with contextlib.suppress(Exception):
context.temp_dir.cleanup()
context.temp_dir = None
# Clean up environment variables set during tests
if hasattr(context, "env_vars_to_clean"):
for key in context.env_vars_to_clean:
@@ -0,0 +1,395 @@
"""Step definitions for ACMS Index Data Model and File Traversal Engine tests."""
from __future__ import annotations
import tempfile
from datetime import datetime, timedelta
from pathlib import Path
from behave import given, then, when
from cleveragents.acms.index import (
ACMSIndex,
FileTraversalEngine,
FileType,
IndexEntry,
TierLevel,
)
@given("I have an ACMS index")
def step_create_index(context):
"""Create a new ACMS index."""
context.index = ACMSIndex()
@given("I have a file traversal engine with chunk size {chunk_size:d}")
def step_create_traversal_engine(context, chunk_size):
"""Create a file traversal engine with specified chunk size."""
context.engine = FileTraversalEngine(chunk_size=chunk_size)
@when("I create an index entry with:")
def step_create_index_entry(context):
"""Create an index entry from table data."""
data = {row["key"]: row["value"] for row in context.table}
file_type = FileType(data.get("file_type", "other"))
size_bytes = int(data.get("size_bytes", "0"))
context.entry = IndexEntry(
path=data["path"],
file_type=file_type,
size_bytes=size_bytes,
created_at=datetime.now(),
modified_at=datetime.now(),
)
@then('the index entry should have path "{path}"')
def step_check_entry_path(context, path):
"""Verify the index entry has the expected path."""
assert context.entry.path == path
@then('the index entry should have file type "{file_type}"')
def step_check_entry_file_type(context, file_type):
"""Verify the index entry has the expected file type."""
assert context.entry.file_type == FileType(file_type)
@then("the index entry should have size {size:d} bytes")
def step_check_entry_size(context, size):
"""Verify the index entry has the expected size."""
assert context.entry.size_bytes == size
@given('I have an index entry with path "{path}"')
def step_create_entry_with_path(context, path):
"""Create an index entry with a specific path."""
context.entry = IndexEntry(
path=path,
file_type=FileType.PYTHON,
size_bytes=1024,
created_at=datetime.now(),
modified_at=datetime.now(),
)
@when('I add tag "{tag}" to the entry')
def step_add_tag_to_entry(context, tag):
"""Add a tag to the current entry."""
context.entry.add_tag(tag)
@then('the entry should have tag "{tag}"')
def step_check_entry_has_tag(context, tag):
"""Verify the entry has a specific tag."""
assert context.entry.has_tag(tag)
@then("the entry should have {count:d} tags")
def step_check_entry_tag_count(context, count):
"""Verify the entry has the expected number of tags."""
assert len(context.entry.tags) == count
@when('I set the tier level to "{tier}"')
def step_set_entry_tier(context, tier):
"""Set the tier level for the entry."""
context.entry.set_tier(TierLevel(tier))
@then('the entry should have tier level "{tier}"')
def step_check_entry_tier(context, tier):
"""Verify the entry has the expected tier level."""
assert context.entry.tier == TierLevel(tier)
@when("I add the entry to the index")
def step_add_entry_to_index(context):
"""Add the current entry to the index."""
context.index.add_entry(context.entry)
@then("the index should contain {count:d} entry")
def step_check_index_entry_count_singular(context, count):
"""Verify the index has the expected number of entries."""
assert context.index.get_entry_count() == count
@then("the index should contain {count:d} entries")
def step_check_index_entry_count(context, count):
"""Verify the index has the expected number of entries."""
assert context.index.get_entry_count() == count
@then('I should be able to retrieve the entry by path "{path}"')
def step_retrieve_entry_by_path(context, path):
"""Verify we can retrieve an entry by path."""
entry = context.index.get_entry(path)
assert entry is not None
assert entry.path == path
@given("I have an index with entries:")
def step_create_index_with_entries(context):
"""Create an index with multiple entries from table data."""
context.index = ACMSIndex()
for row in context.table:
path = row["path"]
file_type = FileType(row.get("file_type", "other"))
entry = IndexEntry(
path=path,
file_type=file_type,
size_bytes=1024,
created_at=datetime.now(),
modified_at=datetime.now(),
)
# Add tags if present
if "tags" in row:
for tag in row["tags"].split(","):
entry.add_tag(tag.strip())
# Set tier if present
if "tier" in row:
entry.set_tier(TierLevel(row["tier"]))
context.index.add_entry(entry)
@given("I have an index with {count:d} entries")
def step_create_index_with_n_entries(context, count):
"""Create an index with a specified number of entries."""
context.index = ACMSIndex()
for i in range(count):
entry = IndexEntry(
path=f"/project/file{i}.py",
file_type=FileType.PYTHON,
size_bytes=1024,
created_at=datetime.now(),
modified_at=datetime.now(),
)
context.index.add_entry(entry)
@when('I query the index by path pattern "{pattern}"')
def step_query_by_path_pattern(context, pattern):
"""Query the index by path pattern."""
context.query_results = context.index.query_by_path(pattern)
@then("I should get {count:d} results")
def step_check_query_result_count(context, count):
"""Verify the query returned the expected number of results."""
assert len(context.query_results) == count, (
f"Expected {count} results, got {len(context.query_results)}"
)
@then("I should get {count:d} result")
def step_check_query_result_count_singular(context, count):
"""Verify the query returned the expected number of results (singular)."""
assert len(context.query_results) == count, (
f"Expected {count} result, got {len(context.query_results)}"
)
@then('the results should include "{path}"')
def step_check_result_includes_path(context, path):
"""Verify the query results include a specific path."""
paths = [entry.path for entry in context.query_results]
assert path in paths
@when('I query the index by file type "{file_type}"')
def step_query_by_file_type(context, file_type):
"""Query the index by file type."""
context.query_results = context.index.query_by_type(FileType(file_type))
@then('all results should have file type "{file_type}"')
def step_check_all_results_file_type(context, file_type):
"""Verify all results have the expected file type."""
expected_type = FileType(file_type)
for entry in context.query_results:
assert entry.file_type == expected_type
@when('I query the index by tag "{tag}"')
def step_query_by_tag(context, tag):
"""Query the index by tag."""
context.query_results = context.index.query_by_tag(tag)
@when('I query the index by tier level "{tier}"')
def step_query_by_tier(context, tier):
"""Query the index by tier level."""
context.query_results = context.index.query_by_tier(TierLevel(tier))
@then('the result should have path "{path}"')
def step_check_single_result_path(context, path):
"""Verify the single result has the expected path."""
assert len(context.query_results) == 1
assert context.query_results[0].path == path
@given("I have an index with entries from different dates")
def step_create_index_with_dated_entries(context):
"""Create an index with entries from different dates."""
context.index = ACMSIndex()
now = datetime.now()
dates = [
now - timedelta(days=10),
now - timedelta(days=5),
now - timedelta(days=1),
now,
]
for i, date in enumerate(dates):
entry = IndexEntry(
path=f"/project/file{i}.py",
file_type=FileType.PYTHON,
size_bytes=1024,
created_at=date,
modified_at=date,
)
context.index.add_entry(entry)
@when('I query the index for entries modified after "{date_str}"')
def step_query_by_recency(context, date_str):
"""Query the index for entries modified after a specific date."""
# Parse date string (format: YYYY-MM-DD)
date = datetime.strptime(date_str, "%Y-%m-%d")
context.query_results = context.index.query_by_recency(date)
@then("I should get entries modified after that date")
def step_check_recency_results(context):
"""Verify the recency query returned valid results."""
assert len(context.query_results) > 0
@given("I have a test directory with {count:d} files")
def step_create_test_directory(context, count):
"""Create a temporary directory with test files."""
context.temp_dir = tempfile.TemporaryDirectory()
temp_path = Path(context.temp_dir.name)
# Create subdirectories and files
for i in range(count):
subdir = temp_path / f"subdir{i % 10}"
subdir.mkdir(exist_ok=True)
file_path = subdir / f"file{i}.py"
file_path.write_text(f"# Test file {i}\nprint('Hello {i}')\n")
@when("I traverse and index the directory")
def step_traverse_and_index(context):
"""Traverse and index the test directory."""
context.engine.reset_index()
context.index = context.engine.traverse_and_index(context.temp_dir.name)
@when("I traverse and index the directory with chunk size {chunk_size:d}")
def step_traverse_and_index_with_chunk_size(context, chunk_size):
"""Traverse and index the test directory with a specific chunk size."""
engine = FileTraversalEngine(chunk_size=chunk_size)
context.index = engine.traverse_and_index(context.temp_dir.name)
@then("all entries should have valid file paths")
def step_check_valid_file_paths(context):
"""Verify all entries have valid file paths."""
for entry in context.index.get_all_entries():
assert entry.path
assert len(entry.path) > 0
@then("the traversal should complete without timeout")
def step_check_no_timeout(context):
"""Verify the traversal completed without timeout."""
# This step passes if we got here without timing out
assert True
@given("I have a test directory with files including:")
def step_create_test_directory_with_specific_files(context):
"""Create a test directory with specific files."""
context.temp_dir = tempfile.TemporaryDirectory()
temp_path = Path(context.temp_dir.name)
for row in context.table:
file_path = temp_path / row["path"].lstrip("/")
file_path.parent.mkdir(parents=True, exist_ok=True)
file_path.write_text("test content")
@when('I traverse and index the directory excluding "{exclude1}" and "{exclude2}"')
def step_traverse_with_exclusions(context, exclude1, exclude2):
"""Traverse and index with exclusion patterns."""
context.engine.reset_index()
context.index = context.engine.traverse_and_index(
context.temp_dir.name,
exclude_patterns=[exclude1, exclude2],
)
@then('the index should not contain "{pattern}" paths')
def step_check_no_excluded_paths(context, pattern):
"""Verify the index doesn't contain paths matching the pattern."""
for entry in context.index.get_all_entries():
assert pattern not in entry.path
@when("I get all entries from the index")
def step_get_all_entries(context):
"""Get all entries from the index."""
context.query_results = context.index.get_all_entries()
@when("I get the entry count")
def step_get_entry_count(context):
"""Get the entry count from the index."""
context.entry_count = context.index.get_entry_count()
@then("the count should be {count:d}")
def step_check_entry_count_value(context, count):
"""Verify the entry count matches the expected value."""
assert context.entry_count == count
@when("I remove an entry by path")
def step_remove_entry(context):
"""Remove an entry from the index."""
# Get the first entry's path
entries = context.index.get_all_entries()
if entries:
context.index.remove_entry(entries[0].path)
@when("I query the index with filters:")
def step_query_with_combined_filters(context):
"""Query the index with multiple filters."""
filters = {row["filter"]: row["value"] for row in context.table}
path_pattern = filters.get("path_pattern")
file_type_str = filters.get("file_type")
tier_str = filters.get("tier")
file_type = FileType(file_type_str) if file_type_str else None
tier = TierLevel(tier_str) if tier_str else None
context.query_results = context.index.query_combined(
path_pattern=path_pattern,
file_type=file_type,
tier=tier,
)
+21 -3
View File
@@ -5,12 +5,22 @@ technology-specific vocabulary extensions, and the DetailLevelMap
inheritance mechanism for resolving named detail levels across the
ontology hierarchy (Layer 3 -> Layer 2 -> Layer 1 -> Layer 0).
Also provides the ACMS index data model and file traversal engine for
indexing large projects.
Based on ``docs/specification.md`` ~lines 42333-42422, 44405-44420.
"""
from __future__ import annotations
from cleveragents.acms import uko as _uko
from cleveragents.acms.index import (
ACMSIndex,
FileTraversalEngine,
FileType,
IndexEntry,
TierLevel,
)
from cleveragents.acms.uko import (
CODE_DETAIL_LEVEL_MAP,
FUNC_DETAIL_LEVEL_MAP,
@@ -62,6 +72,14 @@ from cleveragents.acms.uko import (
resolve_detail_level,
)
# Re-export everything published by the ``uko`` sub-package so the two
# ``__all__`` lists stay in sync automatically.
__all__: list[str] = list(_uko.__all__)
# Combine exports from both uko and index modules
_uko_exports = list(_uko.__all__)
_index_exports = [
"ACMSIndex",
"FileTraversalEngine",
"FileType",
"IndexEntry",
"TierLevel",
]
__all__: list[str] = _uko_exports + _index_exports
+412
View File
@@ -0,0 +1,412 @@
"""ACMS Index Data Model and File Traversal Engine.
Provides the foundational data model for indexed context entries and a
file traversal engine that can handle 10,000+ files without timeout using
chunked processing.
Based on issue #9579 and ``docs/specification.md`` ~lines 44405-44420.
"""
from __future__ import annotations
from collections.abc import Iterator
from dataclasses import dataclass, field
from datetime import datetime
from enum import StrEnum
from pathlib import Path
class FileType(StrEnum):
"""File type enumeration for index entries."""
PYTHON = "python"
JAVASCRIPT = "javascript"
TYPESCRIPT = "typescript"
JAVA = "java"
RUST = "rust"
MARKDOWN = "markdown"
TEXT = "text"
JSON = "json"
YAML = "yaml"
XML = "xml"
OTHER = "other"
class TierLevel(StrEnum):
"""Tier assignment levels for context entries.
Aligns with the hot/warm/cold storage tier vocabulary used throughout
the ACMS specification and milestone v3.4.0 acceptance criteria.
"""
HOT = "hot" # Core/essential
WARM = "warm" # Important
COLD = "cold" # Supporting
ARCHIVE = "archive" # Reference
@dataclass
class IndexEntry:
"""Represents a single indexed context entry.
Attributes:
path: Absolute or relative file path
file_type: Type of file (from FileType enum)
size_bytes: File size in bytes
created_at: File creation timestamp
modified_at: File modification timestamp
tags: Set of tags for categorization
tier: Tier assignment level
metadata: Additional metadata dictionary
"""
path: str
file_type: FileType
size_bytes: int
created_at: datetime
modified_at: datetime
tags: set[str] = field(default_factory=set)
tier: TierLevel = TierLevel.COLD
metadata: dict[str, str] = field(default_factory=dict)
def add_tag(self, tag: str) -> None:
"""Add a tag to this entry.
Args:
tag: Tag string to add. Must be non-empty.
Raises:
ValueError: If tag is empty or whitespace-only.
"""
if not tag or not tag.strip():
raise ValueError("tag must be a non-empty string")
self.tags.add(tag)
def remove_tag(self, tag: str) -> None:
"""Remove a tag from this entry.
Args:
tag: Tag string to remove. Must be non-empty.
Raises:
ValueError: If tag is empty or whitespace-only.
"""
if not tag or not tag.strip():
raise ValueError("tag must be a non-empty string")
self.tags.discard(tag)
def has_tag(self, tag: str) -> bool:
"""Check if entry has a specific tag."""
return tag in self.tags
def set_tier(self, tier: TierLevel) -> None:
"""Set the tier level for this entry."""
self.tier = tier
@dataclass
class ACMSIndex:
"""ACMS Index for storing and querying indexed context entries.
Provides storage and query capabilities for indexed files with support
for filtering by path, tags, type, and recency.
Attributes:
entries: Dictionary mapping file paths to IndexEntry objects
"""
entries: dict[str, IndexEntry] = field(default_factory=dict)
def add_entry(self, entry: IndexEntry) -> None:
"""Add an index entry to the index.
Args:
entry: The IndexEntry to add. Must not be None.
Raises:
TypeError: If entry is not an IndexEntry instance.
"""
if not isinstance(entry, IndexEntry):
raise TypeError(
f"entry must be an IndexEntry instance, got {type(entry).__name__}"
)
self.entries[entry.path] = entry
def remove_entry(self, path: str) -> bool:
"""Remove an entry by path. Returns True if removed, False if not found."""
if path in self.entries:
del self.entries[path]
return True
return False
def get_entry(self, path: str) -> IndexEntry | None:
"""Get an entry by path."""
return self.entries.get(path)
def query_by_path(self, path_pattern: str) -> list[IndexEntry]:
"""Query entries by path pattern (substring match).
Args:
path_pattern: Substring pattern to match against entry paths.
Must be non-empty.
Raises:
ValueError: If path_pattern is empty or whitespace-only.
"""
if not path_pattern or not path_pattern.strip():
raise ValueError("path_pattern must be a non-empty string")
return [entry for entry in self.entries.values() if path_pattern in entry.path]
def query_by_tag(self, tag: str) -> list[IndexEntry]:
"""Query entries by tag."""
return [entry for entry in self.entries.values() if entry.has_tag(tag)]
def query_by_type(self, file_type: FileType) -> list[IndexEntry]:
"""Query entries by file type."""
return [
entry for entry in self.entries.values() if entry.file_type == file_type
]
def query_by_tier(self, tier: TierLevel) -> list[IndexEntry]:
"""Query entries by tier level."""
return [entry for entry in self.entries.values() if entry.tier == tier]
def query_by_recency(
self, after: datetime, before: datetime | None = None
) -> list[IndexEntry]:
"""Query entries by modification recency.
Args:
after: Return entries modified after this datetime.
before: Return entries modified before this datetime (optional).
When both are provided, before must be >= after.
Returns:
List of entries matching the recency criteria.
Raises:
ValueError: If before is earlier than after when both are provided.
"""
if before is not None and before < after:
raise ValueError(
f"before ({before}) must be >= after ({after}) when both are provided"
)
results = [
entry for entry in self.entries.values() if entry.modified_at >= after
]
if before:
results = [entry for entry in results if entry.modified_at <= before]
return results
def query_combined(
self,
path_pattern: str | None = None,
tags: set[str] | None = None,
file_type: FileType | None = None,
tier: TierLevel | None = None,
after: datetime | None = None,
) -> list[IndexEntry]:
"""Query entries with multiple filters (AND logic).
Args:
path_pattern: Filter by path pattern
tags: Filter by any of these tags
file_type: Filter by file type
tier: Filter by tier level
after: Filter by modification date
Returns:
List of entries matching all specified criteria
"""
results = list(self.entries.values())
if path_pattern:
results = [entry for entry in results if path_pattern in entry.path]
if tags:
results = [
entry for entry in results if any(tag in entry.tags for tag in tags)
]
if file_type:
results = [entry for entry in results if entry.file_type == file_type]
if tier:
results = [entry for entry in results if entry.tier == tier]
if after:
results = [entry for entry in results if entry.modified_at >= after]
return results
def get_all_entries(self) -> list[IndexEntry]:
"""Get all entries in the index."""
return list(self.entries.values())
def get_entry_count(self) -> int:
"""Get the total number of entries in the index."""
return len(self.entries)
class FileTraversalEngine:
"""Engine for traversing and indexing files in large projects.
Handles 10,000+ files without timeout using chunked processing to
prevent memory exhaustion.
Attributes:
chunk_size: Number of files to process in each chunk
index: The ACMS index to populate
"""
def __init__(self, chunk_size: int = 100, index: ACMSIndex | None = None) -> None:
"""Initialize the traversal engine.
Args:
chunk_size: Number of files to process per chunk (default: 100).
Must be a positive integer.
index: Optional pre-populated ACMSIndex to use. If None, a new
empty index is created (supports Dependency Inversion).
Raises:
ValueError: If chunk_size is not a positive integer.
"""
if chunk_size <= 0:
raise ValueError(f"chunk_size must be a positive integer, got {chunk_size}")
self.chunk_size = chunk_size
self.index = index if index is not None else ACMSIndex()
def _get_file_type(self, file_path: Path) -> FileType:
"""Determine file type from extension."""
suffix = file_path.suffix.lower()
type_map = {
".py": FileType.PYTHON,
".js": FileType.JAVASCRIPT,
".ts": FileType.TYPESCRIPT,
".tsx": FileType.TYPESCRIPT,
".java": FileType.JAVA,
".rs": FileType.RUST,
".md": FileType.MARKDOWN,
".txt": FileType.TEXT,
".json": FileType.JSON,
".yaml": FileType.YAML,
".yml": FileType.YAML,
".xml": FileType.XML,
}
return type_map.get(suffix, FileType.OTHER)
def _create_index_entry(self, file_path: Path) -> IndexEntry | None:
"""Create an index entry from a file path.
Args:
file_path: Path to the file
Returns:
IndexEntry if successful, None if file cannot be read
"""
try:
stat = file_path.stat()
return IndexEntry(
path=str(file_path),
file_type=self._get_file_type(file_path),
size_bytes=stat.st_size,
created_at=datetime.fromtimestamp(stat.st_ctime),
modified_at=datetime.fromtimestamp(stat.st_mtime),
)
except (OSError, ValueError):
return None
def _traverse_directory(
self,
root_path: Path,
) -> Iterator[Path]:
"""Recursively traverse directory and yield file paths.
Args:
root_path: Root directory to traverse
Yields:
Path objects for each file found
"""
try:
for item in root_path.iterdir():
if item.is_file():
yield item
elif item.is_dir():
# Recursively traverse subdirectories
yield from self._traverse_directory(item)
except (OSError, PermissionError):
# Skip directories we can't read
pass
def traverse_and_index(
self,
root_path: str | Path,
exclude_patterns: list[str] | None = None,
) -> ACMSIndex:
"""Traverse a directory and index all files.
Uses chunked processing to handle large projects without timeout
or memory exhaustion.
Args:
root_path: Root directory to traverse
exclude_patterns: List of patterns to exclude
(e.g., ['.git', '__pycache__'])
Returns:
Populated ACMSIndex
"""
root = Path(root_path)
if not root.exists():
raise ValueError(f"Path does not exist: {root_path}")
exclude_patterns = exclude_patterns or []
chunk: list[IndexEntry] = []
for file_path in self._traverse_directory(root):
# Check if file matches any exclude pattern
if any(pattern in str(file_path) for pattern in exclude_patterns):
continue
# Create index entry
entry = self._create_index_entry(file_path)
if entry:
chunk.append(entry)
# Process chunk when it reaches the size limit
if len(chunk) >= self.chunk_size:
self._process_chunk(chunk)
chunk = []
# Process remaining entries
if chunk:
self._process_chunk(chunk)
return self.index
def _process_chunk(self, chunk: list[IndexEntry]) -> None:
"""Process a chunk of index entries.
Args:
chunk: List of IndexEntry objects to add to the index
"""
for entry in chunk:
self.index.add_entry(entry)
def get_index(self) -> ACMSIndex:
"""Get the current index."""
return self.index
def reset_index(self) -> None:
"""Reset the index to empty state."""
self.index = ACMSIndex()
__all__ = [
"ACMSIndex",
"FileTraversalEngine",
"FileType",
"IndexEntry",
"TierLevel",
]