feat(acms): projects with 10,000+ files index without timeout

Add explicit indexing timeout bounds and propagate timeout_seconds through the repo indexing service, utility walker, and CLI entrypoint so large repositories fail predictably instead of hanging.

Add 10K-scale Behave/Robot verification and a dedicated benchmark path to validate that large-project indexing completes, persists status correctly, and remains measurable for regressions.

Address review feedback by splitting oversized Behave and Robot support files into focused modules, trimming repo_indexing_service.py back under the 500-line limit, and rebasing the branch onto current master.

ISSUES CLOSED: #851

# Conflicts:
#	CHANGELOG.md
#	robot/helper_m5_e2e_verification.py

CHANGELOG.md

@@ -220,6 +220,11 @@
   strategize->execute->apply narrowing diagnostics, Robot acceptance
   verification upgrades, and Behave edge-case coverage for empty,
   single-resource, multi-resource, and budget-constrained contexts. (#849)
+- Added 10,000-file ACMS indexing reliability improvements: configurable
+  runtime timeout bounds propagated through the repository indexing service,
+  utility walker, and CLI (`agents repo index --timeout-seconds`), plus
+  large-scale verification coverage in Behave and Robot and a dedicated
+  10K-file indexing benchmark path for regression tracking.  (#851)
 - Added TDD bug-capture E2E tests for bug #1028 — ACMS indexing pipeline not
   wired into CLI. Four Robot Framework E2E tests prove ContextTierService starts
   empty on every CLI invocation. Tests use `@tdd_expected_fail` until the bug

benchmarks/context_indexing_bench.py

@@ -127,3 +127,29 @@ class TrackRepoIndexing:
         return idx.metadata.token_estimate
 
     track_token_estimate.unit = "tokens"  # type: ignore[attr-defined]
+
+
+class TimeRepoIndexingLarge:
+    """Benchmark indexing throughput for 10,000-file repositories."""
+
+    timeout = 1200
+
+    def setup(self) -> None:
+        """Create a temporary project with exactly 10,000 Python files."""
+        self.svc = _make_service()
+        self.tmpdir = tempfile.mkdtemp()
+        root = Path(self.tmpdir)
+        for i in range(100):
+            module_dir = root / f"module_{i:03d}"
+            module_dir.mkdir(parents=True, exist_ok=True)
+            for j in range(100):
+                (module_dir / f"file_{j:03d}.py").write_text(
+                    f"def f_{i}_{j}():\n    return {i + j}\n"
+                )
+
+    def teardown(self) -> None:
+        shutil.rmtree(self.tmpdir, ignore_errors=True)
+
+    def time_full_index_10000_files(self) -> None:
+        """Full index for a 10,000-file repository."""
+        self.svc.index_resource(_RID, self.tmpdir, timeout_seconds=120.0)

features/repo_indexing.feature

@@ -90,6 +90,15 @@ Feature: Repository indexing with incremental refresh and language detection
     When I run repo-index full index with max total size 100 bytes
     Then the repo-index result file count should be less than 6
 
+  @feature195
+  Scenario: Indexing 10,000 files completes within timeout bound
+    Given a repo-index fresh in-memory indexing service
+    And a repo-index temporary directory with 10000 python files
+    When I run repo-index full index with timeout of 120 seconds
+    Then the repo-index result status should be "ready"
+    And the repo-index result file count should be 10000
+    And the repo-index indexing elapsed seconds should be less than 120
+
   # -- Removal -------------------------------------------------------------
 
   @feature195

features/steps/repo_indexing_assertion_steps.py

@@ -0,0 +1,328 @@
+"""Assertion and cleanup steps for ``features/repo_indexing.feature``."""
+
+from __future__ import annotations
+
+from behave import given, then, when  # type: ignore[import-untyped]
+from behave.runner import Context  # type: ignore[import-untyped]
+
+from cleveragents.application.services.repo_indexing_service import (
+    RepoIndexingService,
+)
+from cleveragents.domain.models.core.repo_index import (
+    FileRecord,
+    IndexMetadata,
+    RepoIndex,
+)
+from cleveragents.infrastructure.database.models import RepoIndexModel
+
+_RESOURCE_ID = "01KK0D8WNATFNEX2JMG5GKF6FM"
+_STALE_RESOURCE_ID = "01KK0D8WNATFNEX2JMG5GKF6FN"
+
+
+@then('the repo-index result status should be "{status}"')
+def step_check_status(context: Context, status: str) -> None:
+    """Assert the index result status."""
+    result: RepoIndex = context.repo_index_result  # type: ignore[attr-defined]
+    actual = result.metadata.status.value
+    assert actual == status, f"Expected status '{status}', got '{actual}'"
+
+
+@then("the repo-index result file count should be {count:d}")
+def step_check_file_count(context: Context, count: int) -> None:
+    """Assert the file count."""
+    result: RepoIndex = context.repo_index_result  # type: ignore[attr-defined]
+    actual = result.metadata.file_count
+    assert actual == count, (
+        f"Expected {count} files, got {actual}. Files: {[f.path for f in result.files]}"
+    )
+
+
+@then("the repo-index result file count should be less than {count:d}")
+def step_check_file_count_less(context: Context, count: int) -> None:
+    """Assert the file count is less than a value."""
+    result: RepoIndex = context.repo_index_result  # type: ignore[attr-defined]
+    actual = result.metadata.file_count
+    assert actual < count, f"Expected fewer than {count} files, got {actual}"
+
+
+@then("the repo-index indexing elapsed seconds should be less than {seconds:d}")
+def step_check_index_elapsed(context: Context, seconds: int) -> None:
+    """Assert indexing runtime remained under the configured bound."""
+    elapsed = context.repo_index_elapsed_seconds  # type: ignore[attr-defined]
+    assert elapsed < float(seconds), (
+        f"Expected elapsed < {seconds}s, got {elapsed:.2f}s"
+    )
+
+
+@then('the repo-index primary language should be "{lang}"')
+def step_check_language(context: Context, lang: str) -> None:
+    """Assert the primary language."""
+    result: RepoIndex = context.repo_index_result  # type: ignore[attr-defined]
+    actual = result.metadata.primary_language
+    assert actual == lang, f"Expected language '{lang}', got '{actual}'"
+
+
+@then("the repo-index token estimate should be positive")
+def step_check_tokens_positive(context: Context) -> None:
+    """Assert token estimate is > 0."""
+    result: RepoIndex = context.repo_index_result  # type: ignore[attr-defined]
+    assert result.metadata.token_estimate > 0, (
+        f"Expected positive token_estimate, got {result.metadata.token_estimate}"
+    )
+
+
+@then('the repo-index file "{filename}" should have language "{lang}"')
+def step_check_file_language(context: Context, filename: str, lang: str) -> None:
+    """Assert a specific file's detected language."""
+    result: RepoIndex = context.repo_index_result  # type: ignore[attr-defined]
+    matches = [f for f in result.files if f.path == filename]
+    assert matches, (
+        f"File '{filename}' not found in index. Files: {[f.path for f in result.files]}"
+    )
+    actual = matches[0].language
+    assert actual == lang, f"Expected '{filename}' language='{lang}', got '{actual}'"
+
+
+@then("the repo-index retrieved index should match the original")
+def step_check_retrieved_matches(context: Context) -> None:
+    """Assert the retrieved index matches the original."""
+    original: RepoIndex = context.repo_index_original_result  # type: ignore[attr-defined]
+    retrieved: RepoIndex = context.repo_index_retrieved  # type: ignore[attr-defined]
+    assert retrieved is not None, "Retrieved index is None"
+    assert retrieved.metadata.file_count == original.metadata.file_count
+    assert retrieved.metadata.token_estimate == original.metadata.token_estimate
+    assert retrieved.metadata.primary_language == original.metadata.primary_language
+    assert len(retrieved.files) == len(original.files)
+
+
+@then("the repo-index retrieved file hashes should match the original")
+def step_check_retrieved_hashes(context: Context) -> None:
+    """Assert the retrieved index has identical per-file content hashes."""
+    original: RepoIndex = context.repo_index_original_result  # type: ignore[attr-defined]
+    retrieved: RepoIndex = context.repo_index_retrieved  # type: ignore[attr-defined]
+    orig_map = {f.path: f.content_hash for f in original.files}
+    retr_map = {f.path: f.content_hash for f in retrieved.files}
+    assert orig_map == retr_map, (
+        f"File hash mismatch.\nOriginal: {orig_map}\nRetrieved: {retr_map}"
+    )
+
+
+@then('the repo-index file "{filename}" content hash should differ from original')
+def step_check_hash_differs(context: Context, filename: str) -> None:
+    """Assert a file's hash changed after modification."""
+    original: RepoIndex = context.repo_index_original_result  # type: ignore[attr-defined]
+    current: RepoIndex = context.repo_index_result  # type: ignore[attr-defined]
+    orig_hash = next(
+        (f.content_hash for f in original.files if f.path == filename),
+        None,
+    )
+    curr_hash = next(
+        (f.content_hash for f in current.files if f.path == filename),
+        None,
+    )
+    assert orig_hash is not None, f"'{filename}' not found in original index"
+    assert curr_hash is not None, f"'{filename}' not found in current index"
+    assert orig_hash != curr_hash, (
+        f"Expected hash for '{filename}' to change, but both are '{orig_hash}'"
+    )
+
+
+@then('the repo-index file "{filename}" content hash should match original')
+def step_check_hash_matches(context: Context, filename: str) -> None:
+    """Assert a file's hash is unchanged."""
+    original: RepoIndex = context.repo_index_original_result  # type: ignore[attr-defined]
+    current: RepoIndex = context.repo_index_result  # type: ignore[attr-defined]
+    orig_hash = next(
+        (f.content_hash for f in original.files if f.path == filename),
+        None,
+    )
+    curr_hash = next(
+        (f.content_hash for f in current.files if f.path == filename),
+        None,
+    )
+    assert orig_hash is not None, f"'{filename}' not found in original index"
+    assert curr_hash is not None, f"'{filename}' not found in current index"
+    assert orig_hash == curr_hash, (
+        f"Expected hash for '{filename}' to match, but got '{orig_hash}' vs '{curr_hash}'"
+    )
+
+
+@then('the repo-index result should not include file "{filename}"')
+def step_check_file_excluded(context: Context, filename: str) -> None:
+    """Assert a file is not in the index."""
+    result: RepoIndex = context.repo_index_result  # type: ignore[attr-defined]
+    paths = [f.path for f in result.files]
+    assert filename not in paths, f"File '{filename}' should be excluded but was found"
+
+
+@then("a repo-index error should be raised")
+def step_check_any_error(context: Context) -> None:
+    """Assert some error was raised."""
+    err = context.repo_index_error  # type: ignore[attr-defined]
+    assert err is not None, "Expected an error but none raised"
+
+
+@then('querying repo-index status for the resource should show "{status}"')
+def step_check_error_status_persisted(context: Context, status: str) -> None:
+    """Assert the persisted index status matches."""
+    service: RepoIndexingService = context.repo_index_service  # type: ignore[attr-defined]
+    meta = service.get_index_status(_RESOURCE_ID)
+    assert meta is not None, (
+        f"Expected persisted status '{status}' but get_index_status returned None"
+    )
+    actual = meta.status.value
+    assert actual == status, f"Expected status '{status}', got '{actual}'"
+
+
+@then("the repo-index previous index should still be retrievable with {count:d} files")
+def step_check_previous_index(context: Context, count: int) -> None:
+    """Assert the previous index is still intact."""
+    service: RepoIndexingService = context.repo_index_service  # type: ignore[attr-defined]
+    idx = service.get_index(_RESOURCE_ID)
+    assert idx is not None, "Previous index was lost"
+    assert idx.metadata.file_count == count, (
+        f"Expected {count} files, got {idx.metadata.file_count}"
+    )
+
+
+@then("the FileRecord last_modified should have UTC timezone")
+def step_check_filerecord_utc(context: Context) -> None:
+    """Assert the FileRecord has UTC timezone on last_modified."""
+    from datetime import UTC
+
+    fr: FileRecord = context.repo_index_filerecord  # type: ignore[attr-defined]
+    assert fr.last_modified.tzinfo is not None, "last_modified has no timezone"
+    assert fr.last_modified.tzinfo == UTC, (
+        f"Expected UTC, got {fr.last_modified.tzinfo}"
+    )
+
+
+@then("the IndexMetadata indexed_at should have UTC timezone")
+def step_check_metadata_utc(context: Context) -> None:
+    """Assert the IndexMetadata has UTC timezone on indexed_at."""
+    from datetime import UTC
+
+    meta: IndexMetadata = context.repo_index_metadata_naive  # type: ignore[attr-defined]
+    assert meta.indexed_at.tzinfo is not None, "indexed_at has no timezone"
+    assert meta.indexed_at.tzinfo == UTC, f"Expected UTC, got {meta.indexed_at.tzinfo}"
+
+
+@then('the repo-index status metadata should have status "{status}"')
+def step_check_status_metadata(context: Context, status: str) -> None:
+    """Assert the status metadata has the expected status."""
+    meta: IndexMetadata = context.repo_index_status  # type: ignore[attr-defined]
+    assert meta is not None, "get_index_status returned None"
+    actual = meta.status.value
+    assert actual == status, f"Expected status '{status}', got '{actual}'"
+
+
+@then("the repo-index status metadata file count should be {count:d}")
+def step_check_status_file_count(context: Context, count: int) -> None:
+    """Assert the status metadata file count."""
+    meta: IndexMetadata = context.repo_index_status  # type: ignore[attr-defined]
+    assert meta is not None, "get_index_status returned None"
+    actual = meta.file_count
+    assert actual == count, f"Expected {count} files, got {actual}"
+
+
+@then("querying repo-index for the resource should return nothing")
+def step_check_removed(context: Context) -> None:
+    """Assert the index is gone."""
+    service: RepoIndexingService = context.repo_index_service  # type: ignore[attr-defined]
+    result = service.get_index(_RESOURCE_ID)
+    assert result is None, f"Expected None after removal, got {result}"
+
+
+@then("the repo-index removal result should be false")
+def step_check_removal_false(context: Context) -> None:
+    """Assert removal returned False."""
+    assert context.repo_index_removal_result is False  # type: ignore[attr-defined]
+
+
+@then('a repo-index ValueError should be raised mentioning "{text}"')
+def step_check_value_error(context: Context, text: str) -> None:
+    """Assert a ValueError was raised with expected message."""
+    err = context.repo_index_error  # type: ignore[attr-defined]
+    assert err is not None, "Expected ValueError but none raised"
+    assert isinstance(err, ValueError), f"Expected ValueError, got {type(err).__name__}"
+    assert text in str(err), f"Expected '{text}' in error: {err}"
+
+
+@then("a repo-index FileNotFoundError should be raised")
+def step_check_fnf_error(context: Context) -> None:
+    """Assert a FileNotFoundError was raised."""
+    err = context.repo_index_error  # type: ignore[attr-defined]
+    assert err is not None, "Expected FileNotFoundError but none raised"
+    assert isinstance(err, FileNotFoundError), (
+        f"Expected FileNotFoundError, got {type(err).__name__}"
+    )
+
+
+@then('the repo-index metadata should have field "{field}"')
+def step_check_metadata_field(context: Context, field: str) -> None:
+    """Assert the metadata has a specific field."""
+    result: RepoIndex = context.repo_index_result  # type: ignore[attr-defined]
+    assert hasattr(result.metadata, field), (
+        f"IndexMetadata missing field '{field}'. "
+        f"Available: {list(result.metadata.model_fields.keys())}"
+    )
+
+
+@then('each repo-index file record should have field "{field}"')
+def step_check_file_field(context: Context, field: str) -> None:
+    """Assert every file record has a specific field."""
+    result: RepoIndex = context.repo_index_result  # type: ignore[attr-defined]
+    assert len(result.files) > 0, "No file records to check"
+    for fr in result.files:
+        assert hasattr(fr, field), (
+            f"FileRecord missing field '{field}'. "
+            f"Available: {list(fr.model_fields.keys())}"
+        )
+
+
+@given("a repo-index stale INDEXING row is inserted for a resource")
+def step_insert_stale_indexing_row(context: Context) -> None:
+    """Manually insert an INDEXING-status row to simulate a crash."""
+    from datetime import UTC, datetime
+
+    svc: RepoIndexingService = context.repo_index_service  # type: ignore[attr-defined]
+    session = svc._session_factory()
+    try:
+        row = RepoIndexModel(
+            index_id="01KK0D8WNATFNEX2JMG5GKF6FZ",
+            resource_id=_STALE_RESOURCE_ID,
+            file_count=0,
+            token_estimate=0,
+            primary_language="unknown",
+            status="indexing",
+            indexed_at=datetime.now(tz=UTC).isoformat(),
+            created_at=datetime.now(tz=UTC).isoformat(),
+        )
+        session.add(row)
+        session.commit()
+    finally:
+        session.close()
+    context.repo_index_stale_resource_id = _STALE_RESOURCE_ID  # type: ignore[attr-defined]
+
+
+@when("I call repo-index cleanup stale indexing")
+def step_call_cleanup(context: Context) -> None:
+    """Call cleanup_stale_indexing on the service."""
+    svc: RepoIndexingService = context.repo_index_service  # type: ignore[attr-defined]
+    context.repo_index_cleanup_count = svc.cleanup_stale_indexing()  # type: ignore[attr-defined]
+
+
+@then("the repo-index cleanup should report {count:d} stale row removed")
+def step_check_cleanup_count(context: Context, count: int) -> None:
+    """Assert the cleanup returned the expected count."""
+    actual = context.repo_index_cleanup_count  # type: ignore[attr-defined]
+    assert actual == count, f"Expected {count} stale rows, got {actual}"
+
+
+@then("querying repo-index for the stale resource should return nothing")
+def step_check_stale_removed(context: Context) -> None:
+    """Assert the stale row no longer exists."""
+    svc: RepoIndexingService = context.repo_index_service  # type: ignore[attr-defined]
+    rid = context.repo_index_stale_resource_id  # type: ignore[attr-defined]
+    result = svc.get_index(rid)
+    assert result is None, f"Expected None, got {result}"

features/steps/repo_indexing_steps.py

@@ -9,10 +9,11 @@ from __future__ import annotations
 
 import shutil
 import tempfile
+import time
 from pathlib import Path
 from unittest.mock import patch
 
-from behave import given, then, when  # type: ignore[import-untyped]
+from behave import given, when  # type: ignore[import-untyped]
 from behave.runner import Context  # type: ignore[import-untyped]
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
@@ -24,9 +25,8 @@ from cleveragents.domain.models.core.repo_index import (
     FileRecord,
     IndexMetadata,
     IndexStatus,
-    RepoIndex,
 )
-from cleveragents.infrastructure.database.models import Base, RepoIndexModel
+from cleveragents.infrastructure.database.models import Base
 
 _RESOURCE_ID = "01KK0D8WNATFNEX2JMG5GKF6FM"
 
@@ -91,6 +91,25 @@ def step_sample_project_with_big_file(context: Context) -> None:
     (tmpdir / "big_file.dat").write_bytes(b"x" * 2_000_000)
 
 
+@given("a repo-index temporary directory with 10000 python files")
+def step_large_directory_10000_files(context: Context) -> None:
+    """Create a temporary directory containing 10,000 small Python files."""
+    tmpdir = Path(tempfile.mkdtemp())
+    created = 0
+    for i in range(100):
+        module_dir = tmpdir / f"module_{i:03d}"
+        module_dir.mkdir(parents=True, exist_ok=True)
+        for j in range(100):
+            (module_dir / f"file_{j:03d}.py").write_text(
+                f"def f_{i}_{j}():\n    return {i + j}\n"
+            )
+            created += 1
+
+    assert created == 10_000, f"Expected 10000 files, created {created}"
+    context.repo_index_tmpdir = str(tmpdir)  # type: ignore[attr-defined]
+    context.add_cleanup(_cleanup_tmpdir, context)
+
+
 @given("a repo-index empty temporary directory")
 def step_empty_directory(context: Context) -> None:
     """Create an empty temporary directory."""
@@ -235,6 +254,20 @@ def step_run_index_max_total_size(context: Context, size: int) -> None:
     context.repo_index_result = result  # type: ignore[attr-defined]
 
 
+@when("I run repo-index full index with timeout of {seconds:d} seconds")
+def step_run_index_with_timeout(context: Context, seconds: int) -> None:
+    """Run index with an explicit timeout and capture elapsed time."""
+    service: RepoIndexingService = context.repo_index_service  # type: ignore[attr-defined]
+    started = time.monotonic()
+    result = service.index_resource(
+        _RESOURCE_ID,
+        context.repo_index_tmpdir,  # type: ignore[attr-defined]
+        timeout_seconds=float(seconds),
+    )
+    context.repo_index_result = result  # type: ignore[attr-defined]
+    context.repo_index_elapsed_seconds = time.monotonic() - started  # type: ignore[attr-defined]
+
+
 @when("I remove the repo-index for the resource")
 def step_remove_index(context: Context) -> None:
     """Remove the index."""
@@ -419,314 +452,3 @@ def step_index_bad_path(context: Context, path: str) -> None:
         service.index_resource(_RESOURCE_ID, path)
     except FileNotFoundError as exc:
         context.repo_index_error = exc  # type: ignore[attr-defined]
-
-
-# ── Then steps ───────────────────────────────────────────────────────
-
-
-@then('the repo-index result status should be "{status}"')
-def step_check_status(context: Context, status: str) -> None:
-    """Assert the index result status."""
-    result: RepoIndex = context.repo_index_result  # type: ignore[attr-defined]
-    actual = result.metadata.status.value
-    assert actual == status, f"Expected status '{status}', got '{actual}'"
-
-
-@then("the repo-index result file count should be {count:d}")
-def step_check_file_count(context: Context, count: int) -> None:
-    """Assert the file count."""
-    result: RepoIndex = context.repo_index_result  # type: ignore[attr-defined]
-    actual = result.metadata.file_count
-    assert actual == count, (
-        f"Expected {count} files, got {actual}. Files: {[f.path for f in result.files]}"
-    )
-
-
-@then("the repo-index result file count should be less than {count:d}")
-def step_check_file_count_less(context: Context, count: int) -> None:
-    """Assert the file count is less than a value."""
-    result: RepoIndex = context.repo_index_result  # type: ignore[attr-defined]
-    actual = result.metadata.file_count
-    assert actual < count, f"Expected fewer than {count} files, got {actual}"
-
-
-@then('the repo-index primary language should be "{lang}"')
-def step_check_language(context: Context, lang: str) -> None:
-    """Assert the primary language."""
-    result: RepoIndex = context.repo_index_result  # type: ignore[attr-defined]
-    actual = result.metadata.primary_language
-    assert actual == lang, f"Expected language '{lang}', got '{actual}'"
-
-
-@then("the repo-index token estimate should be positive")
-def step_check_tokens_positive(context: Context) -> None:
-    """Assert token estimate is > 0."""
-    result: RepoIndex = context.repo_index_result  # type: ignore[attr-defined]
-    assert result.metadata.token_estimate > 0, (
-        f"Expected positive token_estimate, got {result.metadata.token_estimate}"
-    )
-
-
-@then('the repo-index file "{filename}" should have language "{lang}"')
-def step_check_file_language(context: Context, filename: str, lang: str) -> None:
-    """Assert a specific file's detected language."""
-    result: RepoIndex = context.repo_index_result  # type: ignore[attr-defined]
-    matches = [f for f in result.files if f.path == filename]
-    assert matches, (
-        f"File '{filename}' not found in index. Files: {[f.path for f in result.files]}"
-    )
-    actual = matches[0].language
-    assert actual == lang, f"Expected '{filename}' language='{lang}', got '{actual}'"
-
-
-@then("the repo-index retrieved index should match the original")
-def step_check_retrieved_matches(context: Context) -> None:
-    """Assert the retrieved index matches the original."""
-    original: RepoIndex = context.repo_index_original_result  # type: ignore[attr-defined]
-    retrieved: RepoIndex = context.repo_index_retrieved  # type: ignore[attr-defined]
-    assert retrieved is not None, "Retrieved index is None"
-    assert retrieved.metadata.file_count == original.metadata.file_count
-    assert retrieved.metadata.token_estimate == original.metadata.token_estimate
-    assert retrieved.metadata.primary_language == original.metadata.primary_language
-    assert len(retrieved.files) == len(original.files)
-
-
-@then("the repo-index retrieved file hashes should match the original")
-def step_check_retrieved_hashes(context: Context) -> None:
-    """Assert the retrieved index has identical per-file content hashes."""
-    original: RepoIndex = context.repo_index_original_result  # type: ignore[attr-defined]
-    retrieved: RepoIndex = context.repo_index_retrieved  # type: ignore[attr-defined]
-    orig_map = {f.path: f.content_hash for f in original.files}
-    retr_map = {f.path: f.content_hash for f in retrieved.files}
-    assert orig_map == retr_map, (
-        f"File hash mismatch.\nOriginal: {orig_map}\nRetrieved: {retr_map}"
-    )
-
-
-@then('the repo-index file "{filename}" content hash should differ from original')
-def step_check_hash_differs(context: Context, filename: str) -> None:
-    """Assert a file's hash changed after modification."""
-    original: RepoIndex = context.repo_index_original_result  # type: ignore[attr-defined]
-    current: RepoIndex = context.repo_index_result  # type: ignore[attr-defined]
-    orig_hash = next(
-        (f.content_hash for f in original.files if f.path == filename),
-        None,
-    )
-    curr_hash = next(
-        (f.content_hash for f in current.files if f.path == filename),
-        None,
-    )
-    assert orig_hash is not None, f"'{filename}' not found in original index"
-    assert curr_hash is not None, f"'{filename}' not found in current index"
-    assert orig_hash != curr_hash, (
-        f"Expected hash for '{filename}' to change, but both are '{orig_hash}'"
-    )
-
-
-@then('the repo-index file "{filename}" content hash should match original')
-def step_check_hash_matches(context: Context, filename: str) -> None:
-    """Assert a file's hash is unchanged."""
-    original: RepoIndex = context.repo_index_original_result  # type: ignore[attr-defined]
-    current: RepoIndex = context.repo_index_result  # type: ignore[attr-defined]
-    orig_hash = next(
-        (f.content_hash for f in original.files if f.path == filename),
-        None,
-    )
-    curr_hash = next(
-        (f.content_hash for f in current.files if f.path == filename),
-        None,
-    )
-    assert orig_hash is not None, f"'{filename}' not found in original index"
-    assert curr_hash is not None, f"'{filename}' not found in current index"
-    assert orig_hash == curr_hash, (
-        f"Expected hash for '{filename}' to match, but got '{orig_hash}' vs '{curr_hash}'"
-    )
-
-
-@then('the repo-index result should not include file "{filename}"')
-def step_check_file_excluded(context: Context, filename: str) -> None:
-    """Assert a file is not in the index."""
-    result: RepoIndex = context.repo_index_result  # type: ignore[attr-defined]
-    paths = [f.path for f in result.files]
-    assert filename not in paths, f"File '{filename}' should be excluded but was found"
-
-
-@then("a repo-index error should be raised")
-def step_check_any_error(context: Context) -> None:
-    """Assert some error was raised."""
-    err = context.repo_index_error  # type: ignore[attr-defined]
-    assert err is not None, "Expected an error but none raised"
-
-
-@then('querying repo-index status for the resource should show "{status}"')
-def step_check_error_status_persisted(context: Context, status: str) -> None:
-    """Assert the persisted index status matches."""
-    service: RepoIndexingService = context.repo_index_service  # type: ignore[attr-defined]
-    meta = service.get_index_status(_RESOURCE_ID)
-    assert meta is not None, (
-        f"Expected persisted status '{status}' but get_index_status returned None"
-    )
-    actual = meta.status.value
-    assert actual == status, f"Expected status '{status}', got '{actual}'"
-
-
-@then("the repo-index previous index should still be retrievable with {count:d} files")
-def step_check_previous_index(context: Context, count: int) -> None:
-    """Assert the previous index is still intact."""
-    service: RepoIndexingService = context.repo_index_service  # type: ignore[attr-defined]
-    idx = service.get_index(_RESOURCE_ID)
-    assert idx is not None, "Previous index was lost"
-    assert idx.metadata.file_count == count, (
-        f"Expected {count} files, got {idx.metadata.file_count}"
-    )
-
-
-@then("the FileRecord last_modified should have UTC timezone")
-def step_check_filerecord_utc(context: Context) -> None:
-    """Assert the FileRecord has UTC timezone on last_modified."""
-    from datetime import UTC
-
-    fr: FileRecord = context.repo_index_filerecord  # type: ignore[attr-defined]
-    assert fr.last_modified.tzinfo is not None, "last_modified has no timezone"
-    assert fr.last_modified.tzinfo == UTC, (
-        f"Expected UTC, got {fr.last_modified.tzinfo}"
-    )
-
-
-@then("the IndexMetadata indexed_at should have UTC timezone")
-def step_check_metadata_utc(context: Context) -> None:
-    """Assert the IndexMetadata has UTC timezone on indexed_at."""
-    from datetime import UTC
-
-    meta: IndexMetadata = context.repo_index_metadata_naive  # type: ignore[attr-defined]
-    assert meta.indexed_at.tzinfo is not None, "indexed_at has no timezone"
-    assert meta.indexed_at.tzinfo == UTC, f"Expected UTC, got {meta.indexed_at.tzinfo}"
-
-
-@then('the repo-index status metadata should have status "{status}"')
-def step_check_status_metadata(context: Context, status: str) -> None:
-    """Assert the status metadata has the expected status."""
-
-    meta: IndexMetadata = context.repo_index_status  # type: ignore[attr-defined]
-    assert meta is not None, "get_index_status returned None"
-    actual = meta.status.value
-    assert actual == status, f"Expected status '{status}', got '{actual}'"
-
-
-@then("the repo-index status metadata file count should be {count:d}")
-def step_check_status_file_count(context: Context, count: int) -> None:
-    """Assert the status metadata file count."""
-
-    meta: IndexMetadata = context.repo_index_status  # type: ignore[attr-defined]
-    assert meta is not None, "get_index_status returned None"
-    actual = meta.file_count
-    assert actual == count, f"Expected {count} files, got {actual}"
-
-
-@then("querying repo-index for the resource should return nothing")
-def step_check_removed(context: Context) -> None:
-    """Assert the index is gone."""
-    service: RepoIndexingService = context.repo_index_service  # type: ignore[attr-defined]
-    result = service.get_index(_RESOURCE_ID)
-    assert result is None, f"Expected None after removal, got {result}"
-
-
-@then("the repo-index removal result should be false")
-def step_check_removal_false(context: Context) -> None:
-    """Assert removal returned False."""
-    assert context.repo_index_removal_result is False  # type: ignore[attr-defined]
-
-
-@then('a repo-index ValueError should be raised mentioning "{text}"')
-def step_check_value_error(context: Context, text: str) -> None:
-    """Assert a ValueError was raised with expected message."""
-    err = context.repo_index_error  # type: ignore[attr-defined]
-    assert err is not None, "Expected ValueError but none raised"
-    assert isinstance(err, ValueError), f"Expected ValueError, got {type(err).__name__}"
-    assert text in str(err), f"Expected '{text}' in error: {err}"
-
-
-@then("a repo-index FileNotFoundError should be raised")
-def step_check_fnf_error(context: Context) -> None:
-    """Assert a FileNotFoundError was raised."""
-    err = context.repo_index_error  # type: ignore[attr-defined]
-    assert err is not None, "Expected FileNotFoundError but none raised"
-    assert isinstance(err, FileNotFoundError), (
-        f"Expected FileNotFoundError, got {type(err).__name__}"
-    )
-
-
-@then('the repo-index metadata should have field "{field}"')
-def step_check_metadata_field(context: Context, field: str) -> None:
-    """Assert the metadata has a specific field."""
-    result: RepoIndex = context.repo_index_result  # type: ignore[attr-defined]
-    assert hasattr(result.metadata, field), (
-        f"IndexMetadata missing field '{field}'. "
-        f"Available: {list(result.metadata.model_fields.keys())}"
-    )
-
-
-@then('each repo-index file record should have field "{field}"')
-def step_check_file_field(context: Context, field: str) -> None:
-    """Assert every file record has a specific field."""
-    result: RepoIndex = context.repo_index_result  # type: ignore[attr-defined]
-    assert len(result.files) > 0, "No file records to check"
-    for fr in result.files:
-        assert hasattr(fr, field), (
-            f"FileRecord missing field '{field}'. "
-            f"Available: {list(fr.model_fields.keys())}"
-        )
-
-
-# ── Stale indexing cleanup steps ─────────────────────────────────────
-
-
-_STALE_RESOURCE_ID = "01KK0D8WNATFNEX2JMG5GKF6FN"
-
-
-@given("a repo-index stale INDEXING row is inserted for a resource")
-def step_insert_stale_indexing_row(context: Context) -> None:
-    """Manually insert an INDEXING-status row to simulate a crash."""
-    from datetime import UTC, datetime
-
-    svc: RepoIndexingService = context.repo_index_service  # type: ignore[attr-defined]
-    session = svc._session_factory()
-    try:
-        row = RepoIndexModel(
-            index_id="01KK0D8WNATFNEX2JMG5GKF6FZ",
-            resource_id=_STALE_RESOURCE_ID,
-            file_count=0,
-            token_estimate=0,
-            primary_language="unknown",
-            status="indexing",
-            indexed_at=datetime.now(tz=UTC).isoformat(),
-            created_at=datetime.now(tz=UTC).isoformat(),
-        )
-        session.add(row)
-        session.commit()
-    finally:
-        session.close()
-    context.repo_index_stale_resource_id = _STALE_RESOURCE_ID  # type: ignore[attr-defined]
-
-
-@when("I call repo-index cleanup stale indexing")
-def step_call_cleanup(context: Context) -> None:
-    """Call cleanup_stale_indexing on the service."""
-    svc: RepoIndexingService = context.repo_index_service  # type: ignore[attr-defined]
-    context.repo_index_cleanup_count = svc.cleanup_stale_indexing()  # type: ignore[attr-defined]
-
-
-@then("the repo-index cleanup should report {count:d} stale row removed")
-def step_check_cleanup_count(context: Context, count: int) -> None:
-    """Assert the cleanup returned the expected count."""
-    actual = context.repo_index_cleanup_count  # type: ignore[attr-defined]
-    assert actual == count, f"Expected {count} stale rows, got {actual}"
-
-
-@then("querying repo-index for the stale resource should return nothing")
-def step_check_stale_removed(context: Context) -> None:
-    """Assert the stale row no longer exists."""
-    svc: RepoIndexingService = context.repo_index_service  # type: ignore[attr-defined]
-    rid = context.repo_index_stale_resource_id  # type: ignore[attr-defined]
-    result = svc.get_index(rid)
-    assert result is None, f"Expected None, got {result}"

robot/helper_m5_e2e_context.py

@@ -0,0 +1,500 @@
+"""Context-policy verification commands for M5 Robot E2E tests."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from typing import Any, cast
+from unittest.mock import MagicMock
+
+from helper_m5_e2e_support import _create_project, _fail, _setup_db
+
+from cleveragents.application.services.context_phase_analysis import (
+    analyze_phase_summaries,
+)
+from cleveragents.application.services.context_tiers import ContextTierService
+from cleveragents.application.services.execute_phase_context_assembler import (
+    ACMSExecutePhaseContextAssembler,
+)
+from cleveragents.application.services.llm_actors import LLMExecuteActor
+from cleveragents.application.services.plan_executor import StrategyDecision
+from cleveragents.cli.commands.project_context import _read_policy, _write_policy
+from cleveragents.config.settings import Settings
+from cleveragents.domain.models.acms.tiers import ContextTier, TieredFragment
+from cleveragents.domain.models.core.context_policy import (
+    VALID_PHASES,
+    ContextView,
+    ProjectContextPolicy,
+)
+from cleveragents.domain.models.core.project import ContextConfig, NamespacedProject
+
+
+def context_tier_config() -> None:
+    """Verify ContextConfig with hot/warm/cold tier management."""
+    default_cfg = ContextConfig()
+    if default_cfg.hot_max_tokens is not None:
+        _fail(f"default hot_max_tokens should be None: {default_cfg.hot_max_tokens}")
+    if default_cfg.warm_max_decisions is not None:
+        _fail(
+            f"default warm_max_decisions should be None: "
+            f"{default_cfg.warm_max_decisions}"
+        )
+    if default_cfg.cold_max_decisions is not None:
+        _fail(
+            f"default cold_max_decisions should be None: "
+            f"{default_cfg.cold_max_decisions}"
+        )
+
+    tier_cfg = ContextConfig(
+        hot_max_tokens=128_000,
+        warm_max_decisions=50,
+        cold_max_decisions=200,
+        max_file_size=2_000_000,
+        max_total_size=100_000_000,
+    )
+    if tier_cfg.hot_max_tokens != 128_000:
+        _fail(f"hot_max_tokens mismatch: {tier_cfg.hot_max_tokens}")
+    if tier_cfg.warm_max_decisions != 50:
+        _fail(f"warm_max_decisions mismatch: {tier_cfg.warm_max_decisions}")
+    if tier_cfg.cold_max_decisions != 200:
+        _fail(f"cold_max_decisions mismatch: {tier_cfg.cold_max_decisions}")
+    if tier_cfg.max_file_size != 2_000_000:
+        _fail(f"max_file_size mismatch: {tier_cfg.max_file_size}")
+    if tier_cfg.max_total_size != 100_000_000:
+        _fail(f"max_total_size mismatch: {tier_cfg.max_total_size}")
+
+    proj = NamespacedProject(
+        name="tier-test",
+        namespace="local",
+        context_config=tier_cfg,
+    )
+    if proj.context_config.hot_max_tokens != 128_000:
+        _fail("project did not preserve hot_max_tokens")
+    if proj.context_config.warm_max_decisions != 50:
+        _fail("project did not preserve warm_max_decisions")
+    if proj.context_config.cold_max_decisions != 200:
+        _fail("project did not preserve cold_max_decisions")
+
+    print("m5-context-tier-config-ok")
+
+
+def context_policy_set_show() -> None:
+    """Set a context policy via SQL helpers and read it back."""
+    proj_repo, _link_repo, _svc, sf = _setup_db()
+    _create_project(proj_repo, "local/policy-test")
+
+    policy = ProjectContextPolicy(
+        default_view=ContextView(
+            include_resources=["local/repo-*"],
+            exclude_paths=["*.log", "*.tmp"],
+            max_file_size=1_048_576,
+            max_total_size=52_428_800,
+        ),
+        strategize_view=ContextView(
+            include_resources=["local/core-*"],
+            include_paths=["src/**/*.py"],
+            max_file_size=500_000,
+        ),
+    )
+
+    _write_policy(sf, "local/policy-test", policy)
+    loaded = _read_policy(sf, "local/policy-test")
+
+    if loaded.default_view.include_resources != ["local/repo-*"]:
+        _fail(
+            f"default include_resources mismatch: "
+            f"{loaded.default_view.include_resources}"
+        )
+    if loaded.default_view.exclude_paths != ["*.log", "*.tmp"]:
+        _fail(f"default exclude_paths mismatch: {loaded.default_view.exclude_paths}")
+    if loaded.default_view.max_file_size != 1_048_576:
+        _fail(f"default max_file_size mismatch: {loaded.default_view.max_file_size}")
+    if loaded.strategize_view is None:
+        _fail("strategize_view is None after round-trip")
+    if loaded.strategize_view.include_resources != ["local/core-*"]:
+        _fail(
+            f"strategize include_resources mismatch: "
+            f"{loaded.strategize_view.include_resources}"
+        )
+    if loaded.strategize_view.include_paths != ["src/**/*.py"]:
+        _fail(
+            f"strategize include_paths mismatch: {loaded.strategize_view.include_paths}"
+        )
+    if loaded.execute_view is not None:
+        _fail("execute_view should be None")
+    if loaded.apply_view is not None:
+        _fail("apply_view should be None")
+
+    print("m5-context-policy-set-show-ok")
+
+
+def acms_phase_inheritance() -> None:
+    """Verify ACMS view inheritance: default -> strategize -> execute -> apply."""
+    policy = ProjectContextPolicy(
+        default_view=ContextView(
+            include_resources=["*"],
+            max_total_size=50_000_000,
+        ),
+        strategize_view=ContextView(
+            include_resources=["core-*"],
+            max_file_size=1_000_000,
+        ),
+    )
+
+    rv_default = policy.resolve_view("default")
+    if rv_default.include_resources != ["*"]:
+        _fail(f"default resolution wrong: {rv_default.include_resources}")
+    if rv_default.max_total_size != 50_000_000:
+        _fail(f"default max_total_size wrong: {rv_default.max_total_size}")
+
+    rv_strat = policy.resolve_view("strategize")
+    if rv_strat.include_resources != ["core-*"]:
+        _fail(f"strategize resolution wrong: {rv_strat.include_resources}")
+    if rv_strat.max_file_size != 1_000_000:
+        _fail(f"strategize max_file_size wrong: {rv_strat.max_file_size}")
+
+    rv_exec = policy.resolve_view("execute")
+    if rv_exec.include_resources != ["core-*"]:
+        _fail(f"execute should inherit strategize: {rv_exec.include_resources}")
+
+    rv_apply = policy.resolve_view("apply")
+    if rv_apply.include_resources != ["core-*"]:
+        _fail(f"apply should inherit via chain: {rv_apply.include_resources}")
+
+    policy2 = ProjectContextPolicy(
+        default_view=ContextView(include_resources=["*"]),
+        strategize_view=ContextView(include_resources=["core-*"]),
+        execute_view=ContextView(
+            include_resources=["exec-*"],
+            exclude_paths=["tests/**"],
+        ),
+    )
+    rv_apply2 = policy2.resolve_view("apply")
+    if rv_apply2.include_resources != ["exec-*"]:
+        _fail(f"apply should inherit from execute: {rv_apply2.include_resources}")
+    if rv_apply2.exclude_paths != ["tests/**"]:
+        _fail(f"apply exclude_paths wrong: {rv_apply2.exclude_paths}")
+
+    print("m5-acms-phase-inheritance-ok")
+
+
+def acms_scoped_context() -> None:
+    """Verify ACMS produces scoped context output per phase."""
+    policy = ProjectContextPolicy(
+        default_view=ContextView(
+            include_resources=["*"],
+            max_file_size=2_000_000,
+            max_total_size=100_000_000,
+        ),
+        strategize_view=ContextView(
+            include_resources=["local/large-repo"],
+            include_paths=["src/**/*.py", "docs/**/*.md"],
+            exclude_paths=["**/test_*.py"],
+            max_file_size=1_000_000,
+        ),
+        execute_view=ContextView(
+            include_resources=["local/large-repo"],
+            include_paths=["src/**/*.py"],
+            exclude_paths=["**/test_*.py", "**/conftest.py"],
+            max_file_size=500_000,
+            max_total_size=1_200_000,
+        ),
+        apply_view=ContextView(
+            include_resources=["local/large-repo"],
+            include_paths=["src/**/*.py"],
+            exclude_paths=["docs/**", "**/test_*.py", "**/conftest.py"],
+            max_file_size=250_000,
+            max_total_size=700_000,
+        ),
+    )
+
+    for phase in ("default", "strategize", "execute", "apply"):
+        rv = policy.resolve_view(phase)
+        expected_attr = getattr(policy, f"{phase}_view")
+        if expected_attr is None:
+            _fail(f"phase {phase} view should not be None")
+        if rv.include_resources != expected_attr.include_resources:
+            _fail(
+                f"{phase} include_resources mismatch: "
+                f"{rv.include_resources} != {expected_attr.include_resources}"
+            )
+
+    strat_view = policy.resolve_view("strategize")
+    exec_view = policy.resolve_view("execute")
+    apply_view = policy.resolve_view("apply")
+
+    if strat_view.max_file_size is None:
+        _fail("strategize max_file_size is None")
+    if exec_view.max_file_size is None:
+        _fail("execute max_file_size is None")
+    if apply_view.max_file_size is None:
+        _fail("apply max_file_size is None")
+    if not (
+        strat_view.max_file_size > exec_view.max_file_size > apply_view.max_file_size
+    ):
+        _fail(
+            f"size limits should narrow: "
+            f"{strat_view.max_file_size} > {exec_view.max_file_size} > "
+            f"{apply_view.max_file_size}"
+        )
+
+    sample_fragments = [
+        TieredFragment(
+            fragment_id="f-core",
+            content="x" * 900,
+            tier=ContextTier.HOT,
+            resource_id="local/large-repo",
+            project_name="local/scoped-test",
+            token_count=900,
+            metadata={"path": "src/core.py", "byte_size": 900_000},
+        ),
+        TieredFragment(
+            fragment_id="f-engine",
+            content="x" * 500,
+            tier=ContextTier.HOT,
+            resource_id="local/large-repo",
+            project_name="local/scoped-test",
+            token_count=500,
+            metadata={"path": "src/engine.py", "byte_size": 450_000},
+        ),
+        TieredFragment(
+            fragment_id="f-test",
+            content="x" * 120,
+            tier=ContextTier.WARM,
+            resource_id="local/large-repo",
+            project_name="local/scoped-test",
+            token_count=120,
+            metadata={"path": "src/test_core.py", "byte_size": 100_000},
+        ),
+        TieredFragment(
+            fragment_id="f-doc",
+            content="x" * 180,
+            tier=ContextTier.WARM,
+            resource_id="local/large-repo",
+            project_name="local/scoped-test",
+            token_count=180,
+            metadata={"path": "docs/design.md", "byte_size": 200_000},
+        ),
+        TieredFragment(
+            fragment_id="f-conftest",
+            content="x" * 90,
+            tier=ContextTier.COLD,
+            resource_id="local/large-repo",
+            project_name="local/scoped-test",
+            token_count=90,
+            metadata={"path": "src/conftest.py", "byte_size": 80_000},
+        ),
+        TieredFragment(
+            fragment_id="f-huge",
+            content="x" * 1500,
+            tier=ContextTier.COLD,
+            resource_id="local/large-repo",
+            project_name="local/scoped-test",
+            token_count=1500,
+            metadata={"path": "src/huge.py", "byte_size": 1_500_000},
+        ),
+    ]
+    analysis = analyze_phase_summaries(policy, sample_fragments, budget_tokens=2_000)
+
+    for phase in ("strategize", "execute", "apply"):
+        phase_row = analysis["phases"][phase]
+        if not phase_row["summary"]:
+            _fail(f"{phase} summary should be non-empty")
+        if "resource_count" not in phase_row or "total_bytes" not in phase_row:
+            _fail(f"{phase} summary missing resource/size metrics")
+
+    narrowing = analysis["narrowing"]
+    if not narrowing["overall"]:
+        _fail(f"narrowing should hold across phases: {narrowing}")
+    proj_repo, _link_repo, _svc, sf = _setup_db()
+    _create_project(proj_repo, "local/scoped-test")
+    _write_policy(sf, "local/scoped-test", policy)
+    loaded = _read_policy(sf, "local/scoped-test")
+
+    for phase in ("default", "strategize", "execute", "apply"):
+        rv_loaded = loaded.resolve_view(phase)
+        rv_orig = policy.resolve_view(phase)
+        if rv_loaded.include_resources != rv_orig.include_resources:
+            _fail(f"{phase} round-trip include_resources mismatch")
+        if rv_loaded.max_file_size != rv_orig.max_file_size:
+            _fail(f"{phase} round-trip max_file_size mismatch")
+
+    print("m5-acms-scoped-context-ok")
+
+
+def context_policy_clear() -> None:
+    """Verify clearing a phase view causes inheritance fallback."""
+    policy = ProjectContextPolicy(
+        default_view=ContextView(include_resources=["all-*"]),
+        strategize_view=ContextView(include_resources=["strat-*"]),
+        execute_view=ContextView(include_resources=["exec-*"]),
+        apply_view=ContextView(include_resources=["apply-*"]),
+    )
+
+    rv = policy.resolve_view("execute")
+    if rv.include_resources != ["exec-*"]:
+        _fail(f"before clear, execute wrong: {rv.include_resources}")
+
+    cleared = policy.model_copy(update={"execute_view": None})
+    rv_after = cleared.resolve_view("execute")
+    if rv_after.include_resources != ["strat-*"]:
+        _fail(
+            f"after clear, execute should inherit strategize: "
+            f"{rv_after.include_resources}"
+        )
+
+    rv_apply = cleared.resolve_view("apply")
+    if rv_apply.include_resources != ["apply-*"]:
+        _fail(f"apply should still be its own: {rv_apply.include_resources}")
+
+    cleared2 = cleared.model_copy(update={"apply_view": None})
+    rv_apply2 = cleared2.resolve_view("apply")
+    if rv_apply2.include_resources != ["strat-*"]:
+        _fail(
+            f"after clearing apply+execute, apply should inherit strategize: "
+            f"{rv_apply2.include_resources}"
+        )
+
+    proj_repo, _link_repo, _svc, sf = _setup_db()
+    _create_project(proj_repo, "local/clear-test")
+    _write_policy(sf, "local/clear-test", cleared2)
+    loaded = _read_policy(sf, "local/clear-test")
+
+    if loaded.execute_view is not None:
+        _fail("execute_view should be None after round-trip")
+    if loaded.apply_view is not None:
+        _fail("apply_view should be None after round-trip")
+
+    rv_loaded = loaded.resolve_view("execute")
+    if rv_loaded.include_resources != ["strat-*"]:
+        _fail(f"round-trip execute fallback wrong: {rv_loaded.include_resources}")
+
+    print("m5-context-policy-clear-ok")
+
+
+def context_view_validation() -> None:
+    """Verify ContextView validation rules."""
+    cv = ContextView(max_file_size=1, max_total_size=1)
+    if cv.max_file_size != 1:
+        _fail(f"max_file_size should be 1: {cv.max_file_size}")
+
+    try:
+        ContextView(max_file_size=0)
+        _fail("max_file_size=0 should be rejected")
+    except ValueError:
+        pass
+
+    try:
+        ContextView(max_total_size=-1)
+        _fail("max_total_size=-1 should be rejected")
+    except ValueError:
+        pass
+
+    cv_none = ContextView(max_file_size=None, max_total_size=None)
+    if cv_none.max_file_size is not None:
+        _fail("max_file_size should be None")
+
+    expected_phases = {"default", "strategize", "execute", "apply"}
+    if expected_phases != VALID_PHASES:
+        _fail(f"VALID_PHASES mismatch: {VALID_PHASES} != {expected_phases}")
+
+    policy = ProjectContextPolicy()
+    try:
+        policy.resolve_view("invalid-phase")
+        _fail("resolve_view should reject invalid phase")
+    except ValueError:
+        pass
+
+    print("m5-context-view-validation-ok")
+
+
+def llm_execute_acms_context() -> None:
+    """Verify execute-phase LLM prompt uses ACMS assembled execute context."""
+    proj_repo, _link_repo, _svc, sf = _setup_db()
+    _create_project(proj_repo, "local/exec-ctx")
+
+    policy = ProjectContextPolicy(
+        default_view=ContextView(include_paths=["**/*"]),
+        strategize_view=ContextView(include_paths=["docs/**"]),
+        execute_view=ContextView(
+            include_paths=["src/**"],
+            max_file_size=100_000,
+            max_total_size=100_000,
+        ),
+    )
+    _write_policy(sf, "local/exec-ctx", policy)
+
+    tier_service = ContextTierService(settings=Settings())
+    tier_service.store(
+        TieredFragment(
+            fragment_id="frag-docs",
+            content="documentation fragment",
+            tier=ContextTier.HOT,
+            resource_id="local/large-repo",
+            project_name="local/exec-ctx",
+            token_count=20,
+            metadata={"path": "docs/readme.md", "relevance_score": "0.6"},
+        )
+    )
+    tier_service.store(
+        TieredFragment(
+            fragment_id="frag-src",
+            content="def run():\n    return 'ok'",
+            tier=ContextTier.HOT,
+            resource_id="local/large-repo",
+            project_name="local/exec-ctx",
+            token_count=16,
+            metadata={"path": "src/main.py", "relevance_score": "0.9"},
+        )
+    )
+
+    assembler = ACMSExecutePhaseContextAssembler(
+        context_tier_service=tier_service,
+        project_repository=proj_repo,
+        hot_max_tokens=1024,
+    )
+
+    mock_llm = MagicMock()
+    mock_llm.invoke.return_value = SimpleNamespace(
+        content="FILE: src/out.py\n```python\nprint('ok')\n```\n"
+    )
+    registry = cast(Any, SimpleNamespace(create_llm=MagicMock(return_value=mock_llm)))
+    plan = SimpleNamespace(
+        identity=SimpleNamespace(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV"),
+        action_name="local/action",
+        project_links=[SimpleNamespace(project_name="local/exec-ctx")],
+    )
+    action = SimpleNamespace(execution_actor="openai/gpt-4")
+    lifecycle = SimpleNamespace(
+        get_plan=MagicMock(return_value=plan),
+        get_action=MagicMock(return_value=action),
+    )
+
+    actor = LLMExecuteActor(
+        provider_registry=registry,
+        lifecycle_service=lifecycle,
+        context_assembler=assembler,
+    )
+    actor.execute(
+        plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
+        decisions=[
+            StrategyDecision(
+                decision_id="d1",
+                step_text="Implement feature",
+                sequence=0,
+                parent_id=None,
+            )
+        ],
+        stream_callback=None,
+    )
+
+    call_args = mock_llm.invoke.call_args
+    if call_args is None:
+        _fail("LLM invoke was not called")
+    prompt = call_args.args[0][0].content
+    if "ACMS Execute-Phase Context" not in prompt:
+        _fail("prompt missing ACMS execute-phase context section")
+    if "src/main.py" not in prompt:
+        _fail("prompt missing execute-view source fragment")
+    if "docs/readme.md" in prompt:
+        _fail("prompt should exclude strategize-only docs fragment in execute view")
+    print("m5-llm-execute-acms-context-ok")

robot/helper_m5_e2e_support.py

@@ -0,0 +1,94 @@
+"""Shared helpers for the M5 Robot E2E verification commands."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from typing import Any, NoReturn, cast
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+
+from cleveragents.application.services.resource_registry_service import (
+    ResourceRegistryService,
+)
+from cleveragents.domain.models.core.project import (
+    NamespacedProject,
+    parse_namespaced_name,
+)
+from cleveragents.infrastructure.database.models import Base
+from cleveragents.infrastructure.database.repositories import (
+    NamespacedProjectRepository,
+    ProjectResourceLinkRepository,
+)
+
+
+def _fail(msg: str) -> NoReturn:
+    """Print failure message to stderr and exit with code 1."""
+    print(f"FAIL: {msg}", file=sys.stderr)
+    raise SystemExit(1)
+
+
+class _NoClose:
+    """Session wrapper that suppresses ``close()`` for in-memory SQLite."""
+
+    def __init__(self, session: object) -> None:
+        object.__setattr__(self, "_s", session)
+
+    def close(self) -> None:
+        pass
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(object.__getattribute__(self, "_s"), name)
+
+
+def _setup_db() -> tuple[
+    NamespacedProjectRepository,
+    ProjectResourceLinkRepository,
+    ResourceRegistryService,
+    Any,
+]:
+    """Create an in-memory SQLite DB with all tables."""
+    engine = create_engine("sqlite:///:memory:", echo=False)
+    Base.metadata.create_all(engine)
+    session = sessionmaker(bind=engine, expire_on_commit=False)()
+    wrapper = _NoClose(session)
+
+    def factory() -> Session:
+        return cast(Session, wrapper)
+
+    proj_repo = NamespacedProjectRepository(session_factory=factory)
+    link_repo = ProjectResourceLinkRepository(session_factory=factory)
+    svc = ResourceRegistryService(session_factory=factory)
+    svc.bootstrap_builtin_types()
+    return proj_repo, link_repo, svc, factory
+
+
+def _create_project(
+    repo: NamespacedProjectRepository,
+    name: str,
+    description: str | None = None,
+) -> NamespacedProject:
+    """Create and return a NamespacedProject via the repository."""
+    parsed = parse_namespaced_name(name)
+    proj = NamespacedProject(
+        name=parsed.name,
+        namespace=parsed.namespace,
+        description=description,
+    )
+    repo.create(proj)
+    return repo.get(proj.namespaced_name)
+
+
+def _create_large_python_repo(root: Path) -> int:
+    """Populate *root* with 10,000 small Python files and return the count."""
+    file_count = 0
+    for i in range(100):
+        module_dir = root / f"module_{i:03d}"
+        module_dir.mkdir(parents=True, exist_ok=True)
+        for j in range(100):
+            (module_dir / f"file_{j:03d}.py").write_text(
+                f"def f_{i}_{j}():\n    return {i + j}\n"
+            )
+            file_count += 1
+    return file_count

robot/helper_m5_e2e_verification.py

@@ -6,7 +6,7 @@ Exercises the complete M5 success criteria sequence:
   3. Indexing verification (resource linked, project shows correctly)
   4. Context tier management (hot/warm/cold) via ContextConfig
   5. ACMS v1 context policy set/show with persistence round-trip
-  6. Phase view inheritance (default → strategize → execute → apply)
+  6. Phase view inheritance (default -> strategize -> execute -> apply)
   7. Scoped context output per ACMS phase
   8. Policy clear and inheritance fallback
 
@@ -21,128 +21,35 @@ from __future__ import annotations
 
 import sys
 import tempfile
+import time
 from collections.abc import Callable
 from pathlib import Path
-from types import SimpleNamespace
-from typing import Any, NoReturn, cast
-from unittest.mock import MagicMock
-
-from sqlalchemy import create_engine
-from sqlalchemy.orm import Session, sessionmaker
 
 # Ensure src is importable when run from workspace root
 _SRC = str(Path(__file__).resolve().parents[1] / "src")
 if _SRC not in sys.path:
     sys.path.insert(0, _SRC)
 
-from cleveragents.application.services.context_phase_analysis import (  # noqa: E402
-    analyze_phase_summaries,
+from helper_m5_e2e_context import (  # noqa: E402
+    acms_phase_inheritance,
+    acms_scoped_context,
+    context_policy_clear,
+    context_policy_set_show,
+    context_tier_config,
+    context_view_validation,
+    llm_execute_acms_context,
 )
-from cleveragents.application.services.context_tiers import (  # noqa: E402
-    ContextTierService,
-)
-from cleveragents.application.services.execute_phase_context_assembler import (  # noqa: E402
-    ACMSExecutePhaseContextAssembler,
-)
-from cleveragents.application.services.llm_actors import (  # noqa: E402
-    LLMExecuteActor,
-)
-from cleveragents.application.services.plan_executor import (  # noqa: E402
-    StrategyDecision,
-)
-from cleveragents.application.services.resource_registry_service import (  # noqa: E402
-    ResourceRegistryService,
-)
-from cleveragents.cli.commands.project_context import (  # noqa: E402
-    _read_policy,
-    _write_policy,
-)
-from cleveragents.config.settings import Settings  # noqa: E402
-from cleveragents.domain.models.acms.tiers import (  # noqa: E402
-    ContextTier,
-    TieredFragment,
-)
-from cleveragents.domain.models.core.context_policy import (  # noqa: E402
-    VALID_PHASES,
-    ContextView,
-    ProjectContextPolicy,
-)
-from cleveragents.domain.models.core.project import (  # noqa: E402
-    ContextConfig,
-    NamespacedProject,
-    parse_namespaced_name,
-)
-from cleveragents.infrastructure.database.models import Base  # noqa: E402
-from cleveragents.infrastructure.database.repositories import (  # noqa: E402
-    NamespacedProjectRepository,
-    ProjectResourceLinkRepository,
+from helper_m5_e2e_support import (  # noqa: E402
+    _create_large_python_repo,
+    _create_project,
+    _fail,
+    _setup_db,
 )
 
-# -------------------------------------------------------------------
-# Helpers
-# -------------------------------------------------------------------
-
-
-def _fail(msg: str) -> NoReturn:
-    """Print failure message to stderr and exit with code 1."""
-    print(f"FAIL: {msg}", file=sys.stderr)
-    raise SystemExit(1)
-
-
-class _NoClose:
-    """Session wrapper that suppresses ``close()`` for in-memory SQLite."""
-
-    def __init__(self, session: object) -> None:
-        object.__setattr__(self, "_s", session)
-
-    def close(self) -> None:
-        pass
-
-    def __getattr__(self, name: str) -> Any:
-        return getattr(object.__getattribute__(self, "_s"), name)
-
-
-def _setup_db() -> tuple[
-    NamespacedProjectRepository,
-    ProjectResourceLinkRepository,
-    ResourceRegistryService,
-    Any,
-]:
-    """Create an in-memory SQLite DB with all tables.
-
-    Returns:
-        (project_repo, link_repo, registry_service, session_factory)
-    """
-    engine = create_engine("sqlite:///:memory:", echo=False)
-    Base.metadata.create_all(engine)
-    session = sessionmaker(bind=engine, expire_on_commit=False)()
-    wrapper = _NoClose(session)
-
-    def factory() -> Session:
-        return cast(Session, wrapper)
-
-    proj_repo = NamespacedProjectRepository(session_factory=factory)
-    link_repo = ProjectResourceLinkRepository(session_factory=factory)
-    svc = ResourceRegistryService(session_factory=factory)
-    svc.bootstrap_builtin_types()
-    return proj_repo, link_repo, svc, factory
-
-
-def _create_project(
-    repo: NamespacedProjectRepository,
-    name: str,
-    description: str | None = None,
-) -> NamespacedProject:
-    """Create and return a NamespacedProject via the repository."""
-    parsed = parse_namespaced_name(name)
-    proj = NamespacedProject(
-        name=parsed.name,
-        namespace=parsed.namespace,
-        description=description,
-    )
-    repo.create(proj)
-    return repo.get(proj.namespaced_name)
-
+from cleveragents.application.services.repo_indexing_service import (  # noqa: E402
+    RepoIndexingService,
+)
+from cleveragents.domain.models.core.repo_index import IndexStatus  # noqa: E402
 
 # -------------------------------------------------------------------
 # Subcommand: project-create-large
@@ -157,7 +64,6 @@ def project_create_large() -> None:
     """
     proj_repo, link_repo, svc, _sf = _setup_db()
 
-    # Create project
     proj = _create_project(
         proj_repo,
         "local/large-project",
@@ -166,27 +72,14 @@ def project_create_large() -> None:
     if proj.namespaced_name != "local/large-project":
         _fail(f"project name mismatch: {proj.namespaced_name}")
 
-    # Simulate a large repository by creating a temp dir with
-    # a marker file (actual 10K files are not needed for the E2E
-    # CLI path -- the acceptance criterion is exercising the
-    # creation flow with a large-project descriptor).
     with tempfile.TemporaryDirectory() as tmp:
         repo_path = Path(tmp) / "large-repo"
         repo_path.mkdir()
 
-        # Create nested directories with files to simulate scale
-        _FILE_COUNT = 0
-        for i in range(100):
-            d = repo_path / f"module_{i:03d}"
-            d.mkdir()
-            for j in range(100):
-                (d / f"file_{j:03d}.py").write_text(f"# module {i} file {j}\n")
-                _FILE_COUNT += 1
+        file_count = _create_large_python_repo(repo_path)
+        if file_count < 10_000:
+            _fail(f"expected >= 10,000 files, created {file_count}")
 
-        if _FILE_COUNT < 10_000:
-            _fail(f"expected >= 10,000 files, created {_FILE_COUNT}")
-
-        # Register git-checkout resource at that path
         res = svc.register_resource(
             type_name="git-checkout",
             name="local/large-repo",
@@ -198,18 +91,15 @@ def project_create_large() -> None:
         if not res.resource_id:
             _fail("resource_id is empty")
 
-        # Link resource to project
         link_repo.create_link(
             project_name="local/large-project",
             resource_id=res.resource_id,
         )
 
-        # Verify link
         links = link_repo.list_links("local/large-project")
         if len(links) != 1:
             _fail(f"expected 1 link, got {len(links)}")
 
-        # Re-fetch project (links are stored separately)
         fetched = proj_repo.get("local/large-project")
         if fetched.description != "Large project with 10,000+ file repo":
             _fail(f"description mismatch: {fetched.description}")
@@ -228,7 +118,6 @@ def resource_register_link() -> None:
 
     proj = _create_project(proj_repo, "local/link-test")
 
-    # Register two resources
     res1 = svc.register_resource(
         type_name="git-checkout",
         name="local/repo-alpha",
@@ -244,7 +133,6 @@ def resource_register_link() -> None:
         properties={"path": "/tmp/beta", "branch": "develop"},
     )
 
-    # Link both to project
     link_repo.create_link(
         project_name=proj.namespaced_name,
         resource_id=res1.resource_id,
@@ -258,7 +146,6 @@ def resource_register_link() -> None:
     if len(links) != 2:
         _fail(f"expected 2 links, got {len(links)}")
 
-    # Show resource details
     shown = svc.show_resource("local/repo-alpha")
     if shown.name != "local/repo-alpha":
         _fail(f"resource name mismatch: {shown.name}")
@@ -274,612 +161,72 @@ def resource_register_link() -> None:
 
 
 def indexing_complete() -> None:
-    """Verify indexing completes: resource is linked, project shows OK.
-
-    This subcommand simulates the indexing-complete check by verifying
-    the resource link round-trip through the database.
-    """
-    proj_repo, link_repo, svc, _sf = _setup_db()
+    """Verify indexing completes: resource is linked, project shows OK."""
+    proj_repo, link_repo, svc, sf = _setup_db()
 
     proj = _create_project(proj_repo, "local/idx-project")
-    res = svc.register_resource(
-        type_name="git-checkout",
-        name="local/idx-repo",
-        location="/tmp/idx",
-        description="Index test",
-    )
-    link_repo.create_link(
-        project_name=proj.namespaced_name,
-        resource_id=res.resource_id,
-    )
+    with tempfile.TemporaryDirectory(prefix="m5_idx_repo_") as tmp:
+        repo_root = Path(tmp) / "idx-repo"
+        repo_root.mkdir(parents=True, exist_ok=True)
 
-    # Verify link survives re-fetch
-    links = link_repo.list_links(proj.namespaced_name)
-    if len(links) != 1:
-        _fail(f"expected 1 link, got {len(links)}")
+        file_count = _create_large_python_repo(repo_root)
+        if file_count < 10_000:
+            _fail(f"expected >= 10,000 files, created {file_count}")
 
-    # Resource round-trip
-    fetched_res = svc.show_resource("local/idx-repo")
-    if fetched_res.resource_id != res.resource_id:
-        _fail(f"resource_id mismatch after fetch: {fetched_res.resource_id}")
+        res = svc.register_resource(
+            type_name="git-checkout",
+            name="local/idx-repo",
+            location=str(repo_root),
+            description="Index test",
+            properties={"path": str(repo_root), "branch": "main"},
+        )
+        link_repo.create_link(
+            project_name=proj.namespaced_name,
+            resource_id=res.resource_id,
+        )
 
-    # Verify resource details are intact
-    if fetched_res.location != "/tmp/idx":
-        _fail(f"location mismatch: {fetched_res.location}")
+        links = link_repo.list_links(proj.namespaced_name)
+        if len(links) != 1:
+            _fail(f"expected 1 link, got {len(links)}")
+
+        index_service = RepoIndexingService(session_factory=sf)
+        timeout_seconds = 120.0
+        started = time.monotonic()
+        idx = index_service.index_resource(
+            res.resource_id,
+            repo_root,
+            timeout_seconds=timeout_seconds,
+        )
+        elapsed = time.monotonic() - started
+
+        if elapsed > timeout_seconds:
+            _fail(
+                "indexing exceeded timeout bound: "
+                f"{elapsed:.2f}s > {timeout_seconds:.2f}s"
+            )
+        if idx.metadata.file_count < 10_000:
+            _fail(f"expected >= 10,000 indexed files, got {idx.metadata.file_count}")
+        if idx.metadata.status != IndexStatus.READY:
+            _fail(f"expected READY status, got {idx.metadata.status}")
+
+        fetched_res = svc.show_resource("local/idx-repo")
+        if fetched_res.resource_id != res.resource_id:
+            _fail(f"resource_id mismatch after fetch: {fetched_res.resource_id}")
+        if fetched_res.location != str(repo_root):
+            _fail(f"location mismatch: {fetched_res.location}")
+
+        status = index_service.get_index_status(res.resource_id)
+        if status is None:
+            _fail("index status missing after indexing")
+        if status.file_count != idx.metadata.file_count:
+            _fail(
+                "index status file_count mismatch: "
+                f"{status.file_count} != {idx.metadata.file_count}"
+            )
 
     print("m5-indexing-complete-ok")
 
 
-# -------------------------------------------------------------------
-# Subcommand: context-tier-config
-# -------------------------------------------------------------------
-
-
-def context_tier_config() -> None:
-    """Verify ContextConfig with hot/warm/cold tier management.
-
-    Checks that the domain model correctly stores and returns
-    tier configuration values.
-    """
-    # Default config
-    default_cfg = ContextConfig()
-    if default_cfg.hot_max_tokens is not None:
-        _fail(f"default hot_max_tokens should be None: {default_cfg.hot_max_tokens}")
-    if default_cfg.warm_max_decisions is not None:
-        _fail(
-            f"default warm_max_decisions should be None: "
-            f"{default_cfg.warm_max_decisions}"
-        )
-    if default_cfg.cold_max_decisions is not None:
-        _fail(
-            f"default cold_max_decisions should be None: "
-            f"{default_cfg.cold_max_decisions}"
-        )
-
-    # Custom tier config
-    tier_cfg = ContextConfig(
-        hot_max_tokens=128_000,
-        warm_max_decisions=50,
-        cold_max_decisions=200,
-        max_file_size=2_000_000,
-        max_total_size=100_000_000,
-    )
-    if tier_cfg.hot_max_tokens != 128_000:
-        _fail(f"hot_max_tokens mismatch: {tier_cfg.hot_max_tokens}")
-    if tier_cfg.warm_max_decisions != 50:
-        _fail(f"warm_max_decisions mismatch: {tier_cfg.warm_max_decisions}")
-    if tier_cfg.cold_max_decisions != 200:
-        _fail(f"cold_max_decisions mismatch: {tier_cfg.cold_max_decisions}")
-    if tier_cfg.max_file_size != 2_000_000:
-        _fail(f"max_file_size mismatch: {tier_cfg.max_file_size}")
-    if tier_cfg.max_total_size != 100_000_000:
-        _fail(f"max_total_size mismatch: {tier_cfg.max_total_size}")
-
-    # Verify project can carry tier config
-    proj = NamespacedProject(
-        name="tier-test",
-        namespace="local",
-        context_config=tier_cfg,
-    )
-    if proj.context_config.hot_max_tokens != 128_000:
-        _fail("project did not preserve hot_max_tokens")
-    if proj.context_config.warm_max_decisions != 50:
-        _fail("project did not preserve warm_max_decisions")
-    if proj.context_config.cold_max_decisions != 200:
-        _fail("project did not preserve cold_max_decisions")
-
-    print("m5-context-tier-config-ok")
-
-
-# -------------------------------------------------------------------
-# Subcommand: context-policy-set-show
-# -------------------------------------------------------------------
-
-
-def context_policy_set_show() -> None:
-    """Set a context policy via SQL helpers and read it back.
-
-    Verifies the _write_policy / _read_policy round-trip used by
-    the ``agents project context set/show`` commands.
-    """
-    proj_repo, _link_repo, _svc, sf = _setup_db()
-    _create_project(proj_repo, "local/policy-test")
-
-    # Create a policy with default and strategize views
-    policy = ProjectContextPolicy(
-        default_view=ContextView(
-            include_resources=["local/repo-*"],
-            exclude_paths=["*.log", "*.tmp"],
-            max_file_size=1_048_576,
-            max_total_size=52_428_800,
-        ),
-        strategize_view=ContextView(
-            include_resources=["local/core-*"],
-            include_paths=["src/**/*.py"],
-            max_file_size=500_000,
-        ),
-    )
-
-    # Write
-    _write_policy(sf, "local/policy-test", policy)
-
-    # Read back
-    loaded = _read_policy(sf, "local/policy-test")
-
-    # Verify default view
-    if loaded.default_view.include_resources != ["local/repo-*"]:
-        _fail(
-            f"default include_resources mismatch: "
-            f"{loaded.default_view.include_resources}"
-        )
-    if loaded.default_view.exclude_paths != ["*.log", "*.tmp"]:
-        _fail(f"default exclude_paths mismatch: {loaded.default_view.exclude_paths}")
-    if loaded.default_view.max_file_size != 1_048_576:
-        _fail(f"default max_file_size mismatch: {loaded.default_view.max_file_size}")
-
-    # Verify strategize view
-    if loaded.strategize_view is None:
-        _fail("strategize_view is None after round-trip")
-    if loaded.strategize_view.include_resources != ["local/core-*"]:
-        _fail(
-            f"strategize include_resources mismatch: "
-            f"{loaded.strategize_view.include_resources}"
-        )
-    if loaded.strategize_view.include_paths != ["src/**/*.py"]:
-        _fail(
-            f"strategize include_paths mismatch: {loaded.strategize_view.include_paths}"
-        )
-
-    # Verify execute and apply are None (not set)
-    if loaded.execute_view is not None:
-        _fail("execute_view should be None")
-    if loaded.apply_view is not None:
-        _fail("apply_view should be None")
-
-    print("m5-context-policy-set-show-ok")
-
-
-# -------------------------------------------------------------------
-# Subcommand: acms-phase-inheritance
-# -------------------------------------------------------------------
-
-
-def acms_phase_inheritance() -> None:
-    """Verify ACMS view inheritance: default → strategize → execute → apply.
-
-    When a phase view is not set, ``resolve_view`` should walk up
-    the inheritance chain and return the nearest ancestor's view.
-    """
-    # Policy with only default and strategize set
-    policy = ProjectContextPolicy(
-        default_view=ContextView(
-            include_resources=["*"],
-            max_total_size=50_000_000,
-        ),
-        strategize_view=ContextView(
-            include_resources=["core-*"],
-            max_file_size=1_000_000,
-        ),
-        # execute_view: None → inherits from strategize
-        # apply_view: None → inherits from execute → strategize
-    )
-
-    # Default resolves to itself
-    rv_default = policy.resolve_view("default")
-    if rv_default.include_resources != ["*"]:
-        _fail(f"default resolution wrong: {rv_default.include_resources}")
-    if rv_default.max_total_size != 50_000_000:
-        _fail(f"default max_total_size wrong: {rv_default.max_total_size}")
-
-    # Strategize resolves to itself
-    rv_strat = policy.resolve_view("strategize")
-    if rv_strat.include_resources != ["core-*"]:
-        _fail(f"strategize resolution wrong: {rv_strat.include_resources}")
-    if rv_strat.max_file_size != 1_000_000:
-        _fail(f"strategize max_file_size wrong: {rv_strat.max_file_size}")
-
-    # Execute inherits from strategize
-    rv_exec = policy.resolve_view("execute")
-    if rv_exec.include_resources != ["core-*"]:
-        _fail(f"execute should inherit strategize: {rv_exec.include_resources}")
-
-    # Apply inherits from execute → strategize
-    rv_apply = policy.resolve_view("apply")
-    if rv_apply.include_resources != ["core-*"]:
-        _fail(f"apply should inherit via chain: {rv_apply.include_resources}")
-
-    # Now set execute explicitly and verify apply inherits from it
-    policy2 = ProjectContextPolicy(
-        default_view=ContextView(include_resources=["*"]),
-        strategize_view=ContextView(include_resources=["core-*"]),
-        execute_view=ContextView(
-            include_resources=["exec-*"],
-            exclude_paths=["tests/**"],
-        ),
-    )
-    rv_apply2 = policy2.resolve_view("apply")
-    if rv_apply2.include_resources != ["exec-*"]:
-        _fail(f"apply should inherit from execute: {rv_apply2.include_resources}")
-    if rv_apply2.exclude_paths != ["tests/**"]:
-        _fail(f"apply exclude_paths wrong: {rv_apply2.exclude_paths}")
-
-    print("m5-acms-phase-inheritance-ok")
-
-
-# -------------------------------------------------------------------
-# Subcommand: acms-scoped-context
-# -------------------------------------------------------------------
-
-
-def acms_scoped_context() -> None:
-    """Verify ACMS produces scoped context output per phase.
-
-    Sets phase-specific policies and verifies that resolve_view
-    returns the correct scoped view for each phase.
-    """
-    policy = ProjectContextPolicy(
-        default_view=ContextView(
-            include_resources=["*"],
-            max_file_size=2_000_000,
-            max_total_size=100_000_000,
-        ),
-        strategize_view=ContextView(
-            include_resources=["local/large-repo"],
-            include_paths=["src/**/*.py", "docs/**/*.md"],
-            exclude_paths=["**/test_*.py"],
-            max_file_size=1_000_000,
-        ),
-        execute_view=ContextView(
-            include_resources=["local/large-repo"],
-            include_paths=["src/**/*.py"],
-            exclude_paths=["**/test_*.py", "**/conftest.py"],
-            max_file_size=500_000,
-            max_total_size=1_200_000,
-        ),
-        apply_view=ContextView(
-            include_resources=["local/large-repo"],
-            include_paths=["src/**/*.py"],
-            exclude_paths=["docs/**", "**/test_*.py", "**/conftest.py"],
-            max_file_size=250_000,
-            max_total_size=700_000,
-        ),
-    )
-
-    # Verify each phase resolves to its own view
-    for phase in ("default", "strategize", "execute", "apply"):
-        rv = policy.resolve_view(phase)
-        expected_attr = getattr(policy, f"{phase}_view")
-        if expected_attr is None:
-            _fail(f"phase {phase} view should not be None")
-        if rv.include_resources != expected_attr.include_resources:
-            _fail(
-                f"{phase} include_resources mismatch: "
-                f"{rv.include_resources} != {expected_attr.include_resources}"
-            )
-
-    # Verify scoped size limits narrow from strategize → execute → apply
-    strat_view = policy.resolve_view("strategize")
-    exec_view = policy.resolve_view("execute")
-    apply_view = policy.resolve_view("apply")
-
-    if strat_view.max_file_size is None:
-        _fail("strategize max_file_size is None")
-    if exec_view.max_file_size is None:
-        _fail("execute max_file_size is None")
-    if apply_view.max_file_size is None:
-        _fail("apply max_file_size is None")
-
-    if not (
-        strat_view.max_file_size > exec_view.max_file_size > apply_view.max_file_size
-    ):
-        _fail(
-            f"size limits should narrow: "
-            f"{strat_view.max_file_size} > {exec_view.max_file_size} > "
-            f"{apply_view.max_file_size}"
-        )
-
-    # Verify per-phase analysis summaries and narrowing diagnostics
-    sample_fragments = [
-        TieredFragment(
-            fragment_id="f-core",
-            content="x" * 900,
-            tier=ContextTier.HOT,
-            resource_id="local/large-repo",
-            project_name="local/scoped-test",
-            token_count=900,
-            metadata={"path": "src/core.py", "byte_size": 900_000},
-        ),
-        TieredFragment(
-            fragment_id="f-engine",
-            content="x" * 500,
-            tier=ContextTier.HOT,
-            resource_id="local/large-repo",
-            project_name="local/scoped-test",
-            token_count=500,
-            metadata={"path": "src/engine.py", "byte_size": 450_000},
-        ),
-        TieredFragment(
-            fragment_id="f-test",
-            content="x" * 120,
-            tier=ContextTier.WARM,
-            resource_id="local/large-repo",
-            project_name="local/scoped-test",
-            token_count=120,
-            metadata={"path": "src/test_core.py", "byte_size": 100_000},
-        ),
-        TieredFragment(
-            fragment_id="f-doc",
-            content="x" * 180,
-            tier=ContextTier.WARM,
-            resource_id="local/large-repo",
-            project_name="local/scoped-test",
-            token_count=180,
-            metadata={"path": "docs/design.md", "byte_size": 200_000},
-        ),
-        TieredFragment(
-            fragment_id="f-conftest",
-            content="x" * 90,
-            tier=ContextTier.COLD,
-            resource_id="local/large-repo",
-            project_name="local/scoped-test",
-            token_count=90,
-            metadata={"path": "src/conftest.py", "byte_size": 80_000},
-        ),
-        TieredFragment(
-            fragment_id="f-huge",
-            content="x" * 1500,
-            tier=ContextTier.COLD,
-            resource_id="local/large-repo",
-            project_name="local/scoped-test",
-            token_count=1500,
-            metadata={"path": "src/huge.py", "byte_size": 1_500_000},
-        ),
-    ]
-    analysis = analyze_phase_summaries(policy, sample_fragments, budget_tokens=2_000)
-
-    for phase in ("strategize", "execute", "apply"):
-        phase_row = analysis["phases"][phase]
-        if not phase_row["summary"]:
-            _fail(f"{phase} summary should be non-empty")
-        if "resource_count" not in phase_row or "total_bytes" not in phase_row:
-            _fail(f"{phase} summary missing resource/size metrics")
-
-    narrowing = analysis["narrowing"]
-    if not narrowing["overall"]:
-        _fail(f"narrowing should hold across phases: {narrowing}")
-
-    # Verify persistence round-trip
-    proj_repo, _link_repo, _svc, sf = _setup_db()
-    _create_project(proj_repo, "local/scoped-test")
-    _write_policy(sf, "local/scoped-test", policy)
-    loaded = _read_policy(sf, "local/scoped-test")
-
-    for phase in ("default", "strategize", "execute", "apply"):
-        rv_loaded = loaded.resolve_view(phase)
-        rv_orig = policy.resolve_view(phase)
-        if rv_loaded.include_resources != rv_orig.include_resources:
-            _fail(f"{phase} round-trip include_resources mismatch")
-        if rv_loaded.max_file_size != rv_orig.max_file_size:
-            _fail(f"{phase} round-trip max_file_size mismatch")
-
-    print("m5-acms-scoped-context-ok")
-
-
-# -------------------------------------------------------------------
-# Subcommand: context-policy-clear
-# -------------------------------------------------------------------
-
-
-def context_policy_clear() -> None:
-    """Verify clearing a phase view causes inheritance fallback.
-
-    Sets all four views, then clears execute_view. Verifies that
-    ``resolve_view("execute")`` now falls back to strategize_view.
-    """
-    policy = ProjectContextPolicy(
-        default_view=ContextView(include_resources=["all-*"]),
-        strategize_view=ContextView(include_resources=["strat-*"]),
-        execute_view=ContextView(include_resources=["exec-*"]),
-        apply_view=ContextView(include_resources=["apply-*"]),
-    )
-
-    # Before clear: execute resolves to its own view
-    rv = policy.resolve_view("execute")
-    if rv.include_resources != ["exec-*"]:
-        _fail(f"before clear, execute wrong: {rv.include_resources}")
-
-    # Clear execute_view
-    cleared = policy.model_copy(update={"execute_view": None})
-    rv_after = cleared.resolve_view("execute")
-    if rv_after.include_resources != ["strat-*"]:
-        _fail(
-            f"after clear, execute should inherit strategize: "
-            f"{rv_after.include_resources}"
-        )
-
-    # Apply still resolves to its own view (it's still set)
-    rv_apply = cleared.resolve_view("apply")
-    if rv_apply.include_resources != ["apply-*"]:
-        _fail(f"apply should still be its own: {rv_apply.include_resources}")
-
-    # Clear apply too — should fall back to strategize (via execute→strategize)
-    cleared2 = cleared.model_copy(update={"apply_view": None})
-    rv_apply2 = cleared2.resolve_view("apply")
-    if rv_apply2.include_resources != ["strat-*"]:
-        _fail(
-            f"after clearing apply+execute, apply should inherit strategize: "
-            f"{rv_apply2.include_resources}"
-        )
-
-    # Persistence round-trip with cleared views
-    proj_repo, _link_repo, _svc, sf = _setup_db()
-    _create_project(proj_repo, "local/clear-test")
-    _write_policy(sf, "local/clear-test", cleared2)
-    loaded = _read_policy(sf, "local/clear-test")
-
-    if loaded.execute_view is not None:
-        _fail("execute_view should be None after round-trip")
-    if loaded.apply_view is not None:
-        _fail("apply_view should be None after round-trip")
-
-    rv_loaded = loaded.resolve_view("execute")
-    if rv_loaded.include_resources != ["strat-*"]:
-        _fail(f"round-trip execute fallback wrong: {rv_loaded.include_resources}")
-
-    print("m5-context-policy-clear-ok")
-
-
-# -------------------------------------------------------------------
-# Subcommand: context-view-validation
-# -------------------------------------------------------------------
-
-
-def context_view_validation() -> None:
-    """Verify ContextView validation rules.
-
-    Tests that invalid size limits are rejected and that
-    VALID_PHASES is correct.
-    """
-    # Positive size limits work
-    cv = ContextView(max_file_size=1, max_total_size=1)
-    if cv.max_file_size != 1:
-        _fail(f"max_file_size should be 1: {cv.max_file_size}")
-
-    # Zero size limit rejected
-    try:
-        ContextView(max_file_size=0)
-        _fail("max_file_size=0 should be rejected")
-    except ValueError:
-        pass
-
-    # Negative size limit rejected
-    try:
-        ContextView(max_total_size=-1)
-        _fail("max_total_size=-1 should be rejected")
-    except ValueError:
-        pass
-
-    # None is allowed (no limit)
-    cv_none = ContextView(max_file_size=None, max_total_size=None)
-    if cv_none.max_file_size is not None:
-        _fail("max_file_size should be None")
-
-    # VALID_PHASES must contain exactly the four ACMS phases
-    expected_phases = {"default", "strategize", "execute", "apply"}
-    if expected_phases != VALID_PHASES:
-        _fail(f"VALID_PHASES mismatch: {VALID_PHASES} != {expected_phases}")
-
-    # Invalid phase raises ValueError
-    policy = ProjectContextPolicy()
-    try:
-        policy.resolve_view("invalid-phase")
-        _fail("resolve_view should reject invalid phase")
-    except ValueError:
-        pass
-
-    print("m5-context-view-validation-ok")
-
-
-# -------------------------------------------------------------------
-# Subcommand: llm-execute-acms-context
-# -------------------------------------------------------------------
-
-
-def llm_execute_acms_context() -> None:
-    """Verify execute-phase LLM prompt uses ACMS assembled execute context."""
-    proj_repo, _link_repo, _svc, sf = _setup_db()
-    _create_project(proj_repo, "local/exec-ctx")
-
-    policy = ProjectContextPolicy(
-        default_view=ContextView(include_paths=["**/*"]),
-        strategize_view=ContextView(include_paths=["docs/**"]),
-        execute_view=ContextView(
-            include_paths=["src/**"],
-            max_file_size=100_000,
-            max_total_size=100_000,
-        ),
-    )
-    _write_policy(sf, "local/exec-ctx", policy)
-
-    tier_service = ContextTierService(settings=Settings())
-    tier_service.store(
-        TieredFragment(
-            fragment_id="frag-docs",
-            content="documentation fragment",
-            tier=ContextTier.HOT,
-            resource_id="local/large-repo",
-            project_name="local/exec-ctx",
-            token_count=20,
-            metadata={"path": "docs/readme.md", "relevance_score": "0.6"},
-        )
-    )
-    tier_service.store(
-        TieredFragment(
-            fragment_id="frag-src",
-            content="def run():\n    return 'ok'",
-            tier=ContextTier.HOT,
-            resource_id="local/large-repo",
-            project_name="local/exec-ctx",
-            token_count=16,
-            metadata={"path": "src/main.py", "relevance_score": "0.9"},
-        )
-    )
-
-    assembler = ACMSExecutePhaseContextAssembler(
-        context_tier_service=tier_service,
-        project_repository=proj_repo,
-        hot_max_tokens=1024,
-    )
-
-    mock_llm = MagicMock()
-    mock_llm.invoke.return_value = SimpleNamespace(
-        content="FILE: src/out.py\n```python\nprint('ok')\n```\n"
-    )
-    registry = cast(Any, SimpleNamespace(create_llm=MagicMock(return_value=mock_llm)))
-
-    plan = SimpleNamespace(
-        identity=SimpleNamespace(plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV"),
-        action_name="local/action",
-        project_links=[SimpleNamespace(project_name="local/exec-ctx")],
-    )
-    action = SimpleNamespace(execution_actor="openai/gpt-4")
-    lifecycle = SimpleNamespace(
-        get_plan=MagicMock(return_value=plan),
-        get_action=MagicMock(return_value=action),
-    )
-
-    actor = LLMExecuteActor(
-        provider_registry=registry,
-        lifecycle_service=lifecycle,
-        context_assembler=assembler,
-    )
-    actor.execute(
-        plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
-        decisions=[
-            StrategyDecision(
-                decision_id="d1",
-                step_text="Implement feature",
-                sequence=0,
-                parent_id=None,
-            )
-        ],
-        stream_callback=None,
-    )
-
-    call_args = mock_llm.invoke.call_args
-    if call_args is None:
-        _fail("LLM invoke was not called")
-    prompt = call_args.args[0][0].content
-    if "ACMS Execute-Phase Context" not in prompt:
-        _fail("prompt missing ACMS execute-phase context section")
-    if "src/main.py" not in prompt:
-        _fail("prompt missing execute-view source fragment")
-    if "docs/readme.md" in prompt:
-        _fail("prompt should exclude strategize-only docs fragment in execute view")
-
-    print("m5-llm-execute-acms-context-ok")
-
-
 # -------------------------------------------------------------------
 # Dispatcher
 # -------------------------------------------------------------------

src/cleveragents/application/services/repo_indexing_service.py

@@ -1,10 +1,4 @@
-"""Repository indexing service for CleverAgents.
-
-Provides indexing for 10K+ file projects with incremental refresh,
-language detection, and policy enforcement (include/exclude globs,
-size limits).  Utilities in :mod:`~.repo_indexing_utils`; persistence
-in :mod:`~.repo_indexing_persistence`.
-"""
+"""Repository indexing service for full and incremental repo scans."""
 
 from __future__ import annotations
 
@@ -64,21 +58,10 @@ def _serialize_on_resource[F: Callable[..., object]](method: F) -> F:
 
 
 class RepoIndexingService:
-    """Repository indexing service (session-factory pattern).
+    """Repository indexing service using the session-factory pattern.
 
-    Walks the file tree applying globs/size limits, hashes content,
-    detects languages, and persists results.  Concurrent calls for the
-    same *resource_id* are serialized via ``_serialize_on_resource``.
-
-    Each persistence helper opens its own database session, matching the
-    session-factory DI pattern used throughout the codebase.  The
-    per-resource ``RLock`` serializes concurrent operations within a
-    single process; multi-process safety is not required (SQLite).
-
-    INVARIANTS: (1) file_count == IndexedFileModel row count for
-    index_id; (2) token_estimate == sum(token_count); (3)
-    primary_language == mode language or "unknown"; (4) status "ready"
-    iff no error; (5) resource_id has at most one RepoIndexModel row.
+    It walks trees with glob and size filtering, persists the resulting records,
+    and serializes concurrent operations per resource within a process.
     """
 
     def __init__(self, session_factory: Any) -> None:
@@ -113,6 +96,7 @@ class RepoIndexingService:
         exclude_globs: tuple[str, ...] = (),
         max_file_size: int | None = None,
         max_total_size: int | None = None,
+        timeout_seconds: float | None = None,
     ) -> RepoIndex:
         """Index a resource by walking its file tree.
 
@@ -124,6 +108,8 @@ class RepoIndexingService:
             ValueError: If *resource_id* is empty or not a ULID.
         """
         self._validate_resource_id(resource_id)
+        if timeout_seconds is not None and timeout_seconds <= 0:
+            raise ValueError(f"timeout_seconds must be positive, got {timeout_seconds}")
 
         root = Path(root_path)
         if not root.exists():
@@ -158,6 +144,7 @@ class RepoIndexingService:
                 "root_path": str(root),
                 "include_globs": include_globs,
                 "exclude_globs": exclude_globs,
+                "timeout_seconds": timeout_seconds,
             },
         )
 
@@ -168,6 +155,7 @@ class RepoIndexingService:
                 exclude_globs=exclude_globs,
                 max_file_size=max_file_size,
                 max_total_size=max_total_size,
+                timeout_seconds=timeout_seconds,
             )
         except Exception as exc:
             if not has_good_index:
@@ -257,30 +245,16 @@ class RepoIndexingService:
         exclude_globs: tuple[str, ...] = (),
         max_file_size: int | None = None,
         max_total_size: int | None = None,
+        timeout_seconds: float | None = None,
     ) -> RepoIndex:
         """Incrementally refresh an existing index.
 
-        Only files whose content hash has changed (or that are new/deleted)
-        are re-processed.  If no prior index exists, performs a full index.
-        Note: ``last_modified`` for unchanged files retains the value from
-        the previous index (not re-read from disk) to avoid stat overhead.
-
-        Args:
-            resource_id: ULID of the resource.
-            root_path: Filesystem path to the resource root.
-            include_globs: File path globs to include (empty = all).
-            exclude_globs: File path globs to exclude.
-            max_file_size: Max individual file size in bytes.
-            max_total_size: Max total size in bytes.
-
-        Returns:
-            The refreshed :class:`RepoIndex`.
-
-        Raises:
-            FileNotFoundError: If *root_path* does not exist.
-            ValueError: If *resource_id* is empty.
+        Changed or new files are reprocessed, deleted files are dropped, and a
+        missing prior index falls back to a full index.
         """
         self._validate_resource_id(resource_id)
+        if timeout_seconds is not None and timeout_seconds <= 0:
+            raise ValueError(f"timeout_seconds must be positive, got {timeout_seconds}")
 
         existing = self.get_index(resource_id)
         if existing is None:
@@ -291,9 +265,9 @@ class RepoIndexingService:
                 exclude_globs=exclude_globs,
                 max_file_size=max_file_size,
                 max_total_size=max_total_size,
+                timeout_seconds=timeout_seconds,
             )
 
-        # Same validation as index_resource (DRY extraction blocked by 500-line limit).
         root = Path(root_path)
         if not root.exists():
             raise FileNotFoundError(f"Resource root path does not exist: {root}")
@@ -312,6 +286,7 @@ class RepoIndexingService:
                 exclude_globs=exclude_globs,
                 max_file_size=max_file_size,
                 max_total_size=max_total_size,
+                timeout_seconds=timeout_seconds,
             )
         except Exception as exc:
             logger.warning(

src/cleveragents/application/services/repo_indexing_utils.py

@@ -15,6 +15,7 @@ import hashlib
 import logging
 import os
 import stat as stat_mod
+import time
 from collections import Counter
 from datetime import UTC, datetime
 from pathlib import Path, PurePosixPath
@@ -236,6 +237,7 @@ def walk_and_index(
     max_file_size: int | None,
     max_total_size: int | None,
     max_file_count: int | None = None,
+    timeout_seconds: float | None = None,
 ) -> list[FileRecord]:
     """Walk a directory tree and build file records.
 
@@ -248,6 +250,9 @@ def walk_and_index(
     ``max_file_count`` is an internal parameter used only by benchmarks
     and tests — not exposed in the public ``RepoIndexingService`` API.
 
+    If ``timeout_seconds`` is provided, a ``TimeoutError`` is raised when
+    the walk+index operation exceeds the configured bound.
+
     The ``max_total_size`` cutoff uses Phase-1 ``stat`` sizes for the
     budget check; actual byte count (from ``read_and_hash``) may differ
     slightly if files are modified between stat and read.
@@ -256,10 +261,23 @@ def walk_and_index(
     # "../" traversal cannot escape the intended scope (G5 fix).
     root = root.resolve()
 
+    started_at = time.monotonic()
+
+    def _check_timeout() -> None:
+        if timeout_seconds is None:
+            return
+        elapsed = time.monotonic() - started_at
+        if elapsed > timeout_seconds:
+            raise TimeoutError(
+                "Repository indexing timed out after "
+                f"{elapsed:.2f}s (limit: {timeout_seconds:.2f}s)"
+            )
+
     # Phase 1: Collect all candidate files (path, stat) sorted by rel_path
     candidates: list[tuple[str, Path, os.stat_result]] = []
 
     for dirpath, dirnames, filenames in os.walk(root):
+        _check_timeout()
         # Skip hidden directories (startswith(".") covers .git, .venv,
         # .tox, .mypy_cache, etc.) and common non-source directories.
         dirnames[:] = sorted(
@@ -270,6 +288,7 @@ def walk_and_index(
         )
 
         for filename in sorted(filenames):
+            _check_timeout()
             if filename.startswith("."):
                 continue
 
@@ -334,6 +353,7 @@ def walk_and_index(
     total_bytes = 0
 
     for rel_path, full_path, stat in candidates:
+        _check_timeout()
         if max_file_count is not None and len(records) >= max_file_count:
             logger.debug(
                 "Reached max_file_count limit, stopping",

src/cleveragents/cli/commands/repo.py

@@ -101,6 +101,13 @@ def index_command(
         bool,
         typer.Option("--full", help="Force full re-index (not incremental)"),
     ] = False,
+    timeout_seconds: Annotated[
+        float | None,
+        typer.Option(
+            "--timeout-seconds",
+            help=("Abort indexing when elapsed runtime exceeds this many seconds"),
+        ),
+    ] = None,
     fmt: Annotated[
         str,
         typer.Option("--format", "-f", help=_FORMAT_HELP),
@@ -123,9 +130,20 @@ def index_command(
 
     try:
         if full:
-            repo_index = indexing_service.index_resource(resource_id, root_path)
+            repo_index = indexing_service.index_resource(
+                resource_id,
+                root_path,
+                timeout_seconds=timeout_seconds,
+            )
         else:
-            repo_index = indexing_service.refresh_index(resource_id, root_path)
+            repo_index = indexing_service.refresh_index(
+                resource_id,
+                root_path,
+                timeout_seconds=timeout_seconds,
+            )
+    except TimeoutError as exc:
+        console.print(f"[red]Error:[/red] {exc}")
+        raise typer.Exit(code=1) from exc
     except FileNotFoundError as exc:
         console.print(f"[red]Error:[/red] {exc}")
         raise typer.Exit(code=1) from exc