feat(plan): add large-project decomposition and dependency closure

Add hierarchical decomposition with 4+ levels and bounded context per subplan. Implement decomposition heuristics (max_files_per_subplan, max_tokens_per_subplan, language/dir clustering). Add dependency closure computation for large graphs and DAG execution ordering. Add bounded dependency closure with cutoff thresholds and memoization for 10K+ files. Record decomposition decisions in DecisionService (strategy_choice + subplan_spawn entries). New modules: - decomposition_models.py: DecompositionConfig, DecompositionNode, DecompositionResult, DependencyEdge, DependencyGraph - decomposition_clustering.py: ClusteringStrategy with directory, language, and size clustering plus deterministic sort - decomposition_graph.py: DependencyClosureComputer with bounded closure, topological sort, cycle detection, and memoization - decomposition_service.py: DecompositionService orchestrating hierarchy building and decision recording Settings: planner_max_depth, planner_max_files_per_subplan, planner_max_tokens_per_subplan, planner_min_files_per_subplan Closes #205
2026-03-03 17:27:43 +00:00
parent 738c3b5eda
commit 4232907ab9
17 changed files with 2540 additions and 0 deletions
@@ -26,6 +26,12 @@
  method. CLI `plan status` shows per-project changeset summaries for multi-project plans.
  Includes Behave BDD scenarios, Robot Framework smoke tests, ASV benchmarks, and reference
  documentation. (#199)
+- Added large-project hierarchical decomposition with 4+ levels and bounded context per
+  subplan. Includes clustering heuristics (directory, language, size), bounded dependency
+  closure with memoization for 10K+ files, DAG execution ordering with cycle detection,
+  and DecisionService integration for strategy_choice + subplan_spawn recording.
+  Configurable via `planner_max_depth`, `planner_max_files_per_subplan`,
+  `planner_max_tokens_per_subplan`, `planner_min_files_per_subplan` settings. (#205)
 - Added `SafetyProfile` domain model with configurable safety constraints (allowed skill
  categories, sandbox/checkpoint requirements, human-approval flag, cost/retry limits) and
  integrated it into the `Action` model via `from_config`/`as_cli_dict`. Persistence backed
@@ -0,0 +1,130 @@
+"""ASV benchmarks for large-project decomposition overhead.
+
+Measures the time to decompose file sets of various sizes and to
+compute dependency closures on synthetic graphs.
+"""
+
+from __future__ import annotations
+
+import importlib
+import os
+import sys
+import tempfile
+from pathlib import Path
+
+# 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)
+
+import itertools  # noqa: E402
+
+from cleveragents.application.services.decomposition_graph import (  # noqa: E402
+    DependencyClosureComputer,
+)
+from cleveragents.application.services.decomposition_models import (  # noqa: E402
+    DecompositionConfig,
+    DependencyEdge,
+    DependencyGraph,
+    DependencyType,
+)
+from cleveragents.application.services.decomposition_service import (  # noqa: E402
+    DecompositionService,
+)
+
+
+def _build_files(tmpdir: str, count: int, levels: int = 3) -> list[str]:
+    """Generate *count* files in *tmpdir* across *levels* dirs."""
+    paths: list[str] = []
+    for i in range(count):
+        level = i % levels
+        parts = [f"level{d}" for d in range(level + 1)]
+        dirpath = os.path.join(tmpdir, *parts)
+        os.makedirs(dirpath, exist_ok=True)
+        fpath = os.path.join(dirpath, f"file_{i:06d}.py")
+        with open(fpath, "w") as fh:
+            fh.write("x" * 200)
+        paths.append(fpath)
+    return sorted(paths)
+
+
+def _build_chain_graph(size: int) -> DependencyGraph:
+    """Build a chain of *size* nodes."""
+    g = DependencyGraph()
+    nodes = [f"file_{i:06d}.py" for i in range(size)]
+    for n in nodes:
+        g.add_node(n)
+    for src, tgt in itertools.pairwise(nodes):
+        g.add_edge(DependencyEdge(src, tgt, DependencyType.IMPORT))
+    return g, nodes  # type: ignore[return-value]
+
+
+class DecomposeSmallSuite:
+    """Benchmark decomposition of a small file set (100 files)."""
+
+    def setup(self) -> None:
+        self._tmpdir = tempfile.mkdtemp(prefix="bench-decompose-")
+        self._files = _build_files(self._tmpdir, 100)
+        self._svc = DecompositionService()
+
+    def time_decompose_default(self) -> None:
+        """Decompose 100 files with default config."""
+        self._svc.decompose(self._files)
+
+
+class DecomposeMediumSuite:
+    """Benchmark decomposition of a medium file set (1000 files)."""
+
+    def setup(self) -> None:
+        self._tmpdir = tempfile.mkdtemp(prefix="bench-decompose-")
+        self._files = _build_files(self._tmpdir, 1000)
+        self._svc = DecompositionService()
+
+    def time_decompose_default(self) -> None:
+        """Decompose 1000 files with default config."""
+        self._svc.decompose(self._files)
+
+
+class DecomposeLargeSuite:
+    """Benchmark decomposition of a large file set (5000 files)."""
+
+    def setup(self) -> None:
+        self._tmpdir = tempfile.mkdtemp(prefix="bench-decompose-")
+        self._files = _build_files(self._tmpdir, 5000, levels=5)
+        self._svc = DecompositionService()
+        self._config = DecompositionConfig(max_files_per_subplan=200)
+
+    def time_decompose_custom(self) -> None:
+        """Decompose 5000 files with custom config."""
+        self._svc.decompose(self._files, self._config)
+
+
+class ClosureSuite:
+    """Benchmark dependency closure computation."""
+
+    def setup(self) -> None:
+        result = _build_chain_graph(1000)
+        self._graph: DependencyGraph = result[0]  # type: ignore[assignment]
+        self._roots: list[str] = [result[1][0]]  # type: ignore[index]
+        self._computer = DependencyClosureComputer()
+
+    def time_closure_unbounded(self) -> None:
+        """Compute unbounded closure on a 1000-node chain."""
+        self._computer.clear_cache()
+        self._computer.compute_closure(self._graph, self._roots)
+
+    def time_closure_bounded(self) -> None:
+        """Compute bounded closure (cutoff=100) on a 1000-node chain."""
+        self._computer.clear_cache()
+        self._computer.compute_closure(self._graph, self._roots, cutoff=100)
+
+    def time_closure_memoized(self) -> None:
+        """Compute closure twice — second should be memoized."""
+        self._computer.clear_cache()
+        self._computer.compute_closure(self._graph, self._roots)
+        self._computer.compute_closure(self._graph, self._roots)
@@ -0,0 +1,66 @@
+# Large-Project Decomposition
+
+## Overview
+
+The large-project decomposition subsystem breaks large file sets into bounded
+subplans suitable for independent execution. It provides hierarchical
+decomposition with 4+ levels, clustering heuristics, dependency closure
+computation, and DAG execution ordering.
+
+## Configuration
+
+| Setting | Default | Description |
+|---------|---------|-------------|
+| `planner_max_depth` | 4 | Maximum recursion depth |
+| `planner_max_files_per_subplan` | 500 | Upper file count per leaf |
+| `planner_max_tokens_per_subplan` | 100000 | Upper token count per leaf |
+| `planner_min_files_per_subplan` | 10 | Threshold below which decomposition is skipped |
+
+## Clustering Strategies
+
+Three complementary strategies partition files into bounded clusters:
+
+- **Directory** — Groups files by common directory prefix. Splits oversized
+  buckets into chunks of `max_files_per_subplan`.
+- **Language** — Groups files by file extension. Useful for polyglot projects.
+- **Size** — Groups files by cumulative estimated token count.
+
+A deterministic sort ensures clusters are always produced in the same order.
+
+## Dependency Closure
+
+`DependencyClosureComputer` computes bounded transitive closures on file
+dependency graphs:
+
+- `compute_closure(graph, roots, cutoff)` — BFS forward closure with optional
+  size limit. Results are memoised for repeated queries.
+- `topological_sort(graph)` — Kahn's algorithm for DAG execution ordering.
+- `detect_cycles(graph)` — DFS colouring to find elementary cycles.
+
+## Decision Recording
+
+`DecompositionService.record_decisions()` persists two types of decisions:
+
+1. `strategy_choice` — The overall decomposition strategy.
+2. `subplan_spawn` — One per leaf node in the decomposition tree.
+
+## API
+
+```python
+from cleveragents.application.services.decomposition_service import (
+    DecompositionService,
+)
+from cleveragents.application.services.decomposition_models import (
+    DecompositionConfig,
+)
+
+svc = DecompositionService(decision_service=ds)
+result = svc.decompose(files, DecompositionConfig(max_depth=4))
+svc.record_decisions(plan_id, result)
+```
+
+## Related
+
+- [Subplan Service](subplan_service.md)
+- [Decision Service](decision_service.md)
+- Forgejo issue #205
@@ -0,0 +1,217 @@
+Feature: Large-project hierarchical decomposition
+  As a plan orchestrator
+  I want to decompose large file sets into bounded subplans
+  So that each subplan can be executed independently within token budgets
+
+  Background:
+    Given a decomposition service
+
+  # --- decomposition depth -----------------------------------------------
+
+  Scenario: Decomposition produces multiple levels for a deep project
+    Given a project with 2000 files across 5 directory levels
+    When I decompose with max_depth 4
+    Then the decomposition result should have max_depth_reached >= 1
+    And every leaf node should have at most 50 files
+
+  Scenario: Decomposition respects max_depth limit
+    Given a project with 2000 files across 5 directory levels
+    When I decompose with max_depth 2
+    Then the decomposition result should have max_depth_reached <= 2
+
+  # --- bounded context ---------------------------------------------------
+
+  Scenario: Leaf nodes respect max_files_per_subplan
+    Given a project with 1500 files in a single directory
+    When I decompose with max_files_per_subplan 500
+    Then every leaf node should have at most 500 files
+
+  Scenario: Leaf nodes respect max_tokens_per_subplan
+    Given a project with files totalling 500000 estimated tokens
+    When I decompose with max_tokens_per_subplan 100000
+    Then every leaf node should have estimated tokens at most 100000
+
+  # --- language clustering -----------------------------------------------
+
+  Scenario: Files are clustered by language
+    Given a project with 100 Python files and 100 TypeScript files
+    When I decompose with language clustering
+    Then at least two clusters should have different dominant languages
+
+  # --- directory clustering ----------------------------------------------
+
+  Scenario: Files are clustered by directory
+    Given a project with files in src/api and src/web directories
+    When I decompose with directory clustering
+    Then at least two clusters should have different directory prefixes
+
+  # --- dependency closure ------------------------------------------------
+
+  Scenario: Compute transitive closure of root files
+    Given a dependency graph with chain A -> B -> C -> D
+    When I compute closure from A with no cutoff
+    Then the closure should contain A, B, C, D
+
+  Scenario: Bounded closure respects cutoff
+    Given a dependency graph with chain A -> B -> C -> D
+    When I compute closure from A with cutoff 2
+    Then the closure should contain at most 2 files
+
+  # --- DAG ordering ------------------------------------------------------
+
+  Scenario: Topological sort produces valid execution order
+    Given a dependency graph with edges A->B, A->C, B->D, C->D
+    When I compute the topological sort
+    Then A should appear before B and C
+    And B and C should appear before D
+
+  # --- cycle detection ---------------------------------------------------
+
+  Scenario: Detect cycles in dependency graph
+    Given a dependency graph with cycle A -> B -> C -> A
+    When I check for cycles
+    Then at least one cycle should be detected
+
+  Scenario: No cycles in acyclic graph
+    Given a dependency graph with chain A -> B -> C -> D
+    When I check for cycles
+    Then no cycles should be detected
+
+  # --- fallback strategy -------------------------------------------------
+
+  Scenario: Fallback when heuristics produce a single cluster
+    Given a project with 100 files all in the same directory with same extension
+    When I decompose with max_files_per_subplan 200
+    Then the decomposition should produce a single root node
+
+  # --- threshold guard ---------------------------------------------------
+
+  Scenario: Skip decomposition when below threshold
+    Given a project with 5 files
+    When I decompose with min_files_per_subplan 10
+    Then the decomposition should produce a single root node
+    And the metrics should contain skipped equal to 1
+
+  # --- config keys -------------------------------------------------------
+
+  Scenario: Default config has expected values
+    Given default decomposition config
+    Then max_depth should be 4
+    And max_files_per_subplan should be 500
+    And max_tokens_per_subplan should be 100000
+    And min_files_per_subplan should be 10
+
+  Scenario: Invalid config values are rejected
+    When I create a config with max_depth 0
+    Then a decomposition ValueError should be raised
+
+  # --- metrics -----------------------------------------------------------
+
+  Scenario: Decomposition result contains metrics
+    Given a project with 2000 files across 5 directory levels
+    When I decompose with default config
+    Then metrics should contain total_nodes
+    And metrics should contain leaf_nodes
+    And metrics should contain max_depth
+
+  # --- decision recording ------------------------------------------------
+
+  Scenario: Decisions are recorded in DecisionService
+    Given a decomposition service with a mock decision service
+    And a project with 2000 files across 5 directory levels
+    When I decompose and record decisions for plan "01HXBBBBBBBBBBBBBBBBBBBBBB"
+    Then at least one strategy_choice decision should be recorded
+    And at least one subplan_spawn decision should be recorded
+
+  # --- deterministic ordering --------------------------------------------
+
+  Scenario: Deterministic sort produces stable ordering
+    Given two identical file sets presented in different order
+    When I decompose both
+    Then both decomposition results should have identical node ordering
+
+  # --- memoization -------------------------------------------------------
+
+  Scenario: Closure computation is memoised for repeated queries
+    Given a dependency graph with chain A -> B -> C -> D
+    When I compute closure from A twice
+    Then the second computation should return the same result
+
+  # --- additional config validation ----------------------------------------
+
+  Scenario: Config rejects max_files_per_subplan of zero
+    When I create a config with max_files_per_subplan 0
+    Then a decomposition ValueError should be raised
+
+  Scenario: Config rejects max_tokens_per_subplan of zero
+    When I create a config with max_tokens_per_subplan 0
+    Then a decomposition ValueError should be raised
+
+  Scenario: Config rejects min_files_per_subplan of zero
+    When I create a config with min_files_per_subplan 0
+    Then a decomposition ValueError should be raised
+
+  # --- estimate_tokens edge cases ------------------------------------------
+
+  Scenario: Estimate tokens for nonexistent file returns zero
+    When I estimate tokens for a nonexistent file
+    Then the token estimate should be 0
+
+  # --- cluster_by_size with token_map --------------------------------------
+
+  Scenario: Cluster by size uses provided token_map
+    Given a set of files with known token counts
+    When I cluster by size with a token_map and max_tokens 100
+    Then the clusters should respect the token_map values
+
+  # --- closure trimming with multiple roots --------------------------------
+
+  Scenario: Closure trims deterministically when result exceeds cutoff
+    Given a dependency graph with two roots each reaching 3 nodes
+    When I compute closure from both roots with cutoff 3
+    Then the closure should contain at most 3 files
+
+  # --- clear cache ---------------------------------------------------------
+
+  Scenario: Clear cache drops memoised results
+    Given a dependency graph with chain A -> B -> C -> D
+    When I compute closure from A and then clear the cache
+    Then computing closure from A again should still work
+
+  # --- service wrapper methods ---------------------------------------------
+
+  Scenario: Service compute_dependency_order delegates to closure computer
+    Given a dependency graph with edges A->B, A->C, B->D, C->D
+    When I compute dependency order through the service
+    Then A should appear before B and C in the service order
+    And B and C should appear before D in the service order
+
+  Scenario: Service compute_closure delegates to closure computer
+    Given a dependency graph with chain A -> B -> C -> D
+    When I compute closure through the service from A
+    Then the service closure should contain A, B, C, D
+
+  # --- decision_service property -------------------------------------------
+
+  Scenario: Decision service property returns injected service
+    Given a decomposition service with a mock decision service
+    Then the decision service property should return the mock
+
+  Scenario: Decision service property returns None when not injected
+    Then the decision service property should return None
+
+  # --- empty file edge cases -----------------------------------------------
+
+  Scenario: Dominant extension of empty list returns empty string
+    When I compute dominant extension of an empty list
+    Then the dominant extension should be empty
+
+  Scenario: Common prefix of empty list returns empty string
+    When I compute common prefix of an empty list
+    Then the common prefix should be empty
+
+  # --- directory key short path --------------------------------------------
+
+  Scenario: Directory key handles short paths
+    When I compute directory key for a single-component path
+    Then the directory key should be empty string
@@ -0,0 +1,112 @@
+"""Shared test helpers for decomposition BDD step files.
+
+Provides factory functions for creating temporary files, dependency
+graphs, and other test fixtures used by both the primary and coverage
+step implementations.
+"""
+
+from __future__ import annotations
+
+import itertools
+import os
+
+from cleveragents.application.services.decomposition_models import (
+    DependencyEdge,
+    DependencyGraph,
+    DependencyType,
+)
+
+
+def make_files(tmpdir: str, count: int, levels: int = 1) -> list[str]:
+    """Create *count* empty files spread across *levels* directories."""
+    paths: list[str] = []
+    for i in range(count):
+        level = i % levels
+        parts = [f"level{d}" for d in range(level + 1)]
+        dirpath = os.path.join(tmpdir, *parts)
+        os.makedirs(dirpath, exist_ok=True)
+        fpath = os.path.join(dirpath, f"file_{i:05d}.py")
+        with open(fpath, "w") as fh:
+            fh.write("x" * 200)  # ~50 tokens
+        paths.append(fpath)
+    return sorted(paths)
+
+
+def make_lang_files(tmpdir: str, py_count: int, ts_count: int) -> list[str]:
+    """Create Python and TypeScript files in separate dirs."""
+    paths: list[str] = []
+    py_dir = os.path.join(tmpdir, "src", "python")
+    ts_dir = os.path.join(tmpdir, "src", "typescript")
+    os.makedirs(py_dir, exist_ok=True)
+    os.makedirs(ts_dir, exist_ok=True)
+    for i in range(py_count):
+        fpath = os.path.join(py_dir, f"mod_{i:04d}.py")
+        with open(fpath, "w") as fh:
+            fh.write("x" * 200)
+        paths.append(fpath)
+    for i in range(ts_count):
+        fpath = os.path.join(ts_dir, f"mod_{i:04d}.ts")
+        with open(fpath, "w") as fh:
+            fh.write("x" * 200)
+        paths.append(fpath)
+    return sorted(paths)
+
+
+def make_dir_files(tmpdir: str) -> list[str]:
+    """Create files in two distinct directories."""
+    paths: list[str] = []
+    for sub in ("src/api", "src/web"):
+        dirpath = os.path.join(tmpdir, sub)
+        os.makedirs(dirpath, exist_ok=True)
+        for i in range(50):
+            fpath = os.path.join(dirpath, f"f_{i:03d}.py")
+            with open(fpath, "w") as fh:
+                fh.write("x" * 200)
+            paths.append(fpath)
+    return sorted(paths)
+
+
+def make_same_dir_files(tmpdir: str, count: int) -> list[str]:
+    """Create *count* files all in the same directory with same extension."""
+    dirpath = os.path.join(tmpdir, "flat")
+    os.makedirs(dirpath, exist_ok=True)
+    paths: list[str] = []
+    for i in range(count):
+        fpath = os.path.join(dirpath, f"f_{i:04d}.py")
+        with open(fpath, "w") as fh:
+            fh.write("x" * 200)
+        paths.append(fpath)
+    return sorted(paths)
+
+
+def chain_graph(*names: str) -> DependencyGraph:
+    """Build a chain A -> B -> C -> ..."""
+    g = DependencyGraph()
+    for n in names:
+        g.add_node(n)
+    for src, tgt in itertools.pairwise(names):
+        g.add_edge(DependencyEdge(src, tgt, DependencyType.IMPORT))
+    return g
+
+
+def diamond_graph() -> DependencyGraph:
+    """Build A->B, A->C, B->D, C->D."""
+    g = DependencyGraph()
+    for n in ("A", "B", "C", "D"):
+        g.add_node(n)
+    g.add_edge(DependencyEdge("A", "B", DependencyType.IMPORT))
+    g.add_edge(DependencyEdge("A", "C", DependencyType.IMPORT))
+    g.add_edge(DependencyEdge("B", "D", DependencyType.IMPORT))
+    g.add_edge(DependencyEdge("C", "D", DependencyType.IMPORT))
+    return g
+
+
+def cycle_graph() -> DependencyGraph:
+    """Build A -> B -> C -> A."""
+    g = DependencyGraph()
+    for n in ("A", "B", "C"):
+        g.add_node(n)
+    g.add_edge(DependencyEdge("A", "B", DependencyType.IMPORT))
+    g.add_edge(DependencyEdge("B", "C", DependencyType.IMPORT))
+    g.add_edge(DependencyEdge("C", "A", DependencyType.IMPORT))
+    return g
@@ -0,0 +1,242 @@
+"""Additional step implementations for decomposition coverage scenarios.
+
+Exercises edge cases and code paths not covered by the primary
+scenarios: config validation bounds, token estimation errors,
+token_map parameter, closure trimming/clearing, service wrappers,
+property access, empty-list helpers, and short-path directory keys.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from behave import given, then, when  # type: ignore[import-untyped]
+
+from cleveragents.application.services.decomposition_graph import (
+    DependencyClosureComputer,
+)
+from cleveragents.application.services.decomposition_models import (
+    DecompositionConfig,
+    DependencyEdge,
+    DependencyGraph,
+    DependencyType,
+)
+
+# ---------------------------------------------------------------------------
+# Additional config validation
+# ---------------------------------------------------------------------------
+
+
+@when("I create a config with max_files_per_subplan {n:d}")
+def step_when_create_bad_max_files(context: Any, n: int) -> None:
+    try:
+        DecompositionConfig(max_files_per_subplan=n)
+    except ValueError as exc:
+        context.error = exc
+
+
+@when("I create a config with max_tokens_per_subplan {n:d}")
+def step_when_create_bad_max_tokens(context: Any, n: int) -> None:
+    try:
+        DecompositionConfig(max_tokens_per_subplan=n)
+    except ValueError as exc:
+        context.error = exc
+
+
+@when("I create a config with min_files_per_subplan {n:d}")
+def step_when_create_bad_min_files(context: Any, n: int) -> None:
+    try:
+        DecompositionConfig(min_files_per_subplan=n)
+    except ValueError as exc:
+        context.error = exc
+
+
+# ---------------------------------------------------------------------------
+# estimate_tokens edge cases
+# ---------------------------------------------------------------------------
+
+
+@when("I estimate tokens for a nonexistent file")
+def step_when_estimate_nonexistent(context: Any) -> None:
+    from cleveragents.application.services.decomposition_clustering import (
+        estimate_tokens_for_path,
+    )
+
+    context.token_estimate = estimate_tokens_for_path("/no/such/file.py")
+
+
+@then("the token estimate should be {n:d}")
+def step_then_token_estimate(context: Any, n: int) -> None:
+    assert context.token_estimate == n
+
+
+# ---------------------------------------------------------------------------
+# cluster_by_size with token_map
+# ---------------------------------------------------------------------------
+
+
+@given("a set of files with known token counts")
+def step_given_files_with_tokens(context: Any) -> None:
+    context.sized_files = ["a.py", "b.py", "c.py", "d.py"]
+    context.token_map = {"a.py": 40, "b.py": 60, "c.py": 30, "d.py": 70}
+
+
+@when("I cluster by size with a token_map and max_tokens {n:d}")
+def step_when_cluster_by_size_tokenmap(context: Any, n: int) -> None:
+    from cleveragents.application.services.decomposition_clustering import (
+        ClusteringStrategy,
+    )
+
+    context.sized_clusters = ClusteringStrategy.cluster_by_size(
+        context.sized_files, n, token_map=context.token_map
+    )
+
+
+@then("the clusters should respect the token_map values")
+def step_then_clusters_respect_tokenmap(context: Any) -> None:
+    assert len(context.sized_clusters) >= 2
+    for cluster in context.sized_clusters:
+        total = sum(context.token_map[f] for f in cluster)
+        assert total <= 100 + 70  # max_tokens + single large file
+
+
+# ---------------------------------------------------------------------------
+# closure trimming with multiple roots
+# ---------------------------------------------------------------------------
+
+
+@given("a dependency graph with two roots each reaching 3 nodes")
+def step_given_two_root_graph(context: Any) -> None:
+    g = DependencyGraph()
+    for n in ("R1", "R1a", "R1b", "R2", "R2a", "R2b"):
+        g.add_node(n)
+    g.add_edge(DependencyEdge("R1", "R1a", DependencyType.IMPORT))
+    g.add_edge(DependencyEdge("R1", "R1b", DependencyType.IMPORT))
+    g.add_edge(DependencyEdge("R2", "R2a", DependencyType.IMPORT))
+    g.add_edge(DependencyEdge("R2", "R2b", DependencyType.IMPORT))
+    context.graph = g
+
+
+@when("I compute closure from both roots with cutoff {n:d}")
+def step_when_closure_both_roots_cutoff(context: Any, n: int) -> None:
+    computer = DependencyClosureComputer()
+    context.closure = computer.compute_closure(context.graph, ["R1", "R2"], cutoff=n)
+
+
+# ---------------------------------------------------------------------------
+# clear cache
+# ---------------------------------------------------------------------------
+
+
+@when("I compute closure from A and then clear the cache")
+def step_when_closure_and_clear(context: Any) -> None:
+    computer = DependencyClosureComputer()
+    context.closure_before = computer.compute_closure(context.graph, ["A"])
+    computer.clear_cache()
+    context.closure_after = computer.compute_closure(context.graph, ["A"])
+
+
+@then("computing closure from A again should still work")
+def step_then_closure_after_clear(context: Any) -> None:
+    assert context.closure_before == context.closure_after
+
+
+# ---------------------------------------------------------------------------
+# service wrapper methods
+# ---------------------------------------------------------------------------
+
+
+@when("I compute dependency order through the service")
+def step_when_service_topo(context: Any) -> None:
+    context.service_order = context.svc.compute_dependency_order(context.graph)
+
+
+@then("A should appear before B and C in the service order")
+def step_then_service_a_before_bc(context: Any) -> None:
+    order = context.service_order
+    assert order.index("A") < order.index("B")
+    assert order.index("A") < order.index("C")
+
+
+@then("B and C should appear before D in the service order")
+def step_then_service_bc_before_d(context: Any) -> None:
+    order = context.service_order
+    assert order.index("B") < order.index("D")
+    assert order.index("C") < order.index("D")
+
+
+@when("I compute closure through the service from A")
+def step_when_service_closure(context: Any) -> None:
+    context.service_closure = context.svc.compute_closure(context.graph, ["A"])
+
+
+@then("the service closure should contain A, B, C, D")
+def step_then_service_closure_abcd(context: Any) -> None:
+    assert {"A", "B", "C", "D"} <= context.service_closure
+
+
+# ---------------------------------------------------------------------------
+# decision_service property
+# ---------------------------------------------------------------------------
+
+
+@then("the decision service property should return the mock")
+def step_then_ds_property_mock(context: Any) -> None:
+    assert context.svc.decision_service is context.mock_ds
+
+
+@then("the decision service property should return None")
+def step_then_ds_property_none(context: Any) -> None:
+    assert context.svc.decision_service is None
+
+
+# ---------------------------------------------------------------------------
+# empty file edge cases
+# ---------------------------------------------------------------------------
+
+
+@when("I compute dominant extension of an empty list")
+def step_when_dominant_ext_empty(context: Any) -> None:
+    from cleveragents.application.services.decomposition_service import (
+        _dominant_extension,
+    )
+
+    context.dominant_ext = _dominant_extension([])
+
+
+@then("the dominant extension should be empty")
+def step_then_dominant_ext_empty(context: Any) -> None:
+    assert context.dominant_ext == ""
+
+
+@when("I compute common prefix of an empty list")
+def step_when_common_prefix_empty(context: Any) -> None:
+    from cleveragents.application.services.decomposition_service import (
+        _common_prefix,
+    )
+
+    context.common_pfx = _common_prefix([])
+
+
+@then("the common prefix should be empty")
+def step_then_common_prefix_empty(context: Any) -> None:
+    assert context.common_pfx == ""
+
+
+# ---------------------------------------------------------------------------
+# directory key short path
+# ---------------------------------------------------------------------------
+
+
+@when("I compute directory key for a single-component path")
+def step_when_dir_key_short(context: Any) -> None:
+    from cleveragents.application.services.decomposition_clustering import (
+        _directory_key,
+    )
+
+    context.dir_key = _directory_key("file.py", depth=2)
+
+
+@then("the directory key should be empty string")
+def step_then_dir_key_empty(context: Any) -> None:
+    assert context.dir_key == ""
@@ -0,0 +1,451 @@
+"""Step implementations for large_project_decomposition.feature.
+
+Tests hierarchical decomposition, clustering heuristics, dependency
+closure, DAG ordering, cycle detection, fallback, threshold guard,
+config validation, metrics, decision recording, and deterministic
+ordering.
+"""
+
+from __future__ import annotations
+
+import tempfile
+from typing import Any
+from unittest.mock import MagicMock
+
+from behave import given, then, when  # type: ignore[import-untyped]
+
+from cleveragents.application.services.decomposition_graph import (
+    DependencyClosureComputer,
+)
+from cleveragents.application.services.decomposition_models import (
+    DecompositionConfig,
+)
+from cleveragents.application.services.decomposition_service import (
+    DecompositionService,
+)
+from features.steps.decomposition_test_helpers import (
+    chain_graph,
+    cycle_graph,
+    diamond_graph,
+    make_dir_files,
+    make_files,
+    make_lang_files,
+    make_same_dir_files,
+)
+
+# ---------------------------------------------------------------------------
+# Background
+# ---------------------------------------------------------------------------
+
+
+@given("a decomposition service")
+def step_given_decomposition_service(context: Any) -> None:
+    context.svc = DecompositionService()
+    context.tmpdir = tempfile.mkdtemp(prefix="decompose-")
+    context.result = None
+    context.error = None
+    context.graph = None
+    context.closure = None
+    context.topo_order = None
+    context.cycles = None
+    context.files = None
+
+
+# ---------------------------------------------------------------------------
+# Projects
+# ---------------------------------------------------------------------------
+
+
+@given("a project with {count:d} files across {levels:d} directory levels")
+def step_given_multi_level_project(context: Any, count: int, levels: int) -> None:
+    context.files = make_files(context.tmpdir, count, levels)
+
+
+@given("a project with {count:d} files in a single directory")
+def step_given_single_dir_project(context: Any, count: int) -> None:
+    context.files = make_same_dir_files(context.tmpdir, count)
+
+
+@given("a project with files totalling {tokens:d} estimated tokens")
+def step_given_token_project(context: Any, tokens: int) -> None:
+    file_count = tokens // 50
+    context.files = make_files(context.tmpdir, file_count, 3)
+
+
+@given("a project with {py:d} Python files and {ts:d} TypeScript files")
+def step_given_lang_project(context: Any, py: int, ts: int) -> None:
+    context.files = make_lang_files(context.tmpdir, py, ts)
+
+
+@given("a project with files in src/api and src/web directories")
+def step_given_dir_project(context: Any) -> None:
+    context.files = make_dir_files(context.tmpdir)
+
+
+@given("a project with {count:d} files all in the same directory with same extension")
+def step_given_same_dir_same_ext(context: Any, count: int) -> None:
+    context.files = make_same_dir_files(context.tmpdir, count)
+
+
+@given("a project with {count:d} files")
+def step_given_small_project(context: Any, count: int) -> None:
+    context.files = make_files(context.tmpdir, count, 1)
+
+
+# ---------------------------------------------------------------------------
+# Dependency graphs
+# ---------------------------------------------------------------------------
+
+
+@given("a dependency graph with chain A -> B -> C -> D")
+def step_given_chain_graph(context: Any) -> None:
+    context.graph = chain_graph("A", "B", "C", "D")
+
+
+@given("a dependency graph with edges A->B, A->C, B->D, C->D")
+def step_given_diamond_graph(context: Any) -> None:
+    context.graph = diamond_graph()
+
+
+@given("a dependency graph with cycle A -> B -> C -> A")
+def step_given_cycle_graph(context: Any) -> None:
+    context.graph = cycle_graph()
+
+
+# ---------------------------------------------------------------------------
+# Config
+# ---------------------------------------------------------------------------
+
+
+@given("default decomposition config")
+def step_given_default_config(context: Any) -> None:
+    context.decomp_config = DecompositionConfig()
+
+
+@given("a decomposition service with a mock decision service")
+def step_given_svc_with_mock_ds(context: Any) -> None:
+    context.mock_ds = MagicMock()
+    context.svc = DecompositionService(decision_service=context.mock_ds)
+    context.tmpdir = tempfile.mkdtemp(prefix="decompose-")
+
+
+@given("two identical file sets presented in different order")
+def step_given_two_file_sets(context: Any) -> None:
+    context.files = make_files(context.tmpdir, 200, 3)
+    context.files_reversed = list(reversed(context.files))
+
+
+# ---------------------------------------------------------------------------
+# When — decompose
+# ---------------------------------------------------------------------------
+
+
+@when("I decompose with max_depth {depth:d}")
+def step_when_decompose_max_depth(context: Any, depth: int) -> None:
+    cfg = DecompositionConfig(max_depth=depth, max_files_per_subplan=50)
+    context.result = context.svc.decompose(context.files, cfg)
+
+
+@when("I decompose with max_files_per_subplan {n:d}")
+def step_when_decompose_max_files(context: Any, n: int) -> None:
+    cfg = DecompositionConfig(max_files_per_subplan=n)
+    context.result = context.svc.decompose(context.files, cfg)
+
+
+@when("I decompose with max_tokens_per_subplan {n:d}")
+def step_when_decompose_max_tokens(context: Any, n: int) -> None:
+    cfg = DecompositionConfig(max_tokens_per_subplan=n)
+    context.result = context.svc.decompose(context.files, cfg)
+
+
+@when("I decompose with language clustering")
+def step_when_decompose_language(context: Any) -> None:
+    cfg = DecompositionConfig(max_files_per_subplan=80)
+    context.result = context.svc.decompose(context.files, cfg)
+
+
+@when("I decompose with directory clustering")
+def step_when_decompose_directory(context: Any) -> None:
+    cfg = DecompositionConfig(max_files_per_subplan=30)
+    context.result = context.svc.decompose(context.files, cfg)
+
+
+@when("I decompose with min_files_per_subplan {n:d}")
+def step_when_decompose_min_files(context: Any, n: int) -> None:
+    cfg = DecompositionConfig(min_files_per_subplan=n)
+    context.result = context.svc.decompose(context.files, cfg)
+
+
+@when("I decompose with default config")
+def step_when_decompose_default(context: Any) -> None:
+    context.result = context.svc.decompose(context.files)
+
+
+@when('I decompose and record decisions for plan "{plan_id}"')
+def step_when_decompose_and_record(context: Any, plan_id: str) -> None:
+    context.result = context.svc.decompose(context.files)
+    context.svc.record_decisions(plan_id, context.result)
+
+
+@when("I decompose both")
+def step_when_decompose_both(context: Any) -> None:
+    svc = DecompositionService()
+    context.result_a = svc.decompose(context.files)
+    svc2 = DecompositionService()
+    context.result_b = svc2.decompose(context.files_reversed)
+
+
+# ---------------------------------------------------------------------------
+# When — dependency ops
+# ---------------------------------------------------------------------------
+
+
+@when("I compute closure from A with no cutoff")
+def step_when_closure_no_cutoff(context: Any) -> None:
+    computer = DependencyClosureComputer()
+    context.closure = computer.compute_closure(context.graph, ["A"])
+
+
+@when("I compute closure from A with cutoff {n:d}")
+def step_when_closure_cutoff(context: Any, n: int) -> None:
+    computer = DependencyClosureComputer()
+    context.closure = computer.compute_closure(context.graph, ["A"], cutoff=n)
+
+
+@when("I compute the topological sort")
+def step_when_topo_sort(context: Any) -> None:
+    computer = DependencyClosureComputer()
+    context.topo_order = computer.topological_sort(context.graph)
+
+
+@when("I check for cycles")
+def step_when_detect_cycles(context: Any) -> None:
+    computer = DependencyClosureComputer()
+    context.cycles = computer.detect_cycles(context.graph)
+
+
+@when("I compute closure from A twice")
+def step_when_closure_twice(context: Any) -> None:
+    computer = DependencyClosureComputer()
+    context.closure_1 = computer.compute_closure(context.graph, ["A"])
+    context.closure_2 = computer.compute_closure(context.graph, ["A"])
+
+
+# ---------------------------------------------------------------------------
+# When — config
+# ---------------------------------------------------------------------------
+
+
+@when("I create a config with max_depth {n:d}")
+def step_when_create_bad_config(context: Any, n: int) -> None:
+    try:
+        DecompositionConfig(max_depth=n)
+    except ValueError as exc:
+        context.error = exc
+
+
+# ---------------------------------------------------------------------------
+# Then — decomposition assertions
+# ---------------------------------------------------------------------------
+
+
+@then("the decomposition result should have max_depth_reached >= {n:d}")
+def step_then_max_depth_ge(context: Any, n: int) -> None:
+    assert context.result is not None
+    assert context.result.max_depth_reached >= n, (
+        f"expected >= {n}, got {context.result.max_depth_reached}"
+    )
+
+
+@then("the decomposition result should have max_depth_reached <= {n:d}")
+def step_then_max_depth_le(context: Any, n: int) -> None:
+    assert context.result is not None
+    assert context.result.max_depth_reached <= n
+
+
+@then("every leaf node should have at most {n:d} files")
+def step_then_leaf_files_le(context: Any, n: int) -> None:
+    assert context.result is not None
+    for node in context.result.nodes:
+        if node.is_leaf:
+            assert len(node.file_paths) <= n, (
+                f"leaf {node.node_id} has {len(node.file_paths)} files"
+            )
+
+
+@then("every leaf node should have estimated tokens at most {n:d}")
+def step_then_leaf_tokens_le(context: Any, n: int) -> None:
+    assert context.result is not None
+    for node in context.result.nodes:
+        if node.is_leaf:
+            assert node.estimated_tokens <= n, (
+                f"leaf {node.node_id} has {node.estimated_tokens} tokens"
+            )
+
+
+@then("at least two clusters should have different dominant languages")
+def step_then_diff_languages(context: Any) -> None:
+    assert context.result is not None
+    langs = {n.language for n in context.result.nodes if n.is_leaf}
+    assert len(langs) >= 2, f"only {langs}"
+
+
+@then("at least two clusters should have different directory prefixes")
+def step_then_diff_dirs(context: Any) -> None:
+    assert context.result is not None
+    prefixes = {n.directory_prefix for n in context.result.nodes if n.is_leaf}
+    assert len(prefixes) >= 2, f"only {prefixes}"
+
+
+@then("the decomposition should produce a single root node")
+def step_then_single_root(context: Any) -> None:
+    assert context.result is not None
+    roots = [n for n in context.result.nodes if n.parent_id is None]
+    assert len(roots) >= 1
+
+
+@then("the metrics should contain skipped equal to {n:d}")
+def step_then_metrics_skipped(context: Any, n: int) -> None:
+    assert context.result is not None
+    assert context.result.metrics.get("skipped") == n
+
+
+# ---------------------------------------------------------------------------
+# Then — closure / topo / cycle
+# ---------------------------------------------------------------------------
+
+
+@then("the closure should contain A, B, C, D")
+def step_then_closure_abcd(context: Any) -> None:
+    assert context.closure is not None
+    assert {"A", "B", "C", "D"} <= context.closure
+
+
+@then("the closure should contain at most {n:d} files")
+def step_then_closure_le(context: Any, n: int) -> None:
+    assert context.closure is not None
+    assert len(context.closure) <= n
+
+
+@then("A should appear before B and C")
+def step_then_a_before_bc(context: Any) -> None:
+    order = context.topo_order
+    assert order.index("A") < order.index("B")
+    assert order.index("A") < order.index("C")
+
+
+@then("B and C should appear before D")
+def step_then_bc_before_d(context: Any) -> None:
+    order = context.topo_order
+    assert order.index("B") < order.index("D")
+    assert order.index("C") < order.index("D")
+
+
+@then("at least one cycle should be detected")
+def step_then_cycles_detected(context: Any) -> None:
+    assert context.cycles is not None
+    assert len(context.cycles) >= 1
+
+
+@then("no cycles should be detected")
+def step_then_no_cycles(context: Any) -> None:
+    assert context.cycles is not None
+    assert len(context.cycles) == 0
+
+
+# ---------------------------------------------------------------------------
+# Then — config
+# ---------------------------------------------------------------------------
+
+
+@then("max_depth should be {n:d}")
+def step_then_cfg_max_depth(context: Any, n: int) -> None:
+    assert context.decomp_config.max_depth == n
+
+
+@then("max_files_per_subplan should be {n:d}")
+def step_then_cfg_max_files(context: Any, n: int) -> None:
+    assert context.decomp_config.max_files_per_subplan == n
+
+
+@then("max_tokens_per_subplan should be {n:d}")
+def step_then_cfg_max_tokens(context: Any, n: int) -> None:
+    assert context.decomp_config.max_tokens_per_subplan == n
+
+
+@then("min_files_per_subplan should be {n:d}")
+def step_then_cfg_min_files(context: Any, n: int) -> None:
+    assert context.decomp_config.min_files_per_subplan == n
+
+
+@then("a decomposition ValueError should be raised")
+def step_then_decomposition_value_error(context: Any) -> None:
+    assert context.error is not None
+    assert isinstance(context.error, ValueError)
+
+
+# ---------------------------------------------------------------------------
+# Then — metrics
+# ---------------------------------------------------------------------------
+
+
+@then("metrics should contain total_nodes")
+def step_then_metrics_total(context: Any) -> None:
+    assert "total_nodes" in context.result.metrics
+
+
+@then("metrics should contain leaf_nodes")
+def step_then_metrics_leaves(context: Any) -> None:
+    assert "leaf_nodes" in context.result.metrics
+
+
+@then("metrics should contain max_depth")
+def step_then_metrics_depth(context: Any) -> None:
+    assert "max_depth" in context.result.metrics
+
+
+# ---------------------------------------------------------------------------
+# Then — decision recording
+# ---------------------------------------------------------------------------
+
+
+@then("at least one strategy_choice decision should be recorded")
+def step_then_strategy_choice(context: Any) -> None:
+    calls = context.mock_ds.record_decision.call_args_list
+    types = [
+        str(c.kwargs.get("decision_type", c.args[1] if len(c.args) > 1 else ""))
+        for c in calls
+    ]
+    assert any("strategy_choice" in t for t in types)
+
+
+@then("at least one subplan_spawn decision should be recorded")
+def step_then_subplan_spawn(context: Any) -> None:
+    calls = context.mock_ds.record_decision.call_args_list
+    types = [
+        str(c.kwargs.get("decision_type", c.args[1] if len(c.args) > 1 else ""))
+        for c in calls
+    ]
+    assert any("subplan_spawn" in t for t in types)
+
+
+# ---------------------------------------------------------------------------
+# Then — deterministic
+# ---------------------------------------------------------------------------
+
+
+@then("both decomposition results should have identical node ordering")
+def step_then_identical_ordering(context: Any) -> None:
+    ids_a = [n.node_id for n in context.result_a.nodes]
+    ids_b = [n.node_id for n in context.result_b.nodes]
+    assert len(ids_a) == len(ids_b)
+
+
+# ---------------------------------------------------------------------------
+# Then — memoization
+# ---------------------------------------------------------------------------
+
+
+@then("the second computation should return the same result")
+def step_then_same_result(context: Any) -> None:
+    assert context.closure_1 == context.closure_2
@@ -0,0 +1,167 @@
+"""Helper script for large-project decomposition Robot Framework tests.
+
+Usage:
+    python helper_large_project_decompose.py below-threshold
+    python helper_large_project_decompose.py multi-level
+    python helper_large_project_decompose.py closure
+    python helper_large_project_decompose.py topo-sort
+    python helper_large_project_decompose.py cycle-detect
+    python helper_large_project_decompose.py config-validation
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import tempfile
+from pathlib import Path
+
+# Ensure source tree is importable
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+from cleveragents.application.services.decomposition_graph import (  # noqa: E402
+    DependencyClosureComputer,
+)
+from cleveragents.application.services.decomposition_models import (  # noqa: E402
+    DecompositionConfig,
+    DependencyEdge,
+    DependencyGraph,
+    DependencyType,
+)
+from cleveragents.application.services.decomposition_service import (  # noqa: E402
+    DecompositionService,
+)
+
+
+def _make_files(tmpdir: str, count: int, levels: int = 1) -> list[str]:
+    """Create *count* empty files spread across *levels* directories."""
+    paths: list[str] = []
+    for i in range(count):
+        level = i % levels
+        parts = [f"level{d}" for d in range(level + 1)]
+        dirpath = os.path.join(tmpdir, *parts)
+        os.makedirs(dirpath, exist_ok=True)
+        fpath = os.path.join(dirpath, f"file_{i:05d}.py")
+        with open(fpath, "w") as fh:
+            fh.write("x" * 200)
+        paths.append(fpath)
+    return sorted(paths)
+
+
+def cmd_below_threshold() -> None:
+    """Verify decomposition is skipped below threshold."""
+    tmpdir = tempfile.mkdtemp(prefix="decompose-")
+    files = _make_files(tmpdir, 5, 1)
+    svc = DecompositionService()
+    result = svc.decompose(files, DecompositionConfig(min_files_per_subplan=10))
+    assert result.metrics.get("skipped") == 1
+    assert len(result.nodes) == 1
+    print("decompose-below-threshold-ok")
+
+
+def cmd_multi_level() -> None:
+    """Verify multi-level decomposition produces multiple nodes."""
+    tmpdir = tempfile.mkdtemp(prefix="decompose-")
+    files = _make_files(tmpdir, 500, 5)
+    svc = DecompositionService()
+    result = svc.decompose(files, DecompositionConfig(max_files_per_subplan=100))
+    assert len(result.nodes) > 1
+    assert result.max_depth_reached >= 1
+    print(f"decompose-multi-level-ok nodes={len(result.nodes)}")
+
+
+def cmd_closure() -> None:
+    """Verify transitive closure computation."""
+    g = DependencyGraph()
+    for n in ("A", "B", "C", "D"):
+        g.add_node(n)
+    g.add_edge(DependencyEdge("A", "B", DependencyType.IMPORT))
+    g.add_edge(DependencyEdge("B", "C", DependencyType.IMPORT))
+    g.add_edge(DependencyEdge("C", "D", DependencyType.IMPORT))
+
+    computer = DependencyClosureComputer()
+    closure = computer.compute_closure(g, ["A"])
+    assert {"A", "B", "C", "D"} <= closure
+    print(f"decompose-closure-ok size={len(closure)}")
+
+
+def cmd_topo_sort() -> None:
+    """Verify topological sort produces valid order."""
+    g = DependencyGraph()
+    for n in ("A", "B", "C", "D"):
+        g.add_node(n)
+    g.add_edge(DependencyEdge("A", "B", DependencyType.IMPORT))
+    g.add_edge(DependencyEdge("A", "C", DependencyType.IMPORT))
+    g.add_edge(DependencyEdge("B", "D", DependencyType.IMPORT))
+    g.add_edge(DependencyEdge("C", "D", DependencyType.IMPORT))
+
+    computer = DependencyClosureComputer()
+    order = computer.topological_sort(g)
+    assert order.index("A") < order.index("B")
+    assert order.index("A") < order.index("C")
+    assert order.index("B") < order.index("D")
+    assert order.index("C") < order.index("D")
+    print(f"decompose-topo-sort-ok order={order}")
+
+
+def cmd_cycle_detect() -> None:
+    """Verify cycle detection finds a cycle."""
+    g = DependencyGraph()
+    for n in ("A", "B", "C"):
+        g.add_node(n)
+    g.add_edge(DependencyEdge("A", "B", DependencyType.IMPORT))
+    g.add_edge(DependencyEdge("B", "C", DependencyType.IMPORT))
+    g.add_edge(DependencyEdge("C", "A", DependencyType.IMPORT))
+
+    computer = DependencyClosureComputer()
+    cycles = computer.detect_cycles(g)
+    assert len(cycles) >= 1
+    print(f"decompose-cycle-detect-ok cycles={len(cycles)}")
+
+
+def cmd_config_validation() -> None:
+    """Verify invalid config raises ValueError."""
+    for bad_kwargs in (
+        {"max_depth": 0},
+        {"max_files_per_subplan": 0},
+        {"max_tokens_per_subplan": 0},
+        {"min_files_per_subplan": 0},
+    ):
+        try:
+            DecompositionConfig(**bad_kwargs)  # type: ignore[arg-type]
+            print(f"FAIL: {bad_kwargs} did not raise")
+            sys.exit(1)
+        except ValueError:
+            pass
+
+    # Valid config should work
+    DecompositionConfig()
+    print("decompose-config-ok")
+
+
+def main() -> None:
+    """Dispatch subcommand."""
+    if len(sys.argv) < 2:
+        print("Usage: helper_large_project_decompose.py <command>")
+        sys.exit(1)
+
+    cmd = sys.argv[1]
+    dispatch = {
+        "below-threshold": cmd_below_threshold,
+        "multi-level": cmd_multi_level,
+        "closure": cmd_closure,
+        "topo-sort": cmd_topo_sort,
+        "cycle-detect": cmd_cycle_detect,
+        "config-validation": cmd_config_validation,
+    }
+    func = dispatch.get(cmd)
+    if func is None:
+        print(f"Unknown command: {cmd}")
+        sys.exit(1)
+    func()
+
+
+if __name__ == "__main__":
+    main()
@@ -0,0 +1,47 @@
+*** Settings ***
+Documentation    Smoke tests for large-project decomposition service
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    ${CURDIR}/helper_large_project_decompose.py
+
+*** Test Cases ***
+Decompose Small Project Below Threshold
+    [Documentation]    Projects below min_files threshold get a single node
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    below-threshold    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    decompose-below-threshold-ok
+
+Decompose Multi-Level Project
+    [Documentation]    Projects with many files produce multi-level decomposition
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    multi-level    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    decompose-multi-level-ok
+
+Dependency Closure Computation
+    [Documentation]    Transitive closure should collect reachable nodes
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    closure    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    decompose-closure-ok
+
+Topological Sort
+    [Documentation]    Topo sort should produce valid execution order
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    topo-sort    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    decompose-topo-sort-ok
+
+Cycle Detection
+    [Documentation]    Cycles in dependency graph should be detected
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    cycle-detect    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    decompose-cycle-detect-ok
+
+Config Validation
+    [Documentation]    Invalid config should raise ValueError
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    config-validation    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    decompose-config-ok
@@ -20,6 +20,9 @@ from cleveragents.application.services.context_tiers import (
    ContextTierService,
 )
 from cleveragents.application.services.decision_service import DecisionService
+from cleveragents.application.services.decomposition_service import (
+    DecompositionService,
+)
 from cleveragents.application.services.execution_environment_resolver import (
    ExecutionEnvironmentResolver,
 )
@@ -320,6 +323,13 @@ class Container(containers.DeclarativeContainer):
        settings=settings,
    )

+    # Decomposition Service - Factory (uses DecisionService + Settings)
+    decomposition_service = providers.Factory(
+        DecompositionService,
+        decision_service=decision_service,
+        settings=settings,
+    )
+
    # Skeleton Compressor Service - stateless, Singleton is sufficient
    skeleton_compressor_service = providers.Singleton(
        SkeletonCompressorService,
@@ -22,6 +22,24 @@ from cleveragents.application.services.decision_service import (
    SequenceConflictError,
    SnapshotStore,
 )
+from cleveragents.application.services.decomposition_clustering import (
+    ClusteringStrategy,
+)
+from cleveragents.application.services.decomposition_graph import (
+    DependencyClosureComputer,
+)
+from cleveragents.application.services.decomposition_models import (
+    ClusterStrategy,
+    DecompositionConfig,
+    DecompositionNode,
+    DecompositionResult,
+    DependencyEdge,
+    DependencyGraph,
+    DependencyType,
+)
+from cleveragents.application.services.decomposition_service import (
+    DecompositionService,
+)
 from cleveragents.application.services.execution_environment_resolver import (
    ContainerUnavailableError,
    ExecutionEnvironmentResolver,
@@ -122,6 +140,8 @@ __all__ = [
    "AttachmentScope",
    "AutonomyGuardrailService",
    "BrokenReferenceRule",
+    "ClusterStrategy",
+    "ClusteringStrategy",
    "CompressionResult",
    "ConfigEntry",
    "ConfigLevel",
@@ -131,8 +151,16 @@ __all__ = [
    "CorrectionService",
    "DecisionNotFoundError",
    "DecisionService",
+    "DecompositionConfig",
+    "DecompositionNode",
+    "DecompositionResult",
+    "DecompositionService",
    "DefaultValidationRunner",
+    "DependencyClosureComputer",
    "DependencyCycleRule",
+    "DependencyEdge",
+    "DependencyGraph",
+    "DependencyType",
    "DuplicateDecisionError",
    "DuplicateImportRule",
    "ExecutionEnvironmentResolver",
@@ -0,0 +1,216 @@
+"""Clustering strategies for large-project decomposition.
+
+Provides heuristics that partition a flat list of file paths into
+bounded clusters suitable for independent subplan execution.  Three
+complementary strategies are offered:
+
+- **directory** — group by common directory prefix.
+- **language** — group by file extension / programming language.
+- **size** — group by cumulative estimated token count.
+
+A deterministic sort ensures that clusters and their contents are
+always produced in the same order regardless of filesystem iteration
+order.
+
+Based on Forgejo issue #205.
+"""
+
+from __future__ import annotations
+
+import os
+from collections import defaultdict
+
+# ---------------------------------------------------------------------------
+# Token estimation
+# ---------------------------------------------------------------------------
+
+#: Rough bytes-per-token ratio (conservative).  Used when the caller
+#: does not supply pre-computed token counts.
+_BYTES_PER_TOKEN: int = 4
+
+
+def estimate_tokens_for_path(file_path: str) -> int:
+    """Return a rough token estimate for *file_path*.
+
+    Falls back to ``0`` when the file cannot be read (e.g. binary or
+    missing).  This function is intentionally cheap — it reads the
+    file size from the OS, not the actual content.
+    """
+    try:
+        size = os.path.getsize(file_path)
+    except OSError:
+        return 0
+    return max(size // _BYTES_PER_TOKEN, 1)
+
+
+# ---------------------------------------------------------------------------
+# Clustering helpers
+# ---------------------------------------------------------------------------
+
+
+def _extension_of(path: str) -> str:
+    """Return the lowercased file extension (e.g. ``".py"``)."""
+    _, ext = os.path.splitext(path)
+    return ext.lower()
+
+
+def _directory_key(path: str, depth: int = 2) -> str:
+    """Return the first *depth* path components as a grouping key.
+
+    For ``"src/cleveragents/foo/bar.py"`` with *depth=2* the key is
+    ``"src/cleveragents"``.
+    """
+    parts = path.replace("\\", "/").split("/")
+    return "/".join(parts[:depth]) if len(parts) > depth else "/".join(parts[:-1])
+
+
+# ---------------------------------------------------------------------------
+# Public clustering API
+# ---------------------------------------------------------------------------
+
+
+class ClusteringStrategy:
+    """Stateless helper that partitions files into bounded clusters.
+
+    All methods return ``list[list[str]]`` — an ordered sequence of
+    clusters where each inner list contains file paths.
+    """
+
+    @staticmethod
+    def cluster_by_directory(
+        files: list[str],
+        max_per_cluster: int,
+        *,
+        depth: int = 2,
+    ) -> list[list[str]]:
+        """Group *files* by directory prefix.
+
+        Args:
+            files: File paths to partition.
+            max_per_cluster: Maximum files per cluster.
+            depth: Number of leading path components for the grouping key.
+
+        Returns:
+            Ordered list of clusters.
+        """
+        buckets: dict[str, list[str]] = defaultdict(list)
+        for f in sorted(files):
+            buckets[_directory_key(f, depth=depth)].append(f)
+
+        clusters: list[list[str]] = []
+        for key in sorted(buckets):
+            bucket = buckets[key]
+            # Split oversized buckets
+            for i in range(0, len(bucket), max_per_cluster):
+                clusters.append(bucket[i : i + max_per_cluster])
+        return clusters
+
+    @staticmethod
+    def cluster_by_language(
+        files: list[str],
+        max_per_cluster: int,
+    ) -> list[list[str]]:
+        """Group *files* by programming language (file extension).
+
+        Args:
+            files: File paths to partition.
+            max_per_cluster: Maximum files per cluster.
+
+        Returns:
+            Ordered list of clusters.
+        """
+        buckets: dict[str, list[str]] = defaultdict(list)
+        for f in sorted(files):
+            buckets[_extension_of(f)].append(f)
+
+        clusters: list[list[str]] = []
+        for ext in sorted(buckets):
+            bucket = buckets[ext]
+            for i in range(0, len(bucket), max_per_cluster):
+                clusters.append(bucket[i : i + max_per_cluster])
+        return clusters
+
+    @staticmethod
+    def cluster_by_size(
+        files: list[str],
+        max_tokens: int,
+        *,
+        token_map: dict[str, int] | None = None,
+    ) -> list[list[str]]:
+        """Group *files* so each cluster stays within *max_tokens*.
+
+        Args:
+            files: File paths to partition.
+            max_tokens: Maximum estimated tokens per cluster.
+            token_map: Optional pre-computed mapping of path → tokens.
+                When ``None``, :func:`estimate_tokens_for_path` is used.
+
+        Returns:
+            Ordered list of clusters.
+        """
+        sorted_files = sorted(files)
+        clusters: list[list[str]] = []
+        current: list[str] = []
+        current_tokens = 0
+
+        for f in sorted_files:
+            tokens = (
+                token_map[f]
+                if token_map and f in token_map
+                else estimate_tokens_for_path(f)
+            )
+            if current and current_tokens + tokens > max_tokens:
+                clusters.append(current)
+                current = []
+                current_tokens = 0
+            current.append(f)
+            current_tokens += tokens
+
+        if current:
+            clusters.append(current)
+        return clusters
+
+    @staticmethod
+    def deterministic_sort(
+        clusters: list[list[str]],
+        language_priority: list[str] | None = None,
+    ) -> list[list[str]]:
+        """Sort *clusters* for deterministic execution ordering.
+
+        Ordering criteria (most significant first):
+
+        1. Average directory depth (shallower first).
+        2. Dominant language position in *language_priority*.
+        3. Total estimated file size (smaller first).
+        4. First file path (lexicographic tiebreak).
+
+        Args:
+            clusters: Clusters to sort.
+            language_priority: Ordered extensions for language ranking.
+
+        Returns:
+            New list with deterministic ordering.
+        """
+        priority = language_priority or []
+        priority_map: dict[str, int] = {ext: idx for idx, ext in enumerate(priority)}
+        max_priority = len(priority)
+
+        def _sort_key(cluster: list[str]) -> tuple[float, int, int, str]:
+            avg_depth = sum(f.count("/") for f in cluster) / max(len(cluster), 1)
+            # Dominant extension = most frequent
+            ext_counts: dict[str, int] = defaultdict(int)
+            for f in cluster:
+                ext_counts[_extension_of(f)] += 1
+            dominant = max(ext_counts, key=ext_counts.get, default="")  # type: ignore[arg-type]
+            lang_rank = priority_map.get(dominant, max_priority)
+            total_size = len(cluster)
+            first = cluster[0] if cluster else ""
+            return (avg_depth, lang_rank, total_size, first)
+
+        return sorted(clusters, key=_sort_key)
+
+
+__all__: list[str] = [
+    "ClusteringStrategy",
+    "estimate_tokens_for_path",
+]
@@ -0,0 +1,186 @@
+"""Dependency closure and DAG execution ordering for large projects.
+
+Provides :class:`DependencyClosureComputer` which operates on a
+:class:`DependencyGraph` to compute bounded transitive closures,
+topological (DAG) execution orderings, and cycle detection.
+
+Memoization is applied to the closure computation so that graphs
+with 10 000+ files remain tractable.
+
+Based on Forgejo issue #205.
+"""
+
+from __future__ import annotations
+
+from collections import deque
+
+from cleveragents.application.services.decomposition_models import (
+    DependencyGraph,
+)
+
+# ---------------------------------------------------------------------------
+# Dependency closure computer
+# ---------------------------------------------------------------------------
+
+
+class DependencyClosureComputer:
+    """Computes dependency closures and DAG orderings on a file graph.
+
+    All methods are stateless with respect to the graph itself; the
+    memoization cache is instance-scoped and can be reused across
+    multiple calls on the same graph.
+    """
+
+    def __init__(self) -> None:
+        self._closure_cache: dict[str, frozenset[str]] = {}
+
+    # ------------------------------------------------------------------
+    # Public API
+    # ------------------------------------------------------------------
+
+    def compute_closure(
+        self,
+        graph: DependencyGraph,
+        root_files: list[str],
+        cutoff: int = 0,
+    ) -> frozenset[str]:
+        """Return the bounded transitive closure of *root_files*.
+
+        Starting from every file in *root_files*, follows forward edges
+        in *graph* (breadth-first) collecting all reachable nodes.
+        When *cutoff* > 0 the traversal stops once the closure reaches
+        *cutoff* nodes.
+
+        Results are memoised per root file so repeated queries are O(1).
+
+        Args:
+            graph: The dependency graph to traverse.
+            root_files: Starting file paths.
+            cutoff: Maximum closure size.  ``0`` means unlimited.
+
+        Returns:
+            Frozenset of all reachable file paths (including roots).
+        """
+        result: set[str] = set()
+        for root in root_files:
+            if root in self._closure_cache:
+                result.update(self._closure_cache[root])
+                continue
+
+            visited: set[str] = set()
+            queue: deque[str] = deque()
+            if root in graph.file_nodes:
+                queue.append(root)
+
+            while queue:
+                if cutoff > 0 and len(visited) >= cutoff:
+                    break
+                node = queue.popleft()
+                if node in visited:
+                    continue
+                visited.add(node)
+                for succ in graph.successors(node):
+                    if succ not in visited:
+                        queue.append(succ)
+
+            frozen = frozenset(visited)
+            self._closure_cache[root] = frozen
+            result.update(frozen)
+
+        if cutoff > 0 and len(result) > cutoff:
+            # Deterministic trimming — keep the roots plus
+            # lexicographically first nodes up to *cutoff*.
+            root_set = set(root_files)
+            extras = sorted(result - root_set)
+            kept = root_set | set(extras[: cutoff - len(root_set)])
+            return frozenset(kept)
+        return frozenset(result)
+
+    def topological_sort(
+        self,
+        graph: DependencyGraph,
+    ) -> list[str]:
+        """Return a topological (Kahn) ordering of *graph*.
+
+        If the graph contains cycles the returned list will be
+        shorter than the number of nodes — call :meth:`detect_cycles`
+        to diagnose.
+
+        Returns:
+            List of file paths in dependency-safe execution order
+            (nodes with no incoming edges first).
+        """
+        in_degree: dict[str, int] = dict(graph.in_degree)
+        for node in graph.file_nodes:
+            in_degree.setdefault(node, 0)
+
+        queue: deque[str] = deque(sorted(n for n, d in in_degree.items() if d == 0))
+        result: list[str] = []
+
+        while queue:
+            node = queue.popleft()
+            result.append(node)
+            for succ in sorted(graph.successors(node)):
+                in_degree[succ] -= 1
+                if in_degree[succ] == 0:
+                    queue.append(succ)
+
+        return result
+
+    def detect_cycles(
+        self,
+        graph: DependencyGraph,
+    ) -> list[list[str]]:
+        """Return all elementary cycles in *graph*.
+
+        Uses iterative DFS colouring (white/grey/black).  Each time a
+        back-edge is found, the cycle path is extracted from the
+        recursion stack.
+
+        Returns:
+            List of cycles, each cycle being an ordered list of file
+            paths forming the loop.  Empty when the graph is a DAG.
+        """
+        white: set[str] = set(graph.file_nodes)
+        grey: set[str] = set()
+        black: set[str] = set()
+        parent_map: dict[str, str | None] = {}
+        cycles: list[list[str]] = []
+
+        def _dfs(node: str) -> None:
+            white.discard(node)
+            grey.add(node)
+            for succ in sorted(graph.successors(node)):
+                if succ in grey:
+                    # Back-edge found — extract cycle
+                    cycle = [succ, node]
+                    cur = node
+                    while cur != succ:
+                        p = parent_map.get(cur)
+                        if p is None or p == succ:
+                            break
+                        cycle.append(p)
+                        cur = p
+                    cycle.reverse()
+                    cycles.append(cycle)
+                elif succ in white:
+                    parent_map[succ] = node
+                    _dfs(succ)
+            grey.discard(node)
+            black.add(node)
+
+        for start in sorted(graph.file_nodes):
+            if start in white:
+                parent_map[start] = None
+                _dfs(start)
+
+        return cycles
+
+    def clear_cache(self) -> None:
+        """Drop all memoised closure results."""
+        self._closure_cache.clear()
+
+
+__all__: list[str] = [
+    "DependencyClosureComputer",
+]
@@ -0,0 +1,215 @@
+"""Domain models for large-project hierarchical decomposition.
+
+Defines the configuration, tree nodes, dependency edges, and result
+containers used by :class:`DecompositionService` to break a large file
+set into bounded subplans suitable for independent execution.
+
+Key models:
+
+- :class:`DecompositionConfig` — tuneable heuristic parameters.
+- :class:`DecompositionNode` — a single node in the decomposition tree.
+- :class:`DecompositionResult` — full decomposition output with metrics.
+- :class:`DependencyEdge` / :class:`DependencyGraph` — dependency DAG
+  structures consumed by :class:`DependencyClosureComputer`.
+
+Based on Forgejo issue #205.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from enum import StrEnum
+
+from pydantic import BaseModel, ConfigDict, Field
+
+# ---------------------------------------------------------------------------
+# Enumerations
+# ---------------------------------------------------------------------------
+
+
+class ClusterStrategy(StrEnum):
+    """Strategy used to partition files into a cluster."""
+
+    DIRECTORY = "directory"
+    LANGUAGE = "language"
+    SIZE = "size"
+    FALLBACK = "fallback"
+
+
+# ---------------------------------------------------------------------------
+# Configuration
+# ---------------------------------------------------------------------------
+
+
+@dataclass(frozen=True)
+class DecompositionConfig:
+    """Tuneable parameters for hierarchical decomposition.
+
+    Attributes:
+        max_depth: Maximum recursion depth (root = 0).
+        max_files_per_subplan: Upper bound on files in a single leaf.
+        max_tokens_per_subplan: Upper bound on estimated tokens per leaf.
+        min_files_per_subplan: Below this, no further splitting.
+        language_priority: Ordered list of language extensions used for
+            deterministic cluster ordering.
+    """
+
+    max_depth: int = 4
+    max_files_per_subplan: int = 500
+    max_tokens_per_subplan: int = 100_000
+    min_files_per_subplan: int = 10
+    language_priority: list[str] = field(
+        default_factory=lambda: [".py", ".ts", ".js", ".go", ".rs", ".java"]
+    )
+
+    def __post_init__(self) -> None:
+        if self.max_depth < 1:
+            raise ValueError("max_depth must be >= 1")
+        if self.max_files_per_subplan < 1:
+            raise ValueError("max_files_per_subplan must be >= 1")
+        if self.max_tokens_per_subplan < 1:
+            raise ValueError("max_tokens_per_subplan must be >= 1")
+        if self.min_files_per_subplan < 1:
+            raise ValueError("min_files_per_subplan must be >= 1")
+
+
+# ---------------------------------------------------------------------------
+# Decomposition tree
+# ---------------------------------------------------------------------------
+
+
+@dataclass(frozen=True)
+class DecompositionNode:
+    """A single node in the decomposition hierarchy.
+
+    Leaf nodes contain file paths assigned to a subplan.
+    Internal nodes are structural grouping points.
+
+    Attributes:
+        node_id: Stable, deterministic identifier for this node.
+        parent_id: ``node_id`` of the parent (``None`` for root).
+        depth: Distance from root (root = 0).
+        file_paths: Files assigned to this node.
+        language: Dominant language extension (e.g. ``".py"``).
+        directory_prefix: Common directory prefix for contained files.
+        estimated_tokens: Estimated total token count for the files.
+        strategy: Clustering strategy that produced this node.
+        children_ids: Ordered list of child ``node_id`` values.
+    """
+
+    node_id: str
+    parent_id: str | None
+    depth: int
+    file_paths: list[str]
+    language: str
+    directory_prefix: str
+    estimated_tokens: int
+    strategy: ClusterStrategy = ClusterStrategy.DIRECTORY
+    children_ids: list[str] = field(default_factory=list)
+
+    @property
+    def is_leaf(self) -> bool:
+        """Return ``True`` when the node has no children."""
+        return len(self.children_ids) == 0
+
+
+@dataclass(frozen=True)
+class DecompositionResult:
+    """Full output of a hierarchical decomposition pass.
+
+    Attributes:
+        nodes: Every node in the decomposition tree (BFS order).
+        max_depth_reached: Deepest level actually produced.
+        total_files: Number of unique files across all leaves.
+        metrics: Arbitrary key/value diagnostic counters.
+    """
+
+    nodes: list[DecompositionNode]
+    max_depth_reached: int
+    total_files: int
+    metrics: dict[str, int | float] = field(default_factory=dict)
+
+
+# ---------------------------------------------------------------------------
+# Dependency graph
+# ---------------------------------------------------------------------------
+
+
+class DependencyType(StrEnum):
+    """Kind of dependency between two files."""
+
+    IMPORT = "import"
+    INCLUDE = "include"
+    CALL = "call"
+    INHERIT = "inherit"
+    UNKNOWN = "unknown"
+
+
+@dataclass(frozen=True)
+class DependencyEdge:
+    """Directed edge in the file dependency graph.
+
+    Attributes:
+        source_file: The file that depends on *target_file*.
+        target_file: The file being depended upon.
+        dependency_type: Kind of dependency.
+    """
+
+    source_file: str
+    target_file: str
+    dependency_type: DependencyType = DependencyType.IMPORT
+
+
+class DependencyGraph(BaseModel):
+    """Directed graph of file-level dependencies.
+
+    Maintains an adjacency list and per-node in-degree for efficient
+    topological sort and closure computation.
+
+    Attributes:
+        edges: All edges in the graph.
+        file_nodes: Set of all known file paths (nodes).
+        adjacency_list: Forward adjacency: source → [targets].
+        in_degree: Number of incoming edges per node.
+    """
+
+    model_config = ConfigDict(arbitrary_types_allowed=True)
+
+    edges: list[DependencyEdge] = Field(default_factory=list)
+    file_nodes: set[str] = Field(default_factory=set)
+    adjacency_list: dict[str, list[str]] = Field(default_factory=dict)
+    in_degree: dict[str, int] = Field(default_factory=dict)
+
+    # ------------------------------------------------------------------
+    # Mutation helpers
+    # ------------------------------------------------------------------
+
+    def add_edge(self, edge: DependencyEdge) -> None:
+        """Add an edge, updating adjacency and in-degree."""
+        self.edges.append(edge)
+        self.file_nodes.add(edge.source_file)
+        self.file_nodes.add(edge.target_file)
+        self.adjacency_list.setdefault(edge.source_file, []).append(edge.target_file)
+        self.in_degree.setdefault(edge.source_file, 0)
+        self.in_degree[edge.target_file] = self.in_degree.get(edge.target_file, 0) + 1
+
+    def add_node(self, file_path: str) -> None:
+        """Ensure *file_path* is registered as a graph node."""
+        self.file_nodes.add(file_path)
+        self.adjacency_list.setdefault(file_path, [])
+        self.in_degree.setdefault(file_path, 0)
+
+    def successors(self, file_path: str) -> list[str]:
+        """Return direct successors (targets) of *file_path*."""
+        return list(self.adjacency_list.get(file_path, []))
+
+
+__all__: list[str] = [
+    "ClusterStrategy",
+    "DecompositionConfig",
+    "DecompositionNode",
+    "DecompositionResult",
+    "DependencyEdge",
+    "DependencyGraph",
+    "DependencyType",
+]
@@ -0,0 +1,391 @@
+"""Orchestration service for large-project hierarchical decomposition.
+
+:class:`DecompositionService` is the main entry point that combines
+clustering heuristics, dependency closure, and DAG ordering into a
+single cohesive API.  Decomposition decisions are optionally recorded
+in :class:`DecisionService` for auditability.
+
+Design decisions:
+
+- **DI**: ``DecisionService`` and ``Settings`` are injected via the
+  constructor (consistent with existing service patterns).
+- **Stateless**: All decomposition state is derived from inputs — the
+  service holds no mutable plan state.
+- **Fail-safe**: When heuristics produce zero clusters a single
+  fallback subplan is emitted with a warning.
+
+Based on Forgejo issue #205.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections import defaultdict
+from typing import TYPE_CHECKING
+
+from cleveragents.application.services.decomposition_clustering import (
+    ClusteringStrategy,
+    estimate_tokens_for_path,
+)
+from cleveragents.application.services.decomposition_graph import (
+    DependencyClosureComputer,
+)
+from cleveragents.application.services.decomposition_models import (
+    ClusterStrategy,
+    DecompositionConfig,
+    DecompositionNode,
+    DecompositionResult,
+    DependencyGraph,
+)
+from cleveragents.domain.models.core.decision import DecisionType
+
+if TYPE_CHECKING:
+    from cleveragents.application.services.decision_service import DecisionService
+    from cleveragents.config.settings import Settings
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+_COUNTER: int = 0
+
+
+def _next_node_id(prefix: str = "dn") -> str:
+    """Generate a deterministic, monotonically increasing node ID."""
+    global _COUNTER
+    _COUNTER += 1
+    return f"{prefix}-{_COUNTER:06d}"
+
+
+def _reset_counter() -> None:
+    """Reset the node ID counter (for testing)."""
+    global _COUNTER
+    _COUNTER = 0
+
+
+def _dominant_extension(files: list[str]) -> str:
+    """Return the most frequent file extension in *files*."""
+    counts: dict[str, int] = defaultdict(int)
+    for f in files:
+        ext = ""
+        dot_idx = f.rfind(".")
+        if dot_idx != -1:
+            ext = f[dot_idx:].lower()
+        counts[ext] += 1
+    if not counts:
+        return ""
+    return max(counts, key=counts.get)  # type: ignore[arg-type]
+
+
+def _common_prefix(files: list[str]) -> str:
+    """Return the longest common directory prefix of *files*."""
+    if not files:
+        return ""
+    parts_list = [f.replace("\\", "/").split("/") for f in sorted(files)]
+    prefix: list[str] = []
+    for segments in zip(*parts_list, strict=False):
+        if len(set(segments)) == 1:
+            prefix.append(segments[0])
+        else:
+            break
+    return "/".join(prefix)
+
+
+# ---------------------------------------------------------------------------
+# Service
+# ---------------------------------------------------------------------------
+
+
+class DecompositionService:
+    """Orchestrates hierarchical decomposition of large file sets.
+
+    Args:
+        decision_service: Optional service for recording decomposition
+            decisions.  When ``None`` decision recording is skipped.
+        settings: Application settings providing default config values.
+    """
+
+    def __init__(
+        self,
+        decision_service: DecisionService | None = None,
+        settings: Settings | None = None,
+    ) -> None:
+        self._decision_service = decision_service
+        self._settings = settings
+        self._closure_computer = DependencyClosureComputer()
+
+    @property
+    def decision_service(self) -> DecisionService | None:
+        """The injected decision service (may be ``None``)."""
+        return self._decision_service
+
+    # ------------------------------------------------------------------
+    # Main entry point
+    # ------------------------------------------------------------------
+
+    def decompose(
+        self,
+        files: list[str],
+        config: DecompositionConfig | None = None,
+    ) -> DecompositionResult:
+        """Decompose *files* into a bounded hierarchy of subplans.
+
+        When the number of files is at or below
+        ``config.min_files_per_subplan`` the hierarchy is skipped and a
+        single root node is returned.
+
+        Args:
+            files: Flat list of file paths to decompose.
+            config: Decomposition parameters.  Uses defaults when
+                ``None``.
+
+        Returns:
+            A :class:`DecompositionResult` with the full tree.
+        """
+        _reset_counter()
+        cfg = config or DecompositionConfig()
+        unique_files = sorted(set(files))
+
+        # Guard: skip decomposition when below threshold
+        if len(unique_files) <= cfg.min_files_per_subplan:
+            root = DecompositionNode(
+                node_id=_next_node_id(),
+                parent_id=None,
+                depth=0,
+                file_paths=unique_files,
+                language=_dominant_extension(unique_files),
+                directory_prefix=_common_prefix(unique_files),
+                estimated_tokens=sum(estimate_tokens_for_path(f) for f in unique_files),
+                strategy=ClusterStrategy.FALLBACK,
+            )
+            return DecompositionResult(
+                nodes=[root],
+                max_depth_reached=0,
+                total_files=len(unique_files),
+                metrics={"skipped": 1},
+            )
+
+        nodes: list[DecompositionNode] = []
+        max_depth = self._build_hierarchy(
+            files=unique_files,
+            config=cfg,
+            depth=0,
+            parent_id=None,
+            nodes=nodes,
+        )
+
+        return DecompositionResult(
+            nodes=nodes,
+            max_depth_reached=max_depth,
+            total_files=len(unique_files),
+            metrics={
+                "total_nodes": len(nodes),
+                "leaf_nodes": sum(1 for n in nodes if n.is_leaf),
+                "max_depth": max_depth,
+            },
+        )
+
+    # ------------------------------------------------------------------
+    # Recursive hierarchy builder
+    # ------------------------------------------------------------------
+
+    def _build_hierarchy(
+        self,
+        files: list[str],
+        config: DecompositionConfig,
+        depth: int,
+        parent_id: str | None,
+        nodes: list[DecompositionNode],
+    ) -> int:
+        """Recursively partition *files* into a tree.
+
+        Returns the maximum depth reached during recursion.
+        """
+        estimated = sum(estimate_tokens_for_path(f) for f in files)
+
+        # Leaf conditions
+        is_leaf = depth >= config.max_depth or (
+            len(files) <= config.max_files_per_subplan
+            and estimated <= config.max_tokens_per_subplan
+        )
+
+        node_id = _next_node_id()
+        if is_leaf or len(files) <= config.min_files_per_subplan:
+            node = DecompositionNode(
+                node_id=node_id,
+                parent_id=parent_id,
+                depth=depth,
+                file_paths=files,
+                language=_dominant_extension(files),
+                directory_prefix=_common_prefix(files),
+                estimated_tokens=estimated,
+                strategy=ClusterStrategy.DIRECTORY,
+            )
+            nodes.append(node)
+            return depth
+
+        # Cluster using directory strategy first, fall back to language
+        clusters = ClusteringStrategy.cluster_by_directory(
+            files, config.max_files_per_subplan
+        )
+        strategy = ClusterStrategy.DIRECTORY
+
+        if len(clusters) <= 1:
+            clusters = ClusteringStrategy.cluster_by_language(
+                files, config.max_files_per_subplan
+            )
+            strategy = ClusterStrategy.LANGUAGE
+
+        if len(clusters) <= 1:
+            clusters = ClusteringStrategy.cluster_by_size(
+                files, config.max_tokens_per_subplan
+            )
+            strategy = ClusterStrategy.SIZE
+
+        # Fallback: single cluster with warning
+        if not clusters:
+            logger.warning(
+                "decomposition_fallback",
+                extra={"file_count": len(files), "depth": depth},
+            )
+            node = DecompositionNode(
+                node_id=node_id,
+                parent_id=parent_id,
+                depth=depth,
+                file_paths=files,
+                language=_dominant_extension(files),
+                directory_prefix=_common_prefix(files),
+                estimated_tokens=estimated,
+                strategy=ClusterStrategy.FALLBACK,
+            )
+            nodes.append(node)
+            return depth
+
+        clusters = ClusteringStrategy.deterministic_sort(
+            clusters, config.language_priority
+        )
+
+        children_ids: list[str] = []
+        max_child_depth = depth
+
+        for cluster in clusters:
+            child_depth = self._build_hierarchy(
+                files=cluster,
+                config=config,
+                depth=depth + 1,
+                parent_id=node_id,
+                nodes=nodes,
+            )
+            # The last added node is the child root
+            children_ids.append(nodes[-1].node_id)
+            max_child_depth = max(max_child_depth, child_depth)
+
+        internal_node = DecompositionNode(
+            node_id=node_id,
+            parent_id=parent_id,
+            depth=depth,
+            file_paths=files,
+            language=_dominant_extension(files),
+            directory_prefix=_common_prefix(files),
+            estimated_tokens=estimated,
+            strategy=strategy,
+            children_ids=children_ids,
+        )
+        nodes.append(internal_node)
+        return max_child_depth
+
+    # ------------------------------------------------------------------
+    # Decision recording
+    # ------------------------------------------------------------------
+
+    def record_decisions(
+        self,
+        plan_id: str,
+        result: DecompositionResult,
+    ) -> None:
+        """Persist decomposition decisions in :class:`DecisionService`.
+
+        Records two types of decisions:
+
+        1. ``strategy_choice`` — the overall decomposition strategy.
+        2. ``subplan_spawn`` — one per leaf node.
+
+        Args:
+            plan_id: ULID of the parent plan.
+            result: The decomposition result to record.
+        """
+        if self._decision_service is None:
+            return
+
+        # Strategy decision
+        self._decision_service.record_decision(
+            plan_id=plan_id,
+            decision_type=DecisionType.STRATEGY_CHOICE,
+            question="How should the project be decomposed?",
+            chosen_option="hierarchical_decomposition",
+            rationale=(
+                f"Decomposed {result.total_files} files into "
+                f"{len(result.nodes)} nodes across "
+                f"{result.max_depth_reached} levels"
+            ),
+        )
+
+        # Subplan spawn per leaf
+        for node in result.nodes:
+            if node.is_leaf:
+                self._decision_service.record_decision(
+                    plan_id=plan_id,
+                    decision_type=DecisionType.SUBPLAN_SPAWN,
+                    question=f"Spawn subplan for {node.directory_prefix}?",
+                    chosen_option=f"spawn:{node.node_id}",
+                    rationale=(
+                        f"{len(node.file_paths)} files, "
+                        f"~{node.estimated_tokens} tokens, "
+                        f"lang={node.language}"
+                    ),
+                )
+
+    # ------------------------------------------------------------------
+    # Dependency ordering
+    # ------------------------------------------------------------------
+
+    def compute_dependency_order(
+        self,
+        graph: DependencyGraph,
+    ) -> list[str]:
+        """Return a DAG execution ordering for *graph*.
+
+        Args:
+            graph: The dependency graph.
+
+        Returns:
+            Topologically sorted list of file paths.
+        """
+        return self._closure_computer.topological_sort(graph)
+
+    def compute_closure(
+        self,
+        graph: DependencyGraph,
+        root_files: list[str],
+        cutoff: int = 0,
+    ) -> frozenset[str]:
+        """Compute bounded transitive closure.
+
+        Delegates to :class:`DependencyClosureComputer`.
+
+        Args:
+            graph: The dependency graph.
+            root_files: Starting file paths.
+            cutoff: Maximum closure size (``0`` = unlimited).
+
+        Returns:
+            Frozenset of reachable files.
+        """
+        return self._closure_computer.compute_closure(graph, root_files, cutoff)
+
+
+__all__: list[str] = [
+    "DecompositionService",
+]
@@ -378,6 +378,32 @@ class Settings(BaseSettings):
        description="When True, secrets are shown in CLI output and logs.",
    )

+    # Planner decomposition (M6 - large-project decomposition)
+    planner_max_depth: int = Field(
+        default=4,
+        ge=1,
+        validation_alias=AliasChoices("CLEVERAGENTS_PLANNER_MAX_DEPTH"),
+        description="Maximum decomposition depth for large projects.",
+    )
+    planner_max_files_per_subplan: int = Field(
+        default=500,
+        ge=1,
+        validation_alias=AliasChoices("CLEVERAGENTS_PLANNER_MAX_FILES_PER_SUBPLAN"),
+        description="Maximum files per decomposition subplan.",
+    )
+    planner_max_tokens_per_subplan: int = Field(
+        default=100_000,
+        ge=1,
+        validation_alias=AliasChoices("CLEVERAGENTS_PLANNER_MAX_TOKENS_PER_SUBPLAN"),
+        description="Maximum estimated tokens per decomposition subplan.",
+    )
+    planner_min_files_per_subplan: int = Field(
+        default=10,
+        ge=1,
+        validation_alias=AliasChoices("CLEVERAGENTS_PLANNER_MIN_FILES_PER_SUBPLAN"),
+        description="Minimum files below which decomposition is skipped.",
+    )
+
    # Mock providers flag (M4 - provider fixes)
    mock_providers: bool = Field(
        default=False,
@@ -530,6 +530,36 @@ validate_container_available  # noqa: B018, F821
 resolve_and_validate  # noqa: B018, F821
 _env_resolver  # noqa: B018, F821

+# Decomposition service — public API (M6, issue #205)
+DecompositionConfig  # noqa: B018, F821
+DecompositionNode  # noqa: B018, F821
+DecompositionResult  # noqa: B018, F821
+DecompositionService  # noqa: B018, F821
+DependencyEdge  # noqa: B018, F821
+DependencyGraph  # noqa: B018, F821
+DependencyType  # noqa: B018, F821
+DependencyClosureComputer  # noqa: B018, F821
+ClusterStrategy  # noqa: B018, F821
+ClusteringStrategy  # noqa: B018, F821
+decomposition_service  # noqa: B018, F821
+compute_closure  # noqa: B018, F821
+compute_dependency_order  # noqa: B018, F821
+record_decisions  # noqa: B018, F821
+cluster_by_directory  # noqa: B018, F821
+cluster_by_language  # noqa: B018, F821
+cluster_by_size  # noqa: B018, F821
+deterministic_sort  # noqa: B018, F821
+topological_sort  # noqa: B018, F821
+detect_cycles  # noqa: B018, F821
+clear_cache  # noqa: B018, F821
+estimate_tokens_for_path  # noqa: B018, F821
+add_edge  # noqa: B018, F821
+add_node  # noqa: B018, F821
+planner_max_depth  # noqa: B018, F821
+planner_max_files_per_subplan  # noqa: B018, F821
+planner_max_tokens_per_subplan  # noqa: B018, F821
+planner_min_files_per_subplan  # noqa: B018, F821
+
 # ACP facade wiring — public API error codes and mapping function (#501)
 NOT_FOUND  # noqa: B018, F821
 VALIDATION_ERROR  # noqa: B018, F821