Files
cleveragents-core/features/steps/depth_breadth_projection_steps.py
aditya 137d040c4d
CI / benchmark-publish (pull_request) Has been skipped
CI / build (pull_request) Successful in 18s
CI / helm (pull_request) Successful in 22s
CI / quality (pull_request) Successful in 56s
CI / lint (pull_request) Successful in 3m21s
CI / typecheck (pull_request) Successful in 4m0s
CI / security (pull_request) Successful in 4m19s
CI / unit_tests (pull_request) Successful in 9m32s
CI / docker (pull_request) Successful in 1m22s
CI / coverage (pull_request) Successful in 12m27s
CI / e2e_tests (pull_request) Successful in 20m3s
CI / integration_tests (pull_request) Successful in 21m20s
CI / status-check (pull_request) Successful in 1s
CI / benchmark-regression (pull_request) Successful in 54m56s
feat(acms): implement DepthReductionCompressor for skeleton compression
Add a production skeleton compressor that re-renders inherited fragments to overview depths via the UKO detail-level map chain, fits the result within the configured skeleton budget, and wires the pipeline default to the new compressor.

Address prior review feedback by extracting the render visitors into a dedicated module, restoring projected metadata to native runtime types, constraining builtin component resolution with an allowlist, and keeping child-context inheritance compatible with CRP context fragments for the Robot integration path.

Reproduced the Forgejo lint job in a clean python:3.13-slim container with the CI commands All checks passed! and 1740 files already formatted; both passed, so the earlier lint failure appears to have been transient runner behavior rather than a source-level defect.

ISSUES CLOSED: #919
2026-04-01 06:16:41 +00:00

708 lines
25 KiB
Python

"""Step definitions for features/depth_breadth_projection.feature.
Tests the Depth/Breadth Projection System and Skeleton Context Propagation:
ProjectionSpec, ProjectedNode, DepthBreadthProjector, PlanContextInheritance,
ChildContextResult, InheritanceConfig, and built-in DetailLevelMap presets.
"""
from __future__ import annotations
from behave import given, then, when
from behave.runner import Context
from pydantic import ValidationError
from cleveragents.application.services.depth_breadth_projection import (
ChildContextResult,
DepthBreadthProjector,
InheritanceConfig,
PlanContextInheritance,
ProjectedNode,
ProjectionSpec,
code_detail_map,
database_detail_map,
docs_detail_map,
)
from cleveragents.domain.models.acms.crp import ContextRequest
from cleveragents.domain.models.core.context_fragment import (
ContextFragment,
FragmentProvenance,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
_PROV = FragmentProvenance(resource_uri="test://projection")
def _make_frag(
uko_node: str = "test://node",
content: str = "test",
token_count: int = 100,
relevance_score: float = 0.8,
detail_depth: int = 3,
) -> ContextFragment:
return ContextFragment(
uko_node=uko_node,
content=content,
token_count=token_count,
relevance_score=relevance_score,
detail_depth=detail_depth,
provenance=_PROV,
)
class _FakeAssembledContext:
"""Minimal stand-in for AssembledContext."""
def __init__(
self,
fragments: list[ContextFragment],
context_hash: str = "abc123",
) -> None:
self.fragments = fragments
self.context_hash = context_hash
class _MockSkeletonCompressor:
"""Compressor that returns a fixed number of fragments."""
def __init__(self, n_fragments: int = 2) -> None:
self._n = n_fragments
def compress(
self,
fragments: tuple[ContextFragment, ...],
skeleton_budget: int,
) -> tuple[ContextFragment, ...]:
return tuple(fragments[: self._n])
# ---------------------------------------------------------------------------
# Given steps
# ---------------------------------------------------------------------------
@given("the depth/breadth projection modules are available")
def step_modules_available(context: Context) -> None:
pass
@given("a UKO graph with nodes A B C and edges A->B B->C")
def step_graph_abc(context: Context) -> None:
context.adjacency = {
"A": ["B"],
"B": ["C"],
}
@given("a UKO graph with nodes A B and edges A->B")
def step_graph_ab(context: Context) -> None:
context.adjacency = {
"A": ["B"],
}
@given("a UKO graph with nodes A and no edges")
def step_graph_a_only(context: Context) -> None:
adj: dict[str, list[str]] = {}
context.adjacency = adj
@given("a UKO graph with nodes A B C D and edges A->B C->D")
def step_graph_abcd(context: Context) -> None:
context.adjacency = {
"A": ["B"],
"C": ["D"],
}
@given('a code DetailLevelMap is registered for domain "{domain}"')
def step_register_code_map(context: Context, domain: str) -> None:
projector = DepthBreadthProjector()
projector.register_detail_map(domain, code_detail_map())
context.projector = projector
@given("a parent assembled context with {n:d} fragments at detail_depth {depth:d}")
def step_parent_context(context: Context, n: int, depth: int) -> None:
frags = [
_make_frag(
uko_node=f"module://mod_{i}",
content=f"module_{i}",
token_count=100,
detail_depth=depth,
)
for i in range(n)
]
context.parent_context = _FakeAssembledContext(frags, context_hash="parenthash123")
@given("a parent assembled context with the following inheritance fragments:")
def step_parent_context_table(context: Context) -> None:
frags: list[ContextFragment] = []
for row in context.table:
metadata: dict[str, str] = {}
if row["domain"].strip():
metadata["detail_level_domain"] = row["domain"]
relevance_score = float(row["score"]) if "score" in row.headings else 0.8
frags.append(
ContextFragment(
uko_node=row["uko_node"],
content=row["content"],
token_count=int(row["tokens"]),
detail_depth=int(row["depth"]),
relevance_score=relevance_score,
provenance=FragmentProvenance(resource_uri=row["resource_uri"]),
metadata=metadata,
)
)
context.parent_context = _FakeAssembledContext(frags, context_hash="parenthash123")
@given("a PlanContextInheritance service with default config")
def step_inheritance_default(context: Context) -> None:
context.inheritance_service = PlanContextInheritance()
@given(
"a PlanContextInheritance service with default config and min_skeleton_tokens {min_tok:d}"
)
def step_inheritance_default_min_tokens(context: Context, min_tok: int) -> None:
context.inheritance_service = PlanContextInheritance(
config=InheritanceConfig(min_skeleton_tokens=min_tok)
)
@given("a mock skeleton compressor that returns {n:d} fragments")
def step_mock_compressor(context: Context, n: int) -> None:
context.mock_compressor = _MockSkeletonCompressor(n_fragments=n)
@given("a PlanContextInheritance service with the mock compressor")
def step_inheritance_with_compressor(context: Context) -> None:
context.inheritance_service = PlanContextInheritance(
skeleton_compressor=context.mock_compressor,
)
@given(
"a PlanContextInheritance service with the mock compressor and min_skeleton_tokens {min_tok:d}"
)
def step_inheritance_with_min_tokens(context: Context, min_tok: int) -> None:
config = InheritanceConfig(min_skeleton_tokens=min_tok)
context.inheritance_service = PlanContextInheritance(
config=config,
skeleton_compressor=context.mock_compressor,
)
# ---------------------------------------------------------------------------
# When steps — ProjectionSpec
# ---------------------------------------------------------------------------
@when(
'I create a ProjectionSpec with focus "{focus}" and breadth {breadth:d} and depth {depth:d}'
)
def step_create_spec(context: Context, focus: str, breadth: int, depth: int) -> None:
context.proj_spec = ProjectionSpec(
focus=(focus,),
breadth=breadth,
depth=depth,
)
@when("I create a ProjectionSpec with empty focus")
def step_create_spec_empty_focus(context: Context) -> None:
context.proj_error = None
try:
context.proj_spec = ProjectionSpec(focus=(), breadth=2, depth=3)
except ValidationError as exc:
context.proj_error = exc
@when("I create a ProjectionSpec with negative depth")
def step_create_spec_negative_depth(context: Context) -> None:
context.proj_error = None
try:
context.proj_spec = ProjectionSpec(focus=("A",), breadth=2, depth=-1)
except (ValueError, ValidationError) as exc:
context.proj_error = exc
@when('I create a ProjectionSpec with focus "{focus}" and named depth "{depth}"')
def step_create_spec_named(context: Context, focus: str, depth: str) -> None:
context.proj_spec = ProjectionSpec(
focus=(focus,),
depth=depth,
)
# ---------------------------------------------------------------------------
# When steps — ProjectedNode
# ---------------------------------------------------------------------------
@when(
'I create a ProjectedNode with uri "{uri}" and distance {dist:d} and resolved_depth {depth:d}'
)
def step_create_projected_node(
context: Context, uri: str, dist: int, depth: int
) -> None:
context.proj_node = ProjectedNode(uri=uri, distance=dist, resolved_depth=depth)
# ---------------------------------------------------------------------------
# When steps — DepthBreadthProjector
# ---------------------------------------------------------------------------
@when('I project with focus "{focus}" and breadth {breadth:d} and depth {depth:d}')
def step_project(context: Context, focus: str, breadth: int, depth: int) -> None:
projector = getattr(context, "projector", None) or DepthBreadthProjector()
spec = ProjectionSpec(focus=(focus,), breadth=breadth, depth=depth)
context.proj_result = projector.project(spec, context.adjacency)
@when(
'I project with focus "{focus}" and breadth {breadth:d} and depth {depth:d} and gradient {gradient}'
)
def step_project_gradient(
context: Context, focus: str, breadth: int, depth: int, gradient: str
) -> None:
grad = gradient.strip().lower() == "true"
projector = DepthBreadthProjector()
spec = ProjectionSpec(
focus=(focus,), breadth=breadth, depth=depth, depth_gradient=grad
)
context.proj_result = projector.project(spec, context.adjacency)
@when(
'I project with focus "{focus}" and breadth {breadth:d} and named depth "{depth}" in domain "{domain}"'
)
def step_project_named(
context: Context, focus: str, breadth: int, depth: str, domain: str
) -> None:
projector = getattr(context, "projector", None) or DepthBreadthProjector()
spec = ProjectionSpec(focus=(focus,), breadth=breadth, depth=depth, domain=domain)
context.proj_error = None
try:
context.proj_result = projector.project(spec, context.adjacency)
except ValueError as exc:
context.proj_error = exc
@when(
'I project with multi-focus "{f1}" and "{f2}" and breadth {breadth:d} and depth {depth:d}'
)
def step_project_multi_focus(
context: Context, f1: str, f2: str, breadth: int, depth: int
) -> None:
projector = DepthBreadthProjector()
spec = ProjectionSpec(focus=(f1, f2), breadth=breadth, depth=depth)
context.proj_result = projector.project(spec, context.adjacency)
@when(
'I project to fragments with focus "{focus}" and breadth {breadth:d} and depth {depth:d}'
)
def step_project_fragments(
context: Context, focus: str, breadth: int, depth: int
) -> None:
projector = DepthBreadthProjector()
spec = ProjectionSpec(focus=(focus,), breadth=breadth, depth=depth)
context.proj_fragments = projector.project_to_fragments(spec, context.adjacency)
# ---------------------------------------------------------------------------
# When steps — DetailLevelMap presets
# ---------------------------------------------------------------------------
@when("I create the code detail level map preset")
def step_code_map(context: Context) -> None:
context.code_map = code_detail_map()
@when("I create the docs detail level map preset")
def step_docs_map(context: Context) -> None:
context.docs_map = docs_detail_map()
@when("I create the database detail level map preset")
def step_db_map(context: Context) -> None:
context.db_map = database_detail_map()
# ---------------------------------------------------------------------------
# When steps — PlanContextInheritance
# ---------------------------------------------------------------------------
@when('I compute child context with focus "{focus}" and budget {budget:d}')
def step_compute_child(context: Context, focus: str, budget: int) -> None:
context.child_error = None
try:
context.child_result = context.inheritance_service.compute_child_context(
parent_context=context.parent_context,
child_focus=[focus],
child_token_budget=budget,
)
except ValueError as exc:
context.child_error = exc
@when(
'I compute child context with focus "{focus}" and budget {budget:d} at depth_in_tree {pd:d} to {cd:d}'
)
def step_compute_child_depths(
context: Context, focus: str, budget: int, pd: int, cd: int
) -> None:
context.child_result = context.inheritance_service.compute_child_context(
parent_context=context.parent_context,
child_focus=[focus],
child_token_budget=budget,
parent_depth_in_tree=pd,
child_depth_in_tree=cd,
)
@when("I compute child context with parent_depth {pd:d} and child_depth {cd:d}")
def step_compute_child_invalid_depth(context: Context, pd: int, cd: int) -> None:
context.child_error = None
try:
context.child_result = context.inheritance_service.compute_child_context(
parent_context=context.parent_context,
child_focus=["class://X"],
child_token_budget=1000,
parent_depth_in_tree=pd,
child_depth_in_tree=cd,
)
except ValueError as exc:
context.child_error = exc
@when("I create an InheritanceConfig with defaults")
def step_default_config(context: Context) -> None:
context.inheritance_config = InheritanceConfig()
@when('I extract child focus from decisions "{d1}" and "{d2}"')
def step_extract_focus(context: Context, d1: str, d2: str) -> None:
context.child_focus = PlanContextInheritance.extract_child_focus([d1, d2])
@when('I extract child focus from empty decisions with fallback "{fallback}"')
def step_extract_focus_fallback(context: Context, fallback: str) -> None:
context.child_focus = PlanContextInheritance.extract_child_focus(
[], fallback_focus=[fallback]
)
@when("I create a ChildContextResult with a request and skeleton data")
def step_create_child_result(context: Context) -> None:
req = ContextRequest(focus=["class://Test"], breadth=1, depth=5)
frag = _make_frag(uko_node="test://skel", content="skeleton", token_count=50)
context.child_ctx_result = ChildContextResult(
request=req,
skeleton_fragments=(frag,),
parent_context_hash="hash123",
skeleton_budget=200,
depth_delta=1,
)
# ---------------------------------------------------------------------------
# Then steps — ProjectionSpec
# ---------------------------------------------------------------------------
@then('the projection spec focus should contain "{uri}"')
def step_spec_focus(context: Context, uri: str) -> None:
assert uri in context.proj_spec.focus, (
f"Expected {uri!r} in focus {context.proj_spec.focus}"
)
@then("the projection spec breadth should be {expected:d}")
def step_spec_breadth(context: Context, expected: int) -> None:
assert context.proj_spec.breadth == expected
@then("the projection spec depth should be {expected:d}")
def step_spec_depth_int(context: Context, expected: int) -> None:
assert context.proj_spec.depth == expected
@then('the projection spec depth should be "{expected}"')
def step_spec_depth_str(context: Context, expected: str) -> None:
assert context.proj_spec.depth == expected
@then("the projection spec depth_gradient should be True")
def step_spec_gradient_true(context: Context) -> None:
assert context.proj_spec.depth_gradient is True
@then('the projection spec domain should be "{expected}"')
def step_spec_domain(context: Context, expected: str) -> None:
assert context.proj_spec.domain == expected
@then("the projection spec should be immutable")
def step_spec_immutable(context: Context) -> None:
try:
context.proj_spec.breadth = 99 # type: ignore[misc]
assert False, "Should have raised" # noqa: B011
except (ValidationError, TypeError, AttributeError):
pass
@then("a projection ValidationError should be raised")
def step_proj_validation_error(context: Context) -> None:
assert context.proj_error is not None, "Expected ValidationError but none raised"
assert isinstance(context.proj_error, ValidationError)
@then('a projection ValueError should be raised mentioning "{keyword}"')
def step_proj_value_error(context: Context, keyword: str) -> None:
error = getattr(context, "proj_error", None) or getattr(
context, "child_error", None
)
assert error is not None, f"Expected ValueError mentioning {keyword!r}"
assert keyword.lower() in str(error).lower(), f"Expected '{keyword}' in: {error}"
# ---------------------------------------------------------------------------
# Then steps — ProjectedNode
# ---------------------------------------------------------------------------
@then('the projected node uri should be "{expected}"')
def step_node_uri(context: Context, expected: str) -> None:
assert context.proj_node.uri == expected
@then("the projected node distance should be {expected:d}")
def step_node_distance(context: Context, expected: int) -> None:
assert context.proj_node.distance == expected
@then("the projected node resolved_depth should be {expected:d}")
def step_node_resolved_depth(context: Context, expected: int) -> None:
assert context.proj_node.resolved_depth == expected
@then("the projected node should be immutable")
def step_node_immutable(context: Context) -> None:
try:
context.proj_node.distance = 99 # type: ignore[misc]
assert False, "Should have raised" # noqa: B011
except (ValidationError, TypeError, AttributeError):
pass
# ---------------------------------------------------------------------------
# Then steps — DepthBreadthProjector
# ---------------------------------------------------------------------------
@then("the projection should contain exactly {count:d} node")
def step_proj_count_singular(context: Context, count: int) -> None:
assert len(context.proj_result) == count, (
f"Expected {count} nodes, got {len(context.proj_result)}"
)
@then("the projection should contain exactly {count:d} nodes")
def step_proj_count(context: Context, count: int) -> None:
assert len(context.proj_result) == count, (
f"Expected {count} nodes, got {len(context.proj_result)}"
)
@then('the projected nodes should include "{uri}" at distance {dist:d}')
def step_proj_node_at_distance(context: Context, uri: str, dist: int) -> None:
found = [n for n in context.proj_result if n.uri == uri]
assert found, f"Node {uri!r} not found in projection"
assert found[0].distance == dist, (
f"Expected {uri!r} at distance {dist}, got {found[0].distance}"
)
@then('the projected node "{uri}" should have resolved_depth {depth:d}')
def step_proj_node_depth(context: Context, uri: str, depth: int) -> None:
found = [n for n in context.proj_result if n.uri == uri]
assert found, f"Node {uri!r} not found"
assert found[0].resolved_depth == depth, (
f"Expected {uri!r} depth={depth}, got {found[0].resolved_depth}"
)
@then('the projected node "{uri}" should have resolved_depth greater than {depth:d}')
def step_proj_node_depth_gt(context: Context, uri: str, depth: int) -> None:
found = [n for n in context.proj_result if n.uri == uri]
assert found, f"Node {uri!r} not found"
assert found[0].resolved_depth > depth, (
f"Expected {uri!r} depth > {depth}, got {found[0].resolved_depth}"
)
@then('the projected node "{uri}" should have resolved_depth less than {depth:d}')
def step_proj_node_depth_lt(context: Context, uri: str, depth: int) -> None:
found = [n for n in context.proj_result if n.uri == uri]
assert found, f"Node {uri!r} not found"
assert found[0].resolved_depth < depth, (
f"Expected {uri!r} depth < {depth}, got {found[0].resolved_depth}"
)
@then("the result should contain {count:d} ContextFragments")
def step_fragment_count(context: Context, count: int) -> None:
assert len(context.proj_fragments) == count
@then('the first fragment uko_node should be "{expected}"')
def step_first_frag_uko(context: Context, expected: str) -> None:
assert context.proj_fragments[0].uko_node == expected
@then('all fragments should have metadata key "{key}" set to True')
def step_fragments_metadata(context: Context, key: str) -> None:
for frag in context.proj_fragments:
assert frag.metadata.get(key) is True, (
f"Fragment {frag.uko_node} missing metadata key {key!r}"
)
# ---------------------------------------------------------------------------
# Then steps — DetailLevelMap presets
# ---------------------------------------------------------------------------
@then('the code map should resolve "{level}" to {expected:d}')
def step_code_resolve(context: Context, level: str, expected: int) -> None:
assert context.code_map.resolve(level) == expected
@then('the docs map should resolve "{level}" to {expected:d}')
def step_docs_resolve(context: Context, level: str, expected: int) -> None:
assert context.docs_map.resolve(level) == expected
@then('the database map should resolve "{level}" to {expected:d}')
def step_db_resolve(context: Context, level: str, expected: int) -> None:
assert context.db_map.resolve(level) == expected
# ---------------------------------------------------------------------------
# Then steps — PlanContextInheritance
# ---------------------------------------------------------------------------
@then("the child result should contain a ContextRequest")
def step_child_has_request(context: Context) -> None:
assert hasattr(context.child_result, "request"), "No request in result"
assert isinstance(context.child_result.request, ContextRequest)
@then('the child request focus should contain "{uri}"')
def step_child_focus_contains(context: Context, uri: str) -> None:
assert uri in context.child_result.request.focus
@then("the child request depth_gradient should be True")
def step_child_gradient(context: Context) -> None:
assert context.child_result.request.depth_gradient is True
@then("the child result depth_delta should be {expected:d}")
def step_child_depth_delta(context: Context, expected: int) -> None:
assert context.child_result.depth_delta == expected
@then("the child request depth should be greater than {depth:d}")
def step_child_depth_gt(context: Context, depth: int) -> None:
assert context.child_result.request.depth > depth, (
f"Expected depth > {depth}, got {context.child_result.request.depth}"
)
@then("the child request breadth should be at most {max_breadth:d}")
def step_child_breadth_max(context: Context, max_breadth: int) -> None:
assert context.child_result.request.breadth <= max_breadth, (
f"Expected breadth <= {max_breadth}, got {context.child_result.request.breadth}"
)
@then("the child result skeleton_fragments should have {count:d} entries")
def step_child_skeleton_count(context: Context, count: int) -> None:
assert len(context.child_result.skeleton_fragments) == count, (
f"Expected {count} skeleton fragments, got {len(context.child_result.skeleton_fragments)}"
)
@then("all child skeleton fragments should have detail depth at most {max_depth:d}")
def step_child_skeleton_depth_max(context: Context, max_depth: int) -> None:
depths = [
fragment.detail_depth for fragment in context.child_result.skeleton_fragments
]
assert all(depth <= max_depth for depth in depths), (
f"Expected child skeleton fragment depths <= {max_depth}, got {depths}"
)
@then('a child skeleton fragment should contain "{snippet}"')
def step_child_skeleton_contains(context: Context, snippet: str) -> None:
contents = [
fragment.content for fragment in context.child_result.skeleton_fragments
]
assert any(snippet in content for content in contents), (
f"Expected child skeleton fragment to contain {snippet!r}, got {contents!r}"
)
@then('the first child skeleton fragment should contain "{snippet}"')
def step_first_child_skeleton_contains(context: Context, snippet: str) -> None:
first = context.child_result.skeleton_fragments[0].content
assert snippet in first, (
f"Expected first child skeleton fragment to contain {snippet!r}, got {first!r}"
)
@then("the inheritance config skeleton_ratio should be {expected:g}")
def step_config_ratio(context: Context, expected: float) -> None:
assert context.inheritance_config.skeleton_ratio == expected
@then('the child focus should contain "{uri}"')
def step_focus_contains_single(context: Context, uri: str) -> None:
assert uri in context.child_focus
@then('the child focus should include both "{uri1}" and "{uri2}"')
def step_focus_contains_both(context: Context, uri1: str, uri2: str) -> None:
assert uri1 in context.child_focus
assert uri2 in context.child_focus
# ---------------------------------------------------------------------------
# Then steps — ChildContextResult
# ---------------------------------------------------------------------------
@then("the child context result should be immutable")
def step_result_immutable(context: Context) -> None:
try:
context.child_ctx_result.depth_delta = 99 # type: ignore[misc]
assert False, "Should have raised" # noqa: B011
except (ValidationError, TypeError, AttributeError):
pass
@then("the child context result should contain the request")
def step_result_has_request(context: Context) -> None:
assert isinstance(context.child_ctx_result.request, ContextRequest)