feat(perf): large project scaling tests (#984)

## Summary

Add large project scaling benchmarks and tests at production scale (10K–100K files).

### New ASV Benchmarks

**IndexingScalingSuite** (`large_project_scaling_bench.py`):
- `time_walk_and_index` at 1K/10K/50K/100K files
- `time_incremental_refresh` (1% modified files)
- `track_indexed_file_count`, `track_tokens_per_second`

**ContextAssemblyScalingSuite** (`context_assembly_scaling_bench.py`):
- `time_full_pipeline` at 100/1K/5K/10K fragments
- `time_tiered_strategy`, `time_recency_strategy`
- `track_assembled_tokens`, `track_fragments_per_second`

**ExecutionThroughputSuite** (`execution_throughput_bench.py`):
- `time_sequential_plans` at 10/50/100 plans
- `time_executor_construction`, `time_decision_tree_scaling`

### Scale Fixture Updates

- Added `xlarge` (50K files) and `xxlarge` (100K files) profiles to `scale_metadata.json`
- Added 50K/100K thresholds to `baseline_thresholds.json`
- Added `context_assembly` and `execution_throughput` threshold sections

### Tests & Documentation

- 15 Behave scenarios validating profiles, thresholds, monotonicity, memory budgets
- 6 Robot integration tests including live 1K-file indexing throughput check
- `docs/reference/scaling_baselines.md` documenting all baseline metrics

### Quality Gates

| Session | Result |
|---|---|
| `nox -s lint` | PASS |
| `nox -s typecheck` | PASS (0 errors) |
| `nox -s unit_tests` | PASS (10,910 scenarios) |
| `nox -s integration_tests` | PASS (1,526 tests) |
| `nox -s coverage_report` | 97% (>= 97%) |

Closes #859

Reviewed-on: #984
Co-authored-by: Brent E. Edwards <brent.edwards@cleverthis.com>
Co-committed-by: Brent E. Edwards <brent.edwards@cleverthis.com>

benchmarks/bench_uko_layer3.py

@@ -20,11 +20,11 @@ _SRC = str(Path(__file__).resolve().parents[1] / "src")
 if _SRC not in sys.path:
     sys.path.insert(0, _SRC)
 
-from cleveragents.acms.uko.layer3_java import JAVA_DETAIL_LEVELS  # noqa: E402
-from cleveragents.acms.uko.layer3_py import PYTHON_DETAIL_LEVELS  # noqa: E402
-from cleveragents.acms.uko.layer3_rs import RUST_DETAIL_LEVELS  # noqa: E402
-from cleveragents.acms.uko.layer3_ts import TYPESCRIPT_DETAIL_LEVELS  # noqa: E402
-from cleveragents.acms.uko.vocabulary import (  # noqa: E402
+from cleveragents.acms.uko.layer3_java import JAVA_DETAIL_LEVELS
+from cleveragents.acms.uko.layer3_py import PYTHON_DETAIL_LEVELS
+from cleveragents.acms.uko.layer3_rs import RUST_DETAIL_LEVELS
+from cleveragents.acms.uko.layer3_ts import TYPESCRIPT_DETAIL_LEVELS
+from cleveragents.acms.uko.vocabulary import (
     OO_EFFECTIVE_LEVELS,
     ProvenanceInfo,
     UKOClass,

benchmarks/context_assembly_scaling_bench.py

@@ -0,0 +1,138 @@
+"""ASV benchmarks for ACMS context assembly at production scale.
+
+Extends the existing 1K-fragment ceiling from :mod:`acms_pipeline_bench`
+to 5K and 10K fragments, exercising the full 10-stage pipeline,
+deduplication + scoring, and knapsack budget packing.
+"""
+
+from __future__ import annotations
+
+import importlib
+import sys
+import time
+from pathlib import Path
+from typing import ClassVar
+
+# Ensure the local *source* tree is importable even when ASV has an
+# older build of the package installed.
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+import cleveragents  # noqa: E402
+
+importlib.reload(cleveragents)
+
+from cleveragents.application.services.acms_service import ACMSPipeline  # noqa: E402
+from cleveragents.domain.models.core.context_fragment import (  # noqa: E402
+    ContextBudget,
+    ContextFragment,
+    FragmentProvenance,
+)
+
+# Default provenance for benchmark fragments.
+_DEFAULT_PROV = FragmentProvenance(resource_uri="bench://scale")
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _make_fragments(count: int) -> list[ContextFragment]:
+    """Build *count* benchmark fragments with varied scores and tiers."""
+    tiers = ("hot", "warm", "cold")
+    return [
+        ContextFragment(
+            uko_node=f"bench://scale/{i}",
+            content=f"fragment content block {i} " * 5,
+            relevance_score=round((i % 10) / 10, 1),
+            token_count=50,
+            tier=tiers[i % 3],
+            provenance=_DEFAULT_PROV,
+        )
+        for i in range(count)
+    ]
+
+
+# ---------------------------------------------------------------------------
+# Parameterized context assembly suite
+# ---------------------------------------------------------------------------
+
+
+class ContextAssemblyScalingSuite:
+    """Benchmark ACMS context assembly at production fragment counts."""
+
+    params: ClassVar[list[int]] = [100, 1_000, 5_000, 10_000]
+    param_names: ClassVar[list[str]] = ["fragment_count"]
+    timeout = 300  # 5 min for 10K fragments
+
+    _pipeline: ACMSPipeline
+    _budget: ContextBudget
+    _fragments: list[ContextFragment]
+
+    def setup(self, fragment_count: int) -> None:
+        """Create *fragment_count* ContextFragment objects."""
+        self._pipeline = ACMSPipeline()
+        self._budget = ContextBudget(max_tokens=500_000, reserved_tokens=0)
+        self._fragments = _make_fragments(fragment_count)
+
+    # -- timing methods -----------------------------------------------------
+
+    def time_full_pipeline(self, fragment_count: int) -> None:
+        """Time the full ACMS pipeline with *fragment_count* fragments."""
+        self._pipeline.assemble(
+            plan_id="01JQBENCHM00000000000000AA",
+            fragments=self._fragments,
+            budget=self._budget,
+        )
+
+    def time_tiered_strategy(self, fragment_count: int) -> None:
+        """Time tiered strategy assembly."""
+        self._pipeline.assemble(
+            plan_id="01JQBENCHM00000000000000AA",
+            fragments=self._fragments,
+            budget=self._budget,
+            strategy="tiered",
+        )
+
+    def time_recency_strategy(self, fragment_count: int) -> None:
+        """Time recency strategy assembly."""
+        self._pipeline.assemble(
+            plan_id="01JQBENCHM00000000000000AA",
+            fragments=self._fragments,
+            budget=self._budget,
+            strategy="recency",
+        )
+
+    # -- tracking methods ---------------------------------------------------
+
+    def track_assembled_tokens(self, fragment_count: int) -> int:
+        """Track total tokens in assembled context."""
+        result = self._pipeline.assemble(
+            plan_id="01JQBENCHM00000000000000AA",
+            fragments=self._fragments,
+            budget=self._budget,
+        )
+        return result.total_tokens
+
+    def track_fragments_per_second(self, fragment_count: int) -> float:
+        """Track assembly throughput in fragments / second."""
+        t0 = time.perf_counter()
+        self._pipeline.assemble(
+            plan_id="01JQBENCHM00000000000000AA",
+            fragments=self._fragments,
+            budget=self._budget,
+        )
+        elapsed = time.perf_counter() - t0
+        if elapsed <= 0:
+            return float(fragment_count)
+        return fragment_count / elapsed
+
+
+# Attach ASV unit metadata without ``# type: ignore``.
+setattr(  # noqa: B010
+    ContextAssemblyScalingSuite.track_assembled_tokens, "unit", "tokens"
+)
+setattr(  # noqa: B010
+    ContextAssemblyScalingSuite.track_fragments_per_second, "unit", "frags/s"
+)

benchmarks/execution_throughput_bench.py

@@ -0,0 +1,123 @@
+"""ASV benchmarks for plan execution throughput at scale.
+
+Measures sequential and concurrent plan execution overhead at varying
+plan counts (10, 50, 100).  Uses the lightweight in-process executor
+path (no database, no LLM) to isolate execution-dispatch cost.
+"""
+
+from __future__ import annotations
+
+import importlib
+import sys
+from pathlib import Path
+from typing import ClassVar
+from unittest.mock import MagicMock
+
+# Ensure the local *source* tree is importable even when ASV has an
+# older build of the package installed.
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+import cleveragents  # noqa: E402
+
+importlib.reload(cleveragents)
+
+from ulid import ULID  # noqa: E402
+
+from cleveragents.application.services.plan_execution_context import (  # noqa: E402
+    PlanExecutionContext,
+    RuntimeExecuteActor,
+)
+from cleveragents.application.services.plan_executor import (  # noqa: E402
+    PlanExecutor,
+    StrategyDecision,
+)
+from cleveragents.domain.models.core.change import (  # noqa: E402
+    InMemoryChangeSetStore,
+)
+from cleveragents.tool.registry import ToolRegistry  # noqa: E402
+from cleveragents.tool.runner import ToolRunner  # noqa: E402
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+_RESOURCE_ID = "01HGZ6FE0AQDYTR4BXVQZ6EB00"
+
+
+def _make_runner() -> ToolRunner:
+    return ToolRunner(registry=ToolRegistry())
+
+
+def _make_decisions(count: int) -> list[StrategyDecision]:
+    """Build a linear chain of *count* decisions."""
+    root_id = str(ULID())
+    return [
+        StrategyDecision(
+            decision_id=root_id if i == 0 else str(ULID()),
+            step_text=f"Step {i + 1}",
+            sequence=i,
+            parent_id=root_id if i > 0 else None,
+        )
+        for i in range(count)
+    ]
+
+
+def _execute_single_plan(runner: ToolRunner) -> None:
+    """Execute one plan with 3 decisions (fire-and-forget)."""
+    plan_id = str(ULID())
+    ctx = PlanExecutionContext(
+        plan_id=plan_id,
+        changeset_store=InMemoryChangeSetStore(),
+    )
+    actor = RuntimeExecuteActor(tool_runner=runner, execution_context=ctx)
+    actor.execute(decisions=_make_decisions(3))
+
+
+# ---------------------------------------------------------------------------
+# Parameterized execution throughput suite
+# ---------------------------------------------------------------------------
+
+
+class ExecutionThroughputSuite:
+    """Benchmark plan execution throughput at varying plan counts."""
+
+    params: ClassVar[list[int]] = [10, 50, 100]
+    param_names: ClassVar[list[str]] = ["plan_count"]
+    timeout = 300
+
+    _runner: ToolRunner
+
+    def setup(self, plan_count: int) -> None:
+        """Prepare a shared tool runner."""
+        self._runner = _make_runner()
+
+    def time_sequential_plans(self, plan_count: int) -> None:
+        """Execute *plan_count* plans sequentially."""
+        for _ in range(plan_count):
+            _execute_single_plan(self._runner)
+
+    def time_executor_construction(self, plan_count: int) -> None:
+        """Construct *plan_count* PlanExecutor instances."""
+        lifecycle = MagicMock()
+        for _ in range(plan_count):
+            ctx = PlanExecutionContext(
+                plan_id=str(ULID()),
+                changeset_store=InMemoryChangeSetStore(),
+            )
+            PlanExecutor(
+                lifecycle_service=lifecycle,
+                tool_runner=self._runner,
+                execution_context=ctx,
+            )
+
+    def time_decision_tree_scaling(self, plan_count: int) -> None:
+        """Execute one plan with *plan_count* decisions."""
+        plan_id = str(ULID())
+        ctx = PlanExecutionContext(
+            plan_id=plan_id,
+            changeset_store=InMemoryChangeSetStore(),
+        )
+        actor = RuntimeExecuteActor(tool_runner=self._runner, execution_context=ctx)
+        actor.execute(decisions=_make_decisions(plan_count))

benchmarks/large_project_scaling_bench.py

@@ -0,0 +1,181 @@
+"""ASV benchmarks for large-project indexing at production scale.
+
+Measures the performance of ``walk_and_index`` and incremental refresh
+at 1K, 10K, 50K, and 100K file counts, tracking throughput (files and
+tokens per second) and indexed-file totals.
+
+Extends the existing 5K-file ceiling from
+:mod:`large_project_decompose_bench` to production-scale repositories.
+"""
+
+from __future__ import annotations
+
+import importlib
+import os
+import random
+import shutil
+import sys
+import tempfile
+import time
+from pathlib import Path
+from typing import ClassVar
+
+# Ensure the local *source* tree is importable even when ASV has an
+# older build of the package installed.
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+import cleveragents  # noqa: E402
+
+importlib.reload(cleveragents)
+
+from cleveragents.application.services.repo_indexing_utils import (  # noqa: E402
+    walk_and_index,
+)
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+# Realistic directory structure skeleton used for synthetic repos.
+_DIRS = ("src", "tests", "docs", "scripts", "config")
+
+# Extension → content template.  Keeps files realistic so that
+# language detection and token estimation exercise real code paths.
+_EXTENSIONS: dict[str, str] = {
+    ".py": "# auto-generated\ndef func_{i}() -> int:\n    return {i}\n",
+    ".md": "# Document {i}\n\nGenerated documentation paragraph.\n",
+    ".json": '{{"id": {i}, "name": "item_{i}"}}\n',
+    ".ts": "export const value_{i}: number = {i};\n",
+    ".yaml": "key_{i}: value_{i}\n",
+}
+
+_EXT_LIST = list(_EXTENSIONS.keys())
+
+
+def _build_project(root: str, file_count: int) -> None:
+    """Populate *root* with *file_count* files in a realistic layout."""
+    rng = random.Random(42)  # deterministic seed
+    for i in range(file_count):
+        subdir = _DIRS[i % len(_DIRS)]
+        # Create two levels of nesting to mimic real projects.
+        nested = f"pkg{i % 20}"
+        dirpath = os.path.join(root, subdir, nested)
+        os.makedirs(dirpath, exist_ok=True)
+        ext = _EXT_LIST[rng.randint(0, len(_EXT_LIST) - 1)]
+        template = _EXTENSIONS[ext]
+        content = template.format(i=i)
+        fpath = os.path.join(dirpath, f"file_{i:06d}{ext}")
+        with open(fpath, "w") as fh:
+            fh.write(content)
+
+
+def _modify_subset(root: str, file_count: int, pct: float = 0.01) -> int:
+    """Modify *pct* of files under *root* and return count modified."""
+    rng = random.Random(99)  # deterministic seed
+    modify_count = max(1, int(file_count * pct))
+    indices = rng.sample(range(file_count), modify_count)
+    modified = 0
+    for i in indices:
+        subdir = _DIRS[i % len(_DIRS)]
+        nested = f"pkg{i % 20}"
+        ext = _EXT_LIST[rng.randint(0, len(_EXT_LIST) - 1)]
+        fpath = os.path.join(root, subdir, nested, f"file_{i:06d}{ext}")
+        if os.path.exists(fpath):
+            with open(fpath, "a") as fh:
+                fh.write(f"\n# modified at {time.monotonic()}\n")
+            modified += 1
+    return modified
+
+
+# ---------------------------------------------------------------------------
+# Parameterized indexing suite
+# ---------------------------------------------------------------------------
+
+_DEFAULT_INCLUDE: tuple[str, ...] = ()
+_DEFAULT_EXCLUDE: tuple[str, ...] = ("*.pyc", "__pycache__/*")
+
+
+class IndexingScalingSuite:
+    """Benchmark ``walk_and_index`` at production-scale file counts."""
+
+    params: ClassVar[list[int]] = [1_000, 10_000, 50_000, 100_000]
+    param_names: ClassVar[list[str]] = ["file_count"]
+    timeout = 600  # 10 min for 100K files
+
+    _tmpdir: str
+
+    def setup(self, file_count: int) -> None:
+        """Create a temp directory with *file_count* files."""
+        self._tmpdir = tempfile.mkdtemp(prefix="bench-scale-idx-")
+        _build_project(self._tmpdir, file_count)
+
+    def teardown(self, file_count: int) -> None:
+        shutil.rmtree(self._tmpdir, ignore_errors=True)
+
+    # -- timing methods -----------------------------------------------------
+
+    def time_walk_and_index(self, file_count: int) -> None:
+        """Time full ``walk_and_index`` for *file_count* files."""
+        walk_and_index(
+            root=Path(self._tmpdir),
+            include_globs=_DEFAULT_INCLUDE,
+            exclude_globs=_DEFAULT_EXCLUDE,
+            max_file_size=None,
+            max_total_size=None,
+        )
+
+    def time_incremental_refresh(self, file_count: int) -> None:
+        """Modify 1 %% of files, then re-index the full tree.
+
+        This measures incremental *walk* overhead; the actual diff merge
+        happens in ``RepoIndexingService.refresh_index`` which requires
+        a database.  Here we just re-walk to quantify I/O cost.
+        """
+        _modify_subset(self._tmpdir, file_count, pct=0.01)
+        walk_and_index(
+            root=Path(self._tmpdir),
+            include_globs=_DEFAULT_INCLUDE,
+            exclude_globs=_DEFAULT_EXCLUDE,
+            max_file_size=None,
+            max_total_size=None,
+        )
+
+    # -- tracking methods ---------------------------------------------------
+
+    def track_indexed_file_count(self, file_count: int) -> int:
+        """Track the number of files actually indexed."""
+        records = walk_and_index(
+            root=Path(self._tmpdir),
+            include_globs=_DEFAULT_INCLUDE,
+            exclude_globs=_DEFAULT_EXCLUDE,
+            max_file_size=None,
+            max_total_size=None,
+        )
+        return len(records)
+
+    def track_tokens_per_second(self, file_count: int) -> float:
+        """Track indexing throughput in tokens / second."""
+        t0 = time.perf_counter()
+        records = walk_and_index(
+            root=Path(self._tmpdir),
+            include_globs=_DEFAULT_INCLUDE,
+            exclude_globs=_DEFAULT_EXCLUDE,
+            max_file_size=None,
+            max_total_size=None,
+        )
+        elapsed = time.perf_counter() - t0
+        total_tokens = sum(r.token_count for r in records)
+        if elapsed <= 0:
+            return float(total_tokens)
+        return total_tokens / elapsed
+
+
+# Attach ASV unit metadata without ``# type: ignore``.
+setattr(  # noqa: B010
+    IndexingScalingSuite.track_indexed_file_count, "unit", "files"
+)
+setattr(  # noqa: B010
+    IndexingScalingSuite.track_tokens_per_second, "unit", "tokens/s"
+)

benchmarks/phase_reversion_bench.py

@@ -120,7 +120,9 @@ class AutoRevertSuite:
         plan_id = plan.identity.plan_id
         self.service.start_strategize(plan_id)
         self.service.complete_strategize(plan_id)
-        self.service.execute_plan(plan_id)
+        plan = self.service.get_plan(plan_id)
+        if plan.phase != PlanPhase.EXECUTE:
+            self.service.execute_plan(plan_id)
         self.service.start_execute(plan_id)
         self.service.complete_execute(plan_id)
         self.service.apply_plan(plan_id)

benchmarks/plan_explain_bench.py

@@ -92,7 +92,6 @@ class PlanExplainSuite:
             self.decision,
             show_context=True,
             show_reasoning=True,
-            show_alternatives=True,
         )
 
 

benchmarks/resource_cli_tree_bench.py

@@ -38,7 +38,7 @@ _runner = CliRunner()
 
 
 def _mock_resource(
-    resource_id: str = "01HBENCH0000000000RESOURCE",
+    resource_id: str = "01HBENCH0000000000RES0RCE0",
     name: str = "local/bench-res",
     type_name: str = "git-checkout",
 ) -> Resource:

benchmarks/security_async_cleanup_bench.py

@@ -60,8 +60,9 @@ class TimeRegisterBatch:
         pass
 
     def time_register_100(self) -> None:
+        tracker = AsyncResourceTracker()
         for i in range(100):
-            self.tracker.register(f"batch-{i}", _FakeResource())
+            tracker.register(f"batch-{i}", _FakeResource())
 
 
 class TimeCloseAll: