feat(acms): implement depth/breadth projection system #603

Merged
freemo merged 1 commits from feature/m5-depth-breadth-projection into master 2026-03-05 15:47:37 +00:00
6 changed files with 1810 additions and 0 deletions
@@ -0,0 +1,120 @@
"""ASV benchmarks for depth/breadth projection system.
Measures latency of projection, gradient computation, and
skeleton context inheritance.
"""
from __future__ import annotations
from cleveragents.application.services.depth_breadth_projection import (
DepthBreadthProjector,
PlanContextInheritance,
ProjectedNode,
ProjectionSpec,
code_detail_map,
)
from cleveragents.domain.models.acms.crp import (
ContextFragment,
FragmentProvenance,
)
_PROV = FragmentProvenance(resource_uri="bench://projection")
def _make_frag(i: int, depth: int = 3) -> ContextFragment:
return ContextFragment(
uko_node=f"bench://node_{i}",
content=f"node_{i}",
token_count=100,
relevance_score=0.8,
detail_depth=depth,
provenance=_PROV,
)
class _FakeContext:
def __init__(self, n: int = 10) -> None:
self.fragments = [_make_frag(i) for i in range(n)]
self.context_hash = "benchhash"
def _build_linear_graph(n: int) -> dict[str, list[str]]:
"""Build a linear graph: node_0 -> node_1 -> ... -> node_(n-1)."""
adj: dict[str, list[str]] = {}
for i in range(n - 1):
adj[f"node_{i}"] = [f"node_{i + 1}"]
return adj
class TimeProjectionSpec:
"""Benchmark ProjectionSpec creation."""
def time_create_spec(self) -> None:
ProjectionSpec(focus=("class://Auth",), breadth=2, depth=9)
def time_create_spec_named_depth(self) -> None:
ProjectionSpec(focus=("class://Auth",), depth="FULL_SOURCE")
class TimeProjectedNode:
"""Benchmark ProjectedNode creation."""
def time_create_node(self) -> None:
ProjectedNode(uri="class://Foo", distance=1, resolved_depth=4)
class TimeProjector:
"""Benchmark DepthBreadthProjector."""
def setup(self) -> None:
self.projector = DepthBreadthProjector()
self.projector.register_detail_map("uko-code:", code_detail_map())
self.graph_10 = _build_linear_graph(10)
self.graph_100 = _build_linear_graph(100)
def time_project_small_graph(self) -> None:
spec = ProjectionSpec(focus=("node_0",), breadth=5, depth=9)
self.projector.project(spec, self.graph_10)
def time_project_medium_graph(self) -> None:
spec = ProjectionSpec(focus=("node_0",), breadth=10, depth=9)
self.projector.project(spec, self.graph_100)
def time_project_no_gradient(self) -> None:
spec = ProjectionSpec(
focus=("node_0",), breadth=5, depth=9, depth_gradient=False
)
self.projector.project(spec, self.graph_10)
def time_project_to_fragments(self) -> None:
spec = ProjectionSpec(focus=("node_0",), breadth=5, depth=9)
self.projector.project_to_fragments(spec, self.graph_10)
def time_project_named_depth(self) -> None:
spec = ProjectionSpec(
focus=("node_0",),
breadth=3,
depth="FULL_SOURCE",
domain="uko-code:",
)
self.projector.project(spec, self.graph_10)
class TimeInheritance:
"""Benchmark PlanContextInheritance."""
def setup(self) -> None:
self.service = PlanContextInheritance()
self.parent_ctx = _FakeContext(n=10)
def time_compute_child_context(self) -> None:
self.service.compute_child_context(
parent_context=self.parent_ctx,
child_focus=["class://Child"],
child_token_budget=4096,
)
def time_extract_focus(self) -> None:
PlanContextInheritance.extract_child_focus(
["class://A", "class://B", "class://C"]
)
+251
View File
@@ -0,0 +1,251 @@
@phase2 @acms @depth_breadth_projection
Feature: Depth/Breadth Projection System and Skeleton Context Propagation
As a CleverAgents developer
I want a projection system over the UKO graph
So that context can be materialized at varying detail levels
based on distance from focal entities, and child plans can
inherit compressed parent context via skeleton propagation
# ---------------------------------------------------------------------------
# ProjectionSpec model
# ---------------------------------------------------------------------------
@projection @model
Scenario: ProjectionSpec is a frozen Pydantic model
Given the depth/breadth projection modules are available
When I create a ProjectionSpec with focus "class://AuthManager" and breadth 2 and depth 9
Then the projection spec focus should contain "class://AuthManager"
And the projection spec breadth should be 2
And the projection spec depth should be 9
And the projection spec depth_gradient should be True
And the projection spec domain should be "uko-code:"
And the projection spec should be immutable
@projection @model
Scenario: ProjectionSpec rejects empty focus
Given the depth/breadth projection modules are available
When I create a ProjectionSpec with empty focus
Then a projection ValidationError should be raised
@projection @model
Scenario: ProjectionSpec rejects negative integer depth
Given the depth/breadth projection modules are available
When I create a ProjectionSpec with negative depth
Then a projection ValueError should be raised mentioning "non-negative"
@projection @model
Scenario: ProjectionSpec accepts named depth levels
Given the depth/breadth projection modules are available
When I create a ProjectionSpec with focus "module://auth" and named depth "FULL_SOURCE"
Then the projection spec depth should be "FULL_SOURCE"
# ---------------------------------------------------------------------------
# ProjectedNode model
# ---------------------------------------------------------------------------
@projection @model
Scenario: ProjectedNode is a frozen Pydantic model
Given the depth/breadth projection modules are available
When I create a ProjectedNode with uri "class://Foo" and distance 1 and resolved_depth 4
Then the projected node uri should be "class://Foo"
And the projected node distance should be 1
And the projected node resolved_depth should be 4
And the projected node should be immutable
# ---------------------------------------------------------------------------
# DepthBreadthProjector — BFS traversal
# ---------------------------------------------------------------------------
@projection @projector
Scenario: Projector performs BFS from focus with breadth 0
Given the depth/breadth projection modules are available
And a UKO graph with nodes A B C and edges A->B B->C
When I project with focus "A" and breadth 0 and depth 9
Then the projection should contain exactly 1 node
And the projected nodes should include "A" at distance 0
@projection @projector
Scenario: Projector performs BFS from focus with breadth 1
Given the depth/breadth projection modules are available
And a UKO graph with nodes A B C and edges A->B B->C
When I project with focus "A" and breadth 1 and depth 9
Then the projection should contain exactly 2 nodes
And the projected nodes should include "A" at distance 0
And the projected nodes should include "B" at distance 1
@projection @projector
Scenario: Projector performs BFS from focus with breadth 2
Given the depth/breadth projection modules are available
And a UKO graph with nodes A B C and edges A->B B->C
When I project with focus "A" and breadth 2 and depth 9
Then the projection should contain exactly 3 nodes
And the projected nodes should include "A" at distance 0
And the projected nodes should include "B" at distance 1
And the projected nodes should include "C" at distance 2
@projection @projector
Scenario: Projector applies depth gradient
Given the depth/breadth projection modules are available
And a UKO graph with nodes A B C and edges A->B B->C
When I project with focus "A" and breadth 2 and depth 9 and gradient True
Then the projected node "A" should have resolved_depth 9
And the projected node "B" should have resolved_depth greater than 0
And the projected node "B" should have resolved_depth less than 9
And the projected node "C" should have resolved_depth 0
@projection @projector
Scenario: Projector without gradient assigns uniform depth
Given the depth/breadth projection modules are available
And a UKO graph with nodes A B C and edges A->B B->C
When I project with focus "A" and breadth 2 and depth 9 and gradient False
Then the projected node "A" should have resolved_depth 9
And the projected node "B" should have resolved_depth 9
And the projected node "C" should have resolved_depth 9
@projection @projector
Scenario: Projector resolves named depth via DetailLevelMap
Given the depth/breadth projection modules are available
And a code DetailLevelMap is registered for domain "uko-code:"
And a UKO graph with nodes A B and edges A->B
When I project with focus "A" and breadth 1 and named depth "FULL_SOURCE" in domain "uko-code:"
Then the projected node "A" should have resolved_depth 9
And the projected node "B" should have resolved_depth 0
@projection @projector
Scenario: Projector raises error for unresolvable named depth
Given the depth/breadth projection modules are available
And a UKO graph with nodes A and no edges
When I project with focus "A" and breadth 0 and named depth "NONEXISTENT" in domain "unknown:"
Then a projection ValueError should be raised mentioning "no DetailLevelMap"
@projection @projector
Scenario: Projector handles multiple focus nodes
Given the depth/breadth projection modules are available
And a UKO graph with nodes A B C D and edges A->B C->D
When I project with multi-focus "A" and "C" and breadth 1 and depth 5
Then the projection should contain exactly 4 nodes
And the projected nodes should include "A" at distance 0
And the projected nodes should include "C" at distance 0
And the projected nodes should include "B" at distance 1
And the projected nodes should include "D" at distance 1
@projection @projector
Scenario: Projector converts projection to ContextFragments
Given the depth/breadth projection modules are available
And a UKO graph with nodes A B and edges A->B
When I project to fragments with focus "A" and breadth 1 and depth 4
Then the result should contain 2 ContextFragments
And the first fragment uko_node should be "A"
And all fragments should have metadata key "projected" set to True
# ---------------------------------------------------------------------------
# Built-in DetailLevelMap presets
# ---------------------------------------------------------------------------
@projection @presets
Scenario: Code detail level map has correct levels
Given the depth/breadth projection modules are available
When I create the code detail level map preset
Then the code map should resolve "MODULE_LISTING" to 0
And the code map should resolve "SIGNATURES" to 4
And the code map should resolve "FULL_SOURCE" to 9
@projection @presets
Scenario: Docs detail level map has correct levels
Given the depth/breadth projection modules are available
When I create the docs detail level map preset
Then the docs map should resolve "TITLE_ONLY" to 0
And the docs map should resolve "FULL_CONTENT" to 10
@projection @presets
Scenario: Database detail level map has correct levels
Given the depth/breadth projection modules are available
When I create the database detail level map preset
Then the database map should resolve "TABLE_LISTING" to 0
And the database map should resolve "FULL_CATALOG" to 11
# ---------------------------------------------------------------------------
# PlanContextInheritance — skeleton propagation
# ---------------------------------------------------------------------------
@inheritance @skeleton
Scenario: PlanContextInheritance computes child context request
Given the depth/breadth projection modules are available
And a parent assembled context with 3 fragments at detail_depth 3
And a PlanContextInheritance service with default config
When I compute child context with focus "class://Child" and budget 1000
Then the child result should contain a ContextRequest
And the child request focus should contain "class://Child"
And the child request depth_gradient should be True
And the child result depth_delta should be 1
@inheritance @skeleton
Scenario: PlanContextInheritance increases child detail from parent
Given the depth/breadth projection modules are available
And a parent assembled context with 3 fragments at detail_depth 3
And a PlanContextInheritance service with default config
When I compute child context with focus "class://Child" and budget 1000
Then the child request depth should be greater than 3
@inheritance @skeleton
Scenario: PlanContextInheritance narrows child breadth
Given the depth/breadth projection modules are available
And a parent assembled context with 3 fragments at detail_depth 3
And a PlanContextInheritance service with default config
When I compute child context with focus "class://Child" and budget 1000 at depth_in_tree 0 to 1
Then the child request breadth should be at most 2
@inheritance @skeleton
Scenario: PlanContextInheritance injects skeleton fragments
Given the depth/breadth projection modules are available
And a parent assembled context with 3 fragments at detail_depth 3
And a mock skeleton compressor that returns 2 fragments
And a PlanContextInheritance service with the mock compressor
When I compute child context with focus "class://Child" and budget 1000
Then the child result skeleton_fragments should have 2 entries
@inheritance @skeleton
Scenario: PlanContextInheritance skips skeleton when budget too small
Given the depth/breadth projection modules are available
And a parent assembled context with 3 fragments at detail_depth 3
And a mock skeleton compressor that returns 2 fragments
And a PlanContextInheritance service with the mock compressor and min_skeleton_tokens 500
When I compute child context with focus "class://Child" and budget 100
Then the child result skeleton_fragments should have 0 entries
@inheritance @skeleton
Scenario: PlanContextInheritance rejects invalid depth delta
Given the depth/breadth projection modules are available
And a parent assembled context with 3 fragments at detail_depth 3
And a PlanContextInheritance service with default config
When I compute child context with parent_depth 2 and child_depth 1
Then a projection ValueError should be raised mentioning "child_depth_in_tree"
@inheritance @skeleton
Scenario: PlanContextInheritance config uses default skeleton_ratio
Given the depth/breadth projection modules are available
When I create an InheritanceConfig with defaults
Then the inheritance config skeleton_ratio should be 0.2
@inheritance @skeleton
Scenario: PlanContextInheritance extract_child_focus returns parent decisions
Given the depth/breadth projection modules are available
When I extract child focus from decisions "class://A" and "class://B"
Then the child focus should include both "class://A" and "class://B"
@inheritance @skeleton
Scenario: PlanContextInheritance extract_child_focus uses fallback
Given the depth/breadth projection modules are available
When I extract child focus from empty decisions with fallback "class://Default"
Then the child focus should contain "class://Default"
# ---------------------------------------------------------------------------
# ChildContextResult model
# ---------------------------------------------------------------------------
@inheritance @model
Scenario: ChildContextResult is a frozen Pydantic model
Given the depth/breadth projection modules are available
When I create a ChildContextResult with a request and skeleton data
Then the child context result should be immutable
And the child context result should contain the request
@@ -0,0 +1,648 @@
"""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 (
ContextFragment,
ContextRequest,
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 PlanContextInheritance service with default config")
def step_inheritance_default(context: Context) -> None:
context.inheritance_service = PlanContextInheritance()
@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("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)
+49
View File
@@ -0,0 +1,49 @@
*** Settings ***
Documentation Integration tests for depth/breadth projection system
Library helper_depth_breadth_projection.py
*** Test Cases ***
ProjectionSpec Can Be Created
[Documentation] Verify ProjectionSpec model creation
${result}= Create Projection Spec class://Auth 2 9
Should Be True ${result}
ProjectedNode Can Be Created
[Documentation] Verify ProjectedNode model creation
${result}= Create Projected Node class://Foo 1 4
Should Be True ${result}
Projector Performs BFS Traversal
[Documentation] DepthBreadthProjector BFS from focus
${count}= Project And Count Nodes A 2 9
Should Be Equal As Integers ${count} 3
Projector Applies Depth Gradient
[Documentation] Verify gradient reduces depth at distance
${result}= Project With Gradient A 2 9
Should Be True ${result}
Projector Converts To Fragments
[Documentation] project_to_fragments produces ContextFragments
${count}= Project To Fragments Count A 1 4
Should Be Equal As Integers ${count} 2
Code Detail Map Resolves Levels
[Documentation] Built-in code detail map
${depth}= Resolve Code Level FULL_SOURCE
Should Be Equal As Integers ${depth} 9
Inheritance Computes Child Context
[Documentation] PlanContextInheritance produces ChildContextResult
${result}= Compute Child Context Default class://Child 1000
Should Be True ${result}
Inheritance With Skeleton Injection
[Documentation] Skeleton compressor fragments are injected
${count}= Compute Child With Skeleton class://Child 1000
Should Be Equal As Integers ${count} 2
Extract Child Focus From Decisions
[Documentation] extract_child_focus returns decisions
${result}= Extract Focus class://A class://B
Should Be True ${result}
+116
View File
@@ -0,0 +1,116 @@
"""Robot Framework helper library for depth/breadth projection tests."""
from __future__ import annotations
from cleveragents.application.services.depth_breadth_projection import (
ChildContextResult,
DepthBreadthProjector,
PlanContextInheritance,
ProjectedNode,
ProjectionSpec,
code_detail_map,
)
from cleveragents.domain.models.acms.crp import (
ContextFragment,
FragmentProvenance,
)
_PROV = FragmentProvenance(resource_uri="test://robot")
_GRAPH = {"A": ["B"], "B": ["C"]}
def _make_frag(
uko_node: str = "test://node",
content: str = "test",
token_count: int = 100,
detail_depth: int = 3,
) -> ContextFragment:
return ContextFragment(
uko_node=uko_node,
content=content,
token_count=token_count,
relevance_score=0.8,
detail_depth=detail_depth,
provenance=_PROV,
)
class _FakeContext:
def __init__(self, frags: list[ContextFragment], ctx_hash: str = "h123") -> None:
self.fragments = frags
self.context_hash = ctx_hash
class _MockCompressor:
def __init__(self, n: int = 2) -> None:
self._n = n
def compress(
self,
fragments: tuple[ContextFragment, ...],
skeleton_budget: int,
) -> tuple[ContextFragment, ...]:
return tuple(fragments[: self._n])
def create_projection_spec(focus: str, breadth: int, depth: int) -> bool:
spec = ProjectionSpec(focus=(focus,), breadth=breadth, depth=depth)
return spec.focus == (focus,) and spec.breadth == breadth
def create_projected_node(uri: str, distance: int, depth: int) -> bool:
node = ProjectedNode(uri=uri, distance=distance, resolved_depth=depth)
return node.uri == uri and node.distance == distance
def project_and_count_nodes(focus: str, breadth: int, depth: int) -> int:
projector = DepthBreadthProjector()
spec = ProjectionSpec(focus=(focus,), breadth=breadth, depth=depth)
return len(projector.project(spec, _GRAPH))
def project_with_gradient(focus: str, breadth: int, depth: int) -> bool:
projector = DepthBreadthProjector()
spec = ProjectionSpec(
focus=(focus,), breadth=breadth, depth=depth, depth_gradient=True
)
nodes = projector.project(spec, _GRAPH)
by_uri = {n.uri: n for n in nodes}
# Focus at full depth, farthest at 0
return by_uri[focus].resolved_depth == depth and by_uri["C"].resolved_depth == 0
def project_to_fragments_count(focus: str, breadth: int, depth: int) -> int:
projector = DepthBreadthProjector()
spec = ProjectionSpec(focus=(focus,), breadth=breadth, depth=depth)
return len(projector.project_to_fragments(spec, _GRAPH))
def resolve_code_level(level: str) -> int:
return code_detail_map().resolve(level)
def compute_child_context_default(focus: str, budget: int) -> bool:
frags = [_make_frag(uko_node=f"mod://m{i}") for i in range(3)]
ctx = _FakeContext(frags)
svc = PlanContextInheritance()
result = svc.compute_child_context(
parent_context=ctx, child_focus=[focus], child_token_budget=budget
)
return isinstance(result, ChildContextResult) and focus in result.request.focus
def compute_child_with_skeleton(focus: str, budget: int) -> int:
frags = [_make_frag(uko_node=f"mod://m{i}") for i in range(3)]
ctx = _FakeContext(frags)
comp = _MockCompressor(n=2)
svc = PlanContextInheritance(skeleton_compressor=comp)
result = svc.compute_child_context(
parent_context=ctx, child_focus=[focus], child_token_budget=budget
)
return len(result.skeleton_fragments)
def extract_focus(d1: str, d2: str) -> bool:
result = PlanContextInheritance.extract_child_focus([d1, d2])
return d1 in result and d2 in result
@@ -0,0 +1,626 @@
"""Depth/Breadth Projection System and Skeleton Context Propagation.
Implements the projection system described in ``docs/specification.md``
§ Core Concepts > Depth/Breadth Projection (lines 25265-25340) and the
context inheritance mechanism from § Architecture > ACMS > Context
Inheritance Mechanism (lines 43067-43128).
Components
----------
- ``ProjectionSpec`` Frozen Pydantic model capturing a projection request.
- ``ProjectedNode`` Frozen Pydantic model: a UKO node materialized at a
resolved detail depth and distance from focus.
- ``DepthBreadthProjector`` Stateless projector: BFS over a UKO graph
adjacency list, applying depth gradient to produce fragments at varying
detail levels.
- ``PlanContextInheritance`` Service that computes a child plan's
inherited context request from the parent plan's assembled context,
using the SkeletonCompressor for budget-controlled skeleton injection.
"""
from __future__ import annotations
import math
from collections import deque
from collections.abc import Mapping, Sequence
from typing import Any
import structlog
from pydantic import BaseModel, ConfigDict, Field, field_validator
from cleveragents.domain.models.acms.crp import (
ContextFragment,
ContextRequest,
DetailLevelMap,
FragmentProvenance,
)
from cleveragents.domain.models.core.uko import UKONode
logger = structlog.get_logger()
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
DEFAULT_SKELETON_RATIO: float = 0.2
"""Default fraction of child's token budget reserved for parent skeleton."""
EDGE_RELATIONS: tuple[str, ...] = (
"uko:references",
"uko:dependsOn",
"uko:contains",
)
"""UKO relation types traversed during breadth expansion."""
# ---------------------------------------------------------------------------
# ProjectionSpec — request description
# ---------------------------------------------------------------------------
class ProjectionSpec(BaseModel, frozen=True):
"""Captures a depth/breadth projection request.
A ``ProjectionSpec`` is the input to :class:`DepthBreadthProjector`:
it says *what* to focus on, how far to reach, and how much detail
to include.
Spec reference: ``docs/specification.md`` lines 25265-25270.
Attributes:
focus: URIs of entities at distance 0 (the focal items).
breadth: Maximum hop count outward from focus (0 = focus only).
depth: Target detail level (integer or named level string).
depth_gradient: When ``True``, detail decreases with distance.
domain: UKO domain namespace (e.g., ``"uko-code:"``).
"""
focus: tuple[str, ...] = Field(
...,
min_length=1,
description="URIs of focal entities (at least one required)",
)
breadth: int = Field(
default=2,
ge=0,
description="Maximum relationship hops from focus",
)
depth: int | str = Field(
default=3,
description=(
"Target detail level — raw integer or named level "
"resolved via the active DetailLevelMap"
),
)
depth_gradient: bool = Field(
default=True,
description="When True, detail decreases with distance from focus",
)
domain: str = Field(
default="uko-code:",
min_length=1,
description="UKO domain namespace for DetailLevelMap resolution",
)
@field_validator("depth")
@classmethod
def _validate_depth(
cls: type[ProjectionSpec],
v: int | str,
) -> int | str:
if isinstance(v, int) and v < 0:
raise ValueError("depth must be non-negative when integer")
return v
model_config = ConfigDict(str_strip_whitespace=True)
# ---------------------------------------------------------------------------
# ProjectedNode — materialized result
# ---------------------------------------------------------------------------
class ProjectedNode(BaseModel, frozen=True):
"""A UKO node materialized at a specific detail depth and distance.
Produced by :class:`DepthBreadthProjector`. Each projected node
carries the resolved integer depth (after gradient) and the hop
distance from the closest focus item.
Attributes:
uri: The UKO node URI.
distance: Hop count from the nearest focus item.
resolved_depth: Integer depth after gradient/clamping.
label: Human-readable label (from :class:`UKONode`).
"""
uri: str = Field(..., min_length=1)
distance: int = Field(..., ge=0)
resolved_depth: int = Field(..., ge=0)
label: str = ""
model_config = ConfigDict(str_strip_whitespace=True)
# ---------------------------------------------------------------------------
# DepthBreadthProjector — stateless graph projector
# ---------------------------------------------------------------------------
class DepthBreadthProjector:
"""BFS projector over a UKO graph with depth gradient.
Given an adjacency list of UKO edges and a :class:`ProjectionSpec`,
the projector performs a breadth-first traversal from the focus
items, collecting reachable nodes at each distance up to
``spec.breadth``. When ``spec.depth_gradient`` is enabled the
resolved detail level *decreases* linearly with distance:
.. math::
resolved = base - \\lfloor base \\cdot (distance / breadth) \\rfloor
where ``base`` is the resolved base depth at distance 0.
The projector is **stateless** it carries no mutable state and
can be reused across calls.
Spec reference: ``docs/specification.md`` lines 25265-25340.
"""
def __init__(
self, *, detail_maps: Mapping[str, DetailLevelMap] | None = None
) -> None:
self._detail_maps: dict[str, DetailLevelMap] = dict(detail_maps or {})
self._logger = logger.bind(component="DepthBreadthProjector")
@property
def detail_maps(self) -> Mapping[str, DetailLevelMap]:
"""Return registered detail level maps."""
return self._detail_maps
def register_detail_map(self, domain: str, detail_map: DetailLevelMap) -> None:
"""Register a :class:`DetailLevelMap` for a UKO domain."""
self._detail_maps[domain] = detail_map
def project(
self,
spec: ProjectionSpec,
adjacency: Mapping[str, Sequence[str]],
node_index: Mapping[str, UKONode] | None = None,
) -> list[ProjectedNode]:
"""Perform depth/breadth projection.
Args:
spec: The projection request.
adjacency: Maps each UKO URI to its neighbors (bidirectional
edges from ``uko:references``, ``uko:dependsOn``,
``uko:contains``).
node_index: Optional URIUKONode lookup for labels.
Returns:
List of :class:`ProjectedNode` sorted by distance then URI.
"""
base_depth = self._resolve_base_depth(spec)
# BFS from focus items
visited: dict[str, int] = {} # uri → min distance
queue: deque[tuple[str, int]] = deque()
for uri in spec.focus:
if uri not in visited:
visited[uri] = 0
queue.append((uri, 0))
while queue:
current, dist = queue.popleft()
if dist >= spec.breadth:
continue
for neighbor in adjacency.get(current, ()):
if neighbor not in visited:
visited[neighbor] = dist + 1
queue.append((neighbor, dist + 1))
# Build projected nodes
nodes: list[ProjectedNode] = []
for uri, distance in visited.items():
resolved = self._apply_gradient(
base_depth=base_depth,
distance=distance,
max_distance=spec.breadth,
gradient_enabled=spec.depth_gradient,
)
label = ""
if node_index and uri in node_index:
label = node_index[uri].label
nodes.append(
ProjectedNode(
uri=uri,
distance=distance,
resolved_depth=resolved,
label=label,
)
)
nodes.sort(key=lambda n: (n.distance, n.uri))
self._logger.debug(
"Projection complete",
focus_count=len(spec.focus),
breadth=spec.breadth,
base_depth=base_depth,
projected_nodes=len(nodes),
)
return nodes
def project_to_fragments(
self,
spec: ProjectionSpec,
adjacency: Mapping[str, Sequence[str]],
node_index: Mapping[str, UKONode] | None = None,
*,
token_estimator: _TokenEstimator | None = None,
) -> list[ContextFragment]:
"""Project and convert to :class:`ContextFragment` instances.
Convenience wrapper over :meth:`project` that produces fragments
suitable for direct injection into the ACMS pipeline.
Args:
spec: Projection request.
adjacency: Graph adjacency list.
node_index: Optional URIUKONode lookup.
token_estimator: Optional callable ``(uri, depth) -> int``
that estimates token cost. Defaults to depth + 1.
Returns:
List of :class:`ContextFragment` sorted by distance then URI.
"""
projected = self.project(spec, adjacency, node_index)
estimator = token_estimator or _default_token_estimator
fragments: list[ContextFragment] = []
for node in projected:
tokens = estimator(node.uri, node.resolved_depth)
relevance = self._distance_relevance(node.distance, spec.breadth)
fragments.append(
ContextFragment(
uko_node=node.uri,
content=f"{node.label or node.uri} [depth={node.resolved_depth}]",
detail_depth=node.resolved_depth,
token_count=tokens,
relevance_score=relevance,
provenance=FragmentProvenance(resource_uri=node.uri),
metadata={
"distance": node.distance,
"projected": True,
"base_depth": self._resolve_base_depth(spec),
},
)
)
return fragments
# -- internal helpers --------------------------------------------------
def _resolve_base_depth(self, spec: ProjectionSpec) -> int:
"""Resolve the base depth from the spec, using DetailLevelMap if available."""
detail_map = self._detail_maps.get(spec.domain)
if detail_map is not None:
return detail_map.resolve(spec.depth)
if isinstance(spec.depth, int):
return spec.depth
msg = (
f"Cannot resolve named depth {spec.depth!r}: "
f"no DetailLevelMap registered for domain {spec.domain!r}"
)
raise ValueError(msg)
@staticmethod
def _apply_gradient(
*,
base_depth: int,
distance: int,
max_distance: int,
gradient_enabled: bool,
) -> int:
"""Compute the resolved depth at a given distance.
When gradient is disabled, every node gets ``base_depth``.
When enabled, depth decreases linearly from ``base_depth``
at distance 0 to 0 at ``max_distance``. This matches the
spec examples (lines 25276-25304).
"""
if not gradient_enabled or max_distance == 0 or distance == 0:
return base_depth
reduction = math.floor(base_depth * (distance / max_distance))
return max(0, base_depth - reduction)
@staticmethod
def _distance_relevance(distance: int, max_distance: int) -> float:
"""Compute a 0.0-1.0 relevance score inversely proportional to distance."""
if max_distance == 0:
return 1.0
return round(max(0.0, 1.0 - distance / (max_distance + 1)), 4)
# ---------------------------------------------------------------------------
# Token estimation
# ---------------------------------------------------------------------------
_TokenEstimator = Any # callable[[str, int], int]
def _default_token_estimator(_uri: str, depth: int) -> int:
"""Placeholder estimator: tokens = depth + 1."""
return depth + 1
# ---------------------------------------------------------------------------
# PlanContextInheritance — skeleton context propagation
# ---------------------------------------------------------------------------
class InheritanceConfig(BaseModel, frozen=True):
"""Configuration for :class:`PlanContextInheritance`.
Attributes:
skeleton_ratio: Fraction of child's token budget reserved
for the parent skeleton (default 0.2, per spec §43082).
min_skeleton_tokens: Minimum tokens for a skeleton to be
useful. Below this, no skeleton is injected.
max_detail_increase: Maximum detail level increase per
depth-in-tree delta.
"""
skeleton_ratio: float = Field(
default=DEFAULT_SKELETON_RATIO,
ge=0.0,
le=1.0,
)
min_skeleton_tokens: int = Field(default=32, ge=0)
max_detail_increase: int = Field(default=3, ge=0)
model_config = ConfigDict(str_strip_whitespace=True)
class ChildContextResult(BaseModel, frozen=True):
"""Result of a child context inheritance computation.
Contains the :class:`ContextRequest` for the child plan plus
skeleton fragments and metadata from the parent context.
Attributes:
request: The child plan's context request.
skeleton_fragments: Compressed parent fragments for injection.
parent_context_hash: Hash of the parent's assembled context.
skeleton_budget: Tokens reserved for the skeleton.
depth_delta: Depth-in-tree difference between child and parent.
"""
request: ContextRequest
skeleton_fragments: tuple[ContextFragment, ...] = ()
parent_context_hash: str = ""
skeleton_budget: int = 0
depth_delta: int = 0
model_config = ConfigDict(str_strip_whitespace=True)
class PlanContextInheritance:
"""Compute child plan context from parent assembled context.
Implements the ``PlanContextInheritance.compute_child_context()``
pattern from ``docs/specification.md`` lines 43069-43101.
The service:
1. Extracts the child's focus from the parent plan's decisions.
2. Computes depth/breadth adjustments based on tree depth delta.
3. Builds a skeleton from parent context using the pipeline's
:class:`SkeletonCompressor`.
4. Returns a :class:`ChildContextResult` containing the request
and skeleton data.
"""
def __init__(
self,
*,
config: InheritanceConfig | None = None,
skeleton_compressor: _SkeletonCompressorLike | None = None,
) -> None:
self._config = config or InheritanceConfig()
self._compressor = skeleton_compressor
self._logger = logger.bind(component="PlanContextInheritance")
@property
def config(self) -> InheritanceConfig:
"""Return the inheritance configuration."""
return self._config
def compute_child_context(
self,
*,
parent_context: _AssembledContextLike,
child_focus: Sequence[str],
child_token_budget: int,
parent_depth_in_tree: int = 0,
child_depth_in_tree: int = 1,
) -> ChildContextResult:
"""Build the child plan's initial context request.
Args:
parent_context: The parent's :class:`AssembledContext`.
child_focus: URIs the child plan focuses on.
child_token_budget: Total token budget for the child.
parent_depth_in_tree: Parent's depth in the plan tree.
child_depth_in_tree: Child's depth in the plan tree.
Returns:
A :class:`ChildContextResult` containing the request and
skeleton fragments.
"""
depth_delta = child_depth_in_tree - parent_depth_in_tree
if depth_delta < 0:
msg = "child_depth_in_tree must be >= parent_depth_in_tree"
raise ValueError(msg)
# 1. Compute child detail = parent average + delta (clamped)
avg_detail = self._avg_detail(parent_context)
child_detail = min(
avg_detail + depth_delta * self._config.max_detail_increase,
9, # Absolute max detail (FULL_SOURCE)
)
# 2. Compute child breadth = narrower than parent
parent_breadth = getattr(parent_context, "avg_breadth", 2)
child_breadth = max(1, parent_breadth - depth_delta)
# 3. Skeleton budget
skeleton_budget = int(child_token_budget * self._config.skeleton_ratio)
skeleton_fragments: tuple[ContextFragment, ...] = ()
if skeleton_budget >= self._config.min_skeleton_tokens and self._compressor:
parent_fragments = tuple(parent_context.fragments)
skeleton_fragments = self._compressor.compress(
parent_fragments,
skeleton_budget,
)
self._logger.info(
"Skeleton injected",
skeleton_fragments=len(skeleton_fragments),
skeleton_budget=skeleton_budget,
parent_fragments=len(parent_fragments),
)
# 4. Build inherited context request
context_hash = getattr(parent_context, "context_hash", "")
request = ContextRequest(
focus=list(child_focus),
breadth=child_breadth,
depth=child_detail,
depth_gradient=True,
purpose=f"Inherited context from parent (hash={context_hash})",
)
return ChildContextResult(
request=request,
skeleton_fragments=skeleton_fragments,
parent_context_hash=context_hash,
skeleton_budget=skeleton_budget,
depth_delta=depth_delta,
)
@staticmethod
def _avg_detail(context: _AssembledContextLike) -> int:
"""Compute average detail depth of assembled fragments."""
fragments = context.fragments
if not fragments:
return 0
total = sum(getattr(f, "detail_depth", 0) for f in fragments)
return total // len(fragments)
@staticmethod
def extract_child_focus(
parent_decisions: Sequence[str],
fallback_focus: Sequence[str] | None = None,
) -> list[str]:
"""Extract child focus from parent plan decisions.
In the absence of a full decision model, this method accepts
a list of URIs from the parent's decisions and returns them
as the child's focus. A fallback list is used if decisions
are empty.
"""
if parent_decisions:
return list(parent_decisions)
return list(fallback_focus or [])
# ---------------------------------------------------------------------------
# Structural protocols (duck-typed, avoid circular imports)
# ---------------------------------------------------------------------------
_SkeletonCompressorLike = Any
"""Any object with ``compress(fragments, skeleton_budget) -> tuple[...]``."""
_AssembledContextLike = Any
"""Any object with ``fragments``, ``context_hash``, optionally ``avg_breadth``."""
# ---------------------------------------------------------------------------
# Built-in DetailLevelMap presets
# ---------------------------------------------------------------------------
def code_detail_map() -> DetailLevelMap:
"""Return the built-in source-code detail level map.
Maps named levels to integer depths following the spec examples
(lines 25280-25304):
- ``MODULE_LISTING`` = 0
- ``CLASS_NAMES`` = 2
- ``SIGNATURES`` = 4
- ``MEMBER_SUMMARY`` = 6
- ``FULL_SOURCE`` = 9
"""
return DetailLevelMap(
domain="uko-code:",
levels={
"MODULE_LISTING": 0,
"CLASS_NAMES": 2,
"SIGNATURES": 4,
"MEMBER_SUMMARY": 6,
"FULL_SOURCE": 9,
},
max_depth=9,
)
def docs_detail_map() -> DetailLevelMap:
"""Return the built-in documentation detail level map.
Maps named levels to integer depths following the spec examples
(lines 25310-25336):
- ``TITLE_ONLY`` = 0
- ``TABLE_OF_CONTENTS`` = 2
- ``SECTION_HEADINGS`` = 4
- ``TOPIC_SENTENCES`` = 6
- ``SECTION_SUMMARIES`` = 8
- ``FULL_CONTENT`` = 10
"""
return DetailLevelMap(
domain="uko-doc:",
levels={
"TITLE_ONLY": 0,
"TABLE_OF_CONTENTS": 2,
"SECTION_HEADINGS": 4,
"TOPIC_SENTENCES": 6,
"SECTION_SUMMARIES": 8,
"FULL_CONTENT": 10,
},
max_depth=10,
)
def database_detail_map() -> DetailLevelMap:
"""Return the built-in database detail level map.
- ``TABLE_LISTING`` = 0
- ``COLUMN_NAMES`` = 3
- ``SCHEMA_DETAILS`` = 6
- ``INDEX_DETAILS`` = 9
- ``FULL_CATALOG`` = 11
"""
return DetailLevelMap(
domain="uko-data:",
levels={
"TABLE_LISTING": 0,
"COLUMN_NAMES": 3,
"SCHEMA_DETAILS": 6,
"INDEX_DETAILS": 9,
"FULL_CATALOG": 11,
},
max_depth=11,
)