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

feat(acms): implement ACMS pipeline scope resolution and context inheritance for child plans #10774

Merged
HAL9000 merged 1 commits from feat/acms-scope-resolution-context-inheritance into master 2026-06-06 15:26:12 +00:00
Conversation 8 Commits 1 Files Changed 3
+1305
3 changed files with 1305 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
features/acms_scope_resolution.feature
+229
View File
@@ -0,0 +1,229 @@
@acms @acms_scope_resolution @scope_resolution
Feature: ACMS Pipeline Scope Resolution and Context Inheritance
As a CleverAgents developer
I want pipeline components to be overridable at global, project, and plan scope
So that context assembly can be customized per-plan or per-project with correct precedence
# ---------------------------------------------------------------------------
# PipelineScopeResolver -- Global scope
# ---------------------------------------------------------------------------
@scope_chain @global
Scenario: Global default skeleton compressor is resolved when no overrides exist
Given a fresh PipelineScopeResolver
And a global default SkeletonCompressor registered
When I resolve the skeleton compressor with no plan or project context
Then the resolved skeleton compressor should be the global default
And the pipeline scope resolution scope should be "global"
@scope_chain @global
Scenario: Global default preamble generator is resolved when no overrides exist
Given a fresh PipelineScopeResolver
And a global default PreambleGenerator registered
When I resolve the preamble generator with no plan or project context
Then the resolved preamble generator should be the global default
And the pipeline scope resolution scope should be "global"
@scope_chain @global
Scenario: Global default deduplicator is resolved when no overrides exist
Given a fresh PipelineScopeResolver
And a global default FragmentDeduplicator registered
When I resolve the deduplicator with no plan or project context
Then the resolved deduplicator should be the global default
And the pipeline scope resolution scope should be "global"
@scope_chain @global
Scenario: Global default depth resolver is resolved when no overrides exist
Given a fresh PipelineScopeResolver
And a global default DetailDepthResolver registered
When I resolve the depth resolver with no plan or project context
Then the resolved depth resolver should be the global default
And the pipeline scope resolution scope should be "global"
# ---------------------------------------------------------------------------
# PipelineScopeResolver -- Project scope
# ---------------------------------------------------------------------------
@scope_chain @project
Scenario: Project override takes precedence over global for skeleton compressor
Given a fresh PipelineScopeResolver
And a global default SkeletonCompressor registered
And a project-level SkeletonCompressor override for project "my-project"
When I resolve the skeleton compressor for project "my-project"
Then the resolved skeleton compressor should be the project override
And the pipeline scope resolution scope should be "project"
@scope_chain @project
Scenario: Project without override falls through to global for skeleton compressor
Given a fresh PipelineScopeResolver
And a global default SkeletonCompressor registered
When I resolve the skeleton compressor for project "other-project"
Then the resolved skeleton compressor should be the global default
And the pipeline scope resolution scope should be "global"
@scope_chain @project
Scenario: Project override takes precedence over global for preamble generator
Given a fresh PipelineScopeResolver
And a global default PreambleGenerator registered
And a project-level PreambleGenerator override for project "proj-a"
When I resolve the preamble generator for project "proj-a"
Then the resolved preamble generator should be the project override
And the pipeline scope resolution scope should be "project"
# ---------------------------------------------------------------------------
# PipelineScopeResolver -- Plan scope (highest priority)
# ---------------------------------------------------------------------------
@scope_chain @plan
Scenario: Plan override takes highest precedence over project and global
Given a fresh PipelineScopeResolver
And a global default SkeletonCompressor registered
And a project-level SkeletonCompressor override for project "my-project"
And a plan-level SkeletonCompressor override for plan "plan-001"
When I resolve the skeleton compressor for plan "plan-001" and project "my-project"
Then the resolved skeleton compressor should be the plan override
And the pipeline scope resolution scope should be "plan"
@scope_chain @plan
Scenario: Plan without override falls through to project then global
Given a fresh PipelineScopeResolver
And a global default SkeletonCompressor registered
And a project-level SkeletonCompressor override for project "my-project"
When I resolve the skeleton compressor for plan "plan-002" and project "my-project"
Then the resolved skeleton compressor should be the project override
And the pipeline scope resolution scope should be "project"
@scope_chain @plan
Scenario: Plan override for deduplicator takes highest precedence
Given a fresh PipelineScopeResolver
And a global default FragmentDeduplicator registered
And a plan-level FragmentDeduplicator override for plan "plan-dedup"
When I resolve the deduplicator for plan "plan-dedup"
Then the resolved deduplicator should be the plan override
And the pipeline scope resolution scope should be "plan"
@scope_chain @plan
Scenario: Plan override for depth resolver takes highest precedence
Given a fresh PipelineScopeResolver
And a global default DetailDepthResolver registered
And a plan-level DetailDepthResolver override for plan "plan-depth"
When I resolve the depth resolver for plan "plan-depth"
Then the resolved depth resolver should be the plan override
And the pipeline scope resolution scope should be "plan"
# ---------------------------------------------------------------------------
# PipelineScopeResolver -- resolution_scope introspection
# ---------------------------------------------------------------------------
@scope_chain @introspection
Scenario: resolution_scope returns global when only global is registered
Given a fresh PipelineScopeResolver
And a global default SkeletonCompressor registered
When I check the resolution scope for SkeletonCompressor with no context
Then the pipeline scope resolution scope should be "global"
@scope_chain @introspection
Scenario: resolution_scope returns plan when plan override exists
Given a fresh PipelineScopeResolver
And a global default SkeletonCompressor registered
And a plan-level SkeletonCompressor override for plan "scope-plan"
When I check the resolution scope for SkeletonCompressor for plan "scope-plan"
Then the pipeline scope resolution scope should be "plan"
# ---------------------------------------------------------------------------
# ContextInheritanceService -- basic inheritance
# ---------------------------------------------------------------------------
@context_inheritance @basic
Scenario: Child plan receives skeleton context from parent payload
Given a PipelineScopeResolver with a global DefaultSkeletonCompressor
And a ContextInheritanceService with default skeleton ratio 0.15
And a parent payload with 3 fragments totaling 1000 tokens
When I inherit context for child plan "01JQTESTPN00000000000000AA" with budget 2000
Then the skeleton fragments should be non-empty
@context_inheritance @basic
Scenario: Empty parent payload produces empty skeleton
Given a PipelineScopeResolver with a global DefaultSkeletonCompressor
And a ContextInheritanceService with default skeleton ratio 0.15
And an empty parent payload
When I inherit context for child plan "01JQTESTPN00000000000000AA" with budget 2000
Then the skeleton fragments should be empty
@context_inheritance @basic
Scenario: Zero skeleton budget produces empty skeleton
Given a PipelineScopeResolver with a global DefaultSkeletonCompressor
And a ContextInheritanceService with default skeleton ratio 0.0
And a parent payload with 3 fragments totaling 1000 tokens
When I inherit context for child plan "01JQTESTPN00000000000000AA" with budget 2000
Then the skeleton fragments should be empty
@context_inheritance @basic
Scenario: inherit_from_fragments compresses raw fragments for child plan
Given a PipelineScopeResolver with a global DefaultSkeletonCompressor
And a ContextInheritanceService with default skeleton ratio 0.15
And parent fragments with 2 fragments totaling 500 tokens
When I inherit from fragments for child plan "01JQTESTPN00000000000000AA" with budget 1000
Then the skeleton fragments should be non-empty
@context_inheritance @basic
Scenario: Empty parent fragments produces empty skeleton from inherit_from_fragments
Given a PipelineScopeResolver with a global DefaultSkeletonCompressor
And a ContextInheritanceService with default skeleton ratio 0.15
And empty parent fragments
When I inherit from fragments for child plan "01JQTESTPN00000000000000AA" with budget 1000
Then the skeleton fragments should be empty
# ---------------------------------------------------------------------------
# ContextInheritanceService -- scope resolution during inheritance
# ---------------------------------------------------------------------------
@context_inheritance @scope_chain
Scenario: Context inheritance uses plan-level compressor when registered
Given a PipelineScopeResolver with a global DefaultSkeletonCompressor
And a plan-level tracking SkeletonCompressor for plan "01JQTESTPN00000000000000AA"
And a ContextInheritanceService with default skeleton ratio 0.15
And a parent payload with 2 fragments totaling 500 tokens
When I inherit context for child plan "01JQTESTPN00000000000000AA" with budget 1000
Then the plan-level tracking compressor should have been called
@context_inheritance @scope_chain
Scenario: Context inheritance uses project-level compressor when registered
Given a PipelineScopeResolver with a global DefaultSkeletonCompressor
And a project-level tracking SkeletonCompressor for project "my-project"
And a ContextInheritanceService with default skeleton ratio 0.15
And a parent payload with 2 fragments totaling 500 tokens
When I inherit context for child plan "01JQTESTPN00000000000000AA" with budget 1000 and project "my-project"
Then the project-level tracking compressor should have been called
# ---------------------------------------------------------------------------
# ContextInheritanceService -- skeleton ratio
# ---------------------------------------------------------------------------
@context_inheritance @skeleton_ratio
Scenario: Custom skeleton ratio governs the skeleton budget
Given a PipelineScopeResolver with a global DefaultSkeletonCompressor
And a ContextInheritanceService with default skeleton ratio 0.15
And a parent payload with 1 fragment of 100 tokens
When I inherit context for child plan "01JQTESTPN00000000000000AA" with budget 1000 and skeleton ratio 0.5
Then the skeleton fragments should be non-empty
@context_inheritance @skeleton_ratio
Scenario: Default skeleton ratio is used when none specified
Given a PipelineScopeResolver with a global DefaultSkeletonCompressor
And a ContextInheritanceService with default skeleton ratio 0.2
Then the service default skeleton ratio should be 0.2
# ---------------------------------------------------------------------------
# ContextInheritanceService -- budget-respecting compressor
# ---------------------------------------------------------------------------
@context_inheritance @budget
Scenario: Tracking compressor respects budget during inheritance
Given a PipelineScopeResolver with a global DefaultSkeletonCompressor
And a plan-level tracking SkeletonCompressor for plan "01JQTESTPN00000000000000AA"
And a ContextInheritanceService with default skeleton ratio 0.15
And a parent payload with 3 fragments totaling 1000 tokens
When I inherit context for child plan "01JQTESTPN00000000000000AA" with budget 2000
Then the plan-level tracking compressor should have been called
And the skeleton fragments total tokens should be at most 300
features/steps/acms_scope_resolution_steps.py
+595
View File
@@ -0,0 +1,595 @@
"""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}"
src/cleveragents/application/services/acms_scope_resolver.py
+481
View File
@@ -0,0 +1,481 @@
"""ACMS Pipeline Scope Resolver and Context Inheritance Service.
Provides two key services for the ACMS pipeline:
- **PipelineScopeResolver** -- Resolves ACMS pipeline component overrides at
plan > project > global scope using the ``ComponentResolver`` 3-level scope
chain. Supports overriding ``SkeletonCompressor``, ``PreambleGenerator``,
``FragmentDeduplicator``, and ``DetailDepthResolver`` at any scope level.
- **ContextInheritanceService** -- Propagates skeleton context from a parent
plan to child subplans on subplan creation. Uses the ``SkeletonCompressor``
resolved for the child plan's scope to compress parent fragments into a
skeleton suitable for the child plan's token budget.
Based on ``docs/specification.md`` Epic #935 -- pipeline scope resolution
(plan > project > global precedence) and context inheritance for hierarchical
plan decomposition.
ISSUES CLOSED: #10016
"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Any
from cleveragents.application.services.component_resolver import (
ComponentResolver,
ResolutionResult,
ScopeLevel,
)
if TYPE_CHECKING:
from cleveragents.application.services.acms_service import (
SkeletonCompressor,
)
from cleveragents.domain.models.core.context_fragment import (
ContextBudget,
ContextFragment,
ContextPayload,
)
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# PipelineScopeResolver
# ---------------------------------------------------------------------------
class PipelineScopeResolver:
"""Resolve ACMS pipeline component overrides at plan > project > global scope.
Wraps :class:`ComponentResolver` to provide a domain-specific API for
resolving ACMS pipeline components. Supports overriding the following
component types at any scope level:
- ``SkeletonCompressor`` -- compresses parent context for child plan inheritance
- ``PreambleGenerator`` -- generates context preambles from assembled fragments
- ``FragmentDeduplicator`` -- removes duplicate fragments across strategy results
- ``DetailDepthResolver`` -- resolves detail level for each fragment based on budget
Scope precedence (highest to lowest):
1. **Plan scope** -- overrides registered for a specific plan ID
2. **Project scope** -- overrides registered for a specific project ID
3. **Global scope** -- default implementations registered at startup
Example::
resolver = PipelineScopeResolver()
resolver.register_global(SkeletonCompressor, DepthReductionCompressor())
resolver.register_project("my-project", SkeletonCompressor, custom_compressor)
resolver.register_plan("plan-001", SkeletonCompressor, plan_compressor)
# Resolves plan-level override (highest priority)
result = resolver.resolve_skeleton_compressor(
plan_id="plan-001", project_id="my-project"
)
assert result.scope == ScopeLevel.PLAN
Based on ``docs/specification.md`` Epic #935 -- pipeline scope resolution.
"""
def __init__(self, resolver: ComponentResolver | None = None) -> None:
self._resolver = resolver or ComponentResolver()
@property
def resolver(self) -> ComponentResolver:
"""Return the underlying :class:`ComponentResolver`."""
return self._resolver
# ------------------------------------------------------------------
# Registration
# ------------------------------------------------------------------
def register_global(
self,
component_type: type[Any],
implementation: object,
) -> None:
"""Register a global default implementation for a pipeline component.
Args:
component_type: The Protocol type to register (e.g. ``SkeletonCompressor``).
implementation: The implementation instance.
Raises:
ComponentRegistrationError: If *implementation* is ``None``.
"""
self._resolver.register_global(component_type, implementation)
logger.debug(
"Registered global pipeline component: %s",
component_type.__name__,
)
def register_project(
self,
project_id: str,
component_type: type[Any],
implementation: object,
) -> None:
"""Register a project-level override for a pipeline component.
Args:
project_id: The project identifier.
component_type: The Protocol type to register.
implementation: The implementation instance.
Raises:
ComponentRegistrationError: If *project_id* is empty or
*implementation* is ``None``.
"""
self._resolver.register_project(project_id, component_type, implementation)
logger.debug(
"Registered project pipeline component: %s for project %s",
component_type.__name__,
project_id,
)
def register_plan(
self,
plan_id: str,
component_type: type[Any],
implementation: object,
) -> None:
"""Register a plan-level override for a pipeline component.
Args:
plan_id: The plan identifier.
component_type: The Protocol type to register.
implementation: The implementation instance.
Raises:
ComponentRegistrationError: If *plan_id* is empty or
*implementation* is ``None``.
"""
self._resolver.register_plan(plan_id, component_type, implementation)
logger.debug(
"Registered plan pipeline component: %s for plan %s",
component_type.__name__,
plan_id,
)
# ------------------------------------------------------------------
# Resolution helpers
# ------------------------------------------------------------------
def _resolve(
self,
component_type: type[Any],
plan_id: str | None = None,
project_id: str | None = None,
) -> ResolutionResult:
"""Resolve a component through the scope chain."""
return self._resolver.resolve(
component_type,
plan_id=plan_id,
project_id=project_id,
)
def resolve_skeleton_compressor(
self,
plan_id: str | None = None,
project_id: str | None = None,
) -> ResolutionResult:
"""Resolve the ``SkeletonCompressor`` for the given scope.
Applies plan > project > global precedence.
Args:
plan_id: Optional plan identifier for plan-level override lookup.
project_id: Optional project identifier for project-level override lookup.
Returns:
:class:`ResolutionResult` with the resolved component and scope level.
Raises:
ComponentNotFoundError: If no implementation is registered.
"""
from cleveragents.application.services.acms_service import SkeletonCompressor
return self._resolve(SkeletonCompressor, plan_id=plan_id, project_id=project_id)
def resolve_preamble_generator(
self,
plan_id: str | None = None,
project_id: str | None = None,
) -> ResolutionResult:
"""Resolve the ``PreambleGenerator`` for the given scope.
Applies plan > project > global precedence.
Args:
plan_id: Optional plan identifier for plan-level override lookup.
project_id: Optional project identifier for project-level override lookup.
Returns:
:class:`ResolutionResult` with the resolved component and scope level.
Raises:
ComponentNotFoundError: If no implementation is registered.
"""
from cleveragents.application.services.acms_service import PreambleGenerator
return self._resolve(PreambleGenerator, plan_id=plan_id, project_id=project_id)
def resolve_deduplicator(
self,
plan_id: str | None = None,
project_id: str | None = None,
) -> ResolutionResult:
"""Resolve the ``FragmentDeduplicator`` for the given scope.
Applies plan > project > global precedence.
Args:
plan_id: Optional plan identifier for plan-level override lookup.
project_id: Optional project identifier for project-level override lookup.
Returns:
:class:`ResolutionResult` with the resolved component and scope level.
Raises:
ComponentNotFoundError: If no implementation is registered.
"""
from cleveragents.application.services.acms_service import FragmentDeduplicator
return self._resolve(
FragmentDeduplicator, plan_id=plan_id, project_id=project_id
)
def resolve_depth_resolver(
self,
plan_id: str | None = None,
project_id: str | None = None,
) -> ResolutionResult:
"""Resolve the ``DetailDepthResolver`` for the given scope.
Applies plan > project > global precedence.
Args:
plan_id: Optional plan identifier for plan-level override lookup.
project_id: Optional project identifier for project-level override lookup.
Returns:
:class:`ResolutionResult` with the resolved component and scope level.
Raises:
ComponentNotFoundError: If no implementation is registered.
"""
from cleveragents.application.services.acms_service import DetailDepthResolver
return self._resolve(
DetailDepthResolver, plan_id=plan_id, project_id=project_id
)
# ------------------------------------------------------------------
# Scope introspection
# ------------------------------------------------------------------
def resolution_scope(
self,
component_type: type[Any],
plan_id: str | None = None,
project_id: str | None = None,
) -> ScopeLevel:
"""Return the scope level at which a component would be resolved.
Args:
component_type: The Protocol type to check.
plan_id: Optional plan identifier.
project_id: Optional project identifier.
Returns:
The :class:`ScopeLevel` at which the component is registered.
Raises:
ComponentNotFoundError: If no implementation is registered.
"""
result = self._resolve(component_type, plan_id=plan_id, project_id=project_id)
return result.scope
# ---------------------------------------------------------------------------
# ContextInheritanceService
# ---------------------------------------------------------------------------
class ContextInheritanceService:
"""Propagate skeleton context from parent plans to child subplans.
When a child subplan is created, this service compresses the parent
plan's assembled context fragments into a skeleton form suitable for
the child plan's token budget. The compressed skeleton fragments are
then available for injection into the child plan's context assembly.
The ``SkeletonCompressor`` used for compression is resolved through the
``PipelineScopeResolver`` at the child plan's scope (plan > project >
global), allowing per-plan or per-project compression strategies.
Example::
service = ContextInheritanceService(scope_resolver)
skeleton = service.inherit_context(
parent_payload=parent_payload,
child_plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
budget=ContextBudget(max_tokens=2048),
skeleton_ratio=0.15,
)
# skeleton is a tuple of ContextFragment for the child plan
Based on ``docs/specification.md`` Epic #935 -- context inheritance for
hierarchical plan decomposition.
"""
def __init__(
self,
scope_resolver: PipelineScopeResolver,
*,
default_skeleton_ratio: float = 0.15,
) -> None:
self._scope_resolver = scope_resolver
self._default_skeleton_ratio = default_skeleton_ratio
@property
def default_skeleton_ratio(self) -> float:
"""Return the default skeleton ratio used when none is specified."""
return self._default_skeleton_ratio
def inherit_context(
self,
parent_payload: ContextPayload,
child_plan_id: str,
budget: ContextBudget,
*,
skeleton_ratio: float | None = None,
project_id: str | None = None,
) -> tuple[ContextFragment, ...]:
"""Compress parent context into skeleton fragments for a child plan.
Resolves the ``SkeletonCompressor`` at the child plan's scope and
uses it to compress the parent payload's fragments into a skeleton
form that fits within the skeleton budget.
The skeleton budget is computed as::
skeleton_budget = int(budget.available_tokens * skeleton_ratio)
Args:
parent_payload: The assembled context payload from the parent plan.
child_plan_id: The identifier of the child subplan being created.
budget: The token budget for the child plan's context assembly.
skeleton_ratio: Fraction of the budget to allocate for skeleton
context. Defaults to ``default_skeleton_ratio`` (0.15).
project_id: Optional project identifier for scope resolution.
Returns:
A tuple of compressed :class:`ContextFragment` instances for the
child plan. Empty tuple if the parent payload has no fragments or
the skeleton budget is zero.
"""
ratio = (
skeleton_ratio
if skeleton_ratio is not None
else self._default_skeleton_ratio
)
skeleton_budget = int(budget.available_tokens * ratio)
if skeleton_budget <= 0 or not parent_payload.fragments:
logger.debug(
"Skipping context inheritance: no budget or no parent fragments "
"(child_plan_id=%s, skeleton_budget=%d, parent_fragment_count=%d)",
child_plan_id,
skeleton_budget,
len(parent_payload.fragments),
)
return ()
result = self._scope_resolver.resolve_skeleton_compressor(
plan_id=child_plan_id,
project_id=project_id,
)
compressor: SkeletonCompressor = result.component # type: ignore[assignment]
skeleton_fragments = compressor.compress(
parent_payload.fragments,
skeleton_budget,
)
logger.info(
"Context inheritance complete: child_plan_id=%s, scope=%s, "
"skeleton_budget=%d, parent_fragments=%d, skeleton_fragments=%d",
child_plan_id,
result.scope.value,
skeleton_budget,
len(parent_payload.fragments),
len(skeleton_fragments),
)
return skeleton_fragments
def inherit_from_fragments(
self,
parent_fragments: tuple[ContextFragment, ...],
child_plan_id: str,
budget: ContextBudget,
*,
skeleton_ratio: float | None = None,
project_id: str | None = None,
) -> tuple[ContextFragment, ...]:
"""Compress parent fragments directly into skeleton for a child plan.
Variant of :meth:`inherit_context` that accepts raw fragments instead
of a full :class:`ContextPayload`. Useful when the parent context is
available as a fragment tuple rather than an assembled payload.
Args:
parent_fragments: Fragments from the parent plan to compress.
child_plan_id: The identifier of the child subplan being created.
budget: The token budget for the child plan's context assembly.
skeleton_ratio: Fraction of the budget to allocate for skeleton
context. Defaults to ``default_skeleton_ratio`` (0.15).
project_id: Optional project identifier for scope resolution.
Returns:
A tuple of compressed :class:`ContextFragment` instances.
"""
ratio = (
skeleton_ratio
if skeleton_ratio is not None
else self._default_skeleton_ratio
)
skeleton_budget = int(budget.available_tokens * ratio)
if skeleton_budget <= 0 or not parent_fragments:
return ()
result = self._scope_resolver.resolve_skeleton_compressor(
plan_id=child_plan_id,
project_id=project_id,
)
compressor: SkeletonCompressor = result.component # type: ignore[assignment]
skeleton_fragments = compressor.compress(parent_fragments, skeleton_budget)
logger.info(
"Context inheritance from fragments complete: child_plan_id=%s, "
"scope=%s, skeleton_budget=%d, parent_fragments=%d, skeleton_fragments=%d",
child_plan_id,
result.scope.value,
skeleton_budget,
len(parent_fragments),
len(skeleton_fragments),
)
return skeleton_fragments
__all__: list[str] = [
"ContextInheritanceService",
"PipelineScopeResolver",
]