"""Step definitions for acms_scope_resolution.feature. Tests for PipelineScopeResolver (plan > project > global scope chain) and ContextInheritanceService (skeleton context propagation to child plans). ISSUES CLOSED: #10016 """ from __future__ import annotations from collections.abc import Sequence from behave import given, then, when # type: ignore[import-untyped] from behave.runner import Context from cleveragents.application.services.acms_scope_resolver import ( ContextInheritanceService, PipelineScopeResolver, ) from cleveragents.application.services.acms_service import ( DefaultSkeletonCompressor, DetailDepthResolver, FragmentDeduplicator, PreambleGenerator, SkeletonCompressor, ) from cleveragents.application.services.component_resolver import ScopeLevel from cleveragents.domain.models.core.context_fragment import ( ContextBudget, ContextFragment, ContextPayload, FragmentProvenance, build_provenance_map, compute_context_hash, ) # Default provenance for test fragments. _DEFAULT_PROVENANCE = FragmentProvenance(resource_uri="test://default") # --------------------------------------------------------------------------- # Helper factories # --------------------------------------------------------------------------- def _make_frag( uko_node: str = "project://app/main.py", content: str = "hello", score: float = 0.5, tokens: int = 100, ) -> ContextFragment: return ContextFragment( uko_node=uko_node, content=content, relevance_score=score, token_count=tokens, provenance=_DEFAULT_PROVENANCE, ) def _make_payload( fragments: tuple[ContextFragment, ...], plan_id: str = "01JQTESTPN00000000000000AA", ) -> ContextPayload: total_tokens = sum(f.token_count for f in fragments) budget = ContextBudget(max_tokens=max(total_tokens, 1), reserved_tokens=0) available = budget.available_tokens budget_used = total_tokens / available if available > 0 else 0.0 return ContextPayload( plan_id=plan_id, fragments=fragments, total_tokens=total_tokens, budget=budget, budget_used=round(min(budget_used, 1.0), 4), strategies_used=("relevance",), context_hash=compute_context_hash(fragments), preamble=None, provenance_map=build_provenance_map(fragments), ) # --------------------------------------------------------------------------- # Stub implementations for tracking # --------------------------------------------------------------------------- class _TrackingCompressor: """SkeletonCompressor that records calls.""" def __init__(self) -> None: self.called = False self.last_fragments: tuple[ContextFragment, ...] = () self.last_budget: int = 0 def compress( self, fragments: tuple[ContextFragment, ...], skeleton_budget: int, ) -> tuple[ContextFragment, ...]: self.called = True self.last_fragments = fragments self.last_budget = skeleton_budget # Return all fragments that fit within budget result: list[ContextFragment] = [] remaining = skeleton_budget for frag in fragments: if frag.token_count <= remaining: result.append(frag) remaining -= frag.token_count return tuple(result) class _StubCompressor: """Named stub compressor for identity checks.""" def __init__(self, name: str) -> None: self.name = name def compress( self, fragments: tuple[ContextFragment, ...], skeleton_budget: int, ) -> tuple[ContextFragment, ...]: return fragments class _StubPreambleGenerator: """Named stub preamble generator for identity checks.""" def __init__(self, name: str) -> None: self.name = name def generate(self, fragments: Sequence[ContextFragment]) -> str | None: return f"preamble from {self.name}" class _StubDeduplicator: """Named stub deduplicator for identity checks.""" def __init__(self, name: str) -> None: self.name = name def deduplicate( self, fragments: Sequence[ContextFragment] ) -> Sequence[ContextFragment]: return list(fragments) class _StubDepthResolver: """Named stub depth resolver for identity checks.""" def __init__(self, name: str) -> None: self.name = name def resolve( self, fragments: Sequence[ContextFragment], budget: int ) -> Sequence[ContextFragment]: return list(fragments) # --------------------------------------------------------------------------- # Given steps -- PipelineScopeResolver setup # --------------------------------------------------------------------------- @given("a fresh PipelineScopeResolver") def step_fresh_scope_resolver(context: Context) -> None: context.scope_resolver = PipelineScopeResolver() context.global_compressor = _StubCompressor("global") context.project_compressor = _StubCompressor("project") context.plan_compressor = _StubCompressor("plan") context.global_preamble = _StubPreambleGenerator("global") context.project_preamble = _StubPreambleGenerator("project") context.global_dedup = _StubDeduplicator("global") context.plan_dedup = _StubDeduplicator("plan") context.global_depth = _StubDepthResolver("global") context.plan_depth = _StubDepthResolver("plan") @given("a global default SkeletonCompressor registered") def step_register_global_compressor(context: Context) -> None: context.scope_resolver.register_global( SkeletonCompressor, context.global_compressor ) @given("a global default PreambleGenerator registered") def step_register_global_preamble(context: Context) -> None: context.scope_resolver.register_global(PreambleGenerator, context.global_preamble) @given("a global default FragmentDeduplicator registered") def step_register_global_dedup(context: Context) -> None: context.scope_resolver.register_global(FragmentDeduplicator, context.global_dedup) @given("a global default DetailDepthResolver registered") def step_register_global_depth(context: Context) -> None: context.scope_resolver.register_global(DetailDepthResolver, context.global_depth) @given('a project-level SkeletonCompressor override for project "{project_id}"') def step_register_project_compressor(context: Context, project_id: str) -> None: context.scope_resolver.register_project( project_id, SkeletonCompressor, context.project_compressor ) @given('a project-level PreambleGenerator override for project "{project_id}"') def step_register_project_preamble(context: Context, project_id: str) -> None: context.scope_resolver.register_project( project_id, PreambleGenerator, context.project_preamble ) @given('a plan-level SkeletonCompressor override for plan "{plan_id}"') def step_register_plan_compressor(context: Context, plan_id: str) -> None: context.scope_resolver.register_plan( plan_id, SkeletonCompressor, context.plan_compressor ) @given('a plan-level FragmentDeduplicator override for plan "{plan_id}"') def step_register_plan_dedup(context: Context, plan_id: str) -> None: context.scope_resolver.register_plan( plan_id, FragmentDeduplicator, context.plan_dedup ) @given('a plan-level DetailDepthResolver override for plan "{plan_id}"') def step_register_plan_depth(context: Context, plan_id: str) -> None: context.scope_resolver.register_plan( plan_id, DetailDepthResolver, context.plan_depth ) # --------------------------------------------------------------------------- # Given steps -- ContextInheritanceService setup # --------------------------------------------------------------------------- @given("a PipelineScopeResolver with a global DefaultSkeletonCompressor") def step_resolver_with_default_compressor(context: Context) -> None: context.scope_resolver = PipelineScopeResolver() context.scope_resolver.register_global( SkeletonCompressor, DefaultSkeletonCompressor() ) @given("a ContextInheritanceService with default skeleton ratio {ratio:f}") def step_inheritance_service(context: Context, ratio: float) -> None: context.inheritance_service = ContextInheritanceService( context.scope_resolver, default_skeleton_ratio=ratio, ) @given("a parent payload with {count:d} fragments totaling {total:d} tokens") def step_parent_payload(context: Context, count: int, total: int) -> None: tokens_each = total // count frags = tuple( _make_frag( uko_node=f"project://app/file{i}.py", content=f"content {i}", score=0.8, tokens=tokens_each, ) for i in range(count) ) context.parent_payload = _make_payload(frags) @given("an empty parent payload") def step_empty_parent_payload(context: Context) -> None: context.parent_payload = _make_payload(()) @given("parent fragments with {count:d} fragments totaling {total:d} tokens") def step_parent_fragments(context: Context, count: int, total: int) -> None: tokens_each = total // count context.parent_fragments = tuple( _make_frag( uko_node=f"project://app/file{i}.py", content=f"content {i}", score=0.8, tokens=tokens_each, ) for i in range(count) ) @given("empty parent fragments") def step_empty_parent_fragments(context: Context) -> None: context.parent_fragments = () @given('a plan-level tracking SkeletonCompressor for plan "{plan_id}"') def step_plan_tracking_compressor(context: Context, plan_id: str) -> None: context.tracking_compressor = _TrackingCompressor() context.scope_resolver.register_plan( plan_id, SkeletonCompressor, context.tracking_compressor ) @given('a project-level tracking SkeletonCompressor for project "{project_id}"') def step_project_tracking_compressor(context: Context, project_id: str) -> None: context.tracking_compressor = _TrackingCompressor() context.scope_resolver.register_project( project_id, SkeletonCompressor, context.tracking_compressor ) @given("a parent payload with 1 fragment of {tokens:d} tokens") def step_parent_payload_single(context: Context, tokens: int) -> None: frag = _make_frag( uko_node="project://app/main.py", content="content", score=0.8, tokens=tokens, ) context.parent_payload = _make_payload((frag,)) # --------------------------------------------------------------------------- # When steps -- PipelineScopeResolver resolution # --------------------------------------------------------------------------- @when("I resolve the skeleton compressor with no plan or project context") def step_resolve_compressor_no_context(context: Context) -> None: context.resolution_result = context.scope_resolver.resolve_skeleton_compressor() @when("I resolve the preamble generator with no plan or project context") def step_resolve_preamble_no_context(context: Context) -> None: context.resolution_result = context.scope_resolver.resolve_preamble_generator() @when("I resolve the deduplicator with no plan or project context") def step_resolve_dedup_no_context(context: Context) -> None: context.resolution_result = context.scope_resolver.resolve_deduplicator() @when("I resolve the depth resolver with no plan or project context") def step_resolve_depth_no_context(context: Context) -> None: context.resolution_result = context.scope_resolver.resolve_depth_resolver() @when('I resolve the skeleton compressor for project "{project_id}"') def step_resolve_compressor_for_project(context: Context, project_id: str) -> None: context.resolution_result = context.scope_resolver.resolve_skeleton_compressor( project_id=project_id ) @when('I resolve the preamble generator for project "{project_id}"') def step_resolve_preamble_for_project(context: Context, project_id: str) -> None: context.resolution_result = context.scope_resolver.resolve_preamble_generator( project_id=project_id ) @when( 'I resolve the skeleton compressor for plan "{plan_id}" and project "{project_id}"' ) def step_resolve_compressor_for_plan_project( context: Context, plan_id: str, project_id: str ) -> None: context.resolution_result = context.scope_resolver.resolve_skeleton_compressor( plan_id=plan_id, project_id=project_id ) @when('I resolve the skeleton compressor for plan "{plan_id}"') def step_resolve_compressor_for_plan(context: Context, plan_id: str) -> None: context.resolution_result = context.scope_resolver.resolve_skeleton_compressor( plan_id=plan_id ) @when('I resolve the deduplicator for plan "{plan_id}"') def step_resolve_dedup_for_plan(context: Context, plan_id: str) -> None: context.resolution_result = context.scope_resolver.resolve_deduplicator( plan_id=plan_id ) @when('I resolve the depth resolver for plan "{plan_id}"') def step_resolve_depth_for_plan(context: Context, plan_id: str) -> None: context.resolution_result = context.scope_resolver.resolve_depth_resolver( plan_id=plan_id ) @when("I check the resolution scope for SkeletonCompressor with no context") def step_check_scope_no_context(context: Context) -> None: context.resolved_scope = context.scope_resolver.resolution_scope(SkeletonCompressor) @when('I check the resolution scope for SkeletonCompressor for plan "{plan_id}"') def step_check_scope_for_plan(context: Context, plan_id: str) -> None: context.resolved_scope = context.scope_resolver.resolution_scope( SkeletonCompressor, plan_id=plan_id ) # --------------------------------------------------------------------------- # When steps -- ContextInheritanceService # --------------------------------------------------------------------------- @when( 'I inherit context for child plan "{child_plan_id}" with budget {budget_tokens:d}' ) def step_inherit_context( context: Context, child_plan_id: str, budget_tokens: int ) -> None: budget = ContextBudget(max_tokens=budget_tokens, reserved_tokens=0) context.skeleton_fragments = context.inheritance_service.inherit_context( context.parent_payload, child_plan_id, budget, ) @when( 'I inherit context for child plan "{child_plan_id}" with budget {budget_tokens:d}' ' and project "{project_id}"' ) def step_inherit_context_with_project( context: Context, child_plan_id: str, budget_tokens: int, project_id: str ) -> None: budget = ContextBudget(max_tokens=budget_tokens, reserved_tokens=0) context.skeleton_fragments = context.inheritance_service.inherit_context( context.parent_payload, child_plan_id, budget, project_id=project_id, ) @when( 'I inherit context for child plan "{child_plan_id}" with budget {budget_tokens:d}' " and skeleton ratio {ratio:f}" ) def step_inherit_context_with_ratio( context: Context, child_plan_id: str, budget_tokens: int, ratio: float ) -> None: budget = ContextBudget(max_tokens=budget_tokens, reserved_tokens=0) context.skeleton_fragments = context.inheritance_service.inherit_context( context.parent_payload, child_plan_id, budget, skeleton_ratio=ratio, ) @when( 'I inherit from fragments for child plan "{child_plan_id}" with budget {budget_tokens:d}' ) def step_inherit_from_fragments( context: Context, child_plan_id: str, budget_tokens: int ) -> None: budget = ContextBudget(max_tokens=budget_tokens, reserved_tokens=0) context.skeleton_fragments = context.inheritance_service.inherit_from_fragments( context.parent_fragments, child_plan_id, budget, ) # --------------------------------------------------------------------------- # Then steps -- PipelineScopeResolver assertions # --------------------------------------------------------------------------- @then("the resolved skeleton compressor should be the global default") def step_assert_global_compressor(context: Context) -> None: assert context.resolution_result.component is context.global_compressor, ( f"Expected global compressor, got {context.resolution_result.component}" ) @then("the resolved skeleton compressor should be the project override") def step_assert_project_compressor(context: Context) -> None: assert context.resolution_result.component is context.project_compressor, ( f"Expected project compressor, got {context.resolution_result.component}" ) @then("the resolved skeleton compressor should be the plan override") def step_assert_plan_compressor(context: Context) -> None: assert context.resolution_result.component is context.plan_compressor, ( f"Expected plan compressor, got {context.resolution_result.component}" ) @then("the resolved preamble generator should be the global default") def step_assert_global_preamble(context: Context) -> None: assert context.resolution_result.component is context.global_preamble, ( f"Expected global preamble generator, got {context.resolution_result.component}" ) @then("the resolved preamble generator should be the project override") def step_assert_project_preamble(context: Context) -> None: assert context.resolution_result.component is context.project_preamble, ( f"Expected project preamble generator, got {context.resolution_result.component}" ) @then("the resolved deduplicator should be the global default") def step_assert_global_dedup(context: Context) -> None: assert context.resolution_result.component is context.global_dedup, ( f"Expected global deduplicator, got {context.resolution_result.component}" ) @then("the resolved deduplicator should be the plan override") def step_assert_plan_dedup(context: Context) -> None: assert context.resolution_result.component is context.plan_dedup, ( f"Expected plan deduplicator, got {context.resolution_result.component}" ) @then("the resolved depth resolver should be the global default") def step_assert_global_depth(context: Context) -> None: assert context.resolution_result.component is context.global_depth, ( f"Expected global depth resolver, got {context.resolution_result.component}" ) @then("the resolved depth resolver should be the plan override") def step_assert_plan_depth(context: Context) -> None: assert context.resolution_result.component is context.plan_depth, ( f"Expected plan depth resolver, got {context.resolution_result.component}" ) @then('the pipeline scope resolution scope should be "{scope}"') def step_assert_pipeline_resolution_scope(context: Context, scope: str) -> None: # Check both resolution_result and resolved_scope if hasattr(context, "resolution_result"): actual_scope = context.resolution_result.scope else: actual_scope = context.resolved_scope expected = ScopeLevel(scope) assert actual_scope == expected, ( f"Expected scope {expected!r}, got {actual_scope!r}" ) # --------------------------------------------------------------------------- # Then steps -- ContextInheritanceService assertions # --------------------------------------------------------------------------- @then("the skeleton fragments should be non-empty") def step_assert_skeleton_non_empty(context: Context) -> None: assert len(context.skeleton_fragments) > 0, ( "Expected non-empty skeleton fragments, got empty" ) @then("the skeleton fragments should be empty") def step_assert_skeleton_empty(context: Context) -> None: assert len(context.skeleton_fragments) == 0, ( f"Expected empty skeleton fragments, got {len(context.skeleton_fragments)}" ) @then("the skeleton fragments total tokens should be at most {max_tokens:d}") def step_assert_skeleton_tokens(context: Context, max_tokens: int) -> None: total = sum(f.token_count for f in context.skeleton_fragments) assert total <= max_tokens, f"Expected skeleton tokens <= {max_tokens}, got {total}" @then("the plan-level tracking compressor should have been called") def step_assert_plan_tracking_called(context: Context) -> None: assert context.tracking_compressor.called, ( "Expected plan-level tracking compressor to have been called" ) @then("the project-level tracking compressor should have been called") def step_assert_project_tracking_called(context: Context) -> None: assert context.tracking_compressor.called, ( "Expected project-level tracking compressor to have been called" ) @then("the service default skeleton ratio should be {ratio:f}") def step_assert_default_ratio(context: Context, ratio: float) -> None: actual = context.inheritance_service.default_skeleton_ratio assert actual == ratio, f"Expected default skeleton ratio {ratio}, got {actual}"