features/acms_core_pipeline_components.feature

@@ -0,0 +1,364 @@
+@acms @acms_core_pipeline
+Feature: ACMS Core Pipeline Components
+  As a CleverAgents developer
+  I want core ACMS pipeline components implemented
+  So that the pipeline can select strategies, allocate budgets, score fragments,
+  pack within constraints, and order for optimal LLM consumption
+
+  # ===========================================================================
+  # ActorPhaseStrategySelector
+  # ===========================================================================
+
+  @strategy_selector
+  Scenario: Select strategies with no actor type or plan phase
+    Given a set of core pipeline strategies
+    When I select strategies with no actor_type or plan_phase
+    Then all core strategies with positive confidence should be returned
+    And the core strategies should be sorted by confidence descending
+
+  @strategy_selector
+  Scenario: Select strategies boosts preferred strategies for planner actor
+    Given a set of core pipeline strategies
+    When I select strategies with actor_type "planner" only
+    Then the "relevance" strategy should have boosted confidence
+
+  @strategy_selector
+  Scenario: Select strategies boosts preferred strategies for executor actor
+    Given a set of core pipeline strategies
+    When I select strategies with actor_type "executor" only
+    Then the "tiered" strategy should have boosted confidence
+
+  @strategy_selector
+  Scenario: Select strategies boosts preferred strategies for strategize phase
+    Given a set of core pipeline strategies
+    When I select strategies with plan_phase "strategize" only
+    Then the "relevance" strategy should have boosted confidence
+
+  @strategy_selector
+  Scenario: Select strategies applies both actor and phase boosts
+    Given a set of core pipeline strategies
+    When I select strategies with actor_type "planner" and plan_phase "strategize"
+    Then the "relevance" strategy should have maximum boosted confidence
+
+  @strategy_selector
+  Scenario: Select strategies excludes zero-confidence strategies
+    Given a set of core pipeline strategies including a zero-confidence strategy
+    When I select strategies with no actor_type or plan_phase
+    Then the core zero-confidence strategy should not be in the results
+
+  @strategy_selector
+  Scenario: Select strategies with empty strategy list returns empty
+    Given an empty strategy list
+    When I select strategies with no actor_type or plan_phase
+    Then 0 strategies should be selected
+
+  @strategy_selector
+  Scenario: Confidence is clamped to 1.0 after boost
+    Given a strategy with confidence 0.9
+    When I select strategies with actor_type "planner" and plan_phase "strategize"
+    Then the strategy confidence should not exceed 1.0
+
+  # ===========================================================================
+  # SpecBudgetAllocator
+  # ===========================================================================
+
+  @budget_allocator
+  Scenario: Allocate budget to empty candidates returns empty
+    Given an empty candidate list
+    When I allocate budget 1000 with SpecBudgetAllocator
+    Then 0 allocations should be returned
+
+  @budget_allocator
+  Scenario: Allocate budget to single candidate gives full budget
+    Given a single candidate with confidence 0.8
+    When I allocate budget 1000 with SpecBudgetAllocator
+    Then 1 allocation should be returned
+    And the single allocation should receive 1000 tokens
+
+  @budget_allocator
+  Scenario: Allocate budget proportionally to two candidates
+    Given two candidates with equal confidence 0.5
+    When I allocate budget 1000 with SpecBudgetAllocator
+    Then 2 allocations should be returned
+    And the total allocated tokens should equal 1000
+
+  @budget_allocator
+  Scenario: Allocate budget uses spec formula with quality scores
+    Given two candidates with different quality scores
+    When I allocate budget 1000 with SpecBudgetAllocator
+    Then the higher quality candidate should receive more tokens
+
+  @budget_allocator
+  Scenario: Allocate budget with zero total weight falls back to equal split
+    Given two candidates with zero confidence
+    When I allocate budget 1000 with SpecBudgetAllocator
+    Then 2 allocations should be returned
+    And the total allocated tokens should equal 1000
+
+  @budget_allocator
+  Scenario: Allocate budget with min_useful_budget excludes small candidates
+    Given three candidates where one would receive very few tokens
+    When I allocate budget 100 with SpecBudgetAllocator and min_useful_budget 30
+    Then the small candidate should be excluded from allocations
+
+  @budget_allocator
+  Scenario: Allocate budget total never exceeds budget
+    Given three candidates with varying confidence
+    When I allocate budget 500 with SpecBudgetAllocator
+    Then the total allocated tokens should equal 500
+
+  # ===========================================================================
+  # RelevanceRecencyPriorityScorer
+  # ===========================================================================
+
+  @scorer
+  Scenario: Score empty fragment list returns empty
+    Given an empty core pipeline fragment list
+    When I score the fragments with RelevanceRecencyPriorityScorer
+    Then 0 core scored fragments should be returned
+
+  @scorer
+  Scenario: Score single fragment preserves metadata
+    Given a core pipeline fragment with relevance 0.8
+    When I score the fragments with RelevanceRecencyPriorityScorer
+    Then the scored fragment should have _original_relevance metadata "0.8"
+
+  @scorer
+  Scenario: Score produces composite from relevance recency and priority
+    Given a core pipeline fragment with relevance 0.8 and priority 0.9
+    When I score the fragments with RelevanceRecencyPriorityScorer
+    Then the scored fragment composite score should be between 0.0 and 1.0
+
+  @scorer
+  Scenario: Score is deterministic for identical inputs
+    Given two identical core pipeline fragments
+    When I score both fragment sets with RelevanceRecencyPriorityScorer
+    Then both scored fragments should have identical composite scores
+
+  @scorer
+  Scenario: Score recency normalises across fragment timestamps
+    Given two core pipeline fragments with different timestamps
+    When I score the fragments with RelevanceRecencyPriorityScorer
+    Then the newer fragment should have higher recency score
+
+  @scorer
+  Scenario: Score with same timestamps gives recency 1.0
+    Given two core pipeline fragments with identical timestamps
+    When I score the fragments with RelevanceRecencyPriorityScorer
+    Then all scored fragments should have recency score 1.0
+
+  @scorer
+  Scenario: Score composite is clamped to 1.0
+    Given a core pipeline fragment with maximum relevance and priority
+    When I score the fragments with RelevanceRecencyPriorityScorer
+    Then the scored fragment composite score should be at most 1.0
+
+  @scorer
+  Scenario: Score composite is clamped to 0.0
+    Given a core pipeline fragment with zero relevance and zero priority
+    When I score the fragments with RelevanceRecencyPriorityScorer
+    Then the scored fragment composite score should be at least 0.0
+
+  @scorer
+  Scenario: Score with custom weights uses configured weights
+    Given a core pipeline fragment with relevance 1.0 and priority 0.0
+    And scorer configured with relevance_weight 1.0 recency_weight 0.0 priority_weight 0.0
+    When I score the fragments with custom RelevanceRecencyPriorityScorer
+    Then the scored fragment composite score should be 1.0
+
+  @scorer
+  Scenario: Score extracts priority from metadata
+    Given a core pipeline fragment with metadata priority "0.9"
+    When I score the fragments with RelevanceRecencyPriorityScorer
+    Then the scored fragment should have _score_priority metadata near "0.9"
+
+  @scorer
+  Scenario: Score defaults priority to 0.5 when not in metadata
+    Given a core pipeline fragment with no priority metadata
+    When I score the fragments with RelevanceRecencyPriorityScorer
+    Then the scored fragment should have _score_priority metadata "0.5"
+
+  # ===========================================================================
+  # ConstrainedKnapsackPacker
+  # ===========================================================================
+
+  @packer
+  Scenario: Pack empty fragment list returns empty
+    Given an empty core pipeline fragment list
+    And a core pipeline budget with max_tokens 1000 and reserved_tokens 0
+    When I pack the fragments with ConstrainedKnapsackPacker
+    Then 0 core packed fragments should be returned
+
+  @packer
+  Scenario: Pack fragments within token budget
+    Given the following core pipeline fragments:
+      | uko_node              | content | score | tokens | depth |
+      | project://app/main.py | alpha   | 0.9   | 100    | 3     |
+      | project://app/io.py   | beta    | 0.7   | 100    | 3     |
+      | project://app/util.py | gamma   | 0.5   | 100    | 3     |
+    And a core pipeline budget with max_tokens 250 and reserved_tokens 0
+    When I pack the fragments with ConstrainedKnapsackPacker
+    Then 2 core packed fragments should be returned
+    And the core packed total tokens should be at most 250
+
+  @packer
+  Scenario: Pack respects max_file_size constraint
+    Given the following core pipeline fragments:
+      | uko_node              | content                    | score | tokens | depth |
+      | project://app/main.py | short content              | 0.9   | 10     | 3     |
+      | project://app/io.py   | this is a much longer text | 0.7   | 20     | 3     |
+    And a core pipeline budget with max_tokens 1000 and reserved_tokens 0
+    And a ConstrainedKnapsackPacker with max_file_size 15
+    When I pack the fragments with ConstrainedKnapsackPacker
+    Then 1 core packed fragment should be returned
+    And the core packed fragment uko_node should be "project://app/main.py"
+
+  @packer
+  Scenario: Pack respects max_total_size constraint
+    Given the following core pipeline fragments:
+      | uko_node              | content | score | tokens | depth |
+      | project://app/main.py | hello   | 0.9   | 10     | 3     |
+      | project://app/io.py   | world   | 0.7   | 10     | 3     |
+      | project://app/util.py | extra   | 0.5   | 10     | 3     |
+    And a core pipeline budget with max_tokens 1000 and reserved_tokens 0
+    And a ConstrainedKnapsackPacker with max_total_size 4
+    When I pack the fragments with ConstrainedKnapsackPacker
+    Then 0 core packed fragments should be returned
+
+  @packer
+  Scenario: Pack prefers higher scored fragments
+    Given the following core pipeline fragments:
+      | uko_node              | content | score | tokens | depth |
+      | project://app/main.py | alpha   | 0.3   | 100    | 3     |
+      | project://app/io.py   | beta    | 0.9   | 100    | 3     |
+    And a core pipeline budget with max_tokens 150 and reserved_tokens 0
+    When I pack the fragments with ConstrainedKnapsackPacker
+    Then 1 core packed fragment should be returned
+    And the core packed fragment uko_node should be "project://app/io.py"
+
+  @packer
+  Scenario: Pack with context_view uses view constraints
+    Given the following core pipeline fragments:
+      | uko_node              | content | score | tokens | depth |
+      | project://app/main.py | hello   | 0.9   | 10     | 3     |
+      | project://app/io.py   | world   | 0.7   | 10     | 3     |
+    And a core pipeline budget with max_tokens 1000 and reserved_tokens 0
+    And a ConstrainedKnapsackPacker with context_view max_total_size 4
+    When I pack the fragments with ConstrainedKnapsackPacker
+    Then 0 core packed fragments should be returned
+
+  @packer
+  Scenario: Pack with budget=0 returns empty
+    Given the following core pipeline fragments:
+      | uko_node              | content | score | tokens | depth |
+      | project://app/main.py | hello   | 0.9   | 10     | 3     |
+    And a core pipeline budget with max_tokens 1 and reserved_tokens 0
+    When I pack the fragments with ConstrainedKnapsackPacker
+    Then 0 core packed fragments should be returned
+
+  # ===========================================================================
+  # PriorityCoherenceOrderer
+  # ===========================================================================
+
+  @orderer
+  Scenario: Order empty fragment list returns empty
+    Given an empty core pipeline fragment list
+    When I order the fragments with PriorityCoherenceOrderer
+    Then 0 core ordered fragments should be returned
+
+  @orderer
+  Scenario: Order single fragment returns it unchanged
+    Given a core pipeline fragment with relevance 0.8
+    When I order the fragments with PriorityCoherenceOrderer
+    Then 1 core ordered fragment should be returned
+
+  @orderer
+  Scenario: Order preserves all fragments
+    Given the following core pipeline fragments:
+      | uko_node                    | content | score | tokens | depth |
+      | project://app/src/main.py   | alpha   | 0.9   | 10     | 3     |
+      | project://app/src/io.py     | beta    | 0.7   | 10     | 3     |
+      | project://lib/util/util.py  | gamma   | 0.5   | 10     | 3     |
+    When I order the fragments with PriorityCoherenceOrderer
+    Then 3 core ordered fragments should be returned
+
+  @orderer
+  Scenario: Order places high-priority fragments first
+    Given the following core pipeline fragments with priorities:
+      | uko_node                  | content | score | tokens | depth | priority |
+      | project://app/src/main.py | alpha   | 0.5   | 10     | 3     | 0.9      |
+      | project://app/src/io.py   | beta    | 0.9   | 10     | 3     | 0.1      |
+    When I order the fragments with PriorityCoherenceOrderer
+    Then the first core ordered fragment should have uko_node "project://app/src/main.py"
+
+  @orderer
+  Scenario: Order groups related fragments together
+    Given the following core pipeline fragments:
+      | uko_node                    | content | score | tokens | depth |
+      | project://app/src/main.py   | alpha   | 0.9   | 10     | 3     |
+      | project://lib/util/util.py  | beta    | 0.8   | 10     | 3     |
+      | project://app/src/io.py     | gamma   | 0.7   | 10     | 3     |
+    When I order the fragments with PriorityCoherenceOrderer
+    Then fragments from "project://app" should be adjacent
+
+  @orderer
+  Scenario: Order with default priority 0.5 falls back to relevance
+    Given the following core pipeline fragments:
+      | uko_node                    | content | score | tokens | depth |
+      | project://app/src/main.py   | alpha   | 0.3   | 10     | 3     |
+      | project://lib/util/util.py  | beta    | 0.9   | 10     | 3     |
+    When I order the fragments with PriorityCoherenceOrderer
+    Then the first core ordered fragment should have uko_node "project://lib/util/util.py"
+
+  # ===========================================================================
+  # Pipeline Integration
+  # ===========================================================================
+
+  @pipeline_integration
+  Scenario: Inject ActorPhaseStrategySelector into pipeline
+    Given the ACMS pipeline with an ActorPhaseStrategySelector
+    And the following core pipeline fragments:
+      | uko_node              | content | score | tokens | depth |
+      | project://app/main.py | hello   | 0.8   | 10     | 3     |
+    When I assemble context through the core pipeline
+    Then the core pipeline output should contain 1 fragment
+
+  @pipeline_integration
+  Scenario: Inject SpecBudgetAllocator into pipeline
+    Given the ACMS pipeline with a SpecBudgetAllocator
+    And the following core pipeline fragments:
+      | uko_node              | content | score | tokens | depth |
+      | project://app/main.py | hello   | 0.8   | 10     | 3     |
+    When I assemble context through the core pipeline
+    Then the core pipeline output should contain 1 fragment
+
+  @pipeline_integration
+  Scenario: Inject RelevanceRecencyPriorityScorer into pipeline
+    Given the ACMS pipeline with a RelevanceRecencyPriorityScorer
+    And the following core pipeline fragments:
+      | uko_node              | content | score | tokens | depth |
+      | project://app/main.py | hello   | 0.8   | 10     | 3     |
+    When I assemble context through the core pipeline
+    Then the core pipeline output fragments should have updated relevance scores
+
+  @pipeline_integration
+  Scenario: Inject ConstrainedKnapsackPacker into pipeline
+    Given the ACMS pipeline with a ConstrainedKnapsackPacker
+    And the following core pipeline fragments:
+      | uko_node              | content | score | tokens | depth |
+      | project://app/main.py | alpha   | 0.9   | 100    | 3     |
+      | project://app/io.py   | beta    | 0.7   | 100    | 3     |
+      | project://app/util.py | gamma   | 0.5   | 100    | 3     |
+    And a core pipeline budget with max_tokens 250 and reserved_tokens 0
+    When I assemble context through the core pipeline with budget
+    Then the core pipeline output total tokens should be at most 250
+
+  @pipeline_integration
+  Scenario: Inject PriorityCoherenceOrderer into pipeline
+    Given the ACMS pipeline with a PriorityCoherenceOrderer
+    And the following core pipeline fragments:
+      | uko_node              | content | score | tokens | depth |
+      | project://app/main.py | hello   | 0.8   | 10     | 3     |
+      | project://lib/util.py | world   | 0.6   | 15     | 5     |
+    When I assemble context through the core pipeline
+    Then the core pipeline output should contain 2 fragments

features/steps/acms_core_pipeline_components_steps.py

@@ -0,0 +1,842 @@
+"""Step definitions for features/acms_core_pipeline_components.feature.
+
+Tests the ACMS core pipeline components directly in-memory.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+
+from behave import given, then, when
+from behave.runner import Context
+
+from cleveragents.application.services.acms_core_pipeline import (
+    ActorPhaseStrategySelector,
+    ConstrainedKnapsackPacker,
+    PriorityCoherenceOrderer,
+    RelevanceRecencyPriorityScorer,
+    SpecBudgetAllocator,
+)
+from cleveragents.application.services.acms_service import (
+    ACMSPipeline,
+    StrategyCapabilities,
+)
+from cleveragents.domain.models.core.context_fragment import (
+    ContextBudget,
+    ContextFragment,
+    FragmentProvenance,
+)
+from cleveragents.domain.models.core.context_policy import ContextView
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+_DEFAULT_PROVENANCE = FragmentProvenance(resource_uri="test://core-pipeline")
+_TEST_PLAN_ID = "01JQTESTCP00000000000000CC"
+
+
+def _make_core_fragment(**kwargs: Any) -> ContextFragment:
+    """Create a ContextFragment with sensible core pipeline test defaults."""
+    kwargs.setdefault("uko_node", "test://default")
+    kwargs.setdefault("content", "test content")
+    kwargs.setdefault("token_count", 10)
+    kwargs.setdefault("provenance", _DEFAULT_PROVENANCE)
+    return ContextFragment(**kwargs)
+
+
+class _MockStrategy:
+    """A simple mock strategy for testing."""
+
+    def __init__(
+        self,
+        name: str,
+        confidence: float,
+        quality_score: float = 1.0,
+    ) -> None:
+        self._name = name
+        self._confidence = confidence
+        self._quality_score = quality_score
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    @property
+    def capabilities(self) -> StrategyCapabilities:
+        return StrategyCapabilities(quality_score=self._quality_score)
+
+    def can_handle(self, request: dict[str, Any]) -> float:
+        return self._confidence
+
+    def assemble(self, fragments: Any, budget: Any) -> list[Any]:
+        return list(fragments)
+
+    def explain(self) -> str:
+        return f"Mock strategy: {self._name}"
+
+
+# ---------------------------------------------------------------------------
+# ActorPhaseStrategySelector — Given steps
+# ---------------------------------------------------------------------------
+
+
+@given("a set of core pipeline strategies")
+def step_given_core_strategies(context: Context) -> None:
+    context.core_strategies = [
+        _MockStrategy("relevance", 0.8),
+        _MockStrategy("tiered", 0.7),
+        _MockStrategy("recency", 0.6),
+        _MockStrategy("arce", 0.95),
+    ]
+
+
+@given("a set of core pipeline strategies including a zero-confidence strategy")
+def step_given_core_strategies_with_zero(context: Context) -> None:
+    context.core_strategies = [
+        _MockStrategy("relevance", 0.8),
+        _MockStrategy("zero-strategy", 0.0),
+    ]
+
+
+@given("an empty strategy list")
+def step_given_empty_strategy_list(context: Context) -> None:
+    context.core_strategies = []
+
+
+@given("a strategy with confidence {conf:g}")
+def step_given_strategy_with_confidence(context: Context, conf: float) -> None:
+    context.core_strategies = [
+        _MockStrategy("relevance", conf),
+    ]
+
+
+# ---------------------------------------------------------------------------
+# ActorPhaseStrategySelector — When steps
+# ---------------------------------------------------------------------------
+
+
+@when("I select strategies with no actor_type or plan_phase")
+def step_select_no_context(context: Context) -> None:
+    selector = ActorPhaseStrategySelector()
+    context.selector_result = selector.select(context.core_strategies, {})
+
+
+@when('I select strategies with actor_type "{actor_type}" only')
+def step_select_with_actor_only(context: Context, actor_type: str) -> None:
+    selector = ActorPhaseStrategySelector()
+    context.selector_result = selector.select(
+        context.core_strategies, {"actor_type": actor_type}
+    )
+
+
+@when('I select strategies with plan_phase "{plan_phase}" only')
+def step_select_with_phase_only(context: Context, plan_phase: str) -> None:
+    selector = ActorPhaseStrategySelector()
+    context.selector_result = selector.select(
+        context.core_strategies, {"plan_phase": plan_phase}
+    )
+
+
+@when(
+    'I select strategies with actor_type "{actor_type}" '
+    'and plan_phase "{plan_phase_val}"'
+)
+def step_select_with_actor_and_phase(
+    context: Context, actor_type: str, plan_phase_val: str
+) -> None:
+    selector = ActorPhaseStrategySelector()
+    context.selector_result = selector.select(
+        context.core_strategies,
+        {"actor_type": actor_type, "plan_phase": plan_phase_val},
+    )
+
+
+# ---------------------------------------------------------------------------
+# ActorPhaseStrategySelector — Then steps
+# ---------------------------------------------------------------------------
+
+
+@then("all core strategies with positive confidence should be returned")
+def step_all_positive_confidence(context: Context) -> None:
+    result_names = {s.name for s, _ in context.selector_result}
+    for strategy in context.core_strategies:
+        if strategy.can_handle({}) > 0.0:
+            assert strategy.name in result_names, f"Expected {strategy.name} in results"
+
+
+@then("the core strategies should be sorted by confidence descending")
+def step_sorted_by_confidence(context: Context) -> None:
+    confidences = [c for _, c in context.selector_result]
+    assert confidences == sorted(confidences, reverse=True), (
+        f"Not sorted: {confidences}"
+    )
+
+
+@then('the "{name}" strategy should have boosted confidence')
+def step_strategy_boosted(context: Context, name: str) -> None:
+    for strategy, confidence in context.selector_result:
+        if strategy.name == name:
+            base = strategy.can_handle({})
+            assert confidence > base, (
+                f"Expected {name} confidence > {base}, got {confidence}"
+            )
+            return
+    msg = f"Strategy {name} not found in results"
+    raise AssertionError(msg)
+
+
+@then('the "{name}" strategy should have maximum boosted confidence')
+def step_strategy_max_boosted(context: Context, name: str) -> None:
+    for strategy, confidence in context.selector_result:
+        if strategy.name == name:
+            base = strategy.can_handle({})
+            assert confidence > base, (
+                f"Expected {name} confidence > {base}, got {confidence}"
+            )
+            assert confidence <= 1.0, f"Confidence {confidence} exceeds 1.0"
+            return
+    msg = f"Strategy {name} not found in results"
+    raise AssertionError(msg)
+
+
+@then("the core zero-confidence strategy should not be in the results")
+def step_zero_confidence_excluded(context: Context) -> None:
+    result_names = {s.name for s, _ in context.selector_result}
+    assert "zero-strategy" not in result_names, (
+        "Zero-confidence strategy should be excluded"
+    )
+
+
+@then("{count:d} strategies should be selected")
+def step_strategies_count(context: Context, count: int) -> None:
+    assert len(context.selector_result) == count, (
+        f"Expected {count}, got {len(context.selector_result)}"
+    )
+
+
+@then("the strategy confidence should not exceed 1.0")
+def step_confidence_not_exceed_one(context: Context) -> None:
+    for _, confidence in context.selector_result:
+        assert confidence <= 1.0, f"Confidence {confidence} exceeds 1.0"
+
+
+# ---------------------------------------------------------------------------
+# SpecBudgetAllocator — Given steps
+# ---------------------------------------------------------------------------
+
+
+@given("an empty candidate list")
+def step_given_empty_candidates(context: Context) -> None:
+    context.budget_candidates: list[tuple[Any, float]] = []
+
+
+@given("a single candidate with confidence {conf:g}")
+def step_given_single_candidate(context: Context, conf: float) -> None:
+    context.budget_candidates = [(_MockStrategy("relevance", conf), conf)]
+
+
+@given("two candidates with equal confidence {conf:g}")
+def step_given_two_equal_candidates(context: Context, conf: float) -> None:
+    context.budget_candidates = [
+        (_MockStrategy("relevance", conf), conf),
+        (_MockStrategy("tiered", conf), conf),
+    ]
+
+
+@given("two candidates with different quality scores")
+def step_given_two_different_quality(context: Context) -> None:
+    context.budget_candidates = [
+        (_MockStrategy("high-quality", 0.8, quality_score=0.9), 0.8),
+        (_MockStrategy("low-quality", 0.8, quality_score=0.1), 0.8),
+    ]
+
+
+@given("two candidates with zero confidence")
+def step_given_two_zero_confidence(context: Context) -> None:
+    context.budget_candidates = [
+        (_MockStrategy("a", 0.0), 0.0),
+        (_MockStrategy("b", 0.0), 0.0),
+    ]
+
+
+@given("three candidates where one would receive very few tokens")
+def step_given_three_candidates_one_small(context: Context) -> None:
+    context.budget_candidates = [
+        (_MockStrategy("big-a", 0.9), 0.9),
+        (_MockStrategy("big-b", 0.8), 0.8),
+        (_MockStrategy("tiny", 0.01), 0.01),
+    ]
+
+
+@given("three candidates with varying confidence")
+def step_given_three_varying_candidates(context: Context) -> None:
+    context.budget_candidates = [
+        (_MockStrategy("a", 0.9), 0.9),
+        (_MockStrategy("b", 0.6), 0.6),
+        (_MockStrategy("c", 0.3), 0.3),
+    ]
+
+
+# ---------------------------------------------------------------------------
+# SpecBudgetAllocator — When steps
+# ---------------------------------------------------------------------------
+
+
+@when("I allocate budget {budget:d} with SpecBudgetAllocator")
+def step_allocate_budget(context: Context, budget: int) -> None:
+    allocator = SpecBudgetAllocator()
+    context.allocation_result = allocator.allocate(context.budget_candidates, budget)
+
+
+@when(
+    "I allocate budget {budget:d} with SpecBudgetAllocator "
+    "and min_useful_budget {min_budget:d}"
+)
+def step_allocate_budget_with_min(
+    context: Context, budget: int, min_budget: int
+) -> None:
+    allocator = SpecBudgetAllocator(min_useful_budget=min_budget)
+    context.allocation_result = allocator.allocate(context.budget_candidates, budget)
+
+
+# ---------------------------------------------------------------------------
+# SpecBudgetAllocator — Then steps
+# ---------------------------------------------------------------------------
+
+
+@then("{count:d} allocation should be returned")
+def step_allocations_count_singular(context: Context, count: int) -> None:
+    assert len(context.allocation_result) == count, (
+        f"Expected {count}, got {len(context.allocation_result)}"
+    )
+
+
+@then("{count:d} allocations should be returned")
+def step_allocations_count_plural(context: Context, count: int) -> None:
+    assert len(context.allocation_result) == count, (
+        f"Expected {count}, got {len(context.allocation_result)}"
+    )
+
+
+@then("the single allocation should receive {tokens:d} tokens")
+def step_single_allocation_tokens(context: Context, tokens: int) -> None:
+    assert len(context.allocation_result) == 1
+    _, _, allocated = context.allocation_result[0]
+    assert allocated == tokens, f"Expected {tokens}, got {allocated}"
+
+
+@then("the total allocated tokens should equal {total:d}")
+def step_total_allocated_tokens(context: Context, total: int) -> None:
+    actual = sum(t for _, _, t in context.allocation_result)
+    assert actual == total, f"Expected {total}, got {actual}"
+
+
+@then("the higher quality candidate should receive more tokens")
+def step_higher_quality_more_tokens(context: Context) -> None:
+    assert len(context.allocation_result) == 2
+    _, _, high_tokens = context.allocation_result[0]
+    _, _, low_tokens = context.allocation_result[1]
+    assert high_tokens > low_tokens, (
+        f"Expected high-quality ({high_tokens}) > low-quality ({low_tokens})"
+    )
+
+
+@then("the small candidate should be excluded from allocations")
+def step_small_candidate_excluded(context: Context) -> None:
+    names = {s.name for s, _, _ in context.allocation_result}
+    assert "tiny" not in names, f"Expected 'tiny' to be excluded, got {names}"
+
+
+# ---------------------------------------------------------------------------
+# RelevanceRecencyPriorityScorer — Given steps
+# ---------------------------------------------------------------------------
+
+
+@given("an empty core pipeline fragment list")
+def step_given_empty_core_fragments(context: Context) -> None:
+    context.core_fragments: list[ContextFragment] = []
+
+
+@given("a core pipeline fragment with relevance {rel:g}")
+def step_given_core_fragment_relevance(context: Context, rel: float) -> None:
+    context.core_fragments = [_make_core_fragment(relevance_score=rel)]
+
+
+@given("a core pipeline fragment with relevance {rel:g} and priority {pri:g}")
+def step_given_core_fragment_relevance_priority(
+    context: Context, rel: float, pri: float
+) -> None:
+    context.core_fragments = [
+        _make_core_fragment(
+            relevance_score=rel,
+            metadata={"priority": str(pri)},
+        )
+    ]
+
+
+@given("two identical core pipeline fragments")
+def step_given_two_identical_fragments(context: Context) -> None:
+    ts = datetime(2024, 1, 1, tzinfo=UTC)
+    frag = _make_core_fragment(relevance_score=0.7, created_at=ts)
+    context.core_fragments = [frag]
+    context.core_fragments2 = [frag]
+
+
+@given("two core pipeline fragments with different timestamps")
+def step_given_two_different_timestamps(context: Context) -> None:
+    ts_old = datetime(2024, 1, 1, tzinfo=UTC)
+    ts_new = datetime(2024, 6, 1, tzinfo=UTC)
+    context.core_fragments = [
+        _make_core_fragment(
+            uko_node="test://old",
+            relevance_score=0.5,
+            created_at=ts_old,
+        ),
+        _make_core_fragment(
+            uko_node="test://new",
+            relevance_score=0.5,
+            created_at=ts_new,
+        ),
+    ]
+
+
+@given("two core pipeline fragments with identical timestamps")
+def step_given_two_identical_timestamps(context: Context) -> None:
+    ts = datetime(2024, 1, 1, tzinfo=UTC)
+    context.core_fragments = [
+        _make_core_fragment(
+            uko_node="test://a",
+            relevance_score=0.5,
+            created_at=ts,
+        ),
+        _make_core_fragment(
+            uko_node="test://b",
+            relevance_score=0.7,
+            created_at=ts,
+        ),
+    ]
+
+
+@given("a core pipeline fragment with maximum relevance and priority")
+def step_given_max_relevance_priority(context: Context) -> None:
+    context.core_fragments = [
+        _make_core_fragment(
+            relevance_score=1.0,
+            metadata={"priority": "1.0"},
+        )
+    ]
+
+
+@given("a core pipeline fragment with zero relevance and zero priority")
+def step_given_zero_relevance_priority(context: Context) -> None:
+    context.core_fragments = [
+        _make_core_fragment(
+            relevance_score=0.0,
+            metadata={"priority": "0.0"},
+        )
+    ]
+
+
+@given(
+    "scorer configured with relevance_weight {rel:g} recency_weight {rec:g} "
+    "priority_weight {pri:g}"
+)
+def step_given_custom_scorer_weights(
+    context: Context, rel: float, rec: float, pri: float
+) -> None:
+    context.custom_scorer = RelevanceRecencyPriorityScorer(
+        relevance_weight=rel,
+        recency_weight=rec,
+        priority_weight=pri,
+    )
+
+
+@given('a core pipeline fragment with metadata priority "{priority}"')
+def step_given_fragment_with_priority_metadata(context: Context, priority: str) -> None:
+    context.core_fragments = [
+        _make_core_fragment(
+            relevance_score=0.5,
+            metadata={"priority": priority},
+        )
+    ]
+
+
+@given("a core pipeline fragment with no priority metadata")
+def step_given_fragment_no_priority(context: Context) -> None:
+    context.core_fragments = [_make_core_fragment(relevance_score=0.5)]
+
+
+# ---------------------------------------------------------------------------
+# RelevanceRecencyPriorityScorer — When steps
+# ---------------------------------------------------------------------------
+
+
+@when("I score the fragments with RelevanceRecencyPriorityScorer")
+def step_score_with_rrp_scorer(context: Context) -> None:
+    scorer = RelevanceRecencyPriorityScorer()
+    context.core_scored = list(scorer.score(context.core_fragments))
+
+
+@when("I score both fragment sets with RelevanceRecencyPriorityScorer")
+def step_score_both_sets(context: Context) -> None:
+    scorer = RelevanceRecencyPriorityScorer()
+    context.core_scored = list(scorer.score(context.core_fragments))
+    context.core_scored2 = list(scorer.score(context.core_fragments2))
+
+
+@when("I score the fragments with custom RelevanceRecencyPriorityScorer")
+def step_score_with_custom_scorer(context: Context) -> None:
+    scorer = context.custom_scorer
+    context.core_scored = list(scorer.score(context.core_fragments))
+
+
+# ---------------------------------------------------------------------------
+# RelevanceRecencyPriorityScorer — Then steps
+# ---------------------------------------------------------------------------
+
+
+@then("{count:d} core scored fragments should be returned")
+def step_scored_count_core(context: Context, count: int) -> None:
+    assert len(context.core_scored) == count, (
+        f"Expected {count}, got {len(context.core_scored)}"
+    )
+
+
+@then('the scored fragment should have _original_relevance metadata "{val}"')
+def step_scored_original_relevance_core(context: Context, val: str) -> None:
+    assert len(context.core_scored) >= 1
+    meta = context.core_scored[0].metadata
+    assert meta.get("_original_relevance") == val, (
+        f"Expected {val}, got {meta.get('_original_relevance')}"
+    )
+
+
+@then("the scored fragment composite score should be between 0.0 and 1.0")
+def step_scored_composite_range(context: Context) -> None:
+    assert len(context.core_scored) >= 1
+    score = context.core_scored[0].relevance_score
+    assert 0.0 <= score <= 1.0, f"Score {score} out of range"
+
+
+@then("both scored fragments should have identical composite scores")
+def step_scored_identical(context: Context) -> None:
+    assert len(context.core_scored) >= 1
+    assert len(context.core_scored2) >= 1
+    assert (
+        context.core_scored[0].relevance_score
+        == context.core_scored2[0].relevance_score
+    )
+
+
+@then("the newer fragment should have higher recency score")
+def step_newer_higher_recency(context: Context) -> None:
+    assert len(context.core_scored) == 2
+    old_recency = float(context.core_scored[0].metadata.get("_score_recency", "0"))
+    new_recency = float(context.core_scored[1].metadata.get("_score_recency", "0"))
+    assert new_recency > old_recency, (
+        f"Expected new ({new_recency}) > old ({old_recency})"
+    )
+
+
+@then("all scored fragments should have recency score 1.0")
+def step_all_recency_one(context: Context) -> None:
+    for frag in context.core_scored:
+        recency = float(frag.metadata.get("_score_recency", "0"))
+        assert abs(recency - 1.0) < 1e-6, f"Expected recency 1.0, got {recency}"
+
+
+@then("the scored fragment composite score should be at most 1.0")
+def step_scored_at_most_one(context: Context) -> None:
+    assert len(context.core_scored) >= 1
+    assert context.core_scored[0].relevance_score <= 1.0
+
+
+@then("the scored fragment composite score should be at least 0.0")
+def step_scored_at_least_zero(context: Context) -> None:
+    assert len(context.core_scored) >= 1
+    assert context.core_scored[0].relevance_score >= 0.0
+
+
+@then("the scored fragment composite score should be {expected:g}")
+def step_scored_exact_composite(context: Context, expected: float) -> None:
+    assert len(context.core_scored) >= 1
+    actual = context.core_scored[0].relevance_score
+    assert abs(actual - expected) < 1e-4, f"Expected {expected}, got {actual}"
+
+
+@then('the scored fragment should have _score_priority metadata near "{val}"')
+def step_scored_priority_near(context: Context, val: str) -> None:
+    assert len(context.core_scored) >= 1
+    meta = context.core_scored[0].metadata
+    actual = float(meta.get("_score_priority", "0"))
+    expected = float(val)
+    assert abs(actual - expected) < 0.01, f"Expected ~{expected}, got {actual}"
+
+
+@then('the scored fragment should have _score_priority metadata "{val}"')
+def step_scored_priority_exact(context: Context, val: str) -> None:
+    assert len(context.core_scored) >= 1
+    meta = context.core_scored[0].metadata
+    actual = float(meta.get("_score_priority", "0"))
+    expected = float(val)
+    assert abs(actual - expected) < 0.01, f"Expected {expected}, got {actual}"
+
+
+# ---------------------------------------------------------------------------
+# ConstrainedKnapsackPacker — Given steps
+# ---------------------------------------------------------------------------
+
+
+@given("the following core pipeline fragments:")
+def step_given_core_pipeline_fragments(context: Context) -> None:
+    frags: list[ContextFragment] = []
+    for row in context.table:
+        frags.append(
+            _make_core_fragment(
+                uko_node=row["uko_node"],
+                content=row["content"],
+                relevance_score=float(row["score"]),
+                token_count=int(row["tokens"]),
+                detail_depth=int(row["depth"]),
+            )
+        )
+    context.core_fragments = frags
+
+
+@given("the following core pipeline fragments with priorities:")
+def step_given_core_fragments_with_priorities(context: Context) -> None:
+    frags: list[ContextFragment] = []
+    for row in context.table:
+        frags.append(
+            _make_core_fragment(
+                uko_node=row["uko_node"],
+                content=row["content"],
+                relevance_score=float(row["score"]),
+                token_count=int(row["tokens"]),
+                detail_depth=int(row["depth"]),
+                metadata={"priority": row["priority"]},
+            )
+        )
+    context.core_fragments = frags
+
+
+@given("a core pipeline budget with max_tokens {max_t:d} and reserved_tokens {res_t:d}")
+def step_given_core_budget(context: Context, max_t: int, res_t: int) -> None:
+    context.core_budget = ContextBudget(max_tokens=max_t, reserved_tokens=res_t)
+
+
+@given("a ConstrainedKnapsackPacker with max_file_size {max_size:d}")
+def step_given_packer_max_file_size(context: Context, max_size: int) -> None:
+    context.core_packer = ConstrainedKnapsackPacker(max_file_size=max_size)
+
+
+@given("a ConstrainedKnapsackPacker with max_total_size {max_size:d}")
+def step_given_packer_max_total_size(context: Context, max_size: int) -> None:
+    context.core_packer = ConstrainedKnapsackPacker(max_total_size=max_size)
+
+
+@given("a ConstrainedKnapsackPacker with context_view max_total_size {max_size:d}")
+def step_given_packer_context_view(context: Context, max_size: int) -> None:
+    view = ContextView(max_total_size=max_size)
+    context.core_packer = ConstrainedKnapsackPacker(context_view=view)
+
+
+# ---------------------------------------------------------------------------
+# ConstrainedKnapsackPacker — When steps
+# ---------------------------------------------------------------------------
+
+
+@when("I pack the fragments with ConstrainedKnapsackPacker")
+def step_pack_with_constrained_packer(context: Context) -> None:
+    packer = getattr(context, "core_packer", ConstrainedKnapsackPacker())
+    budget = getattr(
+        context, "core_budget", ContextBudget(max_tokens=100000, reserved_tokens=0)
+    )
+    context.core_packed = list(packer.pack(context.core_fragments, budget))
+
+
+# ---------------------------------------------------------------------------
+# ConstrainedKnapsackPacker — Then steps
+# ---------------------------------------------------------------------------
+
+
+@then("{count:d} core packed fragment should be returned")
+def step_packed_count_singular_core(context: Context, count: int) -> None:
+    assert len(context.core_packed) == count, (
+        f"Expected {count}, got {len(context.core_packed)}"
+    )
+
+
+@then("{count:d} core packed fragments should be returned")
+def step_packed_count_plural_core(context: Context, count: int) -> None:
+    assert len(context.core_packed) == count, (
+        f"Expected {count}, got {len(context.core_packed)}"
+    )
+
+
+@then("the core packed total tokens should be at most {max_tokens:d}")
+def step_packed_total_at_most_core(context: Context, max_tokens: int) -> None:
+    total = sum(f.token_count for f in context.core_packed)
+    assert total <= max_tokens, f"Expected <= {max_tokens}, got {total}"
+
+
+@then('the core packed fragment uko_node should be "{expected}"')
+def step_packed_single_node_core(context: Context, expected: str) -> None:
+    assert len(context.core_packed) >= 1
+    assert context.core_packed[0].uko_node == expected, (
+        f"Expected {expected}, got {context.core_packed[0].uko_node}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# PriorityCoherenceOrderer — When steps
+# ---------------------------------------------------------------------------
+
+
+@when("I order the fragments with PriorityCoherenceOrderer")
+def step_order_with_priority_orderer(context: Context) -> None:
+    orderer = PriorityCoherenceOrderer()
+    context.core_ordered = list(orderer.order(context.core_fragments))
+
+
+# ---------------------------------------------------------------------------
+# PriorityCoherenceOrderer — Then steps
+# ---------------------------------------------------------------------------
+
+
+@then("{count:d} core ordered fragment should be returned")
+def step_ordered_count_singular(context: Context, count: int) -> None:
+    assert len(context.core_ordered) == count, (
+        f"Expected {count}, got {len(context.core_ordered)}"
+    )
+
+
+@then("{count:d} core ordered fragments should be returned")
+def step_ordered_count_plural(context: Context, count: int) -> None:
+    assert len(context.core_ordered) == count, (
+        f"Expected {count}, got {len(context.core_ordered)}"
+    )
+
+
+@then('the first core ordered fragment should have uko_node "{expected}"')
+def step_first_ordered_node(context: Context, expected: str) -> None:
+    assert len(context.core_ordered) >= 1
+    actual = context.core_ordered[0].uko_node
+    assert actual == expected, f"Expected {expected}, got {actual}"
+
+
+@then('fragments from "{prefix}" should be adjacent')
+def step_fragments_adjacent(context: Context, prefix: str) -> None:
+    matching_indices = [
+        i for i, f in enumerate(context.core_ordered) if f.uko_node.startswith(prefix)
+    ]
+    if len(matching_indices) <= 1:
+        return
+    for i in range(len(matching_indices) - 1):
+        assert matching_indices[i + 1] == matching_indices[i] + 1, (
+            f"Fragments from {prefix} are not adjacent: {matching_indices}"
+        )
+
+
+# ---------------------------------------------------------------------------
+# Pipeline Integration — Given steps
+# ---------------------------------------------------------------------------
+
+
+@given("the ACMS pipeline with an ActorPhaseStrategySelector")
+def step_pipeline_with_actor_selector(context: Context) -> None:
+    context.core_pipeline = ACMSPipeline(
+        strategy_selector=ActorPhaseStrategySelector(),
+    )
+
+
+@given("the ACMS pipeline with a SpecBudgetAllocator")
+def step_pipeline_with_spec_allocator(context: Context) -> None:
+    context.core_pipeline = ACMSPipeline(
+        budget_allocator=SpecBudgetAllocator(),
+    )
+
+
+@given("the ACMS pipeline with a RelevanceRecencyPriorityScorer")
+def step_pipeline_with_rrp_scorer(context: Context) -> None:
+    context.core_pipeline = ACMSPipeline(
+        scorer=RelevanceRecencyPriorityScorer(),
+    )
+
+
+@given("the ACMS pipeline with a ConstrainedKnapsackPacker")
+def step_pipeline_with_constrained_packer(context: Context) -> None:
+    context.core_pipeline = ACMSPipeline(
+        packer=ConstrainedKnapsackPacker(),
+    )
+
+
+@given("the ACMS pipeline with a PriorityCoherenceOrderer")
+def step_pipeline_with_priority_orderer(context: Context) -> None:
+    context.core_pipeline = ACMSPipeline(
+        orderer=PriorityCoherenceOrderer(),
+    )
+
+
+# ---------------------------------------------------------------------------
+# Pipeline Integration — When steps
+# ---------------------------------------------------------------------------
+
+
+@when("I assemble context through the core pipeline")
+def step_assemble_core_pipeline(context: Context) -> None:
+    budget = getattr(
+        context, "core_budget", ContextBudget(max_tokens=100000, reserved_tokens=0)
+    )
+    payload = context.core_pipeline.assemble(
+        plan_id=_TEST_PLAN_ID,
+        fragments=context.core_fragments,
+        budget=budget,
+    )
+    context.core_payload = payload
+
+
+@when("I assemble context through the core pipeline with budget")
+def step_assemble_core_pipeline_with_budget(context: Context) -> None:
+    payload = context.core_pipeline.assemble(
+        plan_id=_TEST_PLAN_ID,
+        fragments=context.core_fragments,
+        budget=context.core_budget,
+    )
+    context.core_payload = payload
+
+
+# ---------------------------------------------------------------------------
+# Pipeline Integration — Then steps
+# ---------------------------------------------------------------------------
+
+
+@then("the core pipeline output should contain {count:d} fragment")
+def step_core_pipeline_count_singular(context: Context, count: int) -> None:
+    assert len(context.core_payload.fragments) == count, (
+        f"Expected {count}, got {len(context.core_payload.fragments)}"
+    )
+
+
+@then("the core pipeline output should contain {count:d} fragments")
+def step_core_pipeline_count_plural(context: Context, count: int) -> None:
+    assert len(context.core_payload.fragments) == count, (
+        f"Expected {count}, got {len(context.core_payload.fragments)}"
+    )
+
+
+@then("the core pipeline output fragments should have updated relevance scores")
+def step_core_pipeline_scorer_applied(context: Context) -> None:
+    for frag in context.core_payload.fragments:
+        assert "_original_relevance" in frag.metadata, (
+            f"Fragment {frag.uko_node} missing _original_relevance metadata"
+        )
+
+
+@then("the core pipeline output total tokens should be at most {max_tokens:d}")
+def step_core_pipeline_total_at_most(context: Context, max_tokens: int) -> None:
+    assert context.core_payload.total_tokens <= max_tokens, (
+        f"Expected <= {max_tokens}, got {context.core_payload.total_tokens}"
+    )

src/cleveragents/application/services/acms_core_pipeline.py

@@ -0,0 +1,580 @@
+"""ACMS core pipeline component implementations.
+
+Provides the five core ACMS pipeline components that form the central
+assembly logic of the context assembly pipeline:
+
+1. **ActorPhaseStrategySelector** -- Selects context strategies based on
+   actor type and plan phase from the request context.  Applies actor-type
+   and phase-specific confidence boosts to guide strategy selection.
+
+2. **SpecBudgetAllocator** -- Computes per-strategy token budgets using the
+   spec formula: ``budget_i = total_budget * (confidence_i * quality_i) /
+   sum(confidence_j * quality_j)``.  Handles edge cases including empty
+   candidates, zero total weight, and single-candidate scenarios.
+
+3. **RelevanceRecencyPriorityScorer** -- Scores context fragments by a
+   weighted composite of relevance, recency (normalised from ``created_at``),
+   and priority (from fragment metadata).  Produces deterministic scores for
+   identical inputs.
+
+4. **ConstrainedKnapsackPacker** -- Greedy knapsack algorithm that packs
+   fragments within the token budget while respecting ``max_file_size`` and
+   ``max_total_size`` byte-size constraints from a ``ContextView``.
+
+5. **PriorityCoherenceOrderer** -- Orders packed fragments for optimal LLM
+   consumption by grouping related fragments (same UKO node prefix) and
+   ordering groups by priority then relevance.
+
+All components satisfy the Protocol interfaces defined in ``acms_service.py``
+and can be injected into ``ACMSPipeline`` via constructor dependency injection.
+
+Based on ``docs/specification.md`` SS42630-42653.
+
+ISSUES CLOSED: #10015
+"""
+
+from __future__ import annotations
+
+import logging
+from collections import defaultdict
+from collections.abc import Sequence
+from datetime import datetime
+from typing import Any, Final
+
+from cleveragents.application.services.acms_service import (
+    ContextStrategy,
+)
+from cleveragents.domain.models.core.context_fragment import (
+    ContextBudget,
+    ContextFragment,
+)
+from cleveragents.domain.models.core.context_policy import ContextView
+
+logger = logging.getLogger(__name__)
+
+# ---------------------------------------------------------------------------
+# Configuration constants
+# ---------------------------------------------------------------------------
+
+# Default scoring weights for RelevanceRecencyPriorityScorer
+DEFAULT_RELEVANCE_WEIGHT: Final[float] = 0.5
+DEFAULT_RECENCY_WEIGHT: Final[float] = 0.3
+DEFAULT_PRIORITY_WEIGHT: Final[float] = 0.2
+
+# Minimum token count for a fragment to be eligible for packing
+DEFAULT_MIN_FRAGMENT_TOKENS: Final[int] = 1
+
+# Actor-type strategy preference mappings
+_ACTOR_STRATEGY_PREFERENCES: Final[dict[str, list[str]]] = {
+    "planner": ["relevance", "arce", "tiered"],
+    "executor": ["tiered", "relevance", "recency"],
+    "reviewer": ["relevance", "recency", "arce"],
+    "analyst": ["arce", "temporal-archaeology", "relevance"],
+    "corrector": ["plan-decision-context", "relevance", "recency"],
+}
+
+# Plan-phase strategy preference mappings
+_PHASE_STRATEGY_PREFERENCES: Final[dict[str, list[str]]] = {
+    "strategize": ["relevance", "arce"],
+    "execute": ["tiered", "relevance"],
+    "apply": ["tiered", "recency"],
+    "review": ["relevance", "recency"],
+    "correct": ["plan-decision-context", "relevance"],
+}
+
+# Confidence boost applied to preferred strategies
+_PREFERENCE_BOOST: Final[float] = 0.2
+
+
+# ---------------------------------------------------------------------------
+# 1. ActorPhaseStrategySelector  (StrategySelector protocol)
+# ---------------------------------------------------------------------------
+
+
+class ActorPhaseStrategySelector:
+    """Select context strategies based on actor type and plan phase.
+
+    Evaluates all strategies via ``can_handle`` and applies confidence
+    boosts to strategies preferred by the current actor type and plan
+    phase.  The request dict may contain:
+
+    - ``actor_type``: The type of actor making the request.
+    - ``plan_phase``: The current plan phase.
+
+    Implements ``StrategySelector`` protocol from ``acms_service.py``.
+
+    Based on ``docs/specification.md`` SS42630.
+    """
+
+    def __init__(
+        self,
+        *,
+        preference_boost: float = _PREFERENCE_BOOST,
+    ) -> None:
+        self._preference_boost = preference_boost
+
+    @property
+    def preference_boost(self) -> float:
+        """Return the configured preference boost value."""
+        return self._preference_boost
+
+    def select(
+        self,
+        strategies: Sequence[ContextStrategy],
+        request: dict[str, Any],
+    ) -> list[tuple[ContextStrategy, float]]:
+        """Return (strategy, confidence) pairs sorted by priority."""
+        actor_type: str = request.get("actor_type", "")
+        plan_phase: str = request.get("plan_phase", "")
+
+        actor_preferred: list[str] = _ACTOR_STRATEGY_PREFERENCES.get(actor_type, [])
+        phase_preferred: list[str] = _PHASE_STRATEGY_PREFERENCES.get(plan_phase, [])
+
+        scored: list[tuple[ContextStrategy, float]] = []
+
+        for strategy in strategies:
+            base_confidence = strategy.can_handle(request)
+            if base_confidence <= 0.0:
+                continue
+
+            effective = base_confidence
+
+            if strategy.name in actor_preferred:
+                effective = min(effective + self._preference_boost, 1.0)
+
+            if strategy.name in phase_preferred:
+                effective = min(effective + self._preference_boost, 1.0)
+
+            scored.append((strategy, effective))
+
+        scored.sort(key=lambda x: x[1], reverse=True)
+
+        logger.debug(
+            "ActorPhaseStrategySelector: selected %d strategies",
+            len(scored),
+        )
+        return scored
+
+
+# ---------------------------------------------------------------------------
+# 2. SpecBudgetAllocator  (BudgetAllocator protocol)
+# ---------------------------------------------------------------------------
+
+
+class SpecBudgetAllocator:
+    """Compute per-strategy token budgets using the spec formula.
+
+    Implements the spec budget allocation formula:
+
+        budget_i = total_budget * (confidence_i * quality_i) /
+                   sum(confidence_j * quality_j for all j)
+
+    Implements ``BudgetAllocator`` protocol from ``acms_service.py``.
+
+    Based on ``docs/specification.md`` SS42632.
+    """
+
+    def __init__(self, *, min_useful_budget: int = 0) -> None:
+        self._min_useful_budget = min_useful_budget
+
+    @property
+    def min_useful_budget(self) -> int:
+        """Return the minimum useful budget threshold."""
+        return self._min_useful_budget
+
+    @staticmethod
+    def _weight(strategy: ContextStrategy, confidence: float) -> float:
+        """Compute allocation weight: confidence * quality_score."""
+        quality: float = getattr(strategy.capabilities, "quality_score", 1.0)
+        return confidence * quality
+
+    def allocate(
+        self,
+        candidates: list[tuple[ContextStrategy, float]],
+        total_budget: int,
+        request: Any = None,
+    ) -> list[tuple[ContextStrategy, float, int]]:
+        """Return (strategy, confidence, allocated_tokens) triples."""
+        if not candidates:
+            return []
+
+        if len(candidates) == 1:
+            strategy, confidence = candidates[0]
+            return [(strategy, confidence, total_budget)]
+
+        working = self._filter_by_min_budget(candidates, total_budget)
+        if not working:
+            best = max(candidates, key=lambda x: x[1])
+            return [(best[0], best[1], total_budget)]
+
+        total_weight = sum(self._weight(s, c) for s, c in working)
+        if total_weight <= 0.0:
+            return self._equal_split(working, total_budget)
+
+        return self._proportional_split(working, total_budget, total_weight)
+
+    def _filter_by_min_budget(
+        self,
+        candidates: list[tuple[ContextStrategy, float]],
+        total_budget: int,
+    ) -> list[tuple[ContextStrategy, float]]:
+        """Remove candidates whose proportional share < min_useful_budget."""
+        if self._min_useful_budget <= 0:
+            return list(candidates)
+
+        total_weight = sum(self._weight(s, c) for s, c in candidates)
+        if total_weight <= 0.0:
+            return list(candidates)
+
+        kept: list[tuple[ContextStrategy, float]] = []
+        for strategy, confidence in candidates:
+            w = self._weight(strategy, confidence)
+            share = int(total_budget * w / total_weight)
+            if share >= self._min_useful_budget:
+                kept.append((strategy, confidence))
+
+        return kept if kept else list(candidates)
+
+    def _equal_split(
+        self,
+        candidates: list[tuple[ContextStrategy, float]],
+        total_budget: int,
+    ) -> list[tuple[ContextStrategy, float, int]]:
+        """Split budget equally with largest-remainder distribution."""
+        n = len(candidates)
+        share = total_budget // n
+        remainder = total_budget - share * n
+        return [
+            (s, c, share + (1 if i < remainder else 0))
+            for i, (s, c) in enumerate(candidates)
+        ]
+
+    def _proportional_split(
+        self,
+        candidates: list[tuple[ContextStrategy, float]],
+        total_budget: int,
+        total_weight: float,
+    ) -> list[tuple[ContextStrategy, float, int]]:
+        """Proportional allocation with largest-remainder rounding."""
+        weights = [self._weight(s, c) for s, c in candidates]
+        raw = [total_budget * w / total_weight for w in weights]
+        floors = [int(r) for r in raw]
+        remainder = total_budget - sum(floors)
+
+        fractions = sorted(
+            ((r - f, i) for i, (r, f) in enumerate(zip(raw, floors, strict=True))),
+            reverse=True,
+        )
+        for _, i in fractions[:remainder]:
+            floors[i] += 1
+
+        return [(s, c, floors[i]) for i, (s, c) in enumerate(candidates)]
+
+
+# ---------------------------------------------------------------------------
+# 3. RelevanceRecencyPriorityScorer  (FragmentScorer protocol)
+# ---------------------------------------------------------------------------
+
+
+class RelevanceRecencyPriorityScorer:
+    """Score context fragments by relevance, recency, and priority.
+
+    Computes a weighted composite score:
+
+        composite = (relevance_weight * relevance_score
+                     + recency_weight * recency_norm
+                     + priority_weight * priority)
+
+    Produces deterministic scores for identical inputs.
+
+    Implements ``FragmentScorer`` protocol from ``acms_service.py``.
+
+    Based on ``docs/specification.md`` SS42636.
+    """
+
+    def __init__(
+        self,
+        *,
+        relevance_weight: float = DEFAULT_RELEVANCE_WEIGHT,
+        recency_weight: float = DEFAULT_RECENCY_WEIGHT,
+        priority_weight: float = DEFAULT_PRIORITY_WEIGHT,
+    ) -> None:
+        self._relevance_weight = relevance_weight
+        self._recency_weight = recency_weight
+        self._priority_weight = priority_weight
+
+    @property
+    def relevance_weight(self) -> float:
+        """Return the configured relevance weight."""
+        return self._relevance_weight
+
+    @property
+    def recency_weight(self) -> float:
+        """Return the configured recency weight."""
+        return self._recency_weight
+
+    @property
+    def priority_weight(self) -> float:
+        """Return the configured priority weight."""
+        return self._priority_weight
+
+    def score(
+        self,
+        fragments: Sequence[ContextFragment],
+    ) -> Sequence[ContextFragment]:
+        """Return fragments re-scored with composite relevance scores."""
+        if not fragments:
+            return list(fragments)
+
+        timestamps = [f.created_at for f in fragments]
+        min_ts = min(timestamps)
+        max_ts = max(timestamps)
+        ts_range = (max_ts - min_ts).total_seconds()
+
+        scored: list[ContextFragment] = []
+        for frag in fragments:
+            recency_norm = self._recency_norm(frag.created_at, min_ts, ts_range)
+            priority = self._extract_priority(frag)
+            composite = self._compute_composite(
+                frag.relevance_score, recency_norm, priority
+            )
+
+            new_meta = dict(frag.metadata)
+            new_meta["_original_relevance"] = str(frag.relevance_score)
+            new_meta["_score_relevance"] = str(round(frag.relevance_score, 6))
+            new_meta["_score_recency"] = str(round(recency_norm, 6))
+            new_meta["_score_priority"] = str(round(priority, 6))
+
+            scored.append(
+                ContextFragment(
+                    fragment_id=frag.fragment_id,
+                    uko_node=frag.uko_node,
+                    content=frag.content,
+                    detail_depth=frag.detail_depth,
+                    token_count=frag.token_count,
+                    relevance_score=composite,
+                    provenance=frag.provenance,
+                    strategy_source=frag.strategy_source,
+                    tier=frag.tier,
+                    metadata=new_meta,
+                    created_at=frag.created_at,
+                )
+            )
+
+        return scored
+
+    def _recency_norm(
+        self,
+        created_at: datetime,
+        min_ts: datetime,
+        ts_range: float,
+    ) -> float:
+        """Compute normalised recency score in [0.0, 1.0]."""
+        if ts_range <= 0.0:
+            return 1.0
+        elapsed = (created_at - min_ts).total_seconds()
+        return elapsed / ts_range
+
+    def _extract_priority(self, frag: ContextFragment) -> float:
+        """Extract priority from fragment metadata, defaulting to 0.5."""
+        raw = frag.metadata.get("priority", "0.5")
+        try:
+            value = float(raw)
+            return max(0.0, min(1.0, value))
+        except (ValueError, TypeError):
+            return 0.5
+
+    def _compute_composite(
+        self,
+        relevance: float,
+        recency: float,
+        priority: float,
+    ) -> float:
+        """Compute weighted composite score clamped to [0.0, 1.0]."""
+        raw = (
+            self._relevance_weight * relevance
+            + self._recency_weight * recency
+            + self._priority_weight * priority
+        )
+        return max(0.0, min(1.0, round(raw, 6)))
+
+
+# ---------------------------------------------------------------------------
+# 4. ConstrainedKnapsackPacker  (BudgetPacker protocol)
+# ---------------------------------------------------------------------------
+
+
+class ConstrainedKnapsackPacker:
+    """Pack fragments within budget respecting max_file_size and max_total_size.
+
+    Extends the greedy knapsack algorithm with byte-size constraints
+    from a ``ContextView``.
+
+    Implements ``BudgetPacker`` protocol from ``acms_service.py``.
+
+    Based on ``docs/specification.md`` SS42637.
+    """
+
+    def __init__(
+        self,
+        *,
+        max_file_size: int | None = None,
+        max_total_size: int | None = None,
+        min_fragment_tokens: int = DEFAULT_MIN_FRAGMENT_TOKENS,
+        context_view: ContextView | None = None,
+    ) -> None:
+        self._max_file_size = max_file_size
+        self._max_total_size = max_total_size
+        self._min_fragment_tokens = min_fragment_tokens
+        self._context_view = context_view
+
+    @property
+    def max_file_size(self) -> int | None:
+        """Return the configured max_file_size constraint."""
+        return self._max_file_size
+
+    @property
+    def max_total_size(self) -> int | None:
+        """Return the configured max_total_size constraint."""
+        return self._max_total_size
+
+    @property
+    def min_fragment_tokens(self) -> int:
+        """Return the minimum fragment token threshold."""
+        return self._min_fragment_tokens
+
+    def pack(
+        self,
+        fragments: Sequence[ContextFragment],
+        budget: ContextBudget,
+    ) -> Sequence[ContextFragment]:
+        """Pack fragments into budget with byte-size constraints."""
+        if not fragments:
+            return list(fragments)
+
+        max_file = self._max_file_size
+        max_total = self._max_total_size
+
+        if self._context_view is not None:
+            if max_file is None:
+                max_file = self._context_view.max_file_size
+            if max_total is None:
+                max_total = self._context_view.max_total_size
+
+        available_tokens = budget.available_tokens
+
+        eligible = [f for f in fragments if f.token_count >= self._min_fragment_tokens]
+
+        if max_file is not None:
+            eligible = [
+                f
+                for f in eligible
+                if len(f.content.encode("utf-8", errors="replace")) <= max_file
+            ]
+
+        sorted_frags = sorted(eligible, key=lambda f: f.relevance_score, reverse=True)
+
+        packed: list[ContextFragment] = []
+        used_tokens = 0
+        used_bytes = 0
+
+        for frag in sorted_frags:
+            if used_tokens + frag.token_count > available_tokens:
+                continue
+
+            content_bytes = len(frag.content.encode("utf-8", errors="replace"))
+            if max_total is not None and used_bytes + content_bytes > max_total:
+                continue
+
+            packed.append(frag)
+            used_tokens += frag.token_count
+            used_bytes += content_bytes
+
+        return packed
+
+
+# ---------------------------------------------------------------------------
+# 5. PriorityCoherenceOrderer  (FragmentOrderer protocol)
+# ---------------------------------------------------------------------------
+
+
+class PriorityCoherenceOrderer:
+    """Order packed fragments for optimal LLM consumption.
+
+    Groups related fragments (same UKO node prefix) and sorts groups by
+    priority then relevance.
+
+    Implements ``FragmentOrderer`` protocol from ``acms_service.py``.
+
+    Based on ``docs/specification.md`` SS42648.
+    """
+
+    def __init__(
+        self,
+        *,
+        min_shared_segments: int = 2,
+    ) -> None:
+        self._min_shared_segments = min_shared_segments
+
+    @property
+    def min_shared_segments(self) -> int:
+        """Return the minimum shared URI segments for grouping."""
+        return self._min_shared_segments
+
+    def order(
+        self,
+        fragments: Sequence[ContextFragment],
+    ) -> Sequence[ContextFragment]:
+        """Order fragments for optimal LLM consumption."""
+        if not fragments or len(fragments) <= 1:
+            return list(fragments)
+
+        groups: dict[str, list[ContextFragment]] = defaultdict(list)
+        for frag in fragments:
+            prefix = self._extract_prefix(frag.uko_node)
+            groups[prefix].append(frag)
+
+        def group_sort_key(group: list[ContextFragment]) -> tuple[float, float]:
+            max_priority = max(self._extract_priority(f) for f in group)
+            max_relevance = max(f.relevance_score for f in group)
+            return (-max_priority, -max_relevance)
+
+        sorted_groups = sorted(groups.values(), key=group_sort_key)
+
+        ordered: list[ContextFragment] = []
+        for group in sorted_groups:
+            sorted_group = sorted(
+                group,
+                key=lambda f: (-self._extract_priority(f), -f.relevance_score),
+            )
+            ordered.extend(sorted_group)
+
+        return ordered
+
+    def _extract_prefix(self, uko_node: str) -> str:
+        """Extract the grouping prefix from a UKO node URI."""
+        segments = _uri_segments(uko_node)
+        prefix_segments = segments[: self._min_shared_segments]
+        return "/".join(prefix_segments) if prefix_segments else uko_node
+
+    @staticmethod
+    def _extract_priority(frag: ContextFragment) -> float:
+        """Extract priority from fragment metadata, defaulting to 0.5."""
+        raw = frag.metadata.get("priority", "0.5")
+        try:
+            value = float(raw)
+            return max(0.0, min(1.0, value))
+        except (ValueError, TypeError):
+            return 0.5
+
+
+# ---------------------------------------------------------------------------
+# Internal helpers
+# ---------------------------------------------------------------------------
+
+
+def _uri_segments(uri: str) -> list[str]:
+    """Split a UKO URI into path segments for hierarchy comparison."""
+    if "://" in uri:
+        uri = uri.split("://", 1)[1]
+    return [seg for seg in uri.replace("\\", "/").split("/") if seg]