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

fix(acms): align SkeletonCompressorService.compress() with SkeletonCompressor protocol #3057

Merged
HAL9000 merged 1 commits from fix/acms-skeleton-compressor-signature into master 2026-05-30 13:08:56 +00:00
Conversation 28 Commits 1 Files Changed 10
+289 -419
10 changed files with 289 additions and 419 deletions
Download Patch File Download Diff File Expand all files Collapse all files
benchmarks/skeleton_compressor_bench.py
+36 -31
View File
@@ -1,8 +1,13 @@
"""ASV benchmarks for skeleton compressor overhead.
Measures the time to compress context fragments at various ratios
Measures the time to compress context fragments at various budgets
and fragment counts. The benchmark covers the hot path that runs
during subplan context inheritance.
The caller is responsible for converting a skeleton_ratio to an
absolute skeleton_budget before invoking compress(). These benchmarks
use representative absolute token budgets derived from typical ratios
applied to the fragment set sizes.
"""
from __future__ import annotations
@@ -33,9 +38,9 @@ from cleveragents.domain.models.core.context_fragment import ( # noqa: E402
_SKEL_PROV = FragmentProvenance(resource_uri="bench-skeleton://default")
def _build_fragments(count: int, tokens_each: int = 100) -> list[ContextFragment]:
def _build_fragments(count: int, tokens_each: int = 100) -> tuple[ContextFragment, ...]:
"""Generate *count* fragments for benchmarking."""
return [
return tuple(
ContextFragment(
fragment_id=f"bench-{i:05d}",
uko_node=f"bench-skeleton://file/{i}",
@@ -48,60 +53,60 @@ def _build_fragments(count: int, tokens_each: int = 100) -> list[ContextFragment
),
)
for i in range(count)
]
)
class SkeletonCompressorSmallSuite:
"""Benchmark compression of a small fragment set (10 fragments)."""
"""Benchmark compression of a small fragment set (10 fragments, 1000 tokens)."""
def setup(self) -> None:
self._svc = SkeletonCompressorService()
self._fragments = _build_fragments(10)
self._fragments = _build_fragments(10) # 1000 total tokens
def time_compress_ratio_0(self) -> None:
"""No compression (pass-through)."""
self._svc.compress(self._fragments, skeleton_ratio=0.0)
def time_compress_budget_full(self) -> None:
"""No compression (pass-through) — budget equals total tokens."""
self._svc.compress(self._fragments, 1000)
def time_compress_ratio_03(self) -> None:
"""Default ratio compression."""
self._svc.compress(self._fragments, skeleton_ratio=0.3)
def time_compress_budget_70pct(self) -> None:
"""Default-ratio-equivalent compression (budget = 70% of total)."""
self._svc.compress(self._fragments, 700)
def time_compress_ratio_05(self) -> None:
"""Moderate compression."""
self._svc.compress(self._fragments, skeleton_ratio=0.5)
def time_compress_budget_50pct(self) -> None:
"""Moderate compression (budget = 50% of total)."""
self._svc.compress(self._fragments, 500)
def time_compress_ratio_1(self) -> None:
"""Maximum compression."""
self._svc.compress(self._fragments, skeleton_ratio=1.0)
def time_compress_budget_zero(self) -> None:
"""Maximum compression — empty result."""
self._svc.compress(self._fragments, 0)
class SkeletonCompressorMediumSuite:
"""Benchmark compression of a medium fragment set (100 fragments)."""
"""Benchmark compression of a medium fragment set (100 fragments, 10000 tokens)."""
def setup(self) -> None:
self._svc = SkeletonCompressorService()
self._fragments = _build_fragments(100)
self._fragments = _build_fragments(100) # 10000 total tokens
def time_compress_ratio_03(self) -> None:
"""Default ratio compression."""
self._svc.compress(self._fragments, skeleton_ratio=0.3)
def time_compress_budget_70pct(self) -> None:
"""Default-ratio-equivalent compression."""
self._svc.compress(self._fragments, 7000)
def time_compress_ratio_05(self) -> None:
def time_compress_budget_50pct(self) -> None:
"""Moderate compression."""
self._svc.compress(self._fragments, skeleton_ratio=0.5)
self._svc.compress(self._fragments, 5000)
class SkeletonCompressorLargeSuite:
"""Benchmark compression of a large fragment set (1000 fragments)."""
"""Benchmark compression of a large fragment set (1000 fragments, 100000 tokens)."""
def setup(self) -> None:
self._svc = SkeletonCompressorService()
self._fragments = _build_fragments(1000)
self._fragments = _build_fragments(1000) # 100000 total tokens
def time_compress_ratio_05(self) -> None:
def time_compress_budget_50pct(self) -> None:
"""Moderate compression over 1000 fragments."""
self._svc.compress(self._fragments, skeleton_ratio=0.5)
self._svc.compress(self._fragments, 50000)
def time_compress_ratio_08(self) -> None:
def time_compress_budget_20pct(self) -> None:
"""Heavy compression over 1000 fragments."""
self._svc.compress(self._fragments, skeleton_ratio=0.8)
self._svc.compress(self._fragments, 20000)
features/services_init_coverage_r3.feature
-5
View File
@@ -429,11 +429,6 @@ Feature: Services __init__ lazy-import coverage (round 3)
Then svcov3 the attribute "PersistentSessionService" should be resolved
# ---------- skeleton_compressor ----------
Scenario: svcov3 lazy-load CompressionResult
Given svcov3 a fresh services module
When svcov3 I access lazy attribute "CompressionResult"
Then svcov3 the attribute "CompressionResult" should be resolved
Scenario: svcov3 lazy-load SkeletonCompressorService
Given svcov3 a fresh services module
When svcov3 I access lazy attribute "SkeletonCompressorService"
features/skeleton_compressor.feature
+39 -67
View File
@@ -6,134 +6,100 @@ Feature: Skeleton compressor
Background:
Given a skeleton compressor service
# --- ratio validation -------------------------------------------------
# --- budget validation -------------------------------------------------
Scenario: Reject ratio below 0.0
Scenario: Reject negative skeleton_budget
Given context fragments with total tokens 1000
When I compress with skeleton_ratio -0.1
Then the compressor should raise a ValueError for invalid ratio
When I compress with skeleton_budget -1
Then the compressor should raise a ValueError for invalid budget
Scenario: Reject ratio above 1.0
Scenario: Accept budget 0 returns empty
Given context fragments with total tokens 1000
When I compress with skeleton_ratio 1.5
Then the compressor should raise a ValueError for invalid ratio
When I compress with skeleton_budget 0
Then the result should contain zero fragments
Scenario: Accept ratio 0.0
Scenario: Accept large budget returns all fragments
Given context fragments with total tokens 1000
When I compress with skeleton_ratio 0.0
When I compress with skeleton_budget 1000
Then all fragments should be returned unchanged
Scenario: Accept ratio 1.0
Scenario: Accept budget at boundary 500
Given context fragments with total tokens 1000
When I compress with skeleton_ratio 1.0
Then only the highest-relevance fragment should be returned
Scenario: Accept ratio at boundary 0.5
Given context fragments with total tokens 1000
When I compress with skeleton_ratio 0.5
When I compress with skeleton_budget 500
Then compressed tokens should be at most 500
# --- default handling -------------------------------------------------
Scenario: Default ratio applied when None
Given context fragments with total tokens 1000
When I compress with skeleton_ratio not specified
Then the metadata ratio should equal the default 0.15
# --- stable ordering --------------------------------------------------
Scenario: Fragments with equal relevance are ordered by id
Given three fragments with equal relevance 0.5
When I compress with skeleton_ratio 0.0
When I compress with skeleton_budget 10000
Then fragments should be ordered by fragment_id ascending
Scenario: Fragments are ordered by relevance descending
Given fragments with relevances 0.9, 0.3, and 0.7
When I compress with skeleton_ratio 0.0
When I compress with skeleton_budget 10000
Then the first fragment should have relevance 0.9
And the last fragment should have relevance 0.3
# --- metadata ----------------------------------------------------------
Scenario: Metadata records correct token counts
Given context fragments with total tokens 1000
When I compress with skeleton_ratio 0.5
Then metadata original_tokens should be 1000
And metadata compressed_tokens should be at most 500
Scenario: Metadata records source decision IDs
Given fragments with known decision IDs
When I compress with skeleton_ratio 0.0
Then metadata should contain all source decision IDs
Scenario: Metadata ratio matches input
Given context fragments with total tokens 1000
When I compress with skeleton_ratio 0.7
Then metadata ratio should be 0.7
# --- edge cases -------------------------------------------------------
Scenario: Empty fragment list compresses to empty
Given an empty fragment list
When I compress with skeleton_ratio 0.5
When I compress with skeleton_budget 500
Then the result should contain zero fragments
And metadata original_tokens should be 0
And metadata compressed_tokens should equal 0
Scenario: Single fragment at ratio 0.5
Scenario: Single fragment fits within budget
Given a single fragment with 100 tokens
When I compress with skeleton_ratio 0.5
When I compress with skeleton_budget 200
Then the result should contain one fragment
Scenario: Single fragment kept even when it exceeds budget
Given a single fragment with 100 tokens
When I compress with skeleton_budget 50
Then the result should contain one fragment
# --- argument validation ----------------------------------------------
Scenario: Reject non-list fragments argument
When I compress with a non-list fragments argument
Scenario: Reject non-tuple fragments argument
When I compress with a non-tuple fragments argument
Then the compressor should raise a TypeError
Scenario: Reject fragment with negative token count
Given a fragment with negative token count
When I compress with skeleton_ratio 0.5
When I compress with skeleton_budget 500
Then the compressor should raise a ValueError for invalid fragment
Scenario: Reject fragment with empty id
Given a fragment with empty fragment_id
When I compress with skeleton_ratio 0.5
When I compress with skeleton_budget 500
Then the compressor should raise a ValueError for invalid fragment
Scenario: Reject fragment with relevance out of range
Given a fragment with relevance 1.5
When I compress with skeleton_ratio 0.5
When I compress with skeleton_budget 500
Then the compressor should raise a ValueError for invalid fragment
Scenario: Reject non-ContextFragment item in list
When I compress with a list containing a non-fragment item
Scenario: Reject non-ContextFragment item in tuple
When I compress with a tuple containing a non-fragment item
Then the compressor should raise a TypeError for invalid item
Scenario: Reject non-numeric skeleton_ratio
Scenario: Reject non-integer skeleton_budget
Given context fragments with total tokens 1000
When I compress with a non-numeric skeleton_ratio
Then the compressor should raise a TypeError for invalid ratio type
When I compress with a non-integer skeleton_budget
Then the compressor should raise a TypeError for invalid budget type
Scenario: Reject skeleton metadata with compressed exceeding original
When I create skeleton metadata with compressed exceeding original
Then a validation error should be raised for compressed exceeding original
# --- compression summary (original vs compressed) ----------------------
Scenario: Compression summary stored in plan metadata
Given context fragments with total tokens 1000
When I compress with skeleton_ratio 0.6
Then compressed_tokens should be less than original_tokens
# --- skeleton_ratio integration with plan model ------------------------
# --- skeleton_budget integration with plan model -----------------------
Scenario: Plan model accepts skeleton_metadata
Given a skeleton metadata with ratio 0.5 and 1000 original tokens and 500 compressed
When I attach skeleton_metadata to a plan
Then the plan should expose skeleton metadata in cli dict
# --- _validate_fragments edge-cases (lines 153-163) --------------------
# --- _validate_fragments edge-cases ------------------------------------
Scenario: Reject fragment with negative token_count
Given a context fragment constructed with negative token_count
@@ -149,3 +115,9 @@ Feature: Skeleton compressor
Given a context fragment constructed with empty fragment_id
When I validate the invalid fragments
Then the compressor should raise a ValueError mentioning "fragment_id must be non-empty"
# --- structural subtype assertion -------------------------------------
Scenario: SkeletonCompressorService satisfies SkeletonCompressor protocol
When I check that SkeletonCompressorService satisfies the SkeletonCompressor protocol
Then the structural subtype assertion should pass
features/steps/skeleton_compressor_steps.py
+76 -123
View File
@@ -6,8 +6,8 @@ from behave import given, then, when
from behave.runner import Context
from pydantic import ValidationError
from cleveragents.application.services.acms_service import SkeletonCompressor
from cleveragents.application.services.skeleton_compressor import (
DEFAULT_SKELETON_RATIO,
SkeletonCompressorService,
)
from cleveragents.domain.models.core.context_fragment import (
@@ -51,7 +51,7 @@ def _make_skel_fragment(
)
def _make_fragments(total_tokens: int, count: int = 4) -> list[ContextFragment]:
def _make_fragments(total_tokens: int, count: int = 4) -> tuple[ContextFragment, ...]:
"""Create *count* fragments summing to *total_tokens*."""
base = total_tokens // count
remainder = total_tokens - base * count
@@ -68,7 +68,7 @@ def _make_fragments(total_tokens: int, count: int = 4) -> list[ContextFragment]:
source_decision_id=f"01HX{'A' * 22}{i}" if i < 3 else None,
)
)
return frags
return tuple(frags)
def _make_plan_id() -> str:
@@ -94,7 +94,7 @@ def step_fragments_total(context: Context, total: int) -> None:
@given("three fragments with equal relevance {rel:g}")
def step_equal_relevance(context: Context, rel: float) -> None:
context.fragments = [
context.fragments = tuple(
_make_skel_fragment(
fragment_id=f"frag-{chr(ord('c') - i)}",
content="x" * 50,
@@ -102,12 +102,12 @@ def step_equal_relevance(context: Context, rel: float) -> None:
relevance_score=rel,
)
for i in range(3)
]
)
@given("fragments with relevances 0.9, 0.3, and 0.7")
def step_varied_relevances(context: Context) -> None:
context.fragments = [
context.fragments = (
_make_skel_fragment(
fragment_id="f-1", content="a", token_count=100, relevance_score=0.9
),
@@ -117,12 +117,12 @@ def step_varied_relevances(context: Context) -> None:
_make_skel_fragment(
fragment_id="f-3", content="c", token_count=100, relevance_score=0.7
),
]
)
@given("fragments with known decision IDs")
def step_known_ids(context: Context) -> None:
context.fragments = [
context.fragments = (
_make_skel_fragment(
fragment_id="f-1",
content="a",
@@ -137,46 +137,46 @@ def step_known_ids(context: Context) -> None:
relevance_score=0.5,
source_decision_id="01HXDECISION00000000000002",
),
]
)
@given("an empty fragment list")
def step_empty_frags(context: Context) -> None:
context.fragments = []
context.fragments = ()
@given("a single fragment with {tokens:d} tokens")
def step_single_frag(context: Context, tokens: int) -> None:
context.fragments = [
context.fragments = (
_make_skel_fragment(
fragment_id="only",
content="x" * tokens,
token_count=tokens,
relevance_score=0.8,
),
]
)
@given("a fragment with negative token count")
def step_neg_tokens(context: Context) -> None:
try:
context.fragments = [
context.fragments = (
_make_skel_fragment(
fragment_id="bad",
content="x",
token_count=-10,
relevance_score=0.5,
),
]
)
except (ValueError, ValidationError) as exc:
# Model-level validation now catches this at construction time.
context.compressor_error = exc
context.fragments = []
context.fragments = ()
@given("a fragment with empty fragment_id")
def step_empty_id(context: Context) -> None:
context.fragments = [
context.fragments = (
ContextFragment(
fragment_id="",
uko_node="skeleton://empty-id",
@@ -185,24 +185,24 @@ def step_empty_id(context: Context) -> None:
relevance_score=0.5,
provenance=_SKEL_PROV,
),
]
)
@given("a fragment with relevance {rel:g}")
def step_bad_relevance(context: Context, rel: float) -> None:
try:
context.fragments = [
context.fragments = (
_make_skel_fragment(
fragment_id="bad",
content="x",
token_count=10,
relevance_score=rel,
),
]
)
except (ValueError, ValidationError) as exc:
# Model-level validation now catches this at construction time.
context.compressor_error = exc
context.fragments = []
context.fragments = ()
@given(
@@ -220,8 +220,8 @@ def step_make_metadata(context: Context, ratio: float, orig: int, comp: int) ->
# --- When clauses ----------------------------------------------------------
@when("I compress with skeleton_ratio {ratio:g}")
def step_compress_ratio(context: Context, ratio: float) -> None:
@when("I compress with skeleton_budget {budget:d}")
def step_compress_budget(context: Context, budget: int) -> None:
# If model-level validation already caught an error during fragment
# construction (Given step), propagate it as the compress error.
if getattr(context, "compressor_error", None) is not None:
@@ -229,45 +229,37 @@ def step_compress_ratio(context: Context, ratio: float) -> None:
context.result = None
return
try:
context.result = context.service.compress(
context.fragments, skeleton_ratio=ratio
)
context.result = context.service.compress(context.fragments, budget)
context.error = None
except (ValueError, TypeError) as exc:
context.error = exc
context.result = None
@when("I compress with skeleton_ratio not specified")
def step_compress_default(context: Context) -> None:
context.result = context.service.compress(context.fragments)
context.error = None
@when("I compress with a non-list fragments argument")
def step_compress_non_list(context: Context) -> None:
@when("I compress with a non-tuple fragments argument")
def step_compress_non_tuple(context: Context) -> None:
try:
context.result = context.service.compress("not-a-list", skeleton_ratio=0.5) # type: ignore[arg-type]
context.result = context.service.compress("not-a-tuple", 500) # type: ignore[arg-type]
context.error = None
except TypeError as exc:
context.error = exc
context.result = None
@when("I compress with a list containing a non-fragment item")
@when("I compress with a tuple containing a non-fragment item")
def step_compress_non_fragment_item(context: Context) -> None:
try:
context.result = context.service.compress(
[
(
_make_skel_fragment(
fragment_id="ok",
content="x",
token_count=10,
relevance_score=0.5,
),
"not-a-fragment",
], # type: ignore[list-item]
skeleton_ratio=0.5,
"not-a-fragment", # type: ignore[arg-type]
),
500,
)
context.error = None
except TypeError as exc:
@@ -275,12 +267,12 @@ def step_compress_non_fragment_item(context: Context) -> None:
context.result = None
@when("I compress with a non-numeric skeleton_ratio")
def step_compress_non_numeric_ratio(context: Context) -> None:
@when("I compress with a non-integer skeleton_budget")
def step_compress_non_integer_budget(context: Context) -> None:
try:
context.result = context.service.compress(
context.fragments,
skeleton_ratio="bad", # type: ignore[arg-type]
"bad", # type: ignore[arg-type]
)
context.error = None
except TypeError as exc:
@@ -315,11 +307,21 @@ def step_attach_to_plan(context: Context) -> None:
)
@when(
"I check that SkeletonCompressorService satisfies the SkeletonCompressor protocol"
)
def step_check_protocol(context: Context) -> None:
context.protocol_check_result = isinstance(
SkeletonCompressorService(), SkeletonCompressor
)
context.protocol_check_error = None
# --- Then clauses ----------------------------------------------------------
@then("the compressor should raise a ValueError for invalid ratio")
def step_check_value_error_ratio(context: Context) -> None:
@then("the compressor should raise a ValueError for invalid budget")
def step_check_value_error_budget(context: Context) -> None:
assert context.error is not None, "Expected ValueError"
assert isinstance(context.error, ValueError), (
f"Expected ValueError, got {type(context.error)}"
@@ -329,95 +331,33 @@ def step_check_value_error_ratio(context: Context) -> None:
@then("all fragments should be returned unchanged")
def step_all_returned(context: Context) -> None:
assert context.result is not None
assert len(context.result.fragments) == len(context.fragments)
@then("only the highest-relevance fragment should be returned")
def step_top_one(context: Context) -> None:
assert context.result is not None
assert len(context.result.fragments) == 1
top_relevance = max(f.relevance_score for f in context.fragments)
assert context.result.fragments[0].relevance_score == top_relevance
assert len(context.result) == len(context.fragments)
@then("compressed tokens should be at most {limit:d}")
def step_tokens_limit(context: Context, limit: int) -> None:
assert context.result is not None
assert context.result.metadata.compressed_tokens <= limit
@then("the metadata ratio should equal the default {expected:g}")
def step_default_ratio(context: Context, expected: float) -> None:
assert context.result is not None
assert context.result.metadata.ratio == expected
assert expected == DEFAULT_SKELETON_RATIO
total = sum(f.token_count for f in context.result)
assert total <= limit, f"Expected total tokens <= {limit}, got {total}"
@then("fragments should be ordered by fragment_id ascending")
def step_ordered_by_id(context: Context) -> None:
assert context.result is not None
ids = [f.fragment_id for f in context.result.fragments]
ids = [f.fragment_id for f in context.result]
assert ids == sorted(ids), f"Expected sorted IDs, got {ids}"
@then("the first fragment should have relevance {rel:g}")
def step_first_relevance(context: Context, rel: float) -> None:
assert context.result is not None
assert context.result.fragments[0].relevance_score == rel
assert context.result[0].relevance_score == rel
@then("the last fragment should have relevance {rel:g}")
def step_last_relevance(context: Context, rel: float) -> None:
assert context.result is not None
assert context.result.fragments[-1].relevance_score == rel
@then("metadata original_tokens should be {expected:d}")
def step_original_tokens(context: Context, expected: int) -> None:
assert context.result is not None
assert context.result.metadata.original_tokens == expected
@then("metadata compressed_tokens should equal {expected:d}")
def step_compressed_equals(context: Context, expected: int) -> None:
assert context.result is not None
assert context.result.metadata.compressed_tokens == expected
@then("metadata compressed_tokens should be at most {limit:d}")
def step_compressed_at_most(context: Context, limit: int) -> None:
assert context.result is not None
assert context.result.metadata.compressed_tokens <= limit
@then("metadata should contain all source decision IDs")
def step_all_decision_ids(context: Context) -> None:
assert context.result is not None
expected_ids = {
f.metadata["source_decision_id"]
for f in context.fragments
if "source_decision_id" in f.metadata
}
actual_ids = set(context.result.metadata.source_decision_ids)
assert expected_ids == actual_ids
@then("metadata ratio should be {expected:g}")
def step_ratio_value(context: Context, expected: float) -> None:
assert context.result is not None
assert context.result.metadata.ratio == expected
@then("the result should contain zero fragments")
def step_zero_frags(context: Context) -> None:
assert context.result is not None
assert len(context.result.fragments) == 0
@then("the result should contain one fragment")
def step_one_frag(context: Context) -> None:
assert context.result is not None
assert len(context.result.fragments) == 1
assert context.result[-1].relevance_score == rel
@then("the compressor should raise a TypeError")
@@ -444,8 +384,8 @@ def step_type_error_item(context: Context) -> None:
)
@then("the compressor should raise a TypeError for invalid ratio type")
def step_type_error_ratio_type(context: Context) -> None:
@then("the compressor should raise a TypeError for invalid budget type")
def step_type_error_budget_type(context: Context) -> None:
assert context.error is not None, "Expected TypeError"
assert isinstance(context.error, TypeError), (
f"Expected TypeError, got {type(context.error)}"
@@ -457,13 +397,16 @@ def step_validation_error_compressed(context: Context) -> None:
assert context.meta_error is not None, "Expected validation error"
@then("compressed_tokens should be less than original_tokens")
def step_less_tokens(context: Context) -> None:
@then("the result should contain zero fragments")
def step_zero_frags(context: Context) -> None:
assert context.result is not None
assert (
context.result.metadata.compressed_tokens
< context.result.metadata.original_tokens
)
assert len(context.result) == 0
@then("the result should contain one fragment")
def step_one_frag(context: Context) -> None:
assert context.result is not None
assert len(context.result) == 1
@then("the plan should expose skeleton metadata in cli dict")
@@ -476,6 +419,16 @@ def step_plan_cli_dict(context: Context) -> None:
assert skel["compressed_tokens"] == context.skel_meta.compressed_tokens
@then("the structural subtype assertion should pass")
def step_protocol_assertion_passes(context: Context) -> None:
assert context.protocol_check_error is None, (
f"Protocol check raised: {context.protocol_check_error}"
)
assert context.protocol_check_result is True, (
"SkeletonCompressorService does not satisfy the SkeletonCompressor protocol"
)
# --- _validate_fragments edge-case steps --------------------------------
@@ -489,7 +442,7 @@ def step_frag_negative_token(context: Context) -> None:
relevance_score=0.5,
provenance=_SKEL_PROV,
)
context.invalid_fragments = [frag]
context.invalid_fragments = (frag,)
@given("a context fragment constructed with relevance_score {score}")
@@ -502,7 +455,7 @@ def step_frag_bad_relevance(context: Context, score: str) -> None:
relevance_score=float(score),
provenance=_SKEL_PROV,
)
context.invalid_fragments = [frag]
context.invalid_fragments = (frag,)
@given("a context fragment constructed with empty fragment_id")
@@ -515,7 +468,7 @@ def step_frag_empty_id(context: Context) -> None:
relevance_score=0.5,
provenance=_SKEL_PROV,
)
context.invalid_fragments = [frag]
context.invalid_fragments = (frag,)
@when("I validate the invalid fragments")
robot/helper_skeleton_compressor.py
+48 -61
View File
@@ -1,10 +1,10 @@
"""Helper script for skeleton compressor Robot Framework tests.
Usage:
python helper_skeleton_compressor.py compress <ratio>
python helper_skeleton_compressor.py validate-ratio-bounds
python helper_skeleton_compressor.py metadata-fields
python helper_skeleton_compressor.py compress <budget>
python helper_skeleton_compressor.py validate-budget-bounds
python helper_skeleton_compressor.py stable-ordering
python helper_skeleton_compressor.py protocol-check
"""
from __future__ import annotations
@@ -17,6 +17,9 @@ _SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
from cleveragents.application.services.acms_service import ( # noqa: E402
SkeletonCompressor,
)
from cleveragents.application.services.skeleton_compressor import ( # noqa: E402
SkeletonCompressorService,
)
@@ -51,9 +54,9 @@ def _make_skel_fragment(
)
def _sample_fragments() -> list[ContextFragment]:
def _sample_fragments() -> tuple[ContextFragment, ...]:
"""Build a repeatable set of sample fragments."""
return [
return (
_make_skel_fragment(
fragment_id="frag-001",
content="High relevance content " * 20,
@@ -75,73 +78,44 @@ def _sample_fragments() -> list[ContextFragment]:
relevance_score=0.3,
source_decision_id="01HXDECISION00000000000003",
),
]
)
def cmd_compress(ratio_str: str) -> None:
def cmd_compress(budget_str: str) -> None:
"""Compress sample fragments and print summary."""
ratio = float(ratio_str)
budget = int(budget_str)
svc = SkeletonCompressorService()
result = svc.compress(_sample_fragments(), skeleton_ratio=ratio)
meta = result.metadata
print(f"skeleton-compress-ok ratio={meta.ratio}")
print(f"original_tokens={meta.original_tokens}")
print(f"compressed_tokens={meta.compressed_tokens}")
print(f"fragment_count={len(result.fragments)}")
print(f"decision_ids={len(meta.source_decision_ids)}")
result = svc.compress(_sample_fragments(), budget)
total_tokens = sum(f.token_count for f in result)
print(f"skeleton-compress-ok budget={budget}")
print(f"fragment_count={len(result)}")
print(f"total_tokens={total_tokens}")
def cmd_validate_ratio_bounds() -> None:
"""Verify that out-of-range ratios raise ValueError."""
def cmd_validate_budget_bounds() -> None:
"""Verify that negative budgets raise ValueError."""
svc = SkeletonCompressorService()
frags = _sample_fragments()
for bad in (-0.1, 1.5, 2.0):
for bad in (-1, -10, -100):
try:
svc.compress(frags, skeleton_ratio=bad)
print(f"FAIL: ratio {bad} did not raise")
svc.compress(frags, bad)
print(f"FAIL: budget {bad} did not raise")
sys.exit(1)
except ValueError:
pass
# Valid bounds
for good in (0.0, 0.5, 1.0):
svc.compress(frags, skeleton_ratio=good)
for good in (0, 500, 1000):
svc.compress(frags, good)
print("skeleton-ratio-bounds-ok")
def cmd_metadata_fields() -> None:
"""Verify metadata fields are populated correctly."""
svc = SkeletonCompressorService()
result = svc.compress(_sample_fragments(), skeleton_ratio=0.5)
meta = result.metadata
checks_passed = True
if meta.ratio != 0.5:
print(f"FAIL: ratio={meta.ratio}")
checks_passed = False
if meta.original_tokens != 1000:
print(f"FAIL: original_tokens={meta.original_tokens}")
checks_passed = False
if meta.compressed_tokens > 500:
print(f"FAIL: compressed_tokens={meta.compressed_tokens} > 500")
checks_passed = False
if not meta.source_decision_ids:
print("FAIL: no decision IDs")
checks_passed = False
if checks_passed:
print("skeleton-metadata-ok")
else:
sys.exit(1)
print("skeleton-budget-bounds-ok")
def cmd_stable_ordering() -> None:
"""Verify fragments come out in deterministic order."""
svc = SkeletonCompressorService()
frags = [
frags = tuple(
_make_skel_fragment(
fragment_id=f"frag-{chr(ord('c') - i)}",
content="x",
@@ -149,13 +123,13 @@ def cmd_stable_ordering() -> None:
relevance_score=0.5,
)
for i in range(3)
]
)
r1 = svc.compress(frags, skeleton_ratio=0.0)
r2 = svc.compress(list(reversed(frags)), skeleton_ratio=0.0)
r1 = svc.compress(frags, 10000)
r2 = svc.compress(tuple(reversed(frags)), 10000)
ids1 = [f.fragment_id for f in r1.fragments]
ids2 = [f.fragment_id for f in r2.fragments]
ids1 = [f.fragment_id for f in r1]
ids2 = [f.fragment_id for f in r2]
if ids1 == ids2:
print("skeleton-stable-ordering-ok")
@@ -164,6 +138,19 @@ def cmd_stable_ordering() -> None:
sys.exit(1)
def cmd_protocol_check() -> None:
"""Verify SkeletonCompressorService satisfies the SkeletonCompressor protocol."""
svc = SkeletonCompressorService()
if isinstance(svc, SkeletonCompressor):
print("skeleton-protocol-check-ok")
else:
print(
"FAIL: SkeletonCompressorService does not satisfy"
" SkeletonCompressor protocol"
)
sys.exit(1)
def main() -> None:
"""Dispatch subcommand."""
if len(sys.argv) < 2:
@@ -173,15 +160,15 @@ def main() -> None:
cmd = sys.argv[1]
if cmd == "compress":
if len(sys.argv) < 3:
print("Usage: compress <ratio>")
print("Usage: compress <budget>")
sys.exit(1)
cmd_compress(sys.argv[2])
elif cmd == "validate-ratio-bounds":
cmd_validate_ratio_bounds()
elif cmd == "metadata-fields":
cmd_metadata_fields()
elif cmd == "validate-budget-bounds":
cmd_validate_budget_bounds()
elif cmd == "stable-ordering":
cmd_stable_ordering()
elif cmd == "protocol-check":
cmd_protocol_check()
else:
print(f"Unknown command: {cmd}")
sys.exit(1)
robot/skeleton_compressor.robot
+21 -21
View File
@@ -8,39 +8,39 @@ Suite Teardown Cleanup Test Environment
${HELPER} ${CURDIR}/helper_skeleton_compressor.py
*** Test Cases ***
Compress Fragments At Ratio 0.5
[Documentation] Compress sample fragments at 50% ratio and verify output
${result}= Run Process ${PYTHON} ${HELPER} compress 0.5 cwd=${WORKSPACE}
Compress Fragments At Budget 500
[Documentation] Compress sample fragments with budget of 500 tokens and verify output
${result}= Run Process ${PYTHON} ${HELPER} compress 500 cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} skeleton-compress-ok
Should Contain ${result.stdout} skeleton-compress-ok budget=500
Compress Fragments At Ratio 0.0
[Documentation] No compression — all fragments should survive
${result}= Run Process ${PYTHON} ${HELPER} compress 0.0 cwd=${WORKSPACE}
Compress Fragments At Full Budget
HAL9000 commented 2026-04-10 17:14:06 +00:00
Outdated
Review
Copy Link
Copy Source

Typo: comprest should be compress. This is causing the Compress Fragments At Full Budget test to fail with a non-zero exit code because the helper script does not recognise the comprest subcommand.

Typo: `comprest` should be `compress`. This is causing the `Compress Fragments At Full Budget` test to fail with a non-zero exit code because the helper script does not recognise the `comprest` subcommand.
[Documentation] Full budget — all fragments should survive
${result}= Run Process ${PYTHON} ${HELPER} compress 1000 cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} skeleton-compress-ok
Should Contain ${result.stdout} skeleton-compress-ok budget=1000
Should Contain ${result.stdout} fragment_count=3
Compress Fragments At Ratio 1.0
[Documentation] Maximum compression — only top fragment should survive
${result}= Run Process ${PYTHON} ${HELPER} compress 1.0 cwd=${WORKSPACE}
Compress Fragments At Budget Zero
[Documentation] Zero budget — no fragments should survive
${result}= Run Process ${PYTHON} ${HELPER} compress 0 cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} skeleton-compress-ok
Should Contain ${result.stdout} fragment_count=1
Should Contain ${result.stdout} skeleton-compress-ok budget=0
Should Contain ${result.stdout} fragment_count=0
Validate Ratio Bounds
[Documentation] Out-of-range ratios must raise ValueError
${result}= Run Process ${PYTHON} ${HELPER} validate-ratio-bounds cwd=${WORKSPACE}
Validate Budget Bounds
[Documentation] Out-of-range budgets must raise ValueError
${result}= Run Process ${PYTHON} ${HELPER} validate-budget-bounds cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} skeleton-ratio-bounds-ok
Should Contain ${result.stdout} skeleton-budget-bounds-ok
Verify Metadata Fields
[Documentation] Metadata should record ratio, tokens, and decision IDs
${result}= Run Process ${PYTHON} ${HELPER} metadata-fields cwd=${WORKSPACE}
Verify Protocol Conformance
[Documentation] SkeletonCompressorService must satisfy the SkeletonCompressor protocol
${result}= Run Process ${PYTHON} ${HELPER} protocol-check cwd=${WORKSPACE}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} skeleton-metadata-ok
Should Contain ${result.stdout} skeleton-protocol-check-ok
Verify Stable Fragment Ordering
[Documentation] Fragments with equal relevance must be ordered deterministically
src/cleveragents/application/services/__init__.py
-4
View File
@@ -255,9 +255,6 @@ if TYPE_CHECKING:
from cleveragents.application.services.session_service import (
PersistentSessionService as PersistentSessionService,
)
from cleveragents.application.services.skeleton_compressor import (
CompressionResult as CompressionResult,
)
from cleveragents.application.services.skeleton_compressor import (
SkeletonCompressorService as SkeletonCompressorService,
)
@@ -516,7 +513,6 @@ _LAZY_IMPORTS: dict[str, tuple[str, str]] = {
"resolve_severity": ("semantic_validation_service", "resolve_severity"),
"ServiceRetryWiring": ("service_retry_wiring", "ServiceRetryWiring"),
"PersistentSessionService": ("session_service", "PersistentSessionService"),
"CompressionResult": ("skeleton_compressor", "CompressionResult"),
"SkeletonCompressorService": ("skeleton_compressor", "SkeletonCompressorService"),
"SkillRegistryService": ("skill_registry_service", "SkillRegistryService"),
"CoordinationResult": ("strategy_coordinator", "CoordinationResult"),
src/cleveragents/application/services/acms_service.py
+1
View File
@@ -493,6 +493,7 @@ class PreambleGenerator(Protocol):
) -> str | None: ...
@runtime_checkable
class SkeletonCompressor(Protocol):
"""Compress parent context into a skeleton for child plan inheritance.
src/cleveragents/application/services/skeleton_compressor.py
+68 -105
View File
@@ -3,105 +3,91 @@
The ``SkeletonCompressorService`` takes a collection of context
fragments from a parent plan and produces a compressed representation
suitable for propagation to child plans. The compression is governed
by ``skeleton_ratio``:
by ``skeleton_budget`` — an absolute token count:
- **0.0** — no compression; all fragments pass through unchanged.
- **1.0** — maximum compression; only the single highest-relevance
fragment is kept (with minimal content).
- **0** — no fragments are kept (empty result).
- **large value** — all fragments pass through if they fit.
Fragments are sorted by ``relevance_score`` in **descending** order
(highest first). A stable secondary sort on ``fragment_id`` ensures
deterministic output for equal-relevance fragments.
The caller is responsible for converting a ``skeleton_ratio`` (fraction
of the total context budget) to an absolute ``skeleton_budget`` before
invoking ``compress()``. For example::
skeleton_budget = int(total_tokens * skeleton_ratio)
compressed = service.compress(fragments, skeleton_budget)
Based on ``docs/specification.md`` ACMS Skeleton section.
"""
from __future__ import annotations
from dataclasses import dataclass
from cleveragents.application.services.acms_service import SkeletonCompressor
from cleveragents.domain.models.core.context_fragment import ContextFragment
from cleveragents.domain.models.core.skeleton_metadata import SkeletonMetadata
# ---------------------------------------------------------------------------
# Public data structures
# ---------------------------------------------------------------------------
@dataclass(frozen=True, slots=True)
class CompressionResult:
"""Output of a skeleton compression pass.
Attributes:
fragments: The compressed (filtered/truncated) fragments,
ordered by relevance descending then fragment_id ascending.
metadata: Auditable metadata for the compression pass.
"""
fragments: tuple[ContextFragment, ...]
metadata: SkeletonMetadata
# ---------------------------------------------------------------------------
# Service
# ---------------------------------------------------------------------------
# Default skeleton_ratio when the caller does not specify one.
DEFAULT_SKELETON_RATIO: float = 0.15
class SkeletonCompressorService:
"""Compress context fragments for subplan context inheritance.
The service is stateless; all state lives in the arguments and
the returned ``CompressionResult``.
Implements the ``SkeletonCompressor`` protocol defined in
:mod:`cleveragents.application.services.acms_service`. The service
is stateless; all state lives in the arguments and the returned tuple.
The caller is responsible for computing ``skeleton_budget`` from a
ratio and the total available token count before invoking
:meth:`compress`.
"""
def compress(
self,
fragments: list[ContextFragment],
skeleton_ratio: float | None = None,
) -> CompressionResult:
"""Compress *fragments* according to *skeleton_ratio*.
fragments: tuple[ContextFragment, ...],
skeleton_budget: int,
) -> tuple[ContextFragment, ...]:
"""Compress *fragments* to fit within *skeleton_budget* tokens.
Args:
fragments: Context fragments to compress. Each must
have ``token_count >= 0`` and ``relevance_score`` in
``[0.0, 1.0]``.
skeleton_ratio: Compression ratio in ``[0.0, 1.0]``.
``None`` falls back to ``DEFAULT_SKELETON_RATIO``.
skeleton_budget: Maximum total token count for the returned
fragments. Must be a non-negative integer. When zero
an empty tuple is returned.
Returns:
A ``CompressionResult`` containing the filtered fragments
and associated ``SkeletonMetadata``.
A tuple of ``ContextFragment`` objects whose combined
``token_count`` does not exceed *skeleton_budget*, ordered
by relevance descending then ``fragment_id`` ascending.
Raises:
ValueError: If *skeleton_ratio* is outside ``[0.0, 1.0]``
or any fragment has invalid fields.
TypeError: If *fragments* is not a list.
TypeError: If *fragments* is not a tuple or *skeleton_budget*
is not an integer.
ValueError: If *skeleton_budget* is negative or any fragment
has invalid fields.
"""
# -- argument validation ------------------------------------------
if not isinstance(fragments, list):
raise TypeError(f"fragments must be a list, got {type(fragments).__name__}")
effective_ratio = (
skeleton_ratio if skeleton_ratio is not None else DEFAULT_SKELETON_RATIO
)
if not isinstance(effective_ratio, (int, float)):
if not isinstance(fragments, tuple):
raise TypeError(
f"skeleton_ratio must be a float, got {type(effective_ratio).__name__}"
f"fragments must be a tuple, got {type(fragments).__name__}"
)
if effective_ratio < 0.0 or effective_ratio > 1.0:
raise ValueError(
f"skeleton_ratio must be in [0.0, 1.0], got {effective_ratio}"
if not isinstance(skeleton_budget, int):
raise TypeError(
f"skeleton_budget must be an int, got {type(skeleton_budget).__name__}"
)
if skeleton_budget < 0:
raise ValueError(f"skeleton_budget must be >= 0, got {skeleton_budget}")
self._validate_fragments(fragments)
# -- compute totals -----------------------------------------------
original_tokens = sum(f.token_count for f in fragments)
if skeleton_budget == 0 or not fragments:
return ()
# -- stable sort: relevance desc, fragment_id asc -----------------
sorted_fragments = sorted(
@@ -110,35 +96,17 @@ class SkeletonCompressorService:
)
# -- select fragments within budget -------------------------------
kept = self._select_fragments(sorted_fragments, effective_ratio)
kept = self._select_fragments(sorted_fragments, skeleton_budget)
compressed_tokens = sum(f.token_count for f in kept)
source_ids = tuple(
f.metadata["source_decision_id"]
for f in kept
if "source_decision_id" in f.metadata
)
metadata = SkeletonMetadata(
ratio=effective_ratio,
original_tokens=original_tokens,
compressed_tokens=compressed_tokens,
source_decision_ids=source_ids,
)
return CompressionResult(
fragments=tuple(kept),
metadata=metadata,
)
return tuple(kept)
# ------------------------------------------------------------------
# Internal helpers
# ------------------------------------------------------------------
@staticmethod
def _validate_fragments(fragments: list[ContextFragment]) -> None:
"""Validate every fragment in the list.
def _validate_fragments(fragments: tuple[ContextFragment, ...]) -> None:
"""Validate every fragment in the tuple.
Raises:
TypeError: If an element is not a ``ContextFragment``.
@@ -165,31 +133,15 @@ class SkeletonCompressorService:
@staticmethod
def _select_fragments(
sorted_fragments: list[ContextFragment],
ratio: float,
budget: int,
) -> list[ContextFragment]:
"""Select which fragments to keep given the compression ratio.
"""Select which fragments to keep given the token *budget*.
When *ratio* is 0.0 every fragment is kept. When *ratio* is
1.0 only the single highest-relevance fragment survives (or
none if the input is empty). For intermediate values the
token budget is ``original_tokens * (1 - ratio)``; fragments
are added in relevance order until the budget is exhausted.
Fragments are added in relevance order (highest first) until the
budget is exhausted. At least one fragment is always included
when the input is non-empty, even if it exceeds the budget, to
ensure callers always receive some context.
"""
if not sorted_fragments:
return []
if ratio == 0.0:
return list(sorted_fragments)
original_tokens = sum(f.token_count for f in sorted_fragments)
# Budget: fraction of tokens to *keep*
budget = int(original_tokens * (1.0 - ratio))
# At maximum compression keep at most one fragment
if ratio == 1.0:
budget = 0
kept: list[ContextFragment] = []
used = 0
for frag in sorted_fragments:
@@ -198,11 +150,22 @@ class SkeletonCompressorService:
break
kept.append(frag)
used += frag.token_count
if used >= budget and budget > 0:
if used >= budget:
break
# At ratio 1.0, keep exactly the top fragment
if ratio == 1.0 and sorted_fragments:
return [sorted_fragments[0]]
return kept
# ---------------------------------------------------------------------------
# Structural subtype assertion — prevents future protocol drift
# ---------------------------------------------------------------------------
# This assertion verifies at import time that SkeletonCompressorService
# satisfies the SkeletonCompressor Protocol. If the protocol signature
# changes and the service is not updated, this will raise AssertionError
# immediately rather than failing silently at runtime.
assert isinstance(SkeletonCompressorService(), SkeletonCompressor), (
"SkeletonCompressorService does not satisfy the SkeletonCompressor protocol. "
"Ensure compress(fragments: tuple[ContextFragment, ...], skeleton_budget: int) "
"-> tuple[ContextFragment, ...] matches the protocol definition."
)
vulture_whitelist.py
-2
View File
@@ -574,8 +574,6 @@ validate_spawn # noqa: B018, F821
SkeletonMetadata # noqa: B018, F821
SkeletonCompressorService # noqa: B018, F821
ContextFragment # noqa: B018, F821
CompressionResult # noqa: B018, F821
DEFAULT_SKELETON_RATIO # noqa: B018, F821
skeleton_compressor_service # noqa: B018, F821
skeleton_metadata # noqa: B018, F821
source_decision_ids # noqa: B018, F821