feat(acms): implement DepthReductionCompressor for skeleton compression

Add a production skeleton compressor that re-renders inherited fragments to overview depths via the UKO detail-level map chain, fits the result within the configured skeleton budget, and wires the pipeline default to the new compressor.

Address prior review feedback by extracting the render visitors into a dedicated module, restoring projected metadata to native runtime types, constraining builtin component resolution with an allowlist, and keeping child-context inheritance compatible with CRP context fragments for the Robot integration path.

Reproduced the Forgejo lint job in a clean python:3.13-slim container with the CI commands All checks passed! and 1740 files already formatted; both passed, so the earlier lint failure appears to have been transient runner behavior rather than a source-level defect.

ISSUES CLOSED: #919

CHANGELOG.md

@@ -255,6 +255,11 @@
   non-existent ``container.resolve()``, matching corrected production code.
   Activated regression-guard BDD scenarios for ``plan tree``, ``plan explain``,
   and ``plan correct``.  (#647)
+- Added the production ACMS skeleton compression stage via
+  `DepthReductionCompressor`. The pipeline now re-renders inherited parent
+  fragments to overview depths 0-1 using the UKO detail-level map chain,
+  exposes the compressor as the configured builtin, and covers the behavior
+  with BDD scenarios for compressor output and default pipeline wiring. (#919)
 - Added TDD bug-capture tests for bug #1076 — `use_action()` does not
   propagate `automation_profile` to Plan. Three Behave BDD scenarios
   (`@tdd_bug @tdd_bug_1076 @tdd_expected_fail`) verify the full precedence

features/acms_pipeline_orchestrator.feature

@@ -117,6 +117,12 @@ Feature: ACMS Pipeline Orchestrator and Phase 1 Components
     Then the executor should return fragments from the strategy
     And the tracking strategy should have been invoked
 
+  @orchestrator @pipeline
+  Scenario: ContextAssemblyPipeline uses DepthReductionCompressor by default
+    Given the pipeline orchestrator modules are available
+    And a ContextAssemblyPipeline with default components
+    Then the orchestrator pipeline should use DepthReductionCompressor
+
   @orchestrator @executor
   Scenario: ParallelStrategyExecutor skips circuit-broken strategies
     Given the pipeline orchestrator modules are available

features/acms_pipeline_phase3.feature

@@ -109,6 +109,164 @@ Feature: ACMS Pipeline Phase 3 — Context Finalization and Advanced Strategies
     When I generate a preamble with ProvenancePreambleGenerator
     Then the preamble should contain "UKO Nodes: 3"
 
+  # ===========================================================================
+  # DepthReductionCompressor
+  # ===========================================================================
+
+  @compressor
+  Scenario: DepthReductionCompressor reduces Python fragments to inherited overview levels
+    Given the following phase3 skeleton fragments:
+      | uko_node                  | content                     | score | tokens | depth | domain  | resource_type |
+      | project://src/app/main.py | def main(): return runner() | 0.9   | 120    | 7     |         | python-source |
+      | project://src/app/util.py | class Util: pass            | 0.6   | 90     | 5     | uko-py: | python-source |
+    When I compress with DepthReductionCompressor and budget 40
+    Then all compressed fragments should have detail depth at most 1
+    And compressed fragments should fit within budget 40
+    And the compressed fragments should include skeleton level "MODULE_GRAPH"
+    And a compressed fragment should contain "[MODULE_GRAPH]: symbols=main"
+
+  @compressor
+  Scenario: DepthReductionCompressor uses document detail levels for markdown fragments
+    Given the following phase3 skeleton fragments:
+      | uko_node                       | content                         | score | tokens | depth | domain   | resource_type |
+      | project://docs/specification.md | # Spec\n## Overview\n## Details | 0.8   | 100    | 8     |          | markdown      |
+    When I compress with DepthReductionCompressor and budget 20
+    Then all compressed fragments should have detail depth at most 1
+    And compressed fragments should fit within budget 20
+    And the compressed fragments should include skeleton level "TABLE_OF_CONTENTS_L1"
+    And a compressed fragment should contain "Overview; Details"
+
+  @compressor
+  Scenario: DepthReductionCompressor returns no fragments for zero budget
+    Given the following phase3 skeleton fragments:
+      | uko_node               | content           | score | tokens | depth | domain  | resource_type |
+      | project://src/app/a.py | def alpha(): pass | 0.7   | 80     | 6     | uko-py: | python-source |
+    When I compress with DepthReductionCompressor and budget 0
+    Then 0 fragments should remain after compression
+
+  @compressor
+  Scenario: DepthReductionCompressor keeps the highest-relevance fragment when budget is tight
+    Given the following phase3 skeleton fragments:
+      | uko_node                  | content                | score | tokens | depth | domain  | resource_type |
+      | project://src/app/main.py | def main(): return run | 0.9   | 120    | 7     | uko-py: | python-source |
+      | project://src/app/aux.py  | def aux(): return slow | 0.3   | 120    | 7     | uko-py: | python-source |
+    When I compress with DepthReductionCompressor and budget 9
+    Then 1 fragments should remain after compression
+    And compressed fragments should fit within budget 9
+    And a compressed fragment should contain "main.py"
+
+  @compressor
+  Scenario: DepthReductionCompressor renders data fragments with CSV-like summary
+    Given the following phase3 skeleton fragments:
+      | uko_node                  | content                              | score | tokens | depth | domain    | resource_type |
+      | project://db/schema.sql   | users, orders, products, shipments   | 0.7   | 80     | 5     | uko-data: | sql           |
+    When I compress with DepthReductionCompressor and budget 30
+    Then all compressed fragments should have detail depth at most 1
+    And compressed fragments should fit within budget 30
+    And a compressed fragment should contain "schema.sql"
+
+  @compressor
+  Scenario: DepthReductionCompressor renders infrastructure fragments
+    Given the following phase3 skeleton fragments:
+      | uko_node                     | content                           | score | tokens | depth | domain     | resource_type |
+      | project://deploy/main.tf     | resource "aws_instance" "web" {}  | 0.8   | 90     | 6     | uko-infra: | terraform     |
+    When I compress with DepthReductionCompressor and budget 30
+    Then all compressed fragments should have detail depth at most 1
+    And compressed fragments should fit within budget 30
+    And a compressed fragment should contain "main.tf"
+
+  @compressor
+  Scenario: DepthReductionCompressor renders generic fragments at overview depth
+    Given the following phase3 skeleton fragments:
+      | uko_node            | content           | score | tokens | depth | domain | resource_type |
+      | uko:some/resource   | generic resource  | 0.6   | 80     | 4     | uko:   | unknown       |
+    When I compress with DepthReductionCompressor and budget 30
+    Then all compressed fragments should have detail depth at most 1
+    And compressed fragments should fit within budget 30
+
+  @compressor
+  Scenario: DepthReductionCompressor preserves fragment already at target depth
+    Given the following phase3 skeleton fragments:
+      | uko_node                  | content     | score | tokens | depth | domain  | resource_type |
+      | project://src/app/main.py | brief view  | 0.9   | 10     | 0     | uko-py: | python-source |
+    When I compress with DepthReductionCompressor and budget 100
+    Then 1 fragments should remain after compression
+    And a compressed fragment should contain "brief view"
+
+  @compressor
+  Scenario: DepthReductionCompressor clips content when all renderings exceed budget
+    Given the following phase3 skeleton fragments:
+      | uko_node                  | content                                                 | score | tokens | depth | domain  | resource_type |
+      | project://src/app/main.py | def main(): return very_long_function_name_placeholder() | 0.9   | 200    | 7     | uko-py: | python-source |
+    When I compress with DepthReductionCompressor and budget 4
+    Then 1 fragments should remain after compression
+    And compressed fragments should fit within budget 4
+
+  @compressor
+  Scenario: DepthReductionCompressor resolves detail map from provenance URI prefix
+    Given the following phase3 skeleton fragments:
+      | uko_node                | content           | score | tokens | depth | domain | resource_type |
+      | uko-doc:my/design-doc   | # Design Overview | 0.7   | 80     | 5     |        |               |
+    When I compress with DepthReductionCompressor and budget 30
+    Then all compressed fragments should have detail depth at most 1
+
+  @compressor
+  Scenario: DepthReductionCompressor resolves detail map from resource_type
+    Given the following phase3 skeleton fragments:
+      | uko_node                         | content                              | score | tokens | depth | domain | resource_type |
+      | project://data/schema            | CREATE TABLE users (id INTEGER)      | 0.6   | 80     | 4     |        | database      |
+    When I compress with DepthReductionCompressor and budget 30
+    Then all compressed fragments should have detail depth at most 1
+
+  @compressor
+  Scenario: DepthReductionCompressor handles minimal content fragments
+    Given the following phase3 skeleton fragments:
+      | uko_node                   | content | score | tokens | depth | domain  | resource_type |
+      | project://src/app/empty.py | x       | 0.5   | 1      | 3     | uko-py: | python-source |
+    When I compress with DepthReductionCompressor and budget 10
+    Then compressed fragments should fit within budget 10
+
+  @compressor
+  Scenario: DepthReductionCompressor resolves TypeScript and Java from file extension
+    Given the following phase3 skeleton fragments:
+      | uko_node                 | content                      | score | tokens | depth | domain | resource_type |
+      | project://src/app.ts     | export function main() {}    | 0.8   | 80     | 6     |        |               |
+      | project://src/Main.java  | public class Main {}         | 0.7   | 80     | 6     |        |               |
+    When I compress with DepthReductionCompressor and budget 60
+    Then compressed fragments should fit within budget 60
+
+  @compressor
+  Scenario: DepthReductionCompressor resolves Rust from file extension
+    Given the following phase3 skeleton fragments:
+      | uko_node                   | content               | score | tokens | depth | domain | resource_type |
+      | project://src/main.rs      | fn main() {}          | 0.8   | 80     | 6     |        |               |
+    When I compress with DepthReductionCompressor and budget 30
+    Then compressed fragments should fit within budget 30
+
+  @compressor
+  Scenario: DepthReductionCompressor resolves infra from YAML extension
+    Given the following phase3 skeleton fragments:
+      | uko_node                   | content               | score | tokens | depth | domain | resource_type |
+      | project://k8s/deploy.yaml  | apiVersion: apps/v1   | 0.7   | 80     | 5     |        |               |
+    When I compress with DepthReductionCompressor and budget 30
+    Then compressed fragments should fit within budget 30
+
+  @compressor
+  Scenario: DepthReductionCompressor detects doc resource_type
+    Given the following phase3 skeleton fragments:
+      | uko_node                        | content              | score | tokens | depth | domain | resource_type |
+      | project://guides/getting-started | Welcome to the docs | 0.6   | 80     | 5     |        | markdown      |
+    When I compress with DepthReductionCompressor and budget 30
+    Then compressed fragments should fit within budget 30
+
+  @compressor
+  Scenario: DepthReductionCompressor detects infra resource_type
+    Given the following phase3 skeleton fragments:
+      | uko_node                         | content                         | score | tokens | depth | domain | resource_type        |
+      | project://infra/cluster          | kubernetes deployment manifest  | 0.6   | 80     | 5     |        | kubernetes           |
+    When I compress with DepthReductionCompressor and budget 30
+    Then compressed fragments should fit within budget 30
+
   # ===========================================================================
   # ArceStrategy
   # ===========================================================================

features/component_resolver.feature

@@ -263,6 +263,18 @@ Feature: Pluggable Component Resolution with Scope Chain
     When I attempt to import a component from "no_colon_here"
     Then a ComponentRegistrationError should be raised
 
+  @scope_chain @security
+  Scenario: Import builtin component path resolves through application services
+    Given a fresh ComponentResolver instance
+    When I import a component from "builtin:DepthReductionCompressor"
+    Then the imported component class should be "DepthReductionCompressor"
+
+  @scope_chain @security
+  Scenario: Import builtin service outside the allowlist is rejected
+    Given a fresh ComponentResolver instance
+    When I attempt to import a component from "builtin:ConfigService"
+    Then a ComponentRegistrationError should be raised
+
   # ---------------------------------------------------------------------------
   # Edge cases: unknown extension names and failed imports
   # ---------------------------------------------------------------------------

features/depth_breadth_projection.feature

@@ -205,6 +205,30 @@ Feature: Depth/Breadth Projection System and Skeleton Context Propagation
     When I compute child context with focus "class://Child" and budget 1000
     Then the child result skeleton_fragments should have 2 entries
 
+  @inheritance @skeleton
+  Scenario: PlanContextInheritance uses DepthReductionCompressor by default
+    Given the depth/breadth projection modules are available
+    And a parent assembled context with the following inheritance fragments:
+      | uko_node                  | content                     | tokens | depth | domain  | resource_uri                  |
+      | project://src/app/main.py | def main(): return runner() | 120    | 7     | uko-py: | project://src/app/main.py     |
+    And a PlanContextInheritance service with default config
+    When I compute child context with focus "class://Child" and budget 1000
+    Then the child result skeleton_fragments should have 1 entries
+    And all child skeleton fragments should have detail depth at most 1
+    And a child skeleton fragment should contain "[MODULE_GRAPH]: symbols=main"
+
+  @inheritance @skeleton
+  Scenario: PlanContextInheritance prioritises fragments near the child focus
+    Given the depth/breadth projection modules are available
+    And a parent assembled context with the following inheritance fragments:
+      | uko_node                  | content                | tokens | depth | domain | resource_uri                  | score |
+      | project://src/app/main.py | def main(): return run | 120    | 7     |        | project://src/app/main.py     | 0.2   |
+      | project://lib/aux.py      | def aux(): return slow | 120    | 7     |        | project://lib/aux.py          | 0.9   |
+    And a PlanContextInheritance service with default config and min_skeleton_tokens 1
+    When I compute child context with focus "project://src/app/main.py" and budget 60
+    Then the child result skeleton_fragments should have 2 entries
+    And the first child skeleton fragment should contain "main.py"
+
   @inheritance @skeleton
   Scenario: PlanContextInheritance skips skeleton when budget too small
     Given the depth/breadth projection modules are available

features/steps/acms_pipeline_orchestrator_steps.py

@@ -29,6 +29,9 @@ from cleveragents.application.services.acms_service import (
     StrategyCapabilities,
     TieredStrategy,
 )
+from cleveragents.application.services.acms_skeleton_compressor import (
+    DepthReductionCompressor,
+)
 from cleveragents.domain.models.core.context_fragment import (
     ContextBudget,
     ContextFragment,
@@ -246,6 +249,15 @@ def step_default_pipeline(context: Context) -> None:
     context.orch_pipeline = ContextAssemblyPipeline()
 
 
+@then("the orchestrator pipeline should use DepthReductionCompressor")
+def step_pipeline_uses_depth_reduction(context: Context) -> None:
+    compressor = getattr(context.orch_pipeline, "_skeleton_compressor", None)
+    assert isinstance(compressor, DepthReductionCompressor), (
+        "Expected ContextAssemblyPipeline to default to DepthReductionCompressor, "
+        f"got {type(compressor).__name__ if compressor is not None else None}"
+    )
+
+
 @given("the following orchestrator test fragments:")
 def step_orch_fragments_table(context: Context) -> None:
     context.orch_fragments = []

features/steps/acms_pipeline_phase3_steps.py

@@ -21,6 +21,9 @@ from cleveragents.application.services.acms_phase3 import (
     RelevanceCoherenceOrderer,
 )
 from cleveragents.application.services.acms_service import ACMSPipeline
+from cleveragents.application.services.acms_skeleton_compressor import (
+    DepthReductionCompressor,
+)
 from cleveragents.domain.models.core.context_fragment import (
     ContextBudget,
     ContextFragment,
@@ -55,7 +58,7 @@ def step_given_phase3_fragments(context: Context) -> None:
         frags.append(
             _make_phase3_fragment(
                 uko_node=row["uko_node"],
-                content=row["content"],
+                content=row["content"].replace("\\n", "\n"),
                 relevance_score=float(row["score"]),
                 token_count=int(row["tokens"]),
                 detail_depth=int(row["depth"]),
@@ -71,7 +74,7 @@ def step_given_phase3_fragments_with_strategy(context: Context) -> None:
         frags.append(
             _make_phase3_fragment(
                 uko_node=row["uko_node"],
-                content=row["content"],
+                content=row["content"].replace("\\n", "\n"),
                 relevance_score=float(row["score"]),
                 token_count=int(row["tokens"]),
                 detail_depth=int(row["depth"]),
@@ -81,6 +84,30 @@ def step_given_phase3_fragments_with_strategy(context: Context) -> None:
     context.phase3_fragments = frags
 
 
+@given("the following phase3 skeleton fragments:")
+def step_given_phase3_skeleton_fragments(context: Context) -> None:
+    frags: list[ContextFragment] = []
+    for row in context.table:
+        metadata = {}
+        if row["domain"].strip():
+            metadata["detail_level_domain"] = row["domain"]
+        frags.append(
+            _make_phase3_fragment(
+                uko_node=row["uko_node"],
+                content=row["content"].replace("\\n", "\n"),
+                relevance_score=float(row["score"]),
+                token_count=int(row["tokens"]),
+                detail_depth=int(row["depth"]),
+                provenance=FragmentProvenance(
+                    resource_uri=row["uko_node"],
+                    resource_type=row["resource_type"],
+                ),
+                metadata=metadata,
+            )
+        )
+    context.phase3_skeleton_fragments = tuple(frags)
+
+
 @given("an empty phase3 fragment list")
 def step_given_empty_phase3(context: Context) -> None:
     context.phase3_fragments = []
@@ -98,7 +125,7 @@ def step_given_strategy3_fragments(context: Context) -> None:
         frags.append(
             _make_phase3_fragment(
                 uko_node=row["uko_node"],
-                content=row["content"],
+                content=row["content"].replace("\\n", "\n"),
                 relevance_score=float(row["score"]),
                 token_count=int(row["tokens"]),
                 detail_depth=int(row["depth"]),
@@ -114,7 +141,7 @@ def step_given_strategy3_fragments_with_tiers(context: Context) -> None:
         frags.append(
             _make_phase3_fragment(
                 uko_node=row["uko_node"],
-                content=row["content"],
+                content=row["content"].replace("\\n", "\n"),
                 relevance_score=float(row["score"]),
                 token_count=int(row["tokens"]),
                 detail_depth=int(row["depth"]),
@@ -227,6 +254,60 @@ def step_preamble_is_none(context: Context) -> None:
     assert context.phase3_preamble is None
 
 
+# ---------------------------------------------------------------------------
+# DepthReductionCompressor — When / Then
+# ---------------------------------------------------------------------------
+
+
+@when("I compress with DepthReductionCompressor and budget {budget:d}")
+def step_compress_depth_reduction(context: Context, budget: int) -> None:
+    compressor = DepthReductionCompressor()
+    source = getattr(context, "phase3_skeleton_fragments", ())
+    context.phase3_compressed = compressor.compress(source, budget)
+
+
+@then("all compressed fragments should have detail depth at most {max_depth:d}")
+def step_compressed_depth_max(context: Context, max_depth: int) -> None:
+    assert all(f.detail_depth <= max_depth for f in context.phase3_compressed), (
+        f"Expected all compressed fragments to have depth <= {max_depth}, "
+        f"got {[f.detail_depth for f in context.phase3_compressed]}"
+    )
+
+
+@then("compressed fragments should fit within budget {budget:d}")
+def step_compressed_budget(context: Context, budget: int) -> None:
+    used = sum(fragment.token_count for fragment in context.phase3_compressed)
+    assert used <= budget, (
+        f"Compressed fragments used {used} tokens for budget {budget}"
+    )
+
+
+@then('the compressed fragments should include skeleton level "{level}"')
+def step_compressed_level(context: Context, level: str) -> None:
+    levels = {
+        fragment.metadata.get("skeleton_target_level")
+        for fragment in context.phase3_compressed
+    }
+    assert level in levels, (
+        f"Expected level {level!r}, got {sorted(level for level in levels if level is not None)}"
+    )
+
+
+@then("{count:d} fragments should remain after compression")
+def step_compressed_count(context: Context, count: int) -> None:
+    assert len(context.phase3_compressed) == count, (
+        f"Expected {count} compressed fragments, got {len(context.phase3_compressed)}"
+    )
+
+
+@then('a compressed fragment should contain "{snippet}"')
+def step_compressed_content_contains(context: Context, snippet: str) -> None:
+    contents = [fragment.content for fragment in context.phase3_compressed]
+    assert any(snippet in content for content in contents), (
+        f"Expected a compressed fragment to contain {snippet!r}, got {contents!r}"
+    )
+
+
 # ---------------------------------------------------------------------------
 # ArceStrategy — When / Then
 # ---------------------------------------------------------------------------

features/steps/component_resolver_steps.py

@@ -274,11 +274,17 @@ def step_register_none_plan(context: Any) -> None:
 def step_import_disallowed(context: Any, module_path: str) -> None:
     resolver: ComponentResolver = context.resolver
     try:
-        resolver._import_component(module_path)
+        context.imported_component = resolver._import_component(module_path)
     except (ComponentRegistrationError, ImportError, AttributeError) as exc:
         context.error = exc
 
 
+@when('I import a component from "{module_path}"')
+def step_import_component(context: Any, module_path: str) -> None:
+    resolver: ComponentResolver = context.resolver
+    context.imported_component = resolver._import_component(module_path)
+
+
 # ---------------------------------------------------------------------------
 # Then steps
 # ---------------------------------------------------------------------------
@@ -325,6 +331,11 @@ def step_assert_registration_error(context: Any) -> None:
     )
 
 
+@then('the imported component class should be "{class_name}"')
+def step_assert_imported_component_class(context: Any, class_name: str) -> None:
+    assert type(context.imported_component).__name__ == class_name
+
+
 @then("has_global should return True for the test protocol")
 def step_assert_has_global_true(context: Any) -> None:
     resolver: ComponentResolver = context.resolver

features/steps/depth_breadth_projection_steps.py

@@ -22,9 +22,9 @@ from cleveragents.application.services.depth_breadth_projection import (
     database_detail_map,
     docs_detail_map,
 )
-from cleveragents.domain.models.acms.crp import (
+from cleveragents.domain.models.acms.crp import ContextRequest
+from cleveragents.domain.models.core.context_fragment import (
     ContextFragment,
-    ContextRequest,
     FragmentProvenance,
 )
 
@@ -138,11 +138,42 @@ def step_parent_context(context: Context, n: int, depth: int) -> None:
     context.parent_context = _FakeAssembledContext(frags, context_hash="parenthash123")
 
 
+@given("a parent assembled context with the following inheritance fragments:")
+def step_parent_context_table(context: Context) -> None:
+    frags: list[ContextFragment] = []
+    for row in context.table:
+        metadata: dict[str, str] = {}
+        if row["domain"].strip():
+            metadata["detail_level_domain"] = row["domain"]
+        relevance_score = float(row["score"]) if "score" in row.headings else 0.8
+        frags.append(
+            ContextFragment(
+                uko_node=row["uko_node"],
+                content=row["content"],
+                token_count=int(row["tokens"]),
+                detail_depth=int(row["depth"]),
+                relevance_score=relevance_score,
+                provenance=FragmentProvenance(resource_uri=row["resource_uri"]),
+                metadata=metadata,
+            )
+        )
+    context.parent_context = _FakeAssembledContext(frags, context_hash="parenthash123")
+
+
 @given("a PlanContextInheritance service with default config")
 def step_inheritance_default(context: Context) -> None:
     context.inheritance_service = PlanContextInheritance()
 
 
+@given(
+    "a PlanContextInheritance service with default config and min_skeleton_tokens {min_tok:d}"
+)
+def step_inheritance_default_min_tokens(context: Context, min_tok: int) -> None:
+    context.inheritance_service = PlanContextInheritance(
+        config=InheritanceConfig(min_skeleton_tokens=min_tok)
+    )
+
+
 @given("a mock skeleton compressor that returns {n:d} fragments")
 def step_mock_compressor(context: Context, n: int) -> None:
     context.mock_compressor = _MockSkeletonCompressor(n_fragments=n)
@@ -613,6 +644,34 @@ def step_child_skeleton_count(context: Context, count: int) -> None:
     )
 
 
+@then("all child skeleton fragments should have detail depth at most {max_depth:d}")
+def step_child_skeleton_depth_max(context: Context, max_depth: int) -> None:
+    depths = [
+        fragment.detail_depth for fragment in context.child_result.skeleton_fragments
+    ]
+    assert all(depth <= max_depth for depth in depths), (
+        f"Expected child skeleton fragment depths <= {max_depth}, got {depths}"
+    )
+
+
+@then('a child skeleton fragment should contain "{snippet}"')
+def step_child_skeleton_contains(context: Context, snippet: str) -> None:
+    contents = [
+        fragment.content for fragment in context.child_result.skeleton_fragments
+    ]
+    assert any(snippet in content for content in contents), (
+        f"Expected child skeleton fragment to contain {snippet!r}, got {contents!r}"
+    )
+
+
+@then('the first child skeleton fragment should contain "{snippet}"')
+def step_first_child_skeleton_contains(context: Context, snippet: str) -> None:
+    first = context.child_result.skeleton_fragments[0].content
+    assert snippet in first, (
+        f"Expected first child skeleton fragment to contain {snippet!r}, got {first!r}"
+    )
+
+
 @then("the inheritance config skeleton_ratio should be {expected:g}")
 def step_config_ratio(context: Context, expected: float) -> None:
     assert context.inheritance_config.skeleton_ratio == expected

src/cleveragents/application/services/__init__.py

@@ -45,6 +45,9 @@ if TYPE_CHECKING:
     from cleveragents.application.services.acms_phase3 import (
         RelevanceCoherenceOrderer as RelevanceCoherenceOrderer,
     )
+    from cleveragents.application.services.acms_skeleton_compressor import (
+        DepthReductionCompressor as DepthReductionCompressor,
+    )
     from cleveragents.application.services.audit_event_subscriber import (
         AuditEventSubscriber as AuditEventSubscriber,
     )
@@ -402,6 +405,10 @@ _LAZY_IMPORTS: dict[str, tuple[str, str]] = {
     "MaxDepthResolver": ("acms_phase2", "MaxDepthResolver"),
     "ScorerWeights": ("acms_phase2", "ScorerWeights"),
     "WeightedCompositeScorer": ("acms_phase2", "WeightedCompositeScorer"),
+    "DepthReductionCompressor": (
+        "acms_skeleton_compressor",
+        "DepthReductionCompressor",
+    ),
     "ProvenancePreambleGenerator": ("acms_phase3", "ProvenancePreambleGenerator"),
     "RelevanceCoherenceOrderer": ("acms_phase3", "RelevanceCoherenceOrderer"),
     "AuditEventSubscriber": ("audit_event_subscriber", "AuditEventSubscriber"),

src/cleveragents/application/services/acms_pipeline.py

@@ -53,7 +53,6 @@ from cleveragents.application.services.acms_service import (
     DefaultOrderer,
     DefaultPreambleGenerator,
     DefaultScorer,
-    DefaultSkeletonCompressor,
     DetailDepthResolver,
     FragmentDeduplicator,
     FragmentOrderer,
@@ -63,6 +62,9 @@ from cleveragents.application.services.acms_service import (
     StrategyExecutor,
     StrategySelector,
 )
+from cleveragents.application.services.acms_skeleton_compressor import (
+    resolve_configured_skeleton_compressor,
+)
 from cleveragents.domain.models.acms.crp import (
     ContextRequest,
 )
@@ -538,6 +540,10 @@ class ContextAssemblyPipeline(ACMSPipeline):
             max_workers=executor_max_workers,
         )
 
+        effective_skeleton_compressor = (
+            skeleton_compressor or resolve_configured_skeleton_compressor()
+        )
+
         super().__init__(
             default_strategy,
             settings=settings,
@@ -551,7 +557,7 @@ class ContextAssemblyPipeline(ACMSPipeline):
             packer=packer or GreedyKnapsackPacker(),
             orderer=orderer or DefaultOrderer(),
             preamble_generator=preamble_generator or DefaultPreambleGenerator(),
-            skeleton_compressor=skeleton_compressor or DefaultSkeletonCompressor(),
+            skeleton_compressor=effective_skeleton_compressor,
         )
         self._last_timings: StageTimings | None = None
         self._pipeline_logger = logger.bind(service="context_assembly_pipeline")

src/cleveragents/application/services/acms_skeleton_compressor.py

@@ -0,0 +1,376 @@
+"""Production skeleton compression for ACMS child-context inheritance."""
+
+from __future__ import annotations
+
+import logging
+import math
+from pathlib import PurePosixPath
+from typing import TYPE_CHECKING, Final, cast
+
+from cleveragents.acms import (
+    CODE_DETAIL_LEVEL_MAP,
+    FUNC_DETAIL_LEVEL_MAP,
+    JAVA_DETAIL_LEVELS,
+    OO_DETAIL_LEVEL_MAP,
+    PROC_DETAIL_LEVEL_MAP,
+    PYTHON_DETAIL_LEVELS,
+    RUST_DETAIL_LEVELS,
+    TYPESCRIPT_DETAIL_LEVELS,
+)
+from cleveragents.application.services.acms_skeleton_renderers import (
+    FragmentRenderVisitor,
+    get_render_visitor,
+)
+from cleveragents.application.services.component_resolver import (
+    ComponentRegistrationError,
+    ComponentResolver,
+)
+from cleveragents.application.services.config_service import ConfigService
+from cleveragents.domain.models.acms import DetailLevelMap
+from cleveragents.domain.models.acms.ontology_registry import get_domain
+from cleveragents.domain.models.core.context_fragment import ContextFragment
+
+if TYPE_CHECKING:
+    from cleveragents.application.services.acms_service import SkeletonCompressor
+
+logger = logging.getLogger(__name__)
+
+_DETAIL_DEPTH_OVERVIEW: Final[int] = 0
+_DETAIL_DEPTH_SUMMARY: Final[int] = 1
+_CODE_EXTENSIONS: Final[frozenset[str]] = frozenset(
+    {".py", ".pyi", ".ts", ".tsx", ".js", ".jsx", ".java", ".rs"}
+)
+_DOC_EXTENSIONS: Final[frozenset[str]] = frozenset(
+    {".md", ".mdx", ".rst", ".txt", ".adoc"}
+)
+_DATA_EXTENSIONS: Final[frozenset[str]] = frozenset({".sql", ".ddl", ".csv", ".tsv"})
+_INFRA_EXTENSIONS: Final[frozenset[str]] = frozenset(
+    {".yaml", ".yml", ".json", ".toml", ".tf", ".hcl"}
+)
+_PREFIX_FAMILY: Final[dict[str, str]] = {
+    "uko-code:": "code",
+    "uko-oo:": "code",
+    "uko-func:": "code",
+    "uko-proc:": "code",
+    "uko-py:": "code",
+    "uko-ts:": "code",
+    "uko-java:": "code",
+    "uko-rs:": "code",
+    "uko-doc:": "doc",
+    "uko-data:": "data",
+    "uko-infra:": "infra",
+    "uko:": "generic",
+}
+
+
+def _child_detail_map(
+    domain: str,
+    parent: DetailLevelMap,
+    levels: tuple[tuple[str, int], ...],
+) -> DetailLevelMap:
+    """Create a child ``DetailLevelMap`` over an inherited parent chain."""
+
+    depth_map = dict(levels)
+    return DetailLevelMap(
+        domain=domain,
+        parent=parent,
+        levels=depth_map,
+        max_depth=max(depth_map.values()) if depth_map else parent.max_depth,
+    )
+
+
+_DETAIL_MAPS: Final[dict[str, DetailLevelMap]] = {
+    "uko:": get_domain("uko:").detail_map,
+    "uko-code:": CODE_DETAIL_LEVEL_MAP,
+    "uko-doc:": get_domain("uko-doc:").detail_map,
+    "uko-data:": get_domain("uko-data:").detail_map,
+    "uko-infra:": get_domain("uko-infra:").detail_map,
+    "uko-oo:": OO_DETAIL_LEVEL_MAP,
+    "uko-func:": FUNC_DETAIL_LEVEL_MAP,
+    "uko-proc:": PROC_DETAIL_LEVEL_MAP,
+    "uko-py:": _child_detail_map("uko-py:", OO_DETAIL_LEVEL_MAP, PYTHON_DETAIL_LEVELS),
+    "uko-ts:": _child_detail_map(
+        "uko-ts:", OO_DETAIL_LEVEL_MAP, TYPESCRIPT_DETAIL_LEVELS
+    ),
+    "uko-java:": _child_detail_map(
+        "uko-java:", OO_DETAIL_LEVEL_MAP, JAVA_DETAIL_LEVELS
+    ),
+    "uko-rs:": _child_detail_map("uko-rs:", OO_DETAIL_LEVEL_MAP, RUST_DETAIL_LEVELS),
+}
+
+
+class DepthReductionCompressor:
+    """Compress parent fragments by re-rendering them at overview depths 0-1."""
+
+    def compress(
+        self,
+        fragments: tuple[ContextFragment, ...],
+        skeleton_budget: int,
+    ) -> tuple[ContextFragment, ...]:
+        """Return reduced fragments whose total token count fits the budget."""
+
+        if skeleton_budget <= 0 or not fragments:
+            return ()
+
+        normalized = tuple(self._normalise_fragment(fragment) for fragment in fragments)
+        ordered = sorted(
+            normalized,
+            key=lambda fragment: (
+                -fragment.relevance_score,
+                fragment.detail_depth,
+                getattr(fragment, "fragment_id", fragment.uko_node),
+            ),
+        )
+
+        compressed: list[ContextFragment] = []
+        remaining = skeleton_budget
+        for fragment in ordered:
+            reduced = self._reduce_fragment(fragment, remaining)
+            if reduced is None:
+                continue
+            compressed.append(reduced)
+            remaining -= reduced.token_count
+            if remaining <= 0:
+                break
+
+        logger.info(
+            "Compressed skeleton fragments via depth reduction",
+            extra={
+                "input_count": len(fragments),
+                "output_count": len(compressed),
+                "skeleton_budget": skeleton_budget,
+                "used_tokens": sum(fragment.token_count for fragment in compressed),
+            },
+        )
+        return tuple(compressed)
+
+    @staticmethod
+    def _normalise_fragment(fragment: ContextFragment) -> ContextFragment:
+        detail_map = _detail_map_for_fragment(fragment)
+        metadata = dict(fragment.metadata)
+        if metadata.get("detail_level_domain") == detail_map.domain:
+            return fragment
+        metadata["detail_level_domain"] = detail_map.domain
+        return fragment.model_copy(update={"metadata": metadata})
+
+    def _reduce_fragment(
+        self,
+        fragment: ContextFragment,
+        remaining_budget: int,
+    ) -> ContextFragment | None:
+        if remaining_budget <= 0:
+            return None
+
+        detail_map = _detail_map_for_fragment(fragment)
+        family = _domain_family(fragment, detail_map)
+        visitor = get_render_visitor(family)
+
+        candidate_depths = [
+            depth
+            for depth in (_DETAIL_DEPTH_SUMMARY, _DETAIL_DEPTH_OVERVIEW)
+            if depth <= min(fragment.detail_depth, detail_map.max_depth)
+        ]
+        if not candidate_depths:
+            candidate_depths = [_DETAIL_DEPTH_OVERVIEW]
+
+        seen_depths: set[int] = set()
+        fallback_candidate: ContextFragment | None = None
+        for depth in candidate_depths:
+            if depth in seen_depths:
+                continue
+            seen_depths.add(depth)
+            candidate = self._rerender_fragment(
+                fragment,
+                detail_map=detail_map,
+                visitor=visitor,
+                target_depth=depth,
+            )
+            if candidate.token_count <= remaining_budget:
+                return candidate
+            fallback_candidate = candidate
+        if fallback_candidate is None:
+            return None
+        return self._fit_candidate(fallback_candidate, remaining_budget)
+
+    def _rerender_fragment(
+        self,
+        fragment: ContextFragment,
+        *,
+        detail_map: DetailLevelMap,
+        visitor: FragmentRenderVisitor,
+        target_depth: int,
+    ) -> ContextFragment:
+        level_name = _level_name_for_depth(detail_map, target_depth)
+        if fragment.detail_depth <= target_depth:
+            content = fragment.content
+        else:
+            content = visitor.render(
+                fragment,
+                target_depth=target_depth,
+                level_name=level_name,
+            )
+        token_count = _estimate_tokens(content)
+        metadata = {
+            **fragment.metadata,
+            "detail_level_domain": detail_map.domain,
+            "skeleton_target_depth": str(target_depth),
+            "skeleton_target_level": level_name,
+        }
+        return fragment.model_copy(
+            update={
+                "content": content,
+                "detail_depth": target_depth,
+                "token_count": token_count,
+                "metadata": metadata,
+            }
+        )
+
+    @staticmethod
+    def _fit_candidate(
+        fragment: ContextFragment,
+        remaining_budget: int,
+    ) -> ContextFragment | None:
+        if fragment.token_count <= remaining_budget:
+            return fragment
+        if remaining_budget <= 0:
+            return None
+        clipped = _clip_to_token_budget(fragment.content, remaining_budget)
+        if not clipped:
+            return None
+        clipped_tokens = _estimate_tokens(clipped)
+        if clipped_tokens > remaining_budget:
+            return None
+        return fragment.model_copy(
+            update={
+                "content": clipped,
+                "token_count": clipped_tokens,
+            }
+        )
+
+
+def resolve_configured_skeleton_compressor() -> SkeletonCompressor:
+    """Resolve the configured skeleton compressor via the component resolver."""
+
+    resolver = ComponentResolver()
+    builtin = DepthReductionCompressor()
+    from cleveragents.application.services.acms_service import SkeletonCompressor
+
+    resolver.register_global(SkeletonCompressor, builtin)
+    configured = ConfigService().resolve("context.pipeline.skeleton-compressor").value
+    if (
+        not isinstance(configured, str)
+        or configured == "builtin:DepthReductionCompressor"
+    ):
+        return cast(SkeletonCompressor, resolver.resolve(SkeletonCompressor).component)
+
+    instance = resolver._import_component(configured)
+    compress = getattr(instance, "compress", None)
+    if not callable(compress):
+        msg = (
+            f"Configured skeleton compressor {configured!r} does not expose a callable "
+            "compress() method"
+        )
+        raise ComponentRegistrationError(msg)
+
+    resolver.register_global(SkeletonCompressor, cast(SkeletonCompressor, instance))
+    return cast(SkeletonCompressor, resolver.resolve(SkeletonCompressor).component)
+
+
+def _detail_map_for_fragment(fragment: ContextFragment) -> DetailLevelMap:
+    """Infer the most specific ``DetailLevelMap`` chain available for a fragment."""
+
+    metadata = fragment.metadata
+    for key in ("detail_level_domain", "detail_map_domain", "uko_domain"):
+        value = metadata.get(key)
+        if isinstance(value, str) and value in _DETAIL_MAPS:
+            return _DETAIL_MAPS[value]
+
+    for prefix in _DETAIL_MAPS:
+        if fragment.uko_node.startswith(prefix):
+            return _DETAIL_MAPS[prefix]
+        if fragment.provenance.resource_uri.startswith(prefix):
+            return _DETAIL_MAPS[prefix]
+
+    resource_uri = fragment.provenance.resource_uri.lower()
+    resource_type = getattr(fragment.provenance, "resource_type", "").lower()
+    extension = PurePosixPath(resource_uri.split("?", 1)[0]).suffix.lower()
+
+    if extension in _CODE_EXTENSIONS:
+        if extension in {".py", ".pyi"}:
+            return _DETAIL_MAPS["uko-py:"]
+        if extension in {".ts", ".tsx", ".js", ".jsx"}:
+            return _DETAIL_MAPS["uko-ts:"]
+        if extension == ".java":
+            return _DETAIL_MAPS["uko-java:"]
+        if extension == ".rs":
+            return _DETAIL_MAPS["uko-rs:"]
+        return _DETAIL_MAPS["uko-code:"]
+    if extension in _DOC_EXTENSIONS:
+        return _DETAIL_MAPS["uko-doc:"]
+    if extension in _DATA_EXTENSIONS:
+        return _DETAIL_MAPS["uko-data:"]
+    if extension in _INFRA_EXTENSIONS:
+        return _DETAIL_MAPS["uko-infra:"]
+
+    if any(
+        token in resource_type
+        for token in ("python", "typescript", "javascript", "java", "rust", "code")
+    ):
+        if "python" in resource_type:
+            return _DETAIL_MAPS["uko-py:"]
+        if "typescript" in resource_type or "javascript" in resource_type:
+            return _DETAIL_MAPS["uko-ts:"]
+        if "java" in resource_type:
+            return _DETAIL_MAPS["uko-java:"]
+        if "rust" in resource_type:
+            return _DETAIL_MAPS["uko-rs:"]
+        return _DETAIL_MAPS["uko-code:"]
+    if any(token in resource_type for token in ("doc", "markdown", "text")):
+        return _DETAIL_MAPS["uko-doc:"]
+    if any(token in resource_type for token in ("sql", "database", "schema", "table")):
+        return _DETAIL_MAPS["uko-data:"]
+    if any(
+        token in resource_type
+        for token in ("infra", "config", "deploy", "kubernetes", "terraform")
+    ):
+        return _DETAIL_MAPS["uko-infra:"]
+    return _DETAIL_MAPS["uko:"]
+
+
+def _domain_family(fragment: ContextFragment, detail_map: DetailLevelMap) -> str:
+    """Return the rendering family for a fragment."""
+
+    explicit = fragment.metadata.get("detail_level_domain")
+    if isinstance(explicit, str) and explicit in _PREFIX_FAMILY:
+        return _PREFIX_FAMILY[explicit]
+    return _PREFIX_FAMILY.get(detail_map.domain, "generic")
+
+
+def _level_name_for_depth(detail_map: DetailLevelMap, depth: int) -> str:
+    """Resolve the effective named level for a target depth."""
+
+    matches = [
+        name
+        for name, mapped_depth in sorted(
+            detail_map.effective_levels().items(),
+            key=lambda item: (item[1], item[0]),
+        )
+        if mapped_depth == depth
+    ]
+    return matches[0] if matches else f"depth-{depth}"
+
+
+def _estimate_tokens(text: str) -> int:
+    """Estimate token count for reduced fragments conservatively."""
+
+    if not text:
+        return 1
+    return max(1, math.ceil(len(text) / 4))
+
+
+def _clip_to_token_budget(text: str, token_budget: int) -> str:
+    """Clip text so the rendered content itself fits the token budget."""
+
+    if token_budget <= 0:
+        return ""
+    return text[: max(1, token_budget * 4)].rstrip()

src/cleveragents/application/services/acms_skeleton_renderers.py

@@ -0,0 +1,222 @@
+"""Render visitors for ACMS skeleton compression."""
+
+from __future__ import annotations
+
+import re
+from typing import Final
+
+from cleveragents.domain.models.core.context_fragment import ContextFragment
+
+_DETAIL_DEPTH_OVERVIEW: Final[int] = 0
+_MAX_DOC_HEADINGS: Final[int] = 3
+_MAX_DATA_ITEMS: Final[int] = 4
+
+
+class FragmentRenderVisitor:
+    """Render a fragment at a reduced detail depth for one domain family."""
+
+    def render(
+        self,
+        fragment: ContextFragment,
+        *,
+        target_depth: int,
+        level_name: str,
+    ) -> str:
+        raise NotImplementedError
+
+
+class _CodeRenderVisitor(FragmentRenderVisitor):
+    def render(
+        self,
+        fragment: ContextFragment,
+        *,
+        target_depth: int,
+        level_name: str,
+    ) -> str:
+        subject = _resource_label(fragment)
+        summary = _code_structure_summary(fragment.content)
+        if target_depth <= _DETAIL_DEPTH_OVERVIEW:
+            return f"{subject} [{level_name}]"
+        if summary:
+            return f"{subject} [{level_name}]: {summary}"
+        return f"{subject} [{level_name}]"
+
+
+class _DocumentRenderVisitor(FragmentRenderVisitor):
+    def render(
+        self,
+        fragment: ContextFragment,
+        *,
+        target_depth: int,
+        level_name: str,
+    ) -> str:
+        title = _first_non_empty_line(
+            fragment.content,
+            fallback=_resource_label(fragment),
+        )
+        if target_depth <= _DETAIL_DEPTH_OVERVIEW:
+            return f"{title} [{level_name}]"
+        headings = _document_headings(fragment.content)
+        if headings:
+            return f"{title} [{level_name}]: {'; '.join(headings[:_MAX_DOC_HEADINGS])}"
+        return f"{title} [{level_name}]: {_summarise_words(fragment.content, limit=12)}"
+
+
+class _DataRenderVisitor(FragmentRenderVisitor):
+    def render(
+        self,
+        fragment: ContextFragment,
+        *,
+        target_depth: int,
+        level_name: str,
+    ) -> str:
+        label = _resource_label(fragment)
+        items = _csv_like_items(fragment.content)
+        if target_depth <= _DETAIL_DEPTH_OVERVIEW:
+            return f"{label} [{level_name}]"
+        summary = ", ".join(items[:_MAX_DATA_ITEMS]) or _summarise_words(
+            fragment.content,
+            10,
+        )
+        return f"{label} [{level_name}]: {summary}"
+
+
+class _InfrastructureRenderVisitor(FragmentRenderVisitor):
+    def render(
+        self,
+        fragment: ContextFragment,
+        *,
+        target_depth: int,
+        level_name: str,
+    ) -> str:
+        label = _resource_label(fragment)
+        synopsis = _first_non_empty_line(fragment.content)
+        if target_depth <= _DETAIL_DEPTH_OVERVIEW:
+            return f"{label} [{level_name}]"
+        return f"{label} [{level_name}]: {synopsis}"
+
+
+class _GenericRenderVisitor(FragmentRenderVisitor):
+    def render(
+        self,
+        fragment: ContextFragment,
+        *,
+        target_depth: int,
+        level_name: str,
+    ) -> str:
+        label = _resource_label(fragment)
+        if target_depth <= _DETAIL_DEPTH_OVERVIEW:
+            return f"{label} [{level_name}]"
+        return f"{label} [{level_name}]: {_summarise_words(fragment.content, 10)}"
+
+
+_VISITORS: Final[dict[str, FragmentRenderVisitor]] = {
+    "code": _CodeRenderVisitor(),
+    "doc": _DocumentRenderVisitor(),
+    "data": _DataRenderVisitor(),
+    "infra": _InfrastructureRenderVisitor(),
+    "generic": _GenericRenderVisitor(),
+}
+
+
+def get_render_visitor(family: str) -> FragmentRenderVisitor:
+    """Return the visitor for a resolved fragment family."""
+
+    return _VISITORS[family]
+
+
+def _code_structure_summary(text: str) -> str:
+    """Build a structural summary for depth-1 code skeletons."""
+
+    symbols = _extract_code_symbols(text)
+    if symbols:
+        return f"symbols={', '.join(symbols[:5])}"
+    return ""
+
+
+def _extract_code_symbols(text: str) -> list[str]:
+    """Extract class, function, and import targets from source text."""
+
+    patterns = (
+        r"^(?:async\s+)?def\s+([A-Za-z_][A-Za-z0-9_]*)",
+        r"^class\s+([A-Za-z_][A-Za-z0-9_]*)",
+        r"^function\s+([A-Za-z_][A-Za-z0-9_]*)",
+        r"^(?:export\s+)?(?:const|let|var)\s+([A-Za-z_][A-Za-z0-9_]*)\s*=\s*(?:async\s+)?(?:\([^)]*\)\s*=>|function)",
+        r"^fn\s+([A-Za-z_][A-Za-z0-9_]*)",
+        r"^import\s+([A-Za-z_][A-Za-z0-9_]*)",
+    )
+    names: list[str] = []
+    seen: set[str] = set()
+    for line in text.splitlines():
+        stripped = line.strip()
+        if not stripped:
+            continue
+        for pattern in patterns:
+            match = re.match(pattern, stripped)
+            if match is None:
+                continue
+            name = match.group(1)
+            if name not in seen:
+                names.append(name)
+                seen.add(name)
+            break
+    return names
+
+
+def _resource_label(fragment: ContextFragment) -> str:
+    """Return a compact human-readable label for a fragment."""
+
+    resource_uri = fragment.provenance.resource_uri or fragment.uko_node
+    stripped = (
+        resource_uri.split("://", 1)[1] if "://" in resource_uri else resource_uri
+    )
+    stripped = stripped.rstrip("/")
+    if not stripped:
+        return fragment.uko_node
+    return stripped.rsplit("/", 1)[-1]
+
+
+def _first_non_empty_line(text: str, fallback: str = "summary") -> str:
+    """Return the first non-empty line from ``text``."""
+
+    for line in text.splitlines():
+        stripped = line.strip(" #-*\t")
+        if stripped:
+            return stripped
+    return fallback
+
+
+def _summarise_words(text: str, limit: int) -> str:
+    """Return the first ``limit`` words from ``text``."""
+
+    words = re.findall(r"\S+", text)
+    if not words:
+        return "summary unavailable"
+    return " ".join(words[:limit])
+
+
+def _document_headings(text: str) -> list[str]:
+    """Extract likely top-level headings from a document fragment."""
+
+    headings: list[str] = []
+    for line in text.splitlines():
+        stripped = line.strip()
+        if not stripped:
+            continue
+        if stripped.startswith("#"):
+            headings.append(stripped.lstrip("# ").strip())
+            continue
+        if re.match(r"^\d+(?:\.\d+)*\s+", stripped):
+            headings.append(stripped)
+    return headings
+
+
+def _csv_like_items(text: str) -> list[str]:
+    """Extract short comma/newline-separated items from a fragment."""
+
+    items: list[str] = []
+    for raw in re.split(r"[,;\n]+", text):
+        stripped = raw.strip()
+        if stripped:
+            items.append(stripped)
+    return items

src/cleveragents/application/services/component_resolver.py

@@ -32,7 +32,7 @@ import importlib
 import threading
 from dataclasses import dataclass, field
 from enum import Enum
-from typing import Any, TypeVar
+from typing import Any, Final, TypeVar
 
 import structlog
 
@@ -102,6 +102,20 @@ class ComponentRegistrationError(ValueError):
 # ---------------------------------------------------------------------------
 
 _ALLOWED_MODULE_PREFIXES: tuple[str, ...] = ("cleveragents.",)
+_ALLOWED_BUILTIN_COMPONENTS: Final[frozenset[str]] = frozenset(
+    {
+        "ConfidenceWeightedSelector",
+        "ProportionalBudgetAllocator",
+        "ParallelStrategyExecutor",
+        "ContentHashDeduplicator",
+        "MaxDepthResolver",
+        "WeightedCompositeScorer",
+        "GreedyKnapsackPacker",
+        "RelevanceCoherenceOrderer",
+        "ProvenancePreambleGenerator",
+        "DepthReductionCompressor",
+    }
+)
 
 
 # ---------------------------------------------------------------------------
@@ -701,6 +715,18 @@ class ComponentResolver:
 
         module_name, class_name = module_path.rsplit(":", 1)
 
+        if module_name == "builtin":
+            if class_name not in _ALLOWED_BUILTIN_COMPONENTS:
+                msg = (
+                    f"Builtin component {class_name!r} is not in the allowed builtin "
+                    f"list: {sorted(_ALLOWED_BUILTIN_COMPONENTS)}"
+                )
+                raise ComponentRegistrationError(msg)
+            from cleveragents.application import services as builtin_services
+
+            cls = getattr(builtin_services, class_name)
+            return cls()
+
         # Security: restrict to allowed prefixes
         if self.ALLOWED_MODULE_PREFIXES and not any(
             module_name.startswith(prefix) for prefix in self.ALLOWED_MODULE_PREFIXES

src/cleveragents/application/services/depth_breadth_projection.py

@@ -23,13 +23,18 @@ from __future__ import annotations
 import math
 from collections import deque
 from collections.abc import Mapping, Sequence
-from typing import Any
+from typing import Any, cast
 
 import structlog
 from pydantic import BaseModel, ConfigDict, Field, field_validator
 
+from cleveragents.application.services.acms_skeleton_compressor import (
+    resolve_configured_skeleton_compressor,
+)
+from cleveragents.domain.models.acms.crp import (
+    ContextFragment as ProjectionContextFragment,
+)
 from cleveragents.domain.models.acms.crp import (
-    ContextFragment,
     ContextRequest,
     DetailLevelMap,
     FragmentProvenance,
@@ -261,7 +266,7 @@ class DepthBreadthProjector:
         node_index: Mapping[str, UKONode] | None = None,
         *,
         token_estimator: _TokenEstimator | None = None,
-    ) -> list[ContextFragment]:
+    ) -> list[ProjectionContextFragment]:
         """Project and convert to :class:`ContextFragment` instances.
 
         Convenience wrapper over :meth:`project` that produces fragments
@@ -279,12 +284,12 @@ class DepthBreadthProjector:
         """
         projected = self.project(spec, adjacency, node_index)
         estimator = token_estimator or _default_token_estimator
-        fragments: list[ContextFragment] = []
+        fragments: list[ProjectionContextFragment] = []
         for node in projected:
             tokens = estimator(node.uri, node.resolved_depth)
             relevance = self._distance_relevance(node.distance, spec.breadth)
             fragments.append(
-                ContextFragment(
+                ProjectionContextFragment(
                     uko_node=node.uri,
                     content=f"{node.label or node.uri} [depth={node.resolved_depth}]",
                     detail_depth=node.resolved_depth,
@@ -398,7 +403,7 @@ class ChildContextResult(BaseModel, frozen=True):
     """
 
     request: ContextRequest
-    skeleton_fragments: tuple[ContextFragment, ...] = ()
+    skeleton_fragments: tuple[ProjectionContextFragment, ...] = ()
     parent_context_hash: str = ""
     skeleton_budget: int = 0
     depth_delta: int = 0
@@ -428,7 +433,9 @@ class PlanContextInheritance:
         skeleton_compressor: _SkeletonCompressorLike | None = None,
     ) -> None:
         self._config = config or InheritanceConfig()
-        self._compressor = skeleton_compressor
+        self._compressor = (
+            skeleton_compressor or resolve_configured_skeleton_compressor()
+        )
         self._logger = logger.bind(component="PlanContextInheritance")
 
     @property
@@ -476,12 +483,15 @@ class PlanContextInheritance:
 
         # 3. Skeleton budget
         skeleton_budget = int(child_token_budget * self._config.skeleton_ratio)
-        skeleton_fragments: tuple[ContextFragment, ...] = ()
+        skeleton_fragments: tuple[ProjectionContextFragment, ...] = ()
 
         if skeleton_budget >= self._config.min_skeleton_tokens and self._compressor:
-            parent_fragments = tuple(parent_context.fragments)
+            parent_fragments = self._prioritize_parent_fragments(
+                tuple(parent_context.fragments),
+                child_focus,
+            )
             skeleton_fragments = self._compressor.compress(
-                parent_fragments,
+                cast(Any, parent_fragments),
                 skeleton_budget,
             )
             self._logger.info(
@@ -519,6 +529,61 @@ class PlanContextInheritance:
         total = sum(getattr(f, "detail_depth", 0) for f in fragments)
         return total // len(fragments)
 
+    @staticmethod
+    def _prioritize_parent_fragments(
+        fragments: tuple[ProjectionContextFragment, ...],
+        child_focus: Sequence[str],
+    ) -> tuple[ProjectionContextFragment, ...]:
+        """Boost fragments whose provenance is closest to the child focus."""
+
+        if not fragments or not child_focus:
+            return fragments
+        prioritized: list[ProjectionContextFragment] = []
+        for fragment in fragments:
+            proximity = PlanContextInheritance._focus_proximity(fragment, child_focus)
+            adjusted_score = min(
+                1.0,
+                fragment.relevance_score * 0.5 + proximity * 0.5,
+            )
+            if adjusted_score == fragment.relevance_score:
+                prioritized.append(fragment)
+                continue
+            prioritized.append(
+                fragment.model_copy(update={"relevance_score": adjusted_score})
+            )
+        return tuple(prioritized)
+
+    @staticmethod
+    def _focus_proximity(
+        fragment: ProjectionContextFragment,
+        child_focus: Sequence[str],
+    ) -> float:
+        """Return a 0..1 score for how near a fragment is to the child focus."""
+
+        fragment_paths = [fragment.uko_node]
+        resource_uri = fragment.provenance.resource_uri
+        if resource_uri and resource_uri != fragment.uko_node:
+            fragment_paths.append(resource_uri)
+        best = 0.0
+        for fragment_path in fragment_paths:
+            fragment_segments = _uri_segments(fragment_path)
+            if not fragment_segments:
+                continue
+            for focus in child_focus:
+                focus_segments = _uri_segments(focus)
+                shared = 0
+                for left, right in zip(fragment_segments, focus_segments, strict=False):
+                    if left != right:
+                        break
+                    shared += 1
+                if not focus_segments:
+                    continue
+                best = max(
+                    best,
+                    shared / max(len(fragment_segments), len(focus_segments)),
+                )
+        return best
+
     @staticmethod
     def extract_child_focus(
         parent_decisions: Sequence[str],
@@ -536,6 +601,14 @@ class PlanContextInheritance:
         return list(fallback_focus or [])
 
 
+def _uri_segments(uri: str) -> list[str]:
+    """Split a URI into normalized path segments for proximity checks."""
+
+    if "://" in uri:
+        uri = uri.split("://", 1)[1]
+    return [segment for segment in uri.replace("\\", "/").split("/") if segment]
+
+
 # ---------------------------------------------------------------------------
 # Structural protocols (duck-typed, avoid circular imports)
 # ---------------------------------------------------------------------------