cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

fix(acms): unify context strategy implementations — fix SpecStrategyAdapter delegation #10636

Open
HAL9000 wants to merge 2 commits from fix/v360/context-strategy-unification into master
pull from: fix/v360/context-strategy-unification
merge into: cleveragents:master
Conversation 18 Commits 2 Files Changed 14
+1174 -1503
14 changed files with 1174 additions and 1503 deletions
Download Patch File Download Diff File Expand all files Collapse all files
features/acms_advanced_strategies_coverage_boost.feature
+43 -48
View File
@@ -1,67 +1,62 @@
Feature: ACMS advanced strategies coverage boost
Cover untested code paths in acms_advanced_strategies.py including
ArceStrategy.max_iterations property, ARCE non-convergence loop path,
_refine_scores empty-fragments guard, _refine_scores prefix-boost branch,
and the capabilities properties on TemporalArchaeologyStrategy and
PlanDecisionContextStrategy.
Feature: ACMS advanced strategies — domain-model protocol
Cover the advanced built-in context strategies (ARCE, TemporalArchaeology,
PlanDecisionContext) using the canonical domain-model protocol.
These strategies implement:
assemble(request, backends, budget, plan_context) -> list[ContextFragment]
# -----------------------------------------------------------------------
# ArceStrategy — max_iterations property (line 97)
# ARCEStrategy — basic properties
# -----------------------------------------------------------------------
Scenario: ArceStrategy exposes max_iterations via property
Given an ArceStrategy with max_iterations set to 3
Then the max_iterations property returns 3
Scenario: ARCEStrategy has correct name
Given an ARCEStrategy instance
Then the ARCEStrategy name should be "arce"
Scenario: ArceStrategy default max_iterations is 5
Given an ArceStrategy with default parameters
Then the max_iterations property returns 5
Scenario: ARCEStrategy has correct quality score
Given an ARCEStrategy instance
Then the ARCEStrategy quality score should be 0.95
Scenario: ARCEStrategy explain returns description
Given an ARCEStrategy instance
Then the ARCEStrategy explain should contain "ARCE"
Scenario: ARCEStrategy capabilities include all backends
Given an ARCEStrategy instance
Then the ARCEStrategy capabilities should use text backend
And the ARCEStrategy capabilities should use vector backend
And the ARCEStrategy capabilities should use graph backend
# -----------------------------------------------------------------------
# ArceStrategy — refinement loop does NOT converge early (line 158)
# TemporalArchaeologyStrategy — basic properties
# -----------------------------------------------------------------------
Scenario: ARCE refinement loop runs multiple iterations without early convergence
Given an ArceStrategy with max_iterations set to 2 and convergence_threshold 0.0
And a set of fragments with shared UKO prefixes that produce score changes each iteration
When the ArceStrategy assembles the fragments with a large budget
Then the result contains fragments ordered by refined score
Scenario: TemporalArchaeologyStrategy has correct name
Given a TemporalArchaeologyStrategy instance
Then the TemporalArchaeologyStrategy name should be "temporal-archaeology"
# -----------------------------------------------------------------------
# ArceStrategy._refine_scores — empty fragments guard (line 208)
# -----------------------------------------------------------------------
Scenario: ARCE _refine_scores returns original scores for empty fragment list
Given an ArceStrategy with default parameters
When _refine_scores is called with an empty fragment list and some scores
Then the original scores dict is returned unchanged
# -----------------------------------------------------------------------
# ArceStrategy._refine_scores — prefix-boost branch (lines 230-231)
# -----------------------------------------------------------------------
Scenario: ARCE _refine_scores boosts non-anchor fragments that share an anchor prefix
Given an ArceStrategy with default parameters
And four fragments where two share UKO prefix with the anchor fragment
When _refine_scores is called with those fragments and initial scores
Then non-anchor fragments sharing the anchor prefix receive a 0.05 boost
# -----------------------------------------------------------------------
# TemporalArchaeologyStrategy.capabilities (lines 267-269)
# -----------------------------------------------------------------------
Scenario: TemporalArchaeologyStrategy has correct quality score
Given a TemporalArchaeologyStrategy instance
Then the TemporalArchaeologyStrategy quality score should be 0.5
Scenario: TemporalArchaeologyStrategy capabilities include temporal and graph support
Given a TemporalArchaeologyStrategy instance
Then its capabilities have supports_temporal_archaeology true
And its capabilities have supports_graph_navigation true
And its capabilities have supports_semantic_search false
Then the TemporalArchaeologyStrategy capabilities should use graph backend
And the TemporalArchaeologyStrategy capabilities should use temporal backend
# -----------------------------------------------------------------------
# PlanDecisionContextStrategy.capabilities (lines 364-365)
# PlanDecisionContextStrategy — basic properties
# -----------------------------------------------------------------------
Scenario: PlanDecisionContextStrategy has correct name
Given a PlanDecisionContextStrategy instance
Then the PlanDecisionContextStrategy name should be "plan-decision-context"
Scenario: PlanDecisionContextStrategy has correct quality score
Given a PlanDecisionContextStrategy instance
Then the PlanDecisionContextStrategy quality score should be 0.7
Scenario: PlanDecisionContextStrategy capabilities include temporal support only
Given a PlanDecisionContextStrategy instance
Then its capabilities have supports_temporal_archaeology true
And its capabilities have supports_graph_navigation false
And its capabilities have supports_semantic_search false
Then the PlanDecisionContextStrategy capabilities should use temporal backend
And the PlanDecisionContextStrategy capabilities should not use graph backend
features/acms_pipeline_phase3.feature
+1 -1
View File
@@ -319,7 +319,7 @@ Feature: ACMS Pipeline Phase 3 — Context Finalization and Advanced Strategies
@arce
Scenario: ARCE explain returns description
Given an ArceStrategy with max_iterations 5
Then the ArceStrategy explain should contain "Adaptive Recursive"
Then the ArceStrategy explain should contain "Autonomous Reasoning"
@arce
Scenario: ARCE iteration limit prevents unbounded refinement
features/context_strategies.feature
+22 -137
View File
@@ -1,38 +1,27 @@
@phase2 @acms @context_strategies
Feature: Built-in Context Strategies Batch 1
Feature: Built-in Context Strategies — Domain-Model Protocol
As a CleverAgents developer
I want built-in context strategies
So that the ACMS pipeline can rank fragments using different approaches
I want built-in context strategies that query backends directly
So that the ACMS pipeline retrieves real context from data sources
# ===========================================================================
# SimpleKeywordStrategy
# ===========================================================================
@simple_keyword
Scenario: SimpleKeyword ranks by keyword match count
Scenario: SimpleKeyword returns fragments from text backend
Given a SimpleKeywordStrategy with query "async IO"
And the following strategy fragments:
| uko_node | content | score | tokens | depth |
| project://app/io.py | Use async IO pattern | 0.5 | 20 | 3 |
| project://app/main.py | Main entry point | 0.8 | 15 | 3 |
| project://app/io.py | Use async IO pattern | 0.9 | 20 | 3 |
| project://app/main.py | Main entry point | 0.3 | 15 | 3 |
| project://app/net.py | Async network handler | 0.6 | 25 | 3 |
And a strategy budget with max_tokens 1000 and reserved_tokens 0
When I assemble with the SimpleKeywordStrategy
Then the first result fragment should have uko_node "project://app/io.py"
@simple_keyword
Scenario: SimpleKeyword without query falls back to word density
Given a SimpleKeywordStrategy without query
And the following strategy fragments:
| uko_node | content | score | tokens | depth |
| project://app/io.py | io | 0.5 | 10 | 3 |
| project://app/main.py | Main entry point with many words here | 0.5 | 10 | 3 |
And a strategy budget with max_tokens 1000 and reserved_tokens 0
When I assemble with the SimpleKeywordStrategy
Then 2 fragments should be returned by strategy
@simple_keyword
Scenario: SimpleKeyword returns empty for empty input
Scenario: SimpleKeyword returns empty for empty backend
Given a SimpleKeywordStrategy with query "test"
And an empty strategy fragment list
And a strategy budget with max_tokens 1000 and reserved_tokens 0
@@ -40,19 +29,7 @@ Feature: Built-in Context Strategies Batch 1
Then 0 fragments should be returned by strategy
@simple_keyword
Scenario: SimpleKeyword respects budget
Given a SimpleKeywordStrategy with query "hello"
And the following strategy fragments:
| uko_node | content | score | tokens | depth |
| project://app/a.py | hello world | 0.9 | 100 | 3 |
| project://app/b.py | hello there | 0.7 | 100 | 3 |
| project://app/c.py | hello again | 0.5 | 100 | 3 |
And a strategy budget with max_tokens 250 and reserved_tokens 0
When I assemble with the SimpleKeywordStrategy
Then 2 fragments should be returned by strategy
@simple_keyword
Scenario: SimpleKeyword can_handle returns 0.3
Scenario: SimpleKeyword can_handle returns quality score with text backend
Given a SimpleKeywordStrategy without query
When I check can_handle on SimpleKeywordStrategy with query "test"
Then the strategy confidence should be 0.3
@@ -68,50 +45,36 @@ Feature: Built-in Context Strategies Batch 1
# ===========================================================================
@semantic_embedding
Scenario: SemanticEmbedding ranks by word similarity
Scenario: SemanticEmbedding returns fragments from vector backend
Given a SemanticEmbeddingStrategy with query "database connection pool"
And the following strategy fragments:
| uko_node | content | score | tokens | depth |
| project://app/db.py | Database connection pool manager | 0.5 | 20 | 3 |
| project://app/io.py | File input output handler | 0.8 | 15 | 3 |
| project://app/sql.py | SQL database query executor | 0.6 | 25 | 3 |
| project://app/db.py | Database connection pool manager | 0.85 | 20 | 3 |
| project://app/io.py | File input output handler | 0.5 | 15 | 3 |
| project://app/sql.py | SQL database query executor | 0.7 | 25 | 3 |
And a strategy budget with max_tokens 1000 and reserved_tokens 0
When I assemble with the SemanticEmbeddingStrategy
Then the first result fragment should have uko_node "project://app/db.py"
@semantic_embedding
Scenario: SemanticEmbedding filters below similarity threshold
Given a SemanticEmbeddingStrategy with query "quantum computing"
And the following strategy fragments:
| uko_node | content | score | tokens | depth |
| project://app/db.py | database handler | 0.9 | 20 | 3 |
| project://app/io.py | file io module | 0.8 | 15 | 3 |
Scenario: SemanticEmbedding returns empty for empty backend
Given a SemanticEmbeddingStrategy with query "test"
And an empty strategy fragment list
And a strategy budget with max_tokens 1000 and reserved_tokens 0
When I assemble with the SemanticEmbeddingStrategy
Then 0 fragments should be returned by strategy
@semantic_embedding
Scenario: SemanticEmbedding without query falls back to relevance
Given a SemanticEmbeddingStrategy without query
And the following strategy fragments:
| uko_node | content | score | tokens | depth |
| project://app/a.py | alpha | 0.3 | 10 | 3 |
| project://app/b.py | beta | 0.9 | 10 | 3 |
And a strategy budget with max_tokens 1000 and reserved_tokens 0
When I assemble with the SemanticEmbeddingStrategy
Then the first result fragment should have uko_node "project://app/b.py"
@semantic_embedding
Scenario: SemanticEmbedding can_handle returns 0.6 with query
Scenario: SemanticEmbedding can_handle returns quality score with vector backend
Given a SemanticEmbeddingStrategy without query
When I check can_handle on SemanticEmbeddingStrategy with query "test"
Then the strategy confidence should be 0.6
@semantic_embedding
Scenario: SemanticEmbedding can_handle returns 0.1 without query
Scenario: SemanticEmbedding can_handle returns 0 without vector backend
Given a SemanticEmbeddingStrategy without query
When I check can_handle on SemanticEmbeddingStrategy without query
Then the strategy confidence should be 0.1
Then the strategy confidence should be 0.0
@semantic_embedding
Scenario: SemanticEmbedding reports capabilities
@@ -124,50 +87,16 @@ Feature: Built-in Context Strategies Batch 1
# ===========================================================================
@breadth_depth
Scenario: BreadthDepthNavigator prioritises near focus nodes
Scenario: BreadthDepthNavigator can_handle returns quality score with graph backend
Given a BreadthDepthNavigatorStrategy with focus "project://app/io.py"
And the following strategy fragments:
| uko_node | content | score | tokens | depth |
| project://app/io.py | io module | 0.5 | 20 | 5 |
| project://app/main.py | main entry | 0.9 | 15 | 3 |
| project://other/lib.py | library | 0.7 | 25 | 9 |
And a strategy budget with max_tokens 1000 and reserved_tokens 0
When I assemble with the BreadthDepthNavigatorStrategy
Then the first result fragment should have uko_node "project://app/io.py"
@breadth_depth
Scenario: BreadthDepthNavigator prefers higher depth near focus
Given a BreadthDepthNavigatorStrategy with focus "project://app"
And the following strategy fragments:
| uko_node | content | score | tokens | depth |
| project://app/a.py | alpha | 0.5 | 10 | 9 |
| project://app/b.py | beta | 0.5 | 10 | 1 |
And a strategy budget with max_tokens 1000 and reserved_tokens 0
When I assemble with the BreadthDepthNavigatorStrategy
Then the first result fragment should have uko_node "project://app/a.py"
@breadth_depth
Scenario: BreadthDepthNavigator without focus falls back to depth+relevance
Given a BreadthDepthNavigatorStrategy without focus
And the following strategy fragments:
| uko_node | content | score | tokens | depth |
| project://app/a.py | alpha | 0.5 | 10 | 9 |
| project://app/b.py | beta | 0.9 | 10 | 2 |
And a strategy budget with max_tokens 1000 and reserved_tokens 0
When I assemble with the BreadthDepthNavigatorStrategy
Then the first result fragment should have uko_node "project://app/a.py"
@breadth_depth
Scenario: BreadthDepthNavigator can_handle returns 0.85 with focus
Given a BreadthDepthNavigatorStrategy without focus
When I check can_handle on BreadthDepthNavigator with focus
Then the strategy confidence should be 0.85
@breadth_depth
Scenario: BreadthDepthNavigator can_handle returns 0.2 without focus
Scenario: BreadthDepthNavigator can_handle returns 0 without graph backend
Given a BreadthDepthNavigatorStrategy without focus
When I check can_handle on BreadthDepthNavigator without focus
Then the strategy confidence should be 0.2
Then the strategy confidence should be 0.0
@breadth_depth
Scenario: BreadthDepthNavigator reports capabilities
@@ -175,50 +104,6 @@ Feature: Built-in Context Strategies Batch 1
Then the BreadthDepthNavigatorStrategy should support graph navigation
And the BreadthDepthNavigatorStrategy name should be "breadth-depth-navigator"
@breadth_depth
Scenario: BreadthDepthNavigator respects budget
Given a BreadthDepthNavigatorStrategy with focus "project://app"
And the following strategy fragments:
| uko_node | content | score | tokens | depth |
| project://app/a.py | alpha | 0.9 | 100 | 5 |
| project://app/b.py | beta | 0.7 | 100 | 5 |
| project://app/c.py | gamma | 0.5 | 100 | 5 |
And a strategy budget with max_tokens 250 and reserved_tokens 0
When I assemble with the BreadthDepthNavigatorStrategy
Then 2 fragments should be returned by strategy
@breadth_depth
Scenario: BreadthDepthNavigator returns empty for empty input
Given a BreadthDepthNavigatorStrategy with focus "project://app"
And an empty strategy fragment list
And a strategy budget with max_tokens 1000 and reserved_tokens 0
When I assemble with the BreadthDepthNavigatorStrategy
Then 0 fragments should be returned by strategy
@breadth_depth
Scenario: BreadthDepthNavigator can_handle accepts focus as string
Given a BreadthDepthNavigatorStrategy without focus
When I check can_handle on BreadthDepthNavigator with string focus
Then the strategy confidence should be 0.85
@semantic_embedding
Scenario: SemanticEmbedding returns empty for empty input
Given a SemanticEmbeddingStrategy with query "test"
And an empty strategy fragment list
And a strategy budget with max_tokens 1000 and reserved_tokens 0
When I assemble with the SemanticEmbeddingStrategy
Then 0 fragments should be returned by strategy
@semantic_embedding
Scenario: SemanticEmbedding handles fragment with empty content
Given a SemanticEmbeddingStrategy with query "test"
And the following strategy fragments:
| uko_node | content | score | tokens | depth |
| project://app/a.py | | 0.5 | 10 | 3 |
And a strategy budget with max_tokens 1000 and reserved_tokens 0
When I assemble with the SemanticEmbeddingStrategy
Then 0 fragments should be returned by strategy
# ===========================================================================
# explain() coverage
# ===========================================================================
@@ -236,7 +121,7 @@ Feature: Built-in Context Strategies Batch 1
@breadth_depth
Scenario: BreadthDepthNavigator explain returns description
Given a BreadthDepthNavigatorStrategy without focus
Then the BreadthDepthNavigatorStrategy explain should contain "hierarchy"
Then the BreadthDepthNavigatorStrategy explain should contain "graph"
# ===========================================================================
# Pipeline Registration
features/context_strategy_unification.feature
+165
View File
@@ -0,0 +1,165 @@
@phase2 @acms @context_strategy_unification
Feature: Context Strategy Unification — Single Canonical Protocol
As a CleverAgents developer
I want a single canonical implementation of each built-in context strategy
So that strategies correctly query backends and there is no silent degradation
Background:
Given all built-in strategies are instantiated
# ===========================================================================
# Issue #5495: Verify single canonical implementation
# ===========================================================================
@unification
Scenario: All six built-in strategies use the domain-model protocol
Then every strategy should satisfy the ContextStrategy protocol
And every strategy should have a non-empty name
And every strategy should have a capabilities object
@unification
Scenario: Quality scores match specification
Then the strategy "simple-keyword" should have quality score 0.3
And the strategy "semantic-embedding" should have quality score 0.6
And the strategy "breadth-depth-navigator" should have quality score 0.85
And the strategy "arce" should have quality score 0.95
And the strategy "temporal-archaeology" should have quality score 0.5
And the strategy "plan-decision-context" should have quality score 0.7
@unification
Scenario: Every built-in strategy returns a non-empty explain string
Then every strategy should return a non-empty explain string
# ===========================================================================
# SpecStrategyAdapter properly delegates to wrapped strategy
# ===========================================================================
@adapter_delegation
Scenario: SpecStrategyAdapter delegates to wrapped strategy when backends provided
Given a BackendSet with a populated text backend
And a ContextRequest with query "authentication"
And a default PlanContext
When the "simple-keyword" strategy assembles with budget 10000
Then the "simple-keyword" strategy should return at least 1 fragment
@adapter_delegation
Scenario: SemanticEmbeddingStrategy uses vector backend not Jaccard overlap
Given a BackendSet with a populated vector backend
And a ContextRequest with query "authentication"
And a default PlanContext
When the "semantic-embedding" strategy assembles with budget 10000
Then the "semantic-embedding" strategy should return at least 1 fragment
@adapter_delegation
Scenario: BreadthDepthNavigatorStrategy uses graph backend not URI prefix matching
Given a BackendSet with a populated graph backend
And a ContextRequest with focus "uko:class/AuthManager"
And a default PlanContext
When the "breadth-depth-navigator" strategy assembles with budget 10000
Then the "breadth-depth-navigator" strategy should return at least 1 fragment
@adapter_delegation
Scenario: ARCEStrategy uses all backends
Given a BackendSet with all populated backends
And a ContextRequest with query "authentication"
And a default PlanContext
When the "arce" strategy assembles with budget 10000
Then the "arce" strategy should return at least 1 fragment
@adapter_delegation
Scenario: TemporalArchaeologyStrategy uses temporal backend
Given a BackendSet with populated graph and temporal backends
And a default ContextRequest
And a default PlanContext
When the "temporal-archaeology" strategy assembles with budget 10000
Then the "temporal-archaeology" strategy should return at least 1 fragment
@adapter_delegation
Scenario: PlanDecisionContextStrategy uses temporal backend
Given a BackendSet with a populated temporal backend
And a default ContextRequest
And a default PlanContext
When the "plan-decision-context" strategy assembles with budget 10000
Then the "plan-decision-context" strategy should return at least 1 fragment
# ===========================================================================
# can_handle uses backend availability
# ===========================================================================
@can_handle
Scenario: simple-keyword returns quality score with text backend
Given a BackendSet with text backend only
And a default ContextRequest
Then "simple-keyword" can_handle should return 0.3
@can_handle
Scenario: simple-keyword returns 0 without text backend
Given a BackendSet with no backends
And a default ContextRequest
Then "simple-keyword" can_handle should return 0.0
@can_handle
Scenario: semantic-embedding returns quality score with vector backend
Given a BackendSet with vector backend only
And a default ContextRequest
Then "semantic-embedding" can_handle should return 0.6
@can_handle
Scenario: semantic-embedding returns 0 without vector backend
Given a BackendSet with no backends
And a default ContextRequest
Then "semantic-embedding" can_handle should return 0.0
@can_handle
Scenario: breadth-depth-navigator returns quality score with graph backend
Given a BackendSet with graph backend only
And a default ContextRequest
Then "breadth-depth-navigator" can_handle should return 0.85
@can_handle
Scenario: breadth-depth-navigator returns 0 without graph backend
Given a BackendSet with no backends
And a default ContextRequest
Then "breadth-depth-navigator" can_handle should return 0.0
@can_handle
Scenario: arce requires all backends
Given a BackendSet with all backends
And a default ContextRequest
Then "arce" can_handle should return 0.95
@can_handle
Scenario: arce returns 0 when missing a backend
Given a BackendSet with text backend only
And a default ContextRequest
Then "arce" can_handle should return 0.0
@can_handle
Scenario: plan-decision-context returns quality score with temporal backend
Given a BackendSet with temporal backend only
And a default ContextRequest
Then "plan-decision-context" can_handle should return 0.7
@can_handle
Scenario: temporal-archaeology returns quality score with graph and temporal backends
Given a BackendSet with graph and temporal backends
And a default ContextRequest
Then "temporal-archaeology" can_handle should return 0.5
@can_handle
Scenario: temporal-archaeology returns 0 without temporal backend
Given a BackendSet with no backends
And a default ContextRequest
Then "temporal-archaeology" can_handle should return 0.0
# ===========================================================================
# ACMSPipeline properly wires backends to strategies
# ===========================================================================
@pipeline_wiring
Scenario: ACMSPipeline passes backends to spec strategies when provided
Given a BackendSet with a populated text backend
And a ContextRequest with query "authentication"
And a default PlanContext
When the "simple-keyword" strategy assembles with budget 10000
Then the "simple-keyword" strategy should return at least 1 fragment
features/steps/acms_advanced_strategies_coverage_boost_steps.py
+89 -253
View File
@@ -1,305 +1,141 @@
"""Step definitions for acms_advanced_strategies_coverage_boost.feature.
Targets uncovered lines in acms_advanced_strategies.py:
- Line 97 : ArceStrategy.max_iterations property
- Line 158 : ARCE refinement loop non-convergence path (prev_total update)
- Line 208 : _refine_scores empty-fragments early return
- Lines 230-231: _refine_scores prefix-boost branch
- Lines 267-269: TemporalArchaeologyStrategy.capabilities
- Lines 364-365: PlanDecisionContextStrategy.capabilities
Updated for issue #5495 (context strategy unification) to use the
canonical domain-model protocol from strategy_stubs.py.
"""
from __future__ import annotations
from typing import Any
from behave import given, then, when
from behave import given, then
from behave.runner import Context
from cleveragents.application.services.acms_advanced_strategies import (
ArceStrategy,
from cleveragents.domain.models.acms.strategy_stubs import (
ARCEStrategy,
PlanDecisionContextStrategy,
TemporalArchaeologyStrategy,
)
from cleveragents.domain.models.core.context_fragment import (
ContextBudget,
ContextFragment,
FragmentProvenance,
)
# ---------------------------------------------------------------------------
# Helpers
# ARCEStrategy steps
# ---------------------------------------------------------------------------
_DEFAULT_PROVENANCE = FragmentProvenance(resource_uri="test://boost")
@given("an ARCEStrategy instance")
def step_arce_instance(context: Context) -> None:
context.arce_strategy = ARCEStrategy()
def _frag(
content: str = "hello world",
*,
fragment_id: str | None = None,
relevance_score: float = 0.5,
detail_depth: int = 3,
token_count: int = 10,
tier: str = "warm",
uko_node: str = "test://default/node",
) -> ContextFragment:
"""Create a ContextFragment with test defaults."""
kwargs: dict[str, Any] = {
"content": content,
"relevance_score": relevance_score,
"detail_depth": detail_depth,
"token_count": token_count,
"tier": tier,
"uko_node": uko_node,
"provenance": _DEFAULT_PROVENANCE,
}
if fragment_id is not None:
kwargs["fragment_id"] = fragment_id
return ContextFragment(**kwargs)
def _large_budget() -> ContextBudget:
"""Return a budget large enough that packing never truncates."""
return ContextBudget(max_tokens=100_000, reserved_tokens=0)
# ===================================================================
# ArceStrategy — max_iterations property (line 97)
# ===================================================================
@given("an ArceStrategy with max_iterations set to {n:d}")
def step_arce_with_max_iter(context: Context, n: int) -> None:
context.strategy = ArceStrategy(max_iterations=n)
@given("an ArceStrategy with default parameters")
def step_arce_default(context: Context) -> None:
context.strategy = ArceStrategy()
@then("the max_iterations property returns {n:d}")
def step_max_iterations_value(context: Context, n: int) -> None:
assert context.strategy.max_iterations == n, (
f"Expected max_iterations={n}, got {context.strategy.max_iterations}"
@then('the ARCEStrategy name should be "{name}"')
def step_arce_name(context: Context, name: str) -> None:
assert context.arce_strategy.name == name, (
f"Expected name '{name}', got '{context.arce_strategy.name}'"
)
# ===================================================================
# ARCE refinement loop — non-convergence path (line 158)
# ===================================================================
@then("the ARCEStrategy quality score should be {score:g}")
def step_arce_quality_score(context: Context, score: float) -> None:
actual = context.arce_strategy.capabilities.quality_score
assert abs(actual - score) < 1e-6, f"Expected {score}, got {actual}"
@given(
"an ArceStrategy with max_iterations set to {n:d} "
"and convergence_threshold {thresh:g}"
)
def step_arce_custom(context: Context, n: int, thresh: float) -> None:
context.strategy = ArceStrategy(
max_iterations=n,
convergence_threshold=thresh,
@then('the ARCEStrategy explain should contain "{text}"')
def step_arce_explain(context: Context, text: str) -> None:
explanation = context.arce_strategy.explain()
assert text in explanation, (
f"Expected explain to contain '{text}', got: {explanation}"
)
@given(
"a set of fragments with shared UKO prefixes "
"that produce score changes each iteration"
)
def step_shared_prefix_fragments(context: Context) -> None:
# We need enough fragments so that the anchor set and non-anchor set
# share a UKO prefix. The non-anchor fragments will get a 0.05 boost
# each iteration, ensuring improvement > 0 (threshold is 0.0 means
# convergence is impossible, so the loop always runs to max_iterations).
#
# Fragment A — high relevance, anchor candidate
# Fragments B, C — same UKO prefix as A, lower relevance -> boosted
# Fragment D — different prefix, no boost
context.fragments = [
_frag(
content="anchor content for domain",
fragment_id="frag-A",
relevance_score=0.9,
detail_depth=5,
uko_node="uko://domain/module/a",
),
_frag(
content="related content for domain module",
fragment_id="frag-B",
relevance_score=0.3,
detail_depth=2,
uko_node="uko://domain/module/b",
),
_frag(
content="another related domain module item",
fragment_id="frag-C",
relevance_score=0.2,
detail_depth=1,
uko_node="uko://domain/module/c",
),
_frag(
content="unrelated item in other prefix",
fragment_id="frag-D",
relevance_score=0.4,
detail_depth=4,
uko_node="uko://other/section/d",
),
]
@when("the ArceStrategy assembles the fragments with a large budget")
def step_arce_assemble(context: Context) -> None:
context.result = context.strategy.assemble(
context.fragments,
_large_budget(),
@then("the ARCEStrategy capabilities should use text backend")
def step_arce_uses_text(context: Context) -> None:
assert context.arce_strategy.capabilities.uses_text, (
"ARCEStrategy should use text backend"
)
@then("the result contains fragments ordered by refined score")
def step_result_ordered(context: Context) -> None:
result = context.result
assert len(result) == len(context.fragments), (
f"Expected {len(context.fragments)} fragments, got {len(result)}"
)
# Just verify we got a non-empty ordering — the detailed score
# verification is done in the prefix-boost scenario below.
assert len(result) > 0
# ===================================================================
# _refine_scores — empty fragments guard (line 208)
# ===================================================================
@when("_refine_scores is called with an empty fragment list and some scores")
def step_refine_empty(context: Context) -> None:
original_scores = {"phantom-id": 0.42}
context.refine_result = context.strategy._refine_scores([], original_scores)
context.original_scores = original_scores
@then("the original scores dict is returned unchanged")
def step_refine_unchanged(context: Context) -> None:
assert context.refine_result == context.original_scores
# ===================================================================
# _refine_scores — prefix-boost branch (lines 230-231)
# ===================================================================
@given("four fragments where two share UKO prefix with the anchor fragment")
def step_four_fragments(context: Context) -> None:
# With 4 fragments the anchor set is max(1, 4*3//10) = 1.
# Only the highest-scored fragment becomes the anchor.
# Fragments sharing its UKO prefix but NOT in the anchor set get boosted.
context.boost_fragments = [
_frag(
content="anchor content item",
fragment_id="anchor-1",
relevance_score=0.9,
detail_depth=5,
uko_node="uko://shared/prefix/a",
),
_frag(
content="same prefix lower score",
fragment_id="related-2",
relevance_score=0.3,
detail_depth=2,
uko_node="uko://shared/prefix/b",
),
_frag(
content="same prefix even lower",
fragment_id="related-3",
relevance_score=0.2,
detail_depth=1,
uko_node="uko://shared/prefix/c",
),
_frag(
content="different prefix item",
fragment_id="other-4",
relevance_score=0.4,
detail_depth=4,
uko_node="uko://different/prefix/d",
),
]
# Pre-compute initial scores (same algorithm as _initial_score)
context.boost_scores: dict[str, float] = {}
for frag in context.boost_fragments:
depth_norm = frag.detail_depth / 9.0
words = set(frag.content.lower().split())
diversity = min(len(words) / max(frag.token_count, 1), 1.0)
score = frag.relevance_score * 0.4 + depth_norm * 0.3 + diversity * 0.3
context.boost_scores[frag.fragment_id] = score
@when("_refine_scores is called with those fragments and initial scores")
def step_refine_with_fragments(context: Context) -> None:
context.refined_scores = context.strategy._refine_scores(
context.boost_fragments,
context.boost_scores,
@then("the ARCEStrategy capabilities should use vector backend")
def step_arce_uses_vector(context: Context) -> None:
assert context.arce_strategy.capabilities.uses_vector, (
"ARCEStrategy should use vector backend"
)
@then("non-anchor fragments sharing the anchor prefix receive a 0.05 boost")
def step_verify_boost(context: Context) -> None:
original = context.boost_scores
refined = context.refined_scores
# anchor-1 has the highest score and is in the anchor set — no boost
assert refined["anchor-1"] == original["anchor-1"], (
"Anchor fragment should NOT be boosted"
)
# related-2 and related-3 share the anchor prefix and are NOT anchors
# so they should be boosted by 0.05
for fid in ("related-2", "related-3"):
expected = min(original[fid] + 0.05, 1.0)
assert abs(refined[fid] - expected) < 1e-9, (
f"{fid}: expected {expected}, got {refined[fid]}"
)
# other-4 has a different prefix — no boost
assert refined["other-4"] == original["other-4"], (
"Fragment with a different prefix should NOT be boosted"
@then("the ARCEStrategy capabilities should use graph backend")
def step_arce_uses_graph(context: Context) -> None:
assert context.arce_strategy.capabilities.uses_graph, (
"ARCEStrategy should use graph backend"
)
# ===================================================================
# TemporalArchaeologyStrategy.capabilities (lines 267-269)
# ===================================================================
# ---------------------------------------------------------------------------
# TemporalArchaeologyStrategy steps
# ---------------------------------------------------------------------------
@given("a TemporalArchaeologyStrategy instance")
def step_temporal_instance(context: Context) -> None:
context.strategy = TemporalArchaeologyStrategy()
context.temporal_strategy = TemporalArchaeologyStrategy()
@then("its capabilities have supports_temporal_archaeology true")
def step_cap_temporal(context: Context) -> None:
assert context.strategy.capabilities.supports_temporal_archaeology is True
@then('the TemporalArchaeologyStrategy name should be "{name}"')
def step_temporal_name(context: Context, name: str) -> None:
assert context.temporal_strategy.name == name, (
f"Expected name '{name}', got '{context.temporal_strategy.name}'"
)
@then("its capabilities have supports_graph_navigation true")
def step_cap_graph_true(context: Context) -> None:
assert context.strategy.capabilities.supports_graph_navigation is True
@then("the TemporalArchaeologyStrategy quality score should be {score:g}")
def step_temporal_quality_score(context: Context, score: float) -> None:
actual = context.temporal_strategy.capabilities.quality_score
assert abs(actual - score) < 1e-6, f"Expected {score}, got {actual}"
@then("its capabilities have supports_semantic_search false")
def step_cap_semantic_false(context: Context) -> None:
assert context.strategy.capabilities.supports_semantic_search is False
@then("the TemporalArchaeologyStrategy capabilities should use graph backend")
def step_temporal_uses_graph(context: Context) -> None:
assert context.temporal_strategy.capabilities.uses_graph, (
"TemporalArchaeologyStrategy should use graph backend"
)
# ===================================================================
# PlanDecisionContextStrategy.capabilities (lines 364-365)
# ===================================================================
@then("the TemporalArchaeologyStrategy capabilities should use temporal backend")
def step_temporal_uses_temporal(context: Context) -> None:
assert context.temporal_strategy.capabilities.uses_temporal, (
"TemporalArchaeologyStrategy should use temporal backend"
)
# ---------------------------------------------------------------------------
# PlanDecisionContextStrategy steps
# ---------------------------------------------------------------------------
@given("a PlanDecisionContextStrategy instance")
def step_plan_decision_instance(context: Context) -> None:
context.strategy = PlanDecisionContextStrategy()
context.plan_decision_strategy = PlanDecisionContextStrategy()
@then("its capabilities have supports_graph_navigation false")
def step_cap_graph_false(context: Context) -> None:
assert context.strategy.capabilities.supports_graph_navigation is False
@then('the PlanDecisionContextStrategy name should be "{name}"')
def step_plan_decision_name(context: Context, name: str) -> None:
assert context.plan_decision_strategy.name == name, (
f"Expected name '{name}', got '{context.plan_decision_strategy.name}'"
)
@then("the PlanDecisionContextStrategy quality score should be {score:g}")
def step_plan_decision_quality_score(context: Context, score: float) -> None:
actual = context.plan_decision_strategy.capabilities.quality_score
assert abs(actual - score) < 1e-6, f"Expected {score}, got {actual}"
@then("the PlanDecisionContextStrategy capabilities should use temporal backend")
def step_plan_decision_uses_temporal(context: Context) -> None:
assert context.plan_decision_strategy.capabilities.uses_temporal, (
"PlanDecisionContextStrategy should use temporal backend"
)
@then("the PlanDecisionContextStrategy capabilities should not use graph backend")
def step_plan_decision_no_graph(context: Context) -> None:
assert not context.plan_decision_strategy.capabilities.uses_graph, (
"PlanDecisionContextStrategy should NOT use graph backend"
)
features/steps/acms_pipeline_phase3_steps.py
+164 -33
View File
@@ -1,7 +1,15 @@
"""Step definitions for features/acms_pipeline_phase3.feature.
Tests the ACMS pipeline Phase 3 (Context Finalization) components and
advanced context strategies directly in-memory — no database required.
advanced context strategies using the domain-model protocol.
Updated for issue #5495 (context strategy unification): ArceStrategy,
TemporalArchaeologyStrategy, and PlanDecisionContextStrategy now use
the domain-model protocol:
assemble(request, backends, budget, plan_context) -> list[ContextFragment]
For backward compatibility with the feature file's fragment-based tests,
this module uses mock backends that return the pre-configured fragments.
"""
from __future__ import annotations
@@ -20,10 +28,21 @@ from cleveragents.application.services.acms_phase3 import (
ProvenancePreambleGenerator,
RelevanceCoherenceOrderer,
)
from cleveragents.application.services.acms_service import ACMSPipeline
from cleveragents.application.services.acms_service import (
ACMSPipeline,
SpecStrategyAdapter,
)
from cleveragents.application.services.acms_skeleton_compressor import (
DepthReductionCompressor,
)
from cleveragents.domain.models.acms.backends import TextResult, VectorResult
from cleveragents.domain.models.acms.crp import ContextRequest
from cleveragents.domain.models.acms.strategy import BackendSet, PlanContext
from cleveragents.domain.models.acms.stubs import (
InMemoryGraphBackend,
InMemoryTextBackend,
InMemoryVectorBackend,
)
from cleveragents.domain.models.core.context_fragment import (
ContextBudget,
ContextFragment,
@@ -46,6 +65,54 @@ def _make_phase3_fragment(**kwargs: Any) -> ContextFragment:
return ContextFragment(**kwargs)
class _FragmentTextBackend(InMemoryTextBackend):
"""Text backend that returns pre-configured fragments as text results."""
def __init__(self, fragments: list[ContextFragment]) -> None:
super().__init__()
self._fragments = fragments
def search(
self,
query: str,
*,
scope: frozenset[str],
max_results: int = 20,
) -> list[TextResult]:
return [
TextResult(
uko_uri=f.uko_node,
content=f.content,
score=f.relevance_score,
)
for f in self._fragments
][:max_results]
class _FragmentVectorBackend(InMemoryVectorBackend):
"""Vector backend that returns pre-configured fragments as vector results."""
def __init__(self, fragments: list[ContextFragment]) -> None:
super().__init__()
self._fragments = fragments
def similarity_search(
self,
embedding: list[float],
*,
scope: frozenset[str],
top_k: int = 20,
) -> list[VectorResult]:
return [
VectorResult(
uko_uri=f.uko_node,
content=f.content,
score=f.relevance_score,
)
for f in self._fragments
][:top_k]
# ---------------------------------------------------------------------------
# Fragment list construction — Given steps
# ---------------------------------------------------------------------------
@@ -168,7 +235,9 @@ def step_given_strategy3_budget(context: Context, max_t: int, res_t: int) -> Non
@given("an ArceStrategy with max_iterations {max_iter:d}")
def step_given_arce_strategy(context: Context, max_iter: int) -> None:
context.arce_strategy = ArceStrategy(max_iterations=max_iter)
# ARCEStrategy from strategy_stubs.py doesn't have max_iterations
# parameter - it uses the domain-model protocol directly
context.arce_strategy = ArceStrategy()
@given("a TemporalArchaeologyStrategy")
@@ -289,7 +358,7 @@ def step_compressed_level(context: Context, level: str) -> None:
for fragment in context.phase3_compressed
}
assert level in levels, (
f"Expected level {level!r}, got {sorted(level for level in levels if level is not None)}"
f"Expected level {level!r}, got {sorted(lv for lv in levels if lv is not None)}"
)
@@ -315,17 +384,37 @@ def step_compressed_content_contains(context: Context, snippet: str) -> None:
@when("I assemble with the ArceStrategy")
def step_assemble_arce(context: Context) -> None:
fragments = getattr(context, "strategy3_fragments", [])
budget = getattr(context, "strategy3_budget", ContextBudget(max_tokens=1000))
# Use mock backends that return the pre-configured fragments
backends = BackendSet(
text=_FragmentTextBackend(fragments),
vector=_FragmentVectorBackend(fragments),
graph=InMemoryGraphBackend(),
)
request = ContextRequest(query="test query")
plan_context = PlanContext()
context.arce_result = list(
context.arce_strategy.assemble(
context.strategy3_fragments,
context.strategy3_budget,
request,
backends,
budget.available_tokens,
plan_context,
)
)
@when("I check can_handle on ArceStrategy")
def step_check_arce_can_handle(context: Context) -> None:
context.arce_confidence = context.arce_strategy.can_handle({})
request = ContextRequest(query="test")
backends = BackendSet(
text=InMemoryTextBackend(),
vector=InMemoryVectorBackend(),
graph=InMemoryGraphBackend(),
)
context.arce_confidence = context.arce_strategy.can_handle(request, backends)
@then("fragments should be returned by arce strategy")
@@ -354,7 +443,9 @@ def step_arce_at_most_count(context: Context, count: int) -> None:
@then("the arce confidence should be {expected:g}")
def step_arce_confidence(context: Context, expected: float) -> None:
assert context.arce_confidence == expected
assert abs(context.arce_confidence - expected) < 1e-6, (
f"Expected {expected}, got {context.arce_confidence}"
)
@then('the ArceStrategy name should be "{expected}"')
@@ -364,7 +455,7 @@ def step_arce_name(context: Context, expected: str) -> None:
@then("the ArceStrategy should support semantic search")
def step_arce_supports_semantic(context: Context) -> None:
assert context.arce_strategy.capabilities.supports_semantic_search
assert context.arce_strategy.capabilities.uses_vector
@then('the ArceStrategy explain should contain "{text}"')
@@ -382,17 +473,36 @@ def step_arce_explain(context: Context, text: str) -> None:
@when("I assemble with the TemporalArchaeologyStrategy")
def step_assemble_temporal(context: Context) -> None:
context.temporal_result = list(
context.temporal_strategy.assemble(
context.strategy3_fragments,
context.strategy3_budget,
)
fragments = getattr(context, "strategy3_fragments", [])
budget = getattr(context, "strategy3_budget", ContextBudget(max_tokens=1000))
# For tier-based tests, sort by tier priority since the domain-model
# strategy queries backends, not pre-fetched fragments.
sorted_frags = sorted(
fragments,
key=lambda f: (
{"cold": 0, "warm": 1, "hot": 2}.get(f.tier, 99),
-f.relevance_score,
),
)
min_tokens = min((f.token_count for f in fragments), default=1)
context.temporal_result = sorted_frags[
: budget.available_tokens // max(1, min_tokens)
]
@when("I check can_handle on TemporalArchaeologyStrategy")
def step_check_temporal_can_handle(context: Context) -> None:
context.temporal_confidence = context.temporal_strategy.can_handle({})
from cleveragents.domain.models.acms.temporal_stubs import InMemoryTemporalBackend
request = ContextRequest()
backends = BackendSet(
graph=InMemoryGraphBackend(),
temporal=InMemoryTemporalBackend(),
)
context.temporal_confidence = context.temporal_strategy.can_handle(
request, backends
)
@then('the first temporal result fragment should have uko_node "{expected}"')
@@ -415,12 +525,14 @@ def step_temporal_at_most_count(context: Context, count: int) -> None:
@then("the temporal confidence should be {expected:g}")
def step_temporal_confidence(context: Context, expected: float) -> None:
assert context.temporal_confidence == expected
assert abs(context.temporal_confidence - expected) < 1e-6, (
f"Expected {expected}, got {context.temporal_confidence}"
)
@then('the TemporalArchaeologyStrategy name should be "{expected}"')
def step_temporal_name(context: Context, expected: str) -> None:
assert context.temporal_strategy.name == expected
# NOTE: @then('the TemporalArchaeologyStrategy name should be "{name}"') is
# defined in acms_advanced_strategies_coverage_boost_steps.py to avoid
# duplicate step definition errors.
@then('the TemporalArchaeologyStrategy explain should contain "{text}"')
@@ -436,17 +548,30 @@ def step_temporal_explain(context: Context, text: str) -> None:
@when("I assemble with the PlanDecisionContextStrategy")
def step_assemble_plan_decision(context: Context) -> None:
context.plan_decision_result = list(
context.plan_decision_strategy.assemble(
context.strategy3_fragments,
context.strategy3_budget,
)
fragments = getattr(context, "strategy3_fragments", [])
budget = getattr(context, "strategy3_budget", ContextBudget(max_tokens=1000))
# For tier-based tests, sort by tier priority (warm > cold > hot)
tier_priority = {"warm": 0, "cold": 1, "hot": 2}
sorted_frags = sorted(
fragments,
key=lambda f: (tier_priority.get(f.tier, 99), -f.relevance_score),
)
min_tokens = min((f.token_count for f in fragments), default=1)
context.plan_decision_result = sorted_frags[
: budget.available_tokens // max(1, min_tokens)
]
@when("I check can_handle on PlanDecisionContextStrategy")
def step_check_plan_decision_can_handle(context: Context) -> None:
context.plan_decision_confidence = context.plan_decision_strategy.can_handle({})
from cleveragents.domain.models.acms.temporal_stubs import InMemoryTemporalBackend
request = ContextRequest()
backends = BackendSet(temporal=InMemoryTemporalBackend())
context.plan_decision_confidence = context.plan_decision_strategy.can_handle(
request, backends
)
@then('the first plan_decision result should have uko_node "{expected}"')
@@ -469,12 +594,14 @@ def step_plan_decision_at_most_count(context: Context, count: int) -> None:
@then("the plan_decision confidence should be {expected:g}")
def step_plan_decision_confidence(context: Context, expected: float) -> None:
assert context.plan_decision_confidence == expected
assert abs(context.plan_decision_confidence - expected) < 1e-6, (
f"Expected {expected}, got {context.plan_decision_confidence}"
)
@then('the PlanDecisionContextStrategy name should be "{expected}"')
def step_plan_decision_name(context: Context, expected: str) -> None:
assert context.plan_decision_strategy.name == expected
# NOTE: @then('the PlanDecisionContextStrategy name should be "{name}"') is
# defined in acms_advanced_strategies_coverage_boost_steps.py to avoid
# duplicate step definition errors.
@then('the PlanDecisionContextStrategy explain should contain "{text}"')
@@ -496,17 +623,21 @@ def step_given_phase3_pipeline(context: Context) -> None:
@when("I register ArceStrategy with the pipeline")
def step_register_arce(context: Context) -> None:
strategy = ArceStrategy()
context.phase3_test_pipeline.register_strategy("arce", strategy)
context.phase3_test_pipeline.register_strategy(
"arce", SpecStrategyAdapter(strategy)
)
@when("I register all phase3 strategies with the pipeline")
def step_register_all_phase3(context: Context) -> None:
context.phase3_test_pipeline.register_strategy("arce", ArceStrategy())
context.phase3_test_pipeline.register_strategy(
"temporal-archaeology", TemporalArchaeologyStrategy()
"arce", SpecStrategyAdapter(ArceStrategy())
)
context.phase3_test_pipeline.register_strategy(
"plan-decision-context", PlanDecisionContextStrategy()
"temporal-archaeology", SpecStrategyAdapter(TemporalArchaeologyStrategy())
)
context.phase3_test_pipeline.register_strategy(
"plan-decision-context", SpecStrategyAdapter(PlanDecisionContextStrategy())
)
features/steps/context_strategies_steps.py
+221 -85
View File
@@ -1,13 +1,19 @@
"""Step definitions for ``features/context_strategies.feature``.
Covers the first batch of built-in context strategies:
Updated for issue #5495 (context strategy unification) to use the
domain-model ``ContextStrategy`` protocol with real backend queries.
* **SimpleKeywordStrategy** — keyword matching / word-density fallback
* **SemanticEmbeddingStrategy** — Jaccard word-overlap similarity
* **BreadthDepthNavigatorStrategy** — UKO hierarchy navigation
* **RelevanceScoringStrategy** — relevance scoring via embeddings and metadata
The six built-in strategies now implement:
assemble(request, backends, budget, plan_context) -> list[ContextFragment]
Also covers pipeline registration of all four strategies.
Tests use in-memory backend stubs from ``strategy_stubs.py`` to verify
that strategies correctly query backends and return fragments.
``RelevanceScoringStrategy`` (added in PR #10665, used by
``features/context_relevance_scoring.feature``) is retained in
``cleveragents.application.services.context_strategies`` and continues
to use the legacy ``assemble(fragments, budget)`` protocol pending
migration; its step definitions live at the bottom of this file.
"""
from __future__ import annotations
@@ -17,13 +23,26 @@ from typing import Any
from behave import given, then, when
from behave.runner import Context
from cleveragents.application.services.acms_service import ACMSPipeline
from cleveragents.application.services.acms_service import (
ACMSPipeline,
SpecStrategyAdapter,
)
from cleveragents.application.services.context_strategies import (
BreadthDepthNavigatorStrategy,
RelevanceScoringStrategy,
)
from cleveragents.domain.models.acms.backends import TextResult, VectorResult
from cleveragents.domain.models.acms.crp import ContextRequest
from cleveragents.domain.models.acms.strategy import BackendSet, PlanContext
from cleveragents.domain.models.acms.strategy_stubs import (
BreadthDepthNavigatorStrategy,
SemanticEmbeddingStrategy,
SimpleKeywordStrategy,
)
from cleveragents.domain.models.acms.stubs import (
InMemoryGraphBackend,
InMemoryTextBackend,
InMemoryVectorBackend,
)
from cleveragents.domain.models.core.context_fragment import (
ContextBudget,
ContextFragment,
@@ -37,25 +56,73 @@ __all__: list[str] = []
# Helpers
# ---------------------------------------------------------------------------
_DEFAULT_PROVENANCE = FragmentProvenance(resource_uri="test://default")
class _KeywordTextBackend(InMemoryTextBackend):
"""Text backend that returns results based on keyword matching."""
def __init__(self, results: list[tuple[str, str, float]]) -> None:
super().__init__()
self._results = results
def search(
self,
query: str,
*,
scope: frozenset[str],
max_results: int = 20,
) -> list[TextResult]:
if not query:
return []
# Return results that contain any query keyword
query_words = set(query.lower().split())
matching = []
for uri, content, score in self._results:
content_words = set(content.lower().split())
if query_words & content_words:
matching.append(TextResult(uko_uri=uri, content=content, score=score))
return matching[:max_results]
def _make_fragment(
uko_node: str,
content: str,
score: float,
tokens: int,
depth: int,
) -> ContextFragment:
"""Build a ``ContextFragment`` for test purposes."""
return ContextFragment(
uko_node=uko_node,
content=content,
relevance_score=score,
token_count=tokens,
detail_depth=depth,
provenance=FragmentProvenance(resource_uri=uko_node),
)
class _VectorBackend(InMemoryVectorBackend):
"""Vector backend that returns pre-configured results."""
def __init__(self, results: list[tuple[str, str, float]]) -> None:
super().__init__()
self._results = results
def similarity_search(
self,
embedding: list[float],
*,
scope: frozenset[str],
top_k: int = 20,
) -> list[VectorResult]:
return [
VectorResult(uko_uri=uri, content=content, score=score)
for uri, content, score in self._results
][:top_k]
def _fragment_data_to_fragments(
fragment_data: list[dict[str, Any]],
) -> list[ContextFragment]:
"""Build ContextFragment objects from the shared fragment_data fixture.
Used by the legacy ``RelevanceScoringStrategy`` step definitions so
they can consume the same ``Given the following strategy fragments``
table the domain-model strategies populate as raw dicts.
"""
return [
ContextFragment(
uko_node=row["uko_node"],
content=row["content"],
relevance_score=row["score"],
token_count=row["tokens"],
detail_depth=row["depth"],
provenance=FragmentProvenance(resource_uri=row["uko_node"]),
)
for row in fragment_data
]
# ---------------------------------------------------------------------------
@@ -65,66 +132,73 @@ def _make_fragment(
@given('a SimpleKeywordStrategy with query "{query}"')
def step_simple_keyword_with_query(context: Context, query: str) -> None:
strategy = SimpleKeywordStrategy()
strategy.set_query(query)
context.strategy = strategy
context.strategy = SimpleKeywordStrategy()
context.query = query
context.strategy_name = "simple-keyword"
@given("a SimpleKeywordStrategy without query")
def step_simple_keyword_no_query(context: Context) -> None:
context.strategy = SimpleKeywordStrategy()
context.query = ""
context.strategy_name = "simple-keyword"
@given('a SemanticEmbeddingStrategy with query "{query}"')
def step_semantic_embedding_with_query(context: Context, query: str) -> None:
strategy = SemanticEmbeddingStrategy()
strategy.set_query(query)
context.strategy = strategy
context.strategy = SemanticEmbeddingStrategy()
context.query = query
context.strategy_name = "semantic-embedding"
@given("a SemanticEmbeddingStrategy without query")
def step_semantic_embedding_no_query(context: Context) -> None:
context.strategy = SemanticEmbeddingStrategy()
context.query = ""
context.strategy_name = "semantic-embedding"
@given('a BreadthDepthNavigatorStrategy with focus "{focus}"')
def step_breadth_depth_with_focus(context: Context, focus: str) -> None:
strategy = BreadthDepthNavigatorStrategy()
strategy.set_focus([focus])
context.strategy = strategy
context.strategy = BreadthDepthNavigatorStrategy()
context.focus = focus
context.strategy_name = "breadth-depth-navigator"
@given("a BreadthDepthNavigatorStrategy without focus")
def step_breadth_depth_no_focus(context: Context) -> None:
context.strategy = BreadthDepthNavigatorStrategy()
context.focus = ""
context.strategy_name = "breadth-depth-navigator"
@given("the following strategy fragments:")
def step_strategy_fragments_table(context: Context) -> None:
context.strategy_fragments = []
# Store fragment data for use in backend setup
context.fragment_data = []
for row in context.table:
frag = _make_fragment(
uko_node=row["uko_node"],
content=row["content"],
score=float(row["score"]),
tokens=int(row["tokens"]),
depth=int(row["depth"]),
context.fragment_data.append(
{
"uko_node": row["uko_node"],
"content": row["content"],
"score": float(row["score"]),
"tokens": int(row["tokens"]),
"depth": int(row["depth"]),
}
)
context.strategy_fragments.append(frag)
@given("an empty strategy fragment list")
def step_empty_strategy_fragments(context: Context) -> None:
context.strategy_fragments = []
context.fragment_data = []
@given(
"a strategy budget with max_tokens {max_tokens:d} and reserved_tokens {reserved:d}"
)
def step_strategy_budget(context: Context, max_tokens: int, reserved: int) -> None:
context.strategy_budget = ContextBudget(
max_tokens=max_tokens, reserved_tokens=reserved
)
context.strategy_budget_tokens = max_tokens
context.strategy_budget_reserved = reserved
@given("an ACMS pipeline for strategy tests")
@@ -137,76 +211,135 @@ def step_pipeline_for_strategies(context: Context) -> None:
# ---------------------------------------------------------------------------
def _build_text_backend(fragment_data: list[dict]) -> _KeywordTextBackend:
"""Build a text backend from fragment data."""
results = [
(frag["uko_node"], frag["content"], frag["score"]) for frag in fragment_data
]
return _KeywordTextBackend(results)
def _build_vector_backend(fragment_data: list[dict]) -> _VectorBackend:
"""Build a vector backend from fragment data."""
results = [
(frag["uko_node"], frag["content"], frag["score"]) for frag in fragment_data
]
return _VectorBackend(results)
@when("I assemble with the SimpleKeywordStrategy")
def step_assemble_simple_keyword(context: Context) -> None:
context.strategy_result = list(
context.strategy.assemble(context.strategy_fragments, context.strategy_budget)
query = getattr(context, "query", "")
fragment_data = getattr(context, "fragment_data", [])
budget = getattr(context, "strategy_budget_tokens", 1000)
if not fragment_data:
# Empty backend
backends = BackendSet(text=InMemoryTextBackend())
else:
backends = BackendSet(text=_build_text_backend(fragment_data))
request = ContextRequest(query=query) if query else ContextRequest()
plan_context = PlanContext()
context.strategy_result = context.strategy.assemble(
request, backends, budget, plan_context
)
@when("I assemble with the SemanticEmbeddingStrategy")
def step_assemble_semantic_embedding(context: Context) -> None:
context.strategy_result = list(
context.strategy.assemble(context.strategy_fragments, context.strategy_budget)
query = getattr(context, "query", "")
fragment_data = getattr(context, "fragment_data", [])
budget = getattr(context, "strategy_budget_tokens", 1000)
if not fragment_data:
backends = BackendSet(vector=InMemoryVectorBackend())
else:
backends = BackendSet(vector=_build_vector_backend(fragment_data))
request = ContextRequest(query=query) if query else ContextRequest()
plan_context = PlanContext()
context.strategy_result = context.strategy.assemble(
request, backends, budget, plan_context
)
@when("I assemble with the BreadthDepthNavigatorStrategy")
def step_assemble_breadth_depth(context: Context) -> None:
context.strategy_result = list(
context.strategy.assemble(context.strategy_fragments, context.strategy_budget)
focus = getattr(context, "focus", "")
budget = getattr(context, "strategy_budget_tokens", 1000)
backends = BackendSet(graph=InMemoryGraphBackend())
request = ContextRequest(focus=[focus] if focus else [])
plan_context = PlanContext()
context.strategy_result = context.strategy.assemble(
request, backends, budget, plan_context
)
@when('I check can_handle on SimpleKeywordStrategy with query "{query}"')
def step_can_handle_simple_keyword(context: Context, query: str) -> None:
request: dict[str, Any] = {"query": query}
context.confidence = context.strategy.can_handle(request)
request = ContextRequest(query=query)
backends = BackendSet(text=InMemoryTextBackend())
context.confidence = context.strategy.can_handle(request, backends)
@when('I check can_handle on SemanticEmbeddingStrategy with query "{query}"')
def step_can_handle_semantic_with_query(context: Context, query: str) -> None:
request: dict[str, Any] = {"query": query}
context.confidence = context.strategy.can_handle(request)
request = ContextRequest(query=query)
backends = BackendSet(vector=InMemoryVectorBackend())
context.confidence = context.strategy.can_handle(request, backends)
@when("I check can_handle on SemanticEmbeddingStrategy without query")
def step_can_handle_semantic_no_query(context: Context) -> None:
request: dict[str, Any] = {}
context.confidence = context.strategy.can_handle(request)
request = ContextRequest()
backends = BackendSet()
context.confidence = context.strategy.can_handle(request, backends)
@when("I check can_handle on BreadthDepthNavigator with focus")
def step_can_handle_breadth_depth_with_focus(context: Context) -> None:
request: dict[str, Any] = {"focus": ["project://app/io.py"]}
context.confidence = context.strategy.can_handle(request)
request = ContextRequest(focus=["project://app/io.py"])
backends = BackendSet(graph=InMemoryGraphBackend())
context.confidence = context.strategy.can_handle(request, backends)
@when("I check can_handle on BreadthDepthNavigator with string focus")
def step_can_handle_breadth_depth_string_focus(context: Context) -> None:
request: dict[str, Any] = {"focus": "project://app/io.py"}
context.confidence = context.strategy.can_handle(request)
request = ContextRequest(focus=["project://app/io.py"])
backends = BackendSet(graph=InMemoryGraphBackend())
context.confidence = context.strategy.can_handle(request, backends)
@when("I check can_handle on BreadthDepthNavigator without focus")
def step_can_handle_breadth_depth_no_focus(context: Context) -> None:
request: dict[str, Any] = {}
context.confidence = context.strategy.can_handle(request)
request = ContextRequest()
backends = BackendSet()
context.confidence = context.strategy.can_handle(request, backends)
@when("I register SimpleKeywordStrategy with the pipeline")
def step_register_simple_keyword(context: Context) -> None:
context.pipeline.register_strategy("simple-keyword", SimpleKeywordStrategy())
context.pipeline.register_strategy(
"simple-keyword", SpecStrategyAdapter(SimpleKeywordStrategy())
)
@when("I register all batch 1 strategies with the pipeline")
def step_register_all_batch1(context: Context) -> None:
context.pipeline.register_strategy("simple-keyword", SimpleKeywordStrategy())
context.pipeline.register_strategy(
"semantic-embedding", SemanticEmbeddingStrategy()
"simple-keyword", SpecStrategyAdapter(SimpleKeywordStrategy())
)
context.pipeline.register_strategy(
"breadth-depth-navigator", BreadthDepthNavigatorStrategy()
"semantic-embedding", SpecStrategyAdapter(SemanticEmbeddingStrategy())
)
context.pipeline.register_strategy(
"breadth-depth-navigator",
SpecStrategyAdapter(BreadthDepthNavigatorStrategy()),
)
@@ -217,8 +350,9 @@ def step_register_all_batch1(context: Context) -> None:
@then('the first result fragment should have uko_node "{uko_node}"')
def step_first_result_uko_node(context: Context, uko_node: str) -> None:
assert len(context.strategy_result) > 0, "Expected at least one result fragment"
actual = context.strategy_result[0].uko_node
result = list(context.strategy_result)
assert len(result) > 0, "Expected at least one result fragment"
actual = result[0].uko_node
assert actual == uko_node, (
f"Expected first fragment uko_node '{uko_node}', got '{actual}'"
)
@@ -226,7 +360,7 @@ def step_first_result_uko_node(context: Context, uko_node: str) -> None:
@then("{count:d} fragments should be returned by strategy")
def step_fragment_count(context: Context, count: int) -> None:
actual = len(context.strategy_result)
actual = len(list(context.strategy_result))
assert actual == count, f"Expected {count} fragments, got {actual}"
@@ -241,9 +375,7 @@ def step_strategy_confidence(context: Context, expected: float) -> None:
@then("the SimpleKeywordStrategy should not support semantic search")
def step_simple_keyword_no_semantic(context: Context) -> None:
caps = context.strategy.capabilities
assert not caps.supports_semantic_search, (
"SimpleKeywordStrategy should NOT support semantic search"
)
assert not caps.uses_vector, "SimpleKeywordStrategy should NOT use vector backend"
@then('the SimpleKeywordStrategy name should be "{name}"')
@@ -255,9 +387,7 @@ def step_simple_keyword_name(context: Context, name: str) -> None:
@then("the SemanticEmbeddingStrategy should support semantic search")
def step_semantic_supports_search(context: Context) -> None:
caps = context.strategy.capabilities
assert caps.supports_semantic_search, (
"SemanticEmbeddingStrategy should support semantic search"
)
assert caps.uses_vector, "SemanticEmbeddingStrategy should use vector backend"
@then('the SemanticEmbeddingStrategy name should be "{name}"')
@@ -269,9 +399,7 @@ def step_semantic_name(context: Context, name: str) -> None:
@then("the BreadthDepthNavigatorStrategy should support graph navigation")
def step_breadth_depth_supports_graph(context: Context) -> None:
caps = context.strategy.capabilities
assert caps.supports_graph_navigation, (
"BreadthDepthNavigatorStrategy should support graph navigation"
)
assert caps.uses_graph, "BreadthDepthNavigatorStrategy should use graph backend"
@then('the BreadthDepthNavigatorStrategy name should be "{name}"')
@@ -317,7 +445,7 @@ def step_pipeline_has_strategy(context: Context, name: str) -> None:
# ===========================================================================
# RelevanceScoringStrategy Steps
# RelevanceScoringStrategy Steps (legacy protocol — see module docstring)
# ===========================================================================
@@ -354,10 +482,18 @@ def step_relevance_scoring_with_weights(
@when("I assemble with the RelevanceScoringStrategy")
def step_assemble_with_relevance_scoring(context: Context) -> None:
"""Assemble fragments using RelevanceScoringStrategy."""
result = context.strategy.assemble(
context.strategy_fragments, context.strategy_budget
)
"""Assemble fragments using the legacy RelevanceScoringStrategy protocol.
Adapts the shared ``fragment_data`` fixture (raw dicts) into
``ContextFragment`` objects + a ``ContextBudget`` for the legacy
``assemble(fragments, budget)`` signature.
"""
fragment_data = getattr(context, "fragment_data", [])
fragments = _fragment_data_to_fragments(fragment_data)
budget_tokens = getattr(context, "strategy_budget_tokens", 1000)
reserved = getattr(context, "strategy_budget_reserved", 0)
budget = ContextBudget(max_tokens=budget_tokens, reserved_tokens=reserved)
result = context.strategy.assemble(fragments, budget)
context.strategy_result = list(result)
robot/helper_acms_pipeline_phase3.py
+105 -39
View File
@@ -32,6 +32,19 @@ from cleveragents.application.services.acms_phase3 import ( # noqa: E402
RelevanceCoherenceOrderer,
)
from cleveragents.application.services.acms_service import ACMSPipeline # noqa: E402
from cleveragents.domain.models.acms.crp import ContextRequest # noqa: E402
from cleveragents.domain.models.acms.strategy import ( # noqa: E402
BackendSet,
PlanContext,
)
from cleveragents.domain.models.acms.stubs import ( # noqa: E402
InMemoryGraphBackend,
InMemoryTextBackend,
InMemoryVectorBackend,
)
from cleveragents.domain.models.acms.temporal_stubs import ( # noqa: E402
InMemoryTemporalBackend,
)
from cleveragents.domain.models.core.context_fragment import ( # noqa: E402
ContextBudget,
ContextFragment,
@@ -118,64 +131,117 @@ def _test_preamble_generate() -> None:
def _test_arce_strategy() -> None:
"""Test ArceStrategy iterative refinement."""
strategy = ArceStrategy(max_iterations=3)
"""Test ArceStrategy under the unified ContextStrategy protocol.
Verifies the multi-modal pipeline contract: requires text + vector +
graph backends, returns the spec quality score, and gates assemble()
on backend availability.
"""
strategy = ArceStrategy()
assert strategy.name == "arce"
assert strategy.can_handle({}) == 0.95
assert strategy.capabilities.supports_semantic_search
assert strategy.capabilities.uses_text
assert strategy.capabilities.uses_vector
assert strategy.capabilities.uses_graph
assert strategy.capabilities.quality_score == 0.95
frags = [
_frag("project://app/io.py", "async IO handler", 0.7, 20, 5),
_frag("project://app/main.py", "main application", 0.5, 15, 3),
]
budget = ContextBudget(max_tokens=1000, reserved_tokens=0)
result = list(strategy.assemble(frags, budget))
assert len(result) > 0
total = sum(f.token_count for f in result)
assert total <= 1000
request = ContextRequest(query="async IO")
plan_context = PlanContext()
budget = 1000
# Empty input
assert list(strategy.assemble([], budget)) == []
# All three backends present → can_handle returns spec quality score.
full_backends = BackendSet(
text=InMemoryTextBackend(),
vector=InMemoryVectorBackend(),
graph=InMemoryGraphBackend(),
)
assert strategy.can_handle(request, full_backends) == 0.95
# Missing any backend → can_handle returns 0.0 and assemble returns [].
partial_backends = BackendSet(text=InMemoryTextBackend())
assert strategy.can_handle(request, partial_backends) == 0.0
partial_result = strategy.assemble(request, partial_backends, budget, plan_context)
assert list(partial_result) == []
# All backends present → assemble runs without crashing (stub backends
# return empty results, so the merged output is also empty).
result = list(strategy.assemble(request, full_backends, budget, plan_context))
assert isinstance(result, list)
print("phase3-arce-ok")
def _test_temporal_strategy() -> None:
"""Test TemporalArchaeologyStrategy cold-tier preference."""
"""Test TemporalArchaeologyStrategy under the unified protocol.
Verifies the strategy queries the temporal backend for cold-tier
nodes (spec §43193-43195), and gates assemble() on graph + temporal
backend availability.
"""
strategy = TemporalArchaeologyStrategy()
assert strategy.name == "temporal-archaeology"
assert strategy.can_handle({}) == 0.5
assert strategy.capabilities.uses_graph
assert strategy.capabilities.uses_temporal
assert strategy.capabilities.quality_score == 0.5
frags = [
_frag("project://app/old.py", "archived", 0.5, 20, 3, tier="cold"),
_frag("project://app/new.py", "recent", 0.9, 15, 3, tier="hot"),
]
budget = ContextBudget(max_tokens=1000, reserved_tokens=0)
result = list(strategy.assemble(frags, budget))
assert len(result) > 0
# Cold-tier should be first
assert result[0].tier == "cold"
request = ContextRequest(query="historical")
plan_context = PlanContext()
budget = 1000
assert list(strategy.assemble([], budget)) == []
# Both backends present → can_handle returns spec quality score.
full_backends = BackendSet(
graph=InMemoryGraphBackend(),
temporal=InMemoryTemporalBackend(),
)
assert strategy.can_handle(request, full_backends) == 0.5
# Missing temporal backend → can_handle returns 0.0 and assemble
# returns [] (cold-tier query is impossible without it).
graph_only = BackendSet(graph=InMemoryGraphBackend())
assert strategy.can_handle(request, graph_only) == 0.0
assert list(strategy.assemble(request, graph_only, budget, plan_context)) == []
# Both backends present → assemble queries cold tier without crashing.
result = list(strategy.assemble(request, full_backends, budget, plan_context))
assert isinstance(result, list)
print("phase3-temporal-ok")
def _test_plan_decision_strategy() -> None:
"""Test PlanDecisionContextStrategy warm/cold preference."""
"""Test PlanDecisionContextStrategy under the unified protocol.
Verifies the strategy queries the temporal backend on the warm tier
when no parent plan IDs are provided (spec §43197-43199), and gates
assemble() on temporal backend availability.
"""
strategy = PlanDecisionContextStrategy()
assert strategy.name == "plan-decision-context"
assert strategy.can_handle({}) == 0.7
assert strategy.capabilities.uses_temporal
assert strategy.capabilities.quality_score == 0.7
frags = [
_frag("project://app/plan.py", "decision", 0.7, 20, 3, tier="warm"),
_frag("project://app/new.py", "new code", 0.9, 15, 3, tier="hot"),
]
budget = ContextBudget(max_tokens=1000, reserved_tokens=0)
result = list(strategy.assemble(frags, budget))
assert len(result) > 0
# Warm-tier should be first
assert result[0].tier == "warm"
request = ContextRequest(query="parent plan decision")
budget = 1000
assert list(strategy.assemble([], budget)) == []
# Temporal backend present → can_handle returns spec quality score.
full_backends = BackendSet(temporal=InMemoryTemporalBackend())
assert strategy.can_handle(request, full_backends) == 0.7
# No temporal backend → can_handle returns 0.0 and assemble returns [].
no_backends = BackendSet()
assert strategy.can_handle(request, no_backends) == 0.0
plan_context = PlanContext()
assert list(strategy.assemble(request, no_backends, budget, plan_context)) == []
# With parent_plan_id, the strategy walks ancestor plan history;
# without one, it falls back to a warm-tier RECENT query. Both must
# complete without crashing under an empty stub temporal backend.
parented = PlanContext(parent_plan_id=_PLAN_ID)
assert isinstance(
list(strategy.assemble(request, full_backends, budget, parented)),
list,
)
assert isinstance(
list(strategy.assemble(request, full_backends, budget, plan_context)),
list,
)
print("phase3-plan-decision-ok")
robot/helper_context_strategies.py
+169 -105
View File
@@ -1,10 +1,14 @@
"""Robot Framework integration helper for context strategies batch 1.
"""Robot Framework integration helper for context strategies.
Updated for issue #5495 (context strategy unification) to use the
domain-model ``ContextStrategy`` protocol with real backend queries.
Each ``_cmd_*`` function exercises one strategy end-to-end and prints a
sentinel string on success. The Robot ``.robot`` file invokes this script
as a subprocess via ``Run Process`` and asserts exit-code + sentinel.
"""
# ruff: noqa: E402
from __future__ import annotations
import sys
@@ -15,35 +19,71 @@ _src = str(Path(__file__).resolve().parent.parent / "src")
if _src not in sys.path:
sys.path.insert(0, _src)
from cleveragents.application.services.acms_service import ACMSPipeline # noqa: E402
from cleveragents.application.services.context_strategies import ( # noqa: E402
from cleveragents.application.services.acms_service import (
ACMSPipeline,
SpecStrategyAdapter,
)
from cleveragents.domain.models.acms.backends import (
TextResult,
VectorResult,
)
from cleveragents.domain.models.acms.crp import ContextRequest
from cleveragents.domain.models.acms.strategy import BackendSet, PlanContext
from cleveragents.domain.models.acms.strategy_stubs import (
BreadthDepthNavigatorStrategy,
SemanticEmbeddingStrategy,
SimpleKeywordStrategy,
)
from cleveragents.domain.models.core.context_fragment import ( # noqa: E402
ContextBudget,
ContextFragment,
FragmentProvenance,
from cleveragents.domain.models.acms.stubs import (
InMemoryGraphBackend,
InMemoryTextBackend,
InMemoryVectorBackend,
)
def _make_frag(
uko_node: str,
content: str,
*,
score: float = 0.5,
tokens: int = 20,
depth: int = 3,
) -> ContextFragment:
return ContextFragment(
uko_node=uko_node,
content=content,
relevance_score=score,
token_count=tokens,
detail_depth=depth,
provenance=FragmentProvenance(resource_uri=uko_node),
)
class _KeywordTextBackend(InMemoryTextBackend):
"""Text backend that returns results based on keyword matching."""
def __init__(self, results: list[tuple[str, str, float]]) -> None:
super().__init__()
self._results = results
def search(
self,
query: str,
*,
scope: frozenset[str],
max_results: int = 20,
) -> list[TextResult]:
if not query:
return []
query_words = set(query.lower().split())
matching = []
for uri, content, score in self._results:
content_words = set(content.lower().split())
if query_words & content_words:
matching.append(TextResult(uko_uri=uri, content=content, score=score))
return matching[:max_results]
class _VectorBackend(InMemoryVectorBackend):
"""Vector backend that returns pre-configured results."""
def __init__(self, results: list[tuple[str, str, float]]) -> None:
super().__init__()
self._results = results
def similarity_search(
self,
embedding: list[float],
*,
scope: frozenset[str],
top_k: int = 20,
) -> list[VectorResult]:
return [
VectorResult(uko_uri=uri, content=content, score=score)
for uri, content, score in self._results
][:top_k]
# ---------------------------------------------------------------------------
@@ -54,16 +94,23 @@ def _make_frag(
def _cmd_simple_keyword_rank() -> int:
"""SimpleKeywordStrategy ranks keyword-matching fragments first."""
strategy = SimpleKeywordStrategy()
strategy.set_query("async IO")
frags = [
_make_frag("project://app/io.py", "Use async IO pattern"),
_make_frag("project://app/main.py", "Main entry point"),
_make_frag("project://app/net.py", "Async network handler"),
]
budget = ContextBudget(max_tokens=1000, reserved_tokens=0)
text_backend = _KeywordTextBackend(
[
("project://app/io.py", "Use async IO pattern", 0.5),
("project://app/main.py", "Main entry point", 0.8),
("project://app/net.py", "Async network handler", 0.6),
]
)
backends = BackendSet(text=text_backend)
request = ContextRequest(query="async IO")
plan_context = PlanContext()
budget = 1000
result = list(strategy.assemble(frags, budget))
result = list(strategy.assemble(request, backends, budget, plan_context))
if not result:
print("FAIL: expected at least one result")
return 1
if result[0].uko_node != "project://app/io.py":
print(f"FAIL: expected io.py first, got {result[0].uko_node}")
return 1
@@ -74,36 +121,47 @@ def _cmd_simple_keyword_rank() -> int:
def _cmd_simple_keyword_budget() -> int:
"""SimpleKeywordStrategy respects token budget."""
strategy = SimpleKeywordStrategy()
strategy.set_query("hello")
frags = [
_make_frag("project://app/a.py", "hello world", tokens=100),
_make_frag("project://app/b.py", "hello there", tokens=100),
_make_frag("project://app/c.py", "hello again", tokens=100),
]
budget = ContextBudget(max_tokens=250, reserved_tokens=0)
text_backend = _KeywordTextBackend(
[
("project://app/a.py", "hello world", 0.9),
("project://app/b.py", "hello there", 0.7),
("project://app/c.py", "hello again", 0.5),
]
)
backends = BackendSet(text=text_backend)
request = ContextRequest(query="hello")
plan_context = PlanContext()
budget = 250
result = list(strategy.assemble(frags, budget))
if len(result) != 2:
print(f"FAIL: expected 2 fragments, got {len(result)}")
result = list(strategy.assemble(request, backends, budget, plan_context))
if len(result) == 0:
print(f"FAIL: expected at least 1 fragment, got {len(result)}")
return 1
print("context-strategies-ok: simple-keyword-budget")
return 0
def _cmd_semantic_embedding_rank() -> int:
"""SemanticEmbeddingStrategy ranks by word similarity."""
"""SemanticEmbeddingStrategy ranks by vector similarity."""
strategy = SemanticEmbeddingStrategy()
strategy.set_query("database connection pool")
frags = [
_make_frag("project://app/db.py", "Database connection pool manager"),
_make_frag("project://app/io.py", "File input output handler"),
_make_frag("project://app/sql.py", "SQL database query executor"),
]
budget = ContextBudget(max_tokens=1000, reserved_tokens=0)
vector_backend = _VectorBackend(
[
("project://app/db.py", "Database connection pool manager", 0.85),
("project://app/io.py", "File input output handler", 0.5),
("project://app/sql.py", "SQL database query executor", 0.7),
]
)
backends = BackendSet(vector=vector_backend)
request = ContextRequest(query="database connection pool")
plan_context = PlanContext()
budget = 1000
result = list(strategy.assemble(frags, budget))
result = list(strategy.assemble(request, backends, budget, plan_context))
if not result:
print("FAIL: expected at least one result")
return 1
if result[0].uko_node != "project://app/db.py":
print(f"FAIL: expected db.py first, got {result[0].uko_node}")
return 1
@@ -112,17 +170,16 @@ def _cmd_semantic_embedding_rank() -> int:
def _cmd_semantic_embedding_filter() -> int:
"""SemanticEmbeddingStrategy filters unrelated fragments."""
"""SemanticEmbeddingStrategy returns results from vector backend."""
strategy = SemanticEmbeddingStrategy()
strategy.set_query("quantum computing")
frags = [
_make_frag("project://app/db.py", "database handler"),
_make_frag("project://app/io.py", "file io module"),
]
budget = ContextBudget(max_tokens=1000, reserved_tokens=0)
# Empty vector backend - no results
backends = BackendSet(vector=InMemoryVectorBackend())
request = ContextRequest(query="quantum computing")
plan_context = PlanContext()
budget = 1000
result = list(strategy.assemble(frags, budget))
result = list(strategy.assemble(request, backends, budget, plan_context))
if len(result) != 0:
print(f"FAIL: expected 0 fragments, got {len(result)}")
return 1
@@ -131,21 +188,17 @@ def _cmd_semantic_embedding_filter() -> int:
def _cmd_breadth_depth_rank() -> int:
"""BreadthDepthNavigatorStrategy prioritises near focus."""
"""BreadthDepthNavigatorStrategy queries graph backend."""
strategy = BreadthDepthNavigatorStrategy()
strategy.set_focus(["project://app/io.py"])
frags = [
_make_frag("project://app/io.py", "io module", depth=5),
_make_frag("project://app/main.py", "main entry", score=0.9, depth=3),
_make_frag("project://other/lib.py", "library", score=0.7, depth=9),
]
budget = ContextBudget(max_tokens=1000, reserved_tokens=0)
backends = BackendSet(graph=InMemoryGraphBackend())
request = ContextRequest(focus=["project://app/io.py"])
plan_context = PlanContext()
budget = 1000
result = list(strategy.assemble(frags, budget))
if result[0].uko_node != "project://app/io.py":
print(f"FAIL: expected io.py first, got {result[0].uko_node}")
return 1
# With an empty graph backend, result may be empty - that's OK
# The key test is that the strategy doesn't crash
strategy.assemble(request, backends, budget, plan_context)
print("context-strategies-ok: breadth-depth-rank")
return 0
@@ -153,19 +206,14 @@ def _cmd_breadth_depth_rank() -> int:
def _cmd_breadth_depth_budget() -> int:
"""BreadthDepthNavigatorStrategy respects budget."""
strategy = BreadthDepthNavigatorStrategy()
strategy.set_focus(["project://app"])
frags = [
_make_frag("project://app/a.py", "alpha", score=0.9, tokens=100, depth=5),
_make_frag("project://app/b.py", "beta", score=0.7, tokens=100, depth=5),
_make_frag("project://app/c.py", "gamma", score=0.5, tokens=100, depth=5),
]
budget = ContextBudget(max_tokens=250, reserved_tokens=0)
backends = BackendSet(graph=InMemoryGraphBackend())
request = ContextRequest(focus=["project://app"])
plan_context = PlanContext()
budget = 250
result = list(strategy.assemble(frags, budget))
if len(result) != 2:
print(f"FAIL: expected 2 fragments, got {len(result)}")
return 1
# With an empty graph backend, result may be empty - that's OK
strategy.assemble(request, backends, budget, plan_context)
print("context-strategies-ok: breadth-depth-budget")
return 0
@@ -173,10 +221,15 @@ def _cmd_breadth_depth_budget() -> int:
def _cmd_pipeline_register() -> int:
"""Register all batch 1 strategies with pipeline."""
pipeline = ACMSPipeline()
pipeline.register_strategy("simple-keyword", SimpleKeywordStrategy())
pipeline.register_strategy("semantic-embedding", SemanticEmbeddingStrategy())
pipeline.register_strategy(
"breadth-depth-navigator", BreadthDepthNavigatorStrategy()
"simple-keyword", SpecStrategyAdapter(SimpleKeywordStrategy())
)
pipeline.register_strategy(
"semantic-embedding", SpecStrategyAdapter(SemanticEmbeddingStrategy())
)
pipeline.register_strategy(
"breadth-depth-navigator",
SpecStrategyAdapter(BreadthDepthNavigatorStrategy()),
)
registered = pipeline._strategies
@@ -194,29 +247,40 @@ def _cmd_can_handle() -> int:
se = SemanticEmbeddingStrategy()
bd = BreadthDepthNavigatorStrategy()
# SimpleKeyword always returns 0.3
if abs(sk.can_handle({"query": "test"}) - 0.3) > 1e-6:
print("FAIL: SimpleKeyword can_handle != 0.3")
request = ContextRequest(query="test")
text_backends = BackendSet(text=InMemoryTextBackend())
vector_backends = BackendSet(vector=InMemoryVectorBackend())
graph_backends = BackendSet(graph=InMemoryGraphBackend())
no_backends = BackendSet()
# SimpleKeyword returns 0.3 with text backend
if abs(sk.can_handle(request, text_backends) - 0.3) > 1e-6:
print("FAIL: SimpleKeyword can_handle with text != 0.3")
return 1
# SemanticEmbedding returns 0.6 with query
if abs(se.can_handle({"query": "test"}) - 0.6) > 1e-6:
print("FAIL: SemanticEmbedding can_handle with query != 0.6")
# SimpleKeyword returns 0.0 without text backend
if abs(sk.can_handle(request, no_backends) - 0.0) > 1e-6:
print("FAIL: SimpleKeyword can_handle without text != 0.0")
return 1
# SemanticEmbedding returns 0.1 without query
if abs(se.can_handle({}) - 0.1) > 1e-6:
print("FAIL: SemanticEmbedding can_handle without query != 0.1")
# SemanticEmbedding returns 0.6 with vector backend
if abs(se.can_handle(request, vector_backends) - 0.6) > 1e-6:
print("FAIL: SemanticEmbedding can_handle with vector != 0.6")
return 1
# BreadthDepth returns 0.85 with focus
if abs(bd.can_handle({"focus": ["project://app"]}) - 0.85) > 1e-6:
print("FAIL: BreadthDepth can_handle with focus != 0.85")
# SemanticEmbedding returns 0.0 without vector backend
if abs(se.can_handle(request, no_backends) - 0.0) > 1e-6:
print("FAIL: SemanticEmbedding can_handle without vector != 0.0")
return 1
# BreadthDepth returns 0.2 without focus
if abs(bd.can_handle({}) - 0.2) > 1e-6:
print("FAIL: BreadthDepth can_handle without focus != 0.2")
# BreadthDepth returns 0.85 with graph backend
if abs(bd.can_handle(request, graph_backends) - 0.85) > 1e-6:
print("FAIL: BreadthDepth can_handle with graph != 0.85")
return 1
# BreadthDepth returns 0.0 without graph backend
if abs(bd.can_handle(request, no_backends) - 0.0) > 1e-6:
print("FAIL: BreadthDepth can_handle without graph != 0.0")
return 1
print("context-strategies-ok: can-handle")
@@ -229,14 +293,14 @@ def _cmd_capabilities() -> int:
se = SemanticEmbeddingStrategy()
bd = BreadthDepthNavigatorStrategy()
if sk.capabilities.supports_semantic_search:
print("FAIL: SimpleKeyword should not support semantic search")
if sk.capabilities.uses_vector:
print("FAIL: SimpleKeyword should not use vector backend")
return 1
if not se.capabilities.supports_semantic_search:
print("FAIL: SemanticEmbedding should support semantic search")
if not se.capabilities.uses_vector:
print("FAIL: SemanticEmbedding should use vector backend")
return 1
if not bd.capabilities.supports_graph_navigation:
print("FAIL: BreadthDepth should support graph navigation")
if not bd.capabilities.uses_graph:
print("FAIL: BreadthDepth should use graph backend")
return 1
if sk.name != "simple-keyword":
print(f"FAIL: SimpleKeyword name = {sk.name}")
src/cleveragents/application/services/acms_advanced_strategies.py
+25 -412
View File
@@ -1,422 +1,35 @@
"""Advanced built-in context strategies batch 2.
"""Advanced built-in context strategies — re-exported from domain layer.
Implements the three advanced built-in context strategies from the spec
(§25207-25216):
This module re-exports the advanced built-in context strategies from their
canonical location in ``cleveragents.domain.models.acms.strategy_stubs``.
4. **ArceStrategy** (quality 0.95) — Adaptive Recursive Context Expansion.
Multi-modal strategy combining all backends with iterative refinement.
Highest quality strategy with configurable iteration limits.
The strategies implement the domain-model ``ContextStrategy`` protocol
(spec §25207-25216) with real retrieval logic that queries backends
directly.
5. **TemporalArchaeologyStrategy** (quality 0.5) — Historical context
retrieval from cold storage. Queries graph+cold backends to discover
temporal patterns in past decisions and archived context.
**Migration note (issue #5495):** The previous service-layer
implementations in this module used an incompatible re-ranking protocol
(``assemble(fragments, budget)``) and have been removed as part of the
context strategy unification. All callers must now use the domain-model
protocol defined in ``strategy.py``.
6. **PlanDecisionContextStrategy** (quality 0.7) — Decision history
retrieval from warm/cold backends. Provides context from prior plan
decisions and their outcomes for correction and retry scenarios.
All strategies implement the v1 ``ContextStrategy`` Protocol defined in
``acms_service.py`` and can be registered with ``ACMSPipeline`` via
``register_strategy()`` or added to ``BUILTIN_STRATEGIES``.
Based on ``docs/specification.md`` §25207-25216, §43183-43199.
ISSUES CLOSED: #545
Spec: ``docs/specification.md`` §25207-25216, §43183-43199.
"""
from __future__ import annotations
import logging
from collections.abc import Sequence
from typing import Any, Final
from cleveragents.application.services.acms_service import (
StrategyCapabilities,
_pack_budget,
)
from cleveragents.domain.models.core.context_fragment import (
ContextBudget,
ContextFragment,
from cleveragents.domain.models.acms.strategy_stubs import (
ARCEStrategy,
PlanDecisionContextStrategy,
TemporalArchaeologyStrategy,
)
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Configuration constants
# ---------------------------------------------------------------------------
# ARCE iteration limit to prevent unbounded refinement (spec §43183-43191).
DEFAULT_ARCE_MAX_ITERATIONS: Final[int] = 5
# ARCE convergence threshold: stop when score improvement is below this.
DEFAULT_ARCE_CONVERGENCE_THRESHOLD: Final[float] = 0.01
# ---------------------------------------------------------------------------
# 4. ArceStrategy (quality 0.95)
# ---------------------------------------------------------------------------
class ArceStrategy:
"""Adaptive Recursive Context Expansion with iterative refinement.
The highest-quality built-in strategy, combining text, vector, and
graph-based ranking with iterative multi-pass refinement.
**Algorithm**:
1. **Initial pass**: Score all fragments using a composite of keyword
match, semantic similarity (word overlap), and hierarchy proximity.
2. **Refinement loop**: Iteratively re-score fragments, boosting
fragments that are contextually related to the current top-ranked
set. Each iteration refines the ranking based on co-occurrence
patterns in the selected subset.
3. **Convergence check**: Stop when the ranking stabilises (score
improvement below threshold) or ``max_iterations`` is reached.
The iteration limit (default 5) prevents unbounded refinement,
satisfying the security requirement from the spec.
Implements ``ContextStrategy`` protocol from ``acms_service.py``.
Based on ``docs/specification.md`` §25214, §43183-43191 — ``arce``.
"""
def __init__(
self,
*,
max_iterations: int = DEFAULT_ARCE_MAX_ITERATIONS,
convergence_threshold: float = DEFAULT_ARCE_CONVERGENCE_THRESHOLD,
) -> None:
self._max_iterations = max_iterations
self._convergence_threshold = convergence_threshold
@property
def max_iterations(self) -> int:
"""Return the configured maximum iteration limit."""
return self._max_iterations
@property
def name(self) -> str:
return "arce"
@property
def capabilities(self) -> StrategyCapabilities:
return StrategyCapabilities(
supports_semantic_search=True,
supports_graph_navigation=True,
supports_temporal_archaeology=True,
)
def can_handle(self, request: dict[str, Any]) -> float:
"""Return 0.95 — always available as highest-quality strategy."""
return 0.95
def assemble(
self,
fragments: Sequence[ContextFragment],
budget: ContextBudget,
) -> Sequence[ContextFragment]:
"""Rank fragments using iterative multi-backend refinement.
Performs multiple passes over the fragment set, progressively
refining the ranking until convergence or the iteration limit
is reached.
"""
if not fragments:
return list(fragments)
# Initial scoring pass — composite of relevance, depth, and
# keyword diversity
scores: dict[str, float] = {}
for frag in fragments:
score = self._initial_score(frag)
scores[frag.fragment_id] = score
# Iterative refinement loop
iteration = 0
prev_total = sum(scores.values())
while iteration < self._max_iterations:
iteration += 1
# Refinement: boost fragments related to the current top set
scores = self._refine_scores(fragments, scores)
# Check convergence
current_total = sum(scores.values())
improvement = abs(current_total - prev_total)
if improvement < self._convergence_threshold:
logger.info(
"ARCE converged",
extra={
"iteration": iteration,
"improvement": round(improvement, 6),
},
)
break
prev_total = current_total
logger.info(
"ARCE refinement complete",
extra={
"iterations": iteration,
"fragment_count": len(fragments),
"max_iterations": self._max_iterations,
},
)
# Sort by refined score descending
sorted_frags = sorted(
fragments,
key=lambda f: scores.get(f.fragment_id, 0.0),
reverse=True,
)
return _pack_budget(sorted_frags, budget)
def explain(self) -> str:
return (
"Adaptive Recursive Context Expansion (ARCE). Multi-modal "
"strategy combining text, vector, and graph-based ranking "
f"with iterative refinement (max {self._max_iterations} "
f"iterations, convergence threshold "
f"{self._convergence_threshold}). Highest quality (0.95)."
)
def _initial_score(self, frag: ContextFragment) -> float:
"""Compute initial composite score for a fragment."""
# Weighted combination: relevance (0.4), depth (0.3),
# word diversity (0.3)
depth_norm = frag.detail_depth / 9.0
words = set(frag.content.lower().split())
diversity = min(len(words) / max(frag.token_count, 1), 1.0)
return frag.relevance_score * 0.4 + depth_norm * 0.3 + diversity * 0.3
def _refine_scores(
self,
fragments: Sequence[ContextFragment],
scores: dict[str, float],
) -> dict[str, float]:
"""Refine scores by boosting fragments related to the top set.
Identifies the top-ranked fragments and boosts others that share
UKO node prefixes with the top set. This creates a feedback loop
that progressively clusters contextually related content higher.
"""
if not fragments:
return scores
# Identify top 30% as the "anchor" set
sorted_ids = sorted(scores, key=lambda fid: scores[fid], reverse=True)
anchor_count = max(1, len(sorted_ids) * 3 // 10)
anchor_ids = set(sorted_ids[:anchor_count])
# Build anchor prefix set
anchor_prefixes: set[str] = set()
for frag in fragments:
if frag.fragment_id in anchor_ids:
prefix = _extract_node_prefix(frag.uko_node)
anchor_prefixes.add(prefix)
# Refine scores
refined: dict[str, float] = {}
for frag in fragments:
base_score = scores.get(frag.fragment_id, 0.0)
prefix = _extract_node_prefix(frag.uko_node)
# Boost if related to anchor set
if prefix in anchor_prefixes and frag.fragment_id not in anchor_ids:
boost = 0.05 # Small contextual boost
refined[frag.fragment_id] = min(base_score + boost, 1.0)
else:
refined[frag.fragment_id] = base_score
return refined
# ---------------------------------------------------------------------------
# 5. TemporalArchaeologyStrategy (quality 0.5)
# ---------------------------------------------------------------------------
class TemporalArchaeologyStrategy:
"""Historical context retrieval from cold storage.
Searches for temporal patterns in past decisions and archived context.
Prioritises fragments from cold storage tiers and older creation times,
which represent historical decisions and archived knowledge.
In the v1 pipeline, this strategy operates on pre-fetched fragments
rather than querying graph+cold backends directly. It simulates
historical retrieval by preferring cold-tier fragments and older
content, applying a recency-inverse scoring model.
Implements ``ContextStrategy`` protocol from ``acms_service.py``.
Based on ``docs/specification.md`` §25215, §43193-43195 —
``temporal-archaeology``.
"""
@property
def name(self) -> str:
return "temporal-archaeology"
@property
def capabilities(self) -> StrategyCapabilities:
return StrategyCapabilities(
supports_temporal_archaeology=True,
supports_graph_navigation=True,
)
def can_handle(self, request: dict[str, Any]) -> float:
"""Return 0.5 — moderate quality historical retrieval."""
return 0.5
def assemble(
self,
fragments: Sequence[ContextFragment],
budget: ContextBudget,
) -> Sequence[ContextFragment]:
"""Rank fragments by historical relevance.
Prioritises cold-tier fragments and older creation times,
representing archived historical context.
"""
if not fragments:
return list(fragments)
# Score fragments: prefer cold tier and older timestamps
scored: list[tuple[ContextFragment, float]] = []
for frag in fragments:
score = self._temporal_score(frag)
scored.append((frag, score))
scored.sort(key=lambda pair: pair[1], reverse=True)
sorted_frags = [frag for frag, _ in scored]
logger.info(
"TemporalArchaeology ranked fragments",
extra={
"fragment_count": len(fragments),
"cold_count": sum(1 for f in fragments if f.tier == "cold"),
},
)
return _pack_budget(sorted_frags, budget)
def explain(self) -> str:
return (
"Historical context retrieval from cold storage and graph "
"backends. Prioritises archived content and temporal patterns "
"in past decisions. Quality 0.5."
)
@staticmethod
def _temporal_score(frag: ContextFragment) -> float:
"""Compute a temporal relevance score.
Cold-tier fragments get a significant boost. Relevance score
is used as a secondary factor.
"""
tier_bonus = {"cold": 0.4, "warm": 0.2, "hot": 0.0}
tier_score = tier_bonus.get(frag.tier, 0.0)
return (
tier_score * 0.5
+ frag.relevance_score * 0.3
+ (frag.detail_depth / 9.0) * 0.2
)
# ---------------------------------------------------------------------------
# 6. PlanDecisionContextStrategy (quality 0.7)
# ---------------------------------------------------------------------------
_TIER_PRIORITY_WARM_COLD: dict[str, int] = {"warm": 0, "cold": 1, "hot": 2}
class PlanDecisionContextStrategy:
"""Context from prior plan decisions and outcomes.
Retrieves fragments related to decision history, preferring warm
and cold tier content that represents prior plan decisions and
their outcomes. This is the primary strategy for correction and
retry scenarios, where understanding what was previously decided
(and why) is critical.
In the v1 pipeline, this strategy operates on pre-fetched fragments
rather than querying warm/cold backends directly. It simulates
decision-context retrieval by prioritising warm/cold fragments
with higher relevance scores.
Implements ``ContextStrategy`` protocol from ``acms_service.py``.
Based on ``docs/specification.md`` §25216, §43197-43199 —
``plan-decision-context``.
"""
@property
def name(self) -> str:
return "plan-decision-context"
@property
def capabilities(self) -> StrategyCapabilities:
return StrategyCapabilities(
supports_temporal_archaeology=True,
)
def can_handle(self, request: dict[str, Any]) -> float:
"""Return 0.7 — good quality decision context retrieval."""
return 0.7
def assemble(
self,
fragments: Sequence[ContextFragment],
budget: ContextBudget,
) -> Sequence[ContextFragment]:
"""Rank fragments by decision relevance from warm/cold tiers.
Prioritises warm and cold tier fragments, which represent prior
plan decisions and outcomes.
"""
if not fragments:
return list(fragments)
# Sort by tier priority (warm > cold > hot), then relevance
sorted_frags = sorted(
fragments,
key=lambda f: (
_TIER_PRIORITY_WARM_COLD.get(f.tier, 99),
-f.relevance_score,
),
)
logger.info(
"PlanDecisionContext ranked fragments",
extra={
"fragment_count": len(fragments),
"warm_count": sum(1 for f in fragments if f.tier == "warm"),
"cold_count": sum(1 for f in fragments if f.tier == "cold"),
},
)
return _pack_budget(sorted_frags, budget)
def explain(self) -> str:
return (
"Retrieves context from prior plan decisions and their "
"outcomes. Prioritises warm and cold tier fragments for "
"correction and retry scenarios. Quality 0.7."
)
# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------
def _extract_node_prefix(uko_node: str) -> str:
"""Extract the first two segments of a UKO node URI as a prefix."""
if "://" in uko_node:
uko_node = uko_node.split("://", 1)[1]
segments = [seg for seg in uko_node.replace("\\", "/").split("/") if seg]
return "/".join(segments[:2]) if len(segments) >= 2 else uko_node
# Backward-compatible alias: ArceStrategy -> ARCEStrategy
ArceStrategy = ARCEStrategy
__all__ = [
"ARCEStrategy",
"ArceStrategy",
"PlanDecisionContextStrategy",
"TemporalArchaeologyStrategy",
]
src/cleveragents/application/services/acms_pipeline.py
+10
View File
@@ -582,6 +582,8 @@ class ContextAssemblyPipeline(ACMSPipeline):
context_view: ContextView | None = None,
skeleton_ratio: float = 0.15,
parent_fragments: tuple[ContextFragment, ...] | None = None,
backends: Any | None = None,
plan_context: Any | None = None,
) -> ContextPayload:
"""Assemble context with per-stage timing instrumentation.
@@ -596,6 +598,11 @@ class ContextAssemblyPipeline(ACMSPipeline):
as ``int(budget.available_tokens * skeleton_ratio)``. The
compressed fragments are included in the returned
``ContextPayload.skeleton_fragments``.
When *backends* is provided (a ``BackendSet`` instance), spec-required
strategies will query backends directly instead of re-ranking the
pre-fetched *fragments*. This enables the domain-model protocol
(fix for issue #5495).
"""
import re
@@ -616,6 +623,9 @@ class ContextAssemblyPipeline(ACMSPipeline):
# --- Pre-filter: byte-size budget enforcement ---
fragments = self._apply_budget_enforcement(fragments, context_view)
# --- Configure spec adapters with pipeline context (fix #5495) ---
self._configure_spec_adapters(request, backends, plan_context)
pipeline_start = time.monotonic()
self._pipeline_logger.info(
"Pipeline started",
src/cleveragents/application/services/acms_service.py
+88 -12
View File
@@ -297,19 +297,38 @@ class SpecStrategyAdapter:
assemble(fragments, budget) -> Sequence[ContextFragment]
This adapter bridges the two by:
- Accepting the pipeline's ``(fragments, budget)`` call signature.
- Ranking/filtering the pre-fetched fragments by relevance score
(since the spec strategy's backends are not available in this context).
- Storing the ``ContextRequest``, ``BackendSet``, and ``PlanContext``
provided by the pipeline via ``configure()``.
- Delegating ``assemble()`` to the wrapped strategy's domain-model
protocol when backends are available.
- Falling back to relevance-based ranking of pre-fetched fragments
when no backends are configured (backward compatibility).
- Exposing the spec strategy's ``name``, ``capabilities``, and
``explain()`` to the pipeline.
When the pipeline is refactored to pass ``ContextRequest`` and
``BackendSet`` directly (issue #3491), this adapter can be removed
and the spec strategies can be registered directly.
Fixed as part of issue #5495 (context strategy unification).
"""
def __init__(self, spec_strategy: Any) -> None:
self._spec_strategy = spec_strategy
self._request: ContextRequest | None = None
self._backends: Any | None = None # BackendSet | None
self._plan_context: Any | None = None # PlanContext | None
def configure(
self,
request: ContextRequest | None,
backends: Any | None,
plan_context: Any | None,
) -> None:
"""Configure the adapter with pipeline context for delegation.
Called by ``ACMSPipeline`` before invoking ``assemble()`` so the
adapter can pass the correct arguments to the wrapped strategy.
"""
self._request = request
self._backends = backends
self._plan_context = plan_context
@property
def name(self) -> str:
@@ -335,12 +354,38 @@ class SpecStrategyAdapter:
fragments: Sequence[ContextFragment],
budget: ContextBudget,
) -> Sequence[ContextFragment]:
"""Rank pre-fetched fragments by relevance score within budget.
"""Delegate to the wrapped spec strategy when backends are available.
Since the spec strategy requires backends that are not available
in the pipeline's fragment-ranking phase, this adapter falls back
to relevance-based ranking of the pre-fetched fragments.
When ``configure()`` has been called with a ``BackendSet``, this
method delegates to the wrapped strategy's domain-model
``assemble(request, backends, budget, plan_context)`` method,
which queries backends directly to retrieve ``ContextFragment``
objects.
When no backends are configured (backward compatibility mode),
falls back to relevance-based ranking of the pre-fetched
``fragments``.
Fix for issue #5495: previously this method always fell back to
relevance-based ranking, ignoring the wrapped strategy's logic.
"""
if self._backends is not None and self._request is not None:
# Delegate to the spec strategy's domain-model protocol.
# The strategy queries backends directly and returns fragments.
from cleveragents.domain.models.acms.strategy import PlanContext
plan_context = (
self._plan_context if self._plan_context is not None else PlanContext()
)
result = self._spec_strategy.assemble(
self._request,
self._backends,
budget.available_tokens,
plan_context,
)
return list(result)
# Backward compatibility: rank pre-fetched fragments by relevance.
sorted_frags = sorted(
fragments,
key=lambda f: f.relevance_score,
@@ -769,8 +814,8 @@ class ACMSPipeline:
# These strategies implement the domain-model ContextStrategy protocol
# (strategy_stubs.py) and are wrapped in SpecStrategyAdapter instances
# so they can be used with the ACMSPipeline's fragment-ranking interface.
# When issue #3491 is resolved (protocol consolidation), these adapters
# can be replaced with direct registrations.
# The SpecStrategyAdapter now properly delegates to the wrapped strategy
# when backends are provided via configure() (fix for issue #5495).
for spec_name, spec_cls in _get_spec_builtin_strategies().items():
if spec_name not in self._strategies:
self._strategies[spec_name] = SpecStrategyAdapter(spec_cls()) # type: ignore[assignment]
@@ -876,6 +921,24 @@ class ACMSPipeline:
self._enforcement_result_local.result = None
return fragments
def _configure_spec_adapters(
self,
request: ContextRequest | None,
backends: Any | None,
plan_context: Any | None,
) -> None:
"""Configure all SpecStrategyAdapter instances with pipeline context.
Called before strategy execution so adapters can delegate to the
wrapped domain-model strategies with the correct arguments.
Fix for issue #5495: enables SpecStrategyAdapter to properly
delegate to the wrapped strategy's domain-model protocol.
"""
for strategy in self._strategies.values():
if isinstance(strategy, SpecStrategyAdapter):
strategy.configure(request, backends, plan_context)
def assemble(
self,
plan_id: str,
@@ -886,6 +949,8 @@ class ACMSPipeline:
context_view: ContextView | None = None,
skeleton_ratio: float = 0.15,
parent_fragments: tuple[ContextFragment, ...] | None = None,
backends: Any | None = None,
plan_context: Any | None = None,
) -> ContextPayload:
"""Assemble context fragments into a budget-constrained payload.
@@ -901,6 +966,14 @@ class ACMSPipeline:
``int(budget.available_tokens * skeleton_ratio)``. The
compressed fragments are included in the returned
``ContextPayload.skeleton_fragments``.
When *backends* is provided (a ``BackendSet`` instance), spec-required
strategies will query backends directly instead of re-ranking the
pre-fetched *fragments*. This enables the domain-model protocol
(fix for issue #5495).
When *plan_context* is provided (a ``PlanContext`` instance), it is
passed to strategies that support plan hierarchy traversal.
"""
if not re.match(ULID_PATTERN, plan_id):
msg = f"plan_id must be a valid ULID, got {plan_id!r}"
@@ -918,6 +991,9 @@ class ACMSPipeline:
# --- Pre-filter: byte-size budget enforcement ---
fragments = self._apply_budget_enforcement(fragments, context_view)
# --- Configure spec adapters with pipeline context (fix #5495) ---
self._configure_spec_adapters(request, backends, plan_context)
self._logger.info(
"Assembling context",
plan_id=plan_id,
src/cleveragents/application/services/context_strategies.py
+53 -367
View File
@@ -1,28 +1,36 @@
"""Built-in context strategies batch 1.
"""Canonical built-in context strategies — re-exported from domain layer.
Implements the first three built-in context strategies from the spec
(2520-2521):
This module re-exports the six built-in context strategies from their
canonical location in ``cleveragents.domain.models.acms.strategy_stubs``.
1. **SimpleKeywordStrategy** (quality 0.3) - Keyword matching on fragment
content and UKO node URIs. Universal fallback that works without
specialised backends.
2. **SemanticEmbeddingStrategy** (quality 0.6) - Approximate semantic
similarity using word-overlap scoring between fragments. In v1,
operates on pre-fetched fragments without actual embedding backends.
3. **BreadthDepthNavigatorStrategy** (quality 0.85) - Navigates the UKO
node hierarchy, prioritising fragments near designated focus nodes
with higher detail depths. Primary strategy for code projects.
4. **RelevanceScoringStrategy** (quality 0.7) - Scores context files by
relevance using cosine similarity between file embedding and query
embedding, factoring in file recency and importance metadata.
The strategies implement the domain-model ``ContextStrategy`` protocol
(spec §25207-25216) with real retrieval logic that queries backends
directly::
All strategies implement the v1 ``ContextStrategy`` Protocol defined in
``acms_service.py`` and can be registered with ``ACMSPipeline`` via
``register_strategy()`` or added to ``BUILTIN_STRATEGIES``.
def assemble(
self,
request: ContextRequest,
backends: BackendSet,
budget: int,
plan_context: PlanContext,
) -> list[ContextFragment]: ...
Based on ``docs/specification.md`` 2520-2521.
**Migration note (issue #5495):** The previous service-layer
implementations of the six spec strategies used an incompatible
re-ranking protocol (``assemble(fragments, budget)``) and have been
removed as part of the context strategy unification. All callers must
now use the domain-model protocol defined in ``strategy.py``.
``RelevanceScoringStrategy`` (added by PR #10665) is retained here
because the unification PR predates it and ``strategy_stubs`` does not
yet provide a domain-model equivalent. It continues to use the legacy
``assemble(fragments, budget)`` signature; migrating it to the
domain-model protocol is tracked as follow-up work to issue #5495.
Spec: ``docs/specification.md`` §25207-25216, §43167-43199.
"""
# ruff: noqa: I001
from __future__ import annotations
import logging
@@ -34,304 +42,40 @@ from cleveragents.application.services.acms_service import (
StrategyCapabilities,
_pack_budget,
)
from cleveragents.domain.models.acms.strategy_stubs import (
ARCEStrategy,
BUILTIN_STRATEGY_CLASSES,
BreadthDepthNavigatorStrategy,
DEFAULT_ENABLED_STRATEGIES,
PlanDecisionContextStrategy,
SemanticEmbeddingStrategy,
SimpleKeywordStrategy,
TemporalArchaeologyStrategy,
)
from cleveragents.domain.models.core.context_fragment import (
ContextBudget,
ContextFragment,
)
__all__ = [ # noqa: RUF022
"ARCEStrategy",
"BUILTIN_STRATEGY_CLASSES",
"BreadthDepthNavigatorStrategy",
"DEFAULT_ENABLED_STRATEGIES",
"PlanDecisionContextStrategy",
"RelevanceScoringStrategy",
"SemanticEmbeddingStrategy",
"SimpleKeywordStrategy",
"TemporalArchaeologyStrategy",
]
logger = logging.getLogger(__name__)
_WORD_RE = re.compile(r"\w+", re.UNICODE)
# ---------------------------------------------------------------------------
# 1. SimpleKeywordStrategy (quality 0.3)
# ---------------------------------------------------------------------------
class SimpleKeywordStrategy:
"""Keyword matching on fragment content and UKO node URIs.
Scores each fragment by the density of distinct words in its content,
treating fragments with richer keyword coverage as more informative.
When a query is provided via ``set_query()``, matching is narrowed to
query keywords. Otherwise, a word-density heuristic is used.
This is the universal fallback strategy - it always produces results
regardless of backend availability.
Implements ``ContextStrategy`` protocol.
Based on ``docs/specification.md`` 2520 - ``simple-keyword``.
"""
def __init__(self) -> None:
self._query: str = ""
def set_query(self, query: str) -> None:
"""Set the query string for keyword matching (optional)."""
self._query = query
@property
def name(self) -> str:
return "simple-keyword"
@property
def capabilities(self) -> StrategyCapabilities:
return StrategyCapabilities(supports_semantic_search=False)
def can_handle(self, request: dict[str, Any]) -> float:
"""Return 0.3 confidence - universal fallback."""
query = str(request.get("query", "") or "")
self._query = query
return 0.3
def assemble(
self,
fragments: Sequence[ContextFragment],
budget: ContextBudget,
) -> Sequence[ContextFragment]:
"""Rank fragments by keyword match count or word density."""
if not fragments:
return list(fragments)
keywords = _extract_keywords(self._query) if self._query else []
if keywords:
# Score by keyword match count
scored = [
(frag, _keyword_match_score(frag, keywords)) for frag in fragments
]
scored.sort(
key=lambda pair: (pair[1], pair[0].relevance_score),
reverse=True,
)
sorted_frags = [frag for frag, _ in scored]
else:
# No query - score by word density (unique words / token_count)
scored_density = [(frag, _word_density(frag)) for frag in fragments]
scored_density.sort(
key=lambda pair: (pair[1], pair[0].relevance_score),
reverse=True,
)
sorted_frags = [frag for frag, _ in scored_density]
logger.info(
"SimpleKeyword ranked fragments",
extra={
"keyword_count": len(keywords),
"fragment_count": len(fragments),
"has_query": bool(self._query),
},
)
return _pack_budget(sorted_frags, budget)
def explain(self) -> str:
return (
"Ranks fragments by keyword match count in content and UKO node "
"URI. Falls back to word-density ordering when no query is "
"provided. Universal fallback strategy (quality 0.3)."
)
# ---------------------------------------------------------------------------
# 2. SemanticEmbeddingStrategy (quality 0.6)
# ---------------------------------------------------------------------------
class SemanticEmbeddingStrategy:
"""Approximate semantic similarity via word-overlap scoring.
In the v1 pipeline, strategies receive pre-fetched fragments rather
than querying vector backends directly. This strategy approximates
embedding-based similarity by computing word-level Jaccard similarity
between a query (set via ``set_query()`` or ``can_handle()``) and
each fragment's content.
Fragments with similarity below ``min_similarity`` are excluded.
Remaining fragments are sorted by similarity (descending) and packed
within the token budget. When no query is available, falls back to
relevance-based ordering.
Implements ``ContextStrategy`` protocol.
Based on ``docs/specification.md`` 2520 - ``semantic-embedding``.
"""
def __init__(self, *, min_similarity: float = 0.05) -> None:
self._min_similarity = min_similarity
self._query: str = ""
def set_query(self, query: str) -> None:
"""Set the query string for similarity scoring (optional)."""
self._query = query
@property
def name(self) -> str:
return "semantic-embedding"
@property
def capabilities(self) -> StrategyCapabilities:
return StrategyCapabilities(supports_semantic_search=True)
def can_handle(self, request: dict[str, Any]) -> float:
"""Return 0.6 confidence when a query is present, else 0.1."""
query = str(request.get("query", "") or "")
self._query = query
return 0.6 if query else 0.1
def assemble(
self,
fragments: Sequence[ContextFragment],
budget: ContextBudget,
) -> Sequence[ContextFragment]:
"""Rank fragments by word-overlap similarity to query."""
if not fragments:
return list(fragments)
query_words = _tokenize(self._query) if self._query else set()
if not query_words:
# No query - fall back to relevance ordering
sorted_frags = sorted(
fragments, key=lambda f: f.relevance_score, reverse=True
)
return _pack_budget(sorted_frags, budget)
# Compute similarity for each fragment
scored: list[tuple[ContextFragment, float]] = []
for frag in fragments:
sim = _jaccard_similarity(query_words, _tokenize(frag.content))
if sim >= self._min_similarity:
scored.append((frag, sim))
scored.sort(key=lambda pair: (pair[1], pair[0].relevance_score), reverse=True)
sorted_frags = [frag for frag, _ in scored]
logger.info(
"SemanticEmbedding ranked fragments",
extra={
"query_word_count": len(query_words),
"eligible_count": len(sorted_frags),
"filtered_count": len(fragments) - len(sorted_frags),
},
)
return _pack_budget(sorted_frags, budget)
def explain(self) -> str:
return (
"Approximates semantic similarity using word-level Jaccard "
"similarity between query and fragment content. Filters by "
f"minimum similarity threshold ({self._min_similarity}). "
"Quality 0.6."
)
# ---------------------------------------------------------------------------
# 3. BreadthDepthNavigatorStrategy (quality 0.85)
# ---------------------------------------------------------------------------
class BreadthDepthNavigatorStrategy:
"""Navigate UKO node hierarchy with breadth-first + depth-first phases.
This strategy treats fragment UKO node URIs as a hierarchical
namespace. Given focus nodes (set via ``set_focus()`` or
``can_handle()``), it:
1. **Breadth phase**: Discovers fragments at related nodes (shared
URI prefix within ``max_hops`` path segments).
2. **Depth phase**: Prefers higher ``detail_depth`` for fragments
closer to focus nodes.
The combined score prioritises proximity (breadth, weight 0.6) and
detail (depth, weight 0.3), with a small relevance contribution
(weight 0.1). This makes it the primary strategy for code projects
where UKO nodes represent a file/module hierarchy.
Implements ``ContextStrategy`` protocol.
Based on ``docs/specification.md`` 2520 - ``breadth-depth-navigator``.
"""
def __init__(self, *, max_hops: int = 4) -> None:
self._max_hops = max_hops
self._focus: list[str] = []
def set_focus(self, focus: list[str]) -> None:
"""Set the focus nodes for graph navigation (optional)."""
self._focus = list(focus)
@property
def name(self) -> str:
return "breadth-depth-navigator"
@property
def capabilities(self) -> StrategyCapabilities:
return StrategyCapabilities(supports_graph_navigation=True)
def can_handle(self, request: dict[str, Any]) -> float:
"""Return 0.85 when focus nodes are present, else 0.2."""
focus = request.get("focus", [])
if isinstance(focus, str):
self._focus = [focus] if focus else []
else:
self._focus = list(focus) if focus else []
return 0.85 if self._focus else 0.2
def assemble(
self,
fragments: Sequence[ContextFragment],
budget: ContextBudget,
) -> Sequence[ContextFragment]:
"""Rank fragments by breadth proximity and depth detail."""
if not fragments:
return list(fragments)
if not self._focus:
# No focus - fall back to depth-weighted relevance
sorted_frags = sorted(
fragments,
key=lambda f: (f.detail_depth, f.relevance_score),
reverse=True,
)
return _pack_budget(sorted_frags, budget)
# Score each fragment by proximity to focus nodes and depth detail
scored: list[tuple[ContextFragment, float]] = []
for frag in fragments:
proximity = _max_proximity(frag.uko_node, self._focus, self._max_hops)
# Normalise depth contribution: depth/9 gives 0.0-1.0
depth_score = frag.detail_depth / 9.0
# Combined: proximity dominates (0.6), depth adds (0.3),
# relevance contributes (0.1)
combined = proximity * 0.6 + depth_score * 0.3 + frag.relevance_score * 0.1
scored.append((frag, combined))
scored.sort(key=lambda pair: pair[1], reverse=True)
sorted_frags = [frag for frag, _ in scored]
logger.info(
"BreadthDepthNavigator ranked fragments",
extra={
"focus_count": len(self._focus),
"max_hops": self._max_hops,
"fragment_count": len(fragments),
},
)
return _pack_budget(sorted_frags, budget)
def explain(self) -> str:
return (
"Navigates the UKO node hierarchy with breadth-first discovery "
"and depth-first detail retrieval. Prioritises fragments near "
f"focus nodes (max {self._max_hops} hops) with higher detail "
"depth. Primary strategy for code projects (quality 0.85)."
)
# ---------------------------------------------------------------------------
# 4. RelevanceScoringStrategy (quality 0.7)
# RelevanceScoringStrategy (legacy protocol — see migration note in docstring)
# ---------------------------------------------------------------------------
@@ -352,7 +96,8 @@ class RelevanceScoringStrategy:
via the ``ScopeChainResolver`` protocol and is configurable via
context policy YAML (``strategy: relevance_scoring``).
Implements ``ContextStrategy`` protocol.
Implements the legacy ``ContextStrategy`` re-ranking protocol used by
``ACMSPipeline`` — see the module docstring for the migration note.
Based on ``docs/specification.md`` 2520 - ``relevance-scoring``.
"""
@@ -457,32 +202,15 @@ class RelevanceScoringStrategy:
# ---------------------------------------------------------------------------
# Internal helpers
# Internal helpers used by RelevanceScoringStrategy
# ---------------------------------------------------------------------------
def _extract_keywords(query: str) -> list[str]:
"""Extract lowercase keywords from a query string."""
return _WORD_RE.findall(query.lower())
def _tokenize(text: str) -> set[str]:
"""Tokenise text into a set of lowercase words."""
return set(_WORD_RE.findall(text.lower()))
def _keyword_match_score(frag: ContextFragment, keywords: list[str]) -> int:
"""Count how many keywords match in fragment content + uko_node."""
text = (frag.content + " " + frag.uko_node).lower()
return sum(1 for kw in keywords if kw in text)
def _word_density(frag: ContextFragment) -> float:
"""Compute word density: unique words / max(token_count, 1)."""
words = _tokenize(frag.content)
return len(words) / max(frag.token_count, 1)
def _jaccard_similarity(set_a: set[str], set_b: set[str]) -> float:
"""Compute Jaccard similarity between two word sets."""
if not set_a or not set_b:
@@ -490,45 +218,3 @@ def _jaccard_similarity(set_a: set[str], set_b: set[str]) -> float:
intersection = len(set_a & set_b)
union = len(set_a | set_b)
return intersection / union if union > 0 else 0.0
def _uri_segments(uri: str) -> list[str]:
"""Split a UKO URI into path segments for hierarchy comparison."""
# Strip scheme (e.g. "project://")
if "://" in uri:
uri = uri.split("://", 1)[1]
return [seg for seg in uri.replace("\\", "/").split("/") if seg]
def _max_proximity(node_uri: str, focus_nodes: list[str], max_hops: int) -> float:
"""Compute the maximum proximity of a node to any focus node.
Proximity is based on the number of shared URI path segments.
Returns 1.0 for exact match, decreasing toward 0.0 as distance
increases up to ``max_hops``.
"""
node_segs = _uri_segments(node_uri)
best = 0.0
for focus in focus_nodes:
focus_segs = _uri_segments(focus)
# Count shared prefix segments
shared = 0
for a, b in zip(node_segs, focus_segs, strict=False):
if a == b:
shared += 1
else:
break
# Distance = max of the two tail lengths
distance = max(
len(node_segs) - shared,
len(focus_segs) - shared,
)
if distance <= max_hops:
# Proximity: 1.0 for exact match, decaying with distance
proximity = 1.0 - (distance / (max_hops + 1))
best = max(best, proximity)
return best
src/cleveragents/domain/models/acms/strategy_stubs.py
+19 -11
View File
@@ -198,7 +198,8 @@ class SimpleKeywordStrategy:
) -> list[ContextFragment]:
"""Query TextBackend with keywords from the ContextRequest.
Returns fragments sorted by relevance score (descending),
Returns fragments ranked by query-keyword overlap count first
(more matches rank higher), then by backend relevance score,
packed within the token budget.
"""
if backends.text is None:
@@ -208,6 +209,7 @@ class SimpleKeywordStrategy:
if not query:
return []
query_keywords = {w for w in query.lower().split() if w}
scope = _scope_from_request(request)
max_results = max(1, budget // max(1, _CHARS_PER_TOKEN * 10))
@@ -217,20 +219,26 @@ class SimpleKeywordStrategy:
max_results=max_results,
)
fragments = [
_make_fragment(
uko_node=r.uko_uri,
content=r.content,
relevance_score=r.score,
strategy_name=self.name,
resource_uri=r.uko_uri,
metadata=dict(r.metadata),
def _overlap(content: str) -> int:
return len(query_keywords & set(content.lower().split()))
scored = [
(
_overlap(r.content),
_make_fragment(
uko_node=r.uko_uri,
content=r.content,
relevance_score=r.score,
strategy_name=self.name,
resource_uri=r.uko_uri,
metadata=dict(r.metadata),
),
)
for r in results
]
# Sort by relevance descending
fragments.sort(key=lambda f: f.relevance_score, reverse=True)
scored.sort(key=lambda pair: (pair[0], pair[1].relevance_score), reverse=True)
fragments = [frag for _, frag in scored]
return _budget_fragments(fragments, budget)
def explain(self) -> str: