diff --git a/features/acms_core_pipeline_components.feature b/features/acms_core_pipeline_components.feature new file mode 100644 index 000000000..f7ba1b3ac --- /dev/null +++ b/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 diff --git a/features/steps/acms_core_pipeline_components_steps.py b/features/steps/acms_core_pipeline_components_steps.py new file mode 100644 index 000000000..b57dd3f50 --- /dev/null +++ b/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}" + ) diff --git a/src/cleveragents/application/services/acms_core_pipeline.py b/src/cleveragents/application/services/acms_core_pipeline.py new file mode 100644 index 000000000..f3b2581b5 --- /dev/null +++ b/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]