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

feat(acms): add ACMS v1 context pipeline #465

Merged
hamza.khyari merged 1 commits from feature/m5-acms-context into master 2026-03-05 01:33:54 +00:00
Conversation 8 Commits 1 Files Changed 12
+3543 -1
12 changed files with 3543 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+3
View File
@@ -193,6 +193,9 @@
names, full-URI layer detection, multi-parent `rdfs:subClassOf` (DAG
traversal via BFS), `rdfs:domain`/`rdfs:range`/`rdfs:subPropertyOf`
resolution, and non-existent parent validation. (#189)
- Added ACMS v1 context assembly pipeline with UKO and CRP integration, three
fusion strategies (relevance, recency, tiered), budget-constrained assembly,
and extensible strategy registration. (#188)
- Added `AgentSkillSpec` loader that parses SKILL.md frontmatter and progressive disclosure
sections into structured `SkillStep` objects with stable 1-based ordering. Supports
namespaced naming (`namespace/short_name`), optional `steps`, `version`, `compatibility`,
benchmarks/acms_pipeline_bench.py
+213
View File
@@ -0,0 +1,213 @@
"""ASV benchmarks for ACMS v1 context assembly pipeline.
Measures the performance of:
- ContextFragment creation (with and without metadata)
- ACMSPipeline.assemble with varying fragment counts
- Tiered fusion strategy ranking
- Recency strategy ranking
"""
from __future__ import annotations
import importlib
import sys
from datetime import UTC, datetime
from pathlib import Path
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
import cleveragents # noqa: E402
importlib.reload(cleveragents)
from cleveragents.application.services.acms_service import ACMSPipeline # noqa: E402
from cleveragents.domain.models.core.context_fragment import ( # noqa: E402
ContextBudget,
ContextFragment,
FragmentProvenance,
compute_context_hash,
)
# Default provenance for benchmark fragments.
_DEFAULT_PROV = FragmentProvenance(resource_uri="bench://default")
class ContextFragmentSuite:
"""Benchmark ContextFragment creation throughput."""
def time_fragment_creation(self) -> None:
"""Benchmark creating a ContextFragment with defaults."""
ContextFragment(
uko_node="bench://file",
content="benchmark content",
token_count=10,
provenance=_DEFAULT_PROV,
)
def time_fragment_with_metadata(self) -> None:
"""Benchmark creating a ContextFragment with metadata."""
ContextFragment(
uko_node="bench://decision",
content="benchmark with metadata",
relevance_score=0.85,
token_count=150,
tier="hot",
metadata={"author": "bench", "priority": "high"},
provenance=_DEFAULT_PROV,
)
class ACMSPipelineSuite:
"""Benchmark ACMSPipeline.assemble throughput."""
def setup(self) -> None:
"""Set up pipeline and fragment lists for assembly benchmarks."""
self._pipeline = ACMSPipeline()
self._budget = ContextBudget(max_tokens=100_000, reserved_tokens=0)
self._frags_10 = [
ContextFragment(
uko_node=f"bench://file/{i}",
content=f"fragment {i}",
relevance_score=round(i / 10, 1),
token_count=50,
provenance=_DEFAULT_PROV,
)
for i in range(10)
]
self._frags_100 = [
ContextFragment(
uko_node=f"bench://file/{i}",
content=f"fragment {i}",
relevance_score=round((i % 10) / 10, 1),
token_count=50,
provenance=_DEFAULT_PROV,
)
for i in range(100)
]
self._frags_1000 = [
ContextFragment(
uko_node=f"bench://file/{i}",
content=f"fragment {i}",
relevance_score=round((i % 10) / 10, 1),
token_count=50,
provenance=_DEFAULT_PROV,
)
for i in range(1000)
]
tiers = ("hot", "warm", "cold")
self._tiered_frags = [
ContextFragment(
uko_node=f"bench://file/{i}",
content=f"fragment {i}",
relevance_score=round((i % 10) / 10, 1),
token_count=50,
tier=tiers[i % 3],
provenance=_DEFAULT_PROV,
)
for i in range(100)
]
# Recency benchmark fragments with distinct timestamps
self._recency_frags = [
ContextFragment(
uko_node=f"bench://file/{i}",
content=f"fragment {i}",
token_count=50,
created_at=datetime(2024, 1, 1 + (i % 28), tzinfo=UTC),
provenance=_DEFAULT_PROV,
)
for i in range(100)
]
def time_assemble_10_fragments(self) -> None:
"""Benchmark assembling 10 fragments with relevance strategy."""
self._pipeline.assemble(
plan_id="01JQBENCHM00000000000000AA",
fragments=self._frags_10,
budget=self._budget,
)
def time_assemble_100_fragments(self) -> None:
"""Benchmark assembling 100 fragments with relevance strategy."""
self._pipeline.assemble(
plan_id="01JQBENCHM00000000000000AA",
fragments=self._frags_100,
budget=self._budget,
)
def time_assemble_1000_fragments(self) -> None:
"""Benchmark assembling 1000 fragments with relevance strategy."""
self._pipeline.assemble(
plan_id="01JQBENCHM00000000000000AA",
fragments=self._frags_1000,
budget=self._budget,
)
def time_tiered_strategy(self) -> None:
"""Benchmark assembling with tiered strategy."""
self._pipeline.assemble(
plan_id="01JQBENCHM00000000000000AA",
fragments=self._tiered_frags,
budget=self._budget,
strategy="tiered",
)
def time_recency_strategy(self) -> None:
"""Benchmark assembling with recency strategy."""
self._pipeline.assemble(
plan_id="01JQBENCHM00000000000000AA",
fragments=self._recency_frags,
budget=self._budget,
strategy="recency",
)
class ContextHashSuite:
"""Benchmark compute_context_hash in isolation.
Measures the length-prefixed SHA-256 hash function at various
fragment counts to detect performance regressions.
"""
def setup(self) -> None:
"""Build fragment tuples of varying sizes."""
self._frags_10 = tuple(
ContextFragment(
uko_node=f"bench://hash/{i}",
content=f"fragment content {i}" * 10,
token_count=50,
provenance=_DEFAULT_PROV,
)
for i in range(10)
)
self._frags_100 = tuple(
ContextFragment(
uko_node=f"bench://hash/{i}",
content=f"fragment content {i}" * 10,
token_count=50,
provenance=_DEFAULT_PROV,
)
for i in range(100)
)
self._frags_1000 = tuple(
ContextFragment(
uko_node=f"bench://hash/{i}",
content=f"fragment content {i}" * 10,
token_count=50,
provenance=_DEFAULT_PROV,
)
for i in range(1000)
)
def time_hash_10_fragments(self) -> None:
"""Benchmark hashing 10 fragments."""
compute_context_hash(self._frags_10)
def time_hash_100_fragments(self) -> None:
"""Benchmark hashing 100 fragments."""
compute_context_hash(self._frags_100)
def time_hash_1000_fragments(self) -> None:
"""Benchmark hashing 1000 fragments."""
compute_context_hash(self._frags_1000)
docs/reference/acms.md
+274
View File
@@ -0,0 +1,274 @@
# ACMS v1 Context Assembly Pipeline
This document covers the Adaptive Context Management System (ACMS) v1
context assembly pipeline, which integrates UKO (Unified Knowledge Ontology)
and CRP (Context Request Protocol) components with pluggable context
strategies.
## Overview
The ACMS pipeline assembles context fragments into a budget-constrained
payload for actor consumption. It supports multiple context strategies
that determine how fragments are ranked and selected to fit within the
token budget. The pipeline implements the spec's 10-component architecture
across three phases: Strategy Orchestration, Fragment Fusion, and Context
Finalization.
| Component | Description |
|-----------|-------------|
| `ContextFragment` | A single piece of context with UKO node, relevance, provenance, and tier |
| `FragmentProvenance` | Provenance trace linking a fragment back to its originating resource |
| `ContextBudget` | Token budget with max and reserved tokens |
| `ContextPayload` | Assembled payload with fragments, token count, budget usage, context hash, and provenance map |
| `ACMSPipeline` | Pipeline that applies a context strategy to assemble payloads |
## Context Fragments
Each `ContextFragment` represents a single piece of context:
| Field | Type | Description |
|-------|------|-------------|
| `fragment_id` | `str` | ULID identifier (auto-generated) |
| `uko_node` | `str` | UKO URI of the source node (required) |
| `content` | `str` | Rendered text content (max 1,000,000 chars) |
| `detail_depth` | `int` | Resolved depth: 0 (MODULE_LISTING) through 9 (FULL_SOURCE), default 0 |
| `token_count` | `int` | Actual token count of content (required, >= 0) |
| `relevance_score` | `float` | Score from 0.0 to 1.0 (default 0.5) |
| `provenance` | `FragmentProvenance` | Provenance trace (required) |
| `tier` | `str` | Priority tier: `"hot"`, `"warm"`, `"cold"` (default `"warm"`) |
| `metadata` | `dict[str, str]` | Arbitrary key-value metadata (max 64 entries) |
| `created_at` | `datetime` | UTC timestamp (auto-generated) |
### FragmentProvenance
| Field | Type | Description |
|-------|------|-------------|
| `resource_uri` | `str` | URI of the originating resource (required) |
| `location` | `str` | Location within the resource (default `""`) |
| `resource_type` | `str` | Type of originating resource (default `"unknown"`) |
### Tiers
Fragments are classified into three tiers:
- **hot** -- Critical context that should always be included first
- **warm** -- Standard context included when budget allows (default)
- **cold** -- Low-priority context included only if space remains
> **Note:** In v1, tiers are *sort-priority labels* used for ranking during
> assembly. They do **not** represent storage tiers with retention policies,
> promotion, or eviction semantics. Full hot/warm/cold storage-tier semantics
> will be implemented separately in `ContextTierService`.
## Budget Management
The `ContextBudget` defines the token budget for assembly:
- `max_tokens` -- Maximum total tokens (default 4096, must be >= 1)
- `reserved_tokens` -- Tokens reserved for system prompt (default 512, must be >= 0)
- `available_tokens` -- Computed as `max_tokens - reserved_tokens`
Validation: `reserved_tokens` must be strictly less than `max_tokens`.
The pipeline never exceeds the available token budget. Fragments that
would push the total over the limit are skipped, with early termination
once the budget is fully consumed.
## Context Payload
The assembled `ContextPayload` includes:
| Field | Type | Description |
|-------|------|-------------|
| `payload_id` | `str` | ULID identifier (auto-generated) |
| `plan_id` | `str` | Plan this payload was assembled for |
| `fragments` | `tuple[ContextFragment, ...]` | Selected fragments |
| `total_tokens` | `int` | Sum of fragment `token_count` values |
| `budget` | `ContextBudget` | Budget used for assembly |
| `budget_used` | `float` | Fraction of budget consumed (0.0-1.0) |
| `strategies_used` | `tuple[str, ...]` | Strategy names that contributed (immutable) |
| `context_hash` | `str` | SHA-256 hash of assembled content |
| `preamble` | `str \| None` | Optional structure summary |
| `provenance_map` | `dict[str, Any]` | Fragment ID -> provenance mapping |
| `assembled_at` | `datetime` | UTC timestamp |
Properties:
- `is_within_budget` -- `True` if total tokens do not exceed available budget
- `remaining_tokens` -- Tokens still available in the budget
## Context Strategies
Strategies implement the `ContextStrategy` protocol:
| Method | Description |
|--------|-------------|
| `name` (property) | Strategy identifier |
| `capabilities` (property) | `StrategyCapabilities` dataclass |
| `can_handle(request)` | Confidence (0.0-1.0) for handling a request |
| `assemble(fragments, budget)` | Rank/filter fragments to fit budget |
| `explain()` | Human-readable explanation |
### Relevance (default)
Sorts fragments by `relevance_score` descending. Highest-relevance
fragments are selected first until the budget is exhausted.
### Recency
Sorts fragments by `created_at` descending (datetime comparison, not
string comparison). Most recent fragments are selected first.
### Tiered
Groups fragments by tier priority (`hot` > `warm` > `cold`). Within
each tier, fragments are sorted by `relevance_score` descending.
## 10-Component Pipeline Architecture
The pipeline runs three phases:
All 10 components have explicit Protocol definitions and injectable
Default implementations. Inject custom components via the
`ACMSPipeline` constructor.
**Phase 1 — Strategy Orchestration:** `StrategySelector`,
`BudgetAllocator`, `StrategyExecutor` (v1 defaults: single-strategy
selection, full-budget allocation, synchronous execution)
**Phase 2 — Fragment Fusion:** `FragmentDeduplicator`,
`DetailDepthResolver`, `FragmentScorer`, `BudgetPacker`,
`FragmentOrderer` (v1 defaults: pass-through / no-op)
**Phase 3 — Context Finalization:** `PreambleGenerator`,
`SkeletonCompressor` (v1 defaults: no-op preamble, identity compression)
## v1 Known Limitations
The following are known deviations from the full specification, accepted
for the v1 implementation with a path to spec conformance in future
milestones:
| Area | v1 Behaviour | Spec Target | Planned |
|------|-------------|-------------|---------|
| `ContextStrategy.can_handle` signature | `(request: dict[str, Any]) -> float` | `(request: ContextRequest, backends: BackendSet) -> float` | M6 strategy registry aligns signatures |
| `ContextStrategy.assemble` signature | `(fragments, budget)` — receives pre-fetched fragments | `(request, backends, budget, plan_context)` — queries backends directly | M6 strategy registry aligns signatures |
| `StrategyCapabilities` fields | `supports_semantic_search`, `supports_graph_navigation`, `supports_temporal_archaeology`, `max_fragments` | `uses_text`, `uses_vector`, `uses_graph`, `uses_temporal`, `uko_levels`, `resource_types`, `quality_score` | M6 strategy registry uses spec field names |
| Pipeline components | All 10 Protocol + Default classes defined; defaults are pass-through stubs | Production implementations (parallel execution, dedup, scoring, compression) | Future milestone |
| Tiers | Sort-priority labels for ranking (`hot > warm > cold`) | Storage tiers with retention policies, promotion/demotion | `ContextTierService` in future milestone |
| `StrategySelector.select()` | `(strategies, request: dict)` | `(strategies, request: ContextRequest, backends: BackendSet)` — spec §42666 | M6 strategy registry aligns signatures |
| `BudgetAllocator.allocate()` | `(candidates, total_budget: int)` | `(candidates, total_budget, request: ContextRequest)` — spec §42682 | Future milestone |
| `StrategyExecutor.execute()` | `(allocations, fragments, budget)` | `(allocations, request: ContextRequest, backends: BackendSet, plan_context: PlanContext)` — spec §42698 | M6 strategy registry aligns signatures |
| `BudgetPacker.pack()` | `(fragments: Sequence[ContextFragment], budget: ContextBudget)` | `(scored_fragments: list[ScoredFragment], budget: int, detail_level_maps)` — spec §42763 | Future milestone |
| `SkeletonCompressor.compress()` | `(fragments: tuple, skeleton_budget: int) -> tuple` | `(parent_context: AssembledContext, child_focus: list[str], skeleton_budget: int) -> AssembledContext` — spec §42811 | Future milestone |
| `DetailDepthResolver.resolve()` | Missing `budget` parameter | `resolve(fragments, budget)` | Future milestone |
| `FragmentScorer.score()` | Missing `plan_context` param; returns `Sequence[ContextFragment]` | Returns `list[ScoredFragment]` with `plan_context` | Future milestone |
| `PreambleGenerator.generate()` | Missing `strategies_used`, `budget_used`, `max_tokens` params | Full parameter set per spec §42795 | Future milestone |
| `provenance_map` keying | Keyed by `fragment_id` (ULID) | Spec keys by `uko_node` (UKO URI) | Under review — `fragment_id` avoids collisions when multiple fragments share a `uko_node` |
## Extension Points
Register custom context strategies at runtime:
```python
from cleveragents.application.services.acms_service import (
ACMSPipeline,
ContextStrategy,
StrategyCapabilities,
)
from cleveragents.domain.models.core.context_fragment import (
ContextBudget,
ContextFragment,
FragmentProvenance,
)
from collections.abc import Sequence
from typing import Any
class MyCustomStrategy:
@property
def name(self) -> str:
return "custom"
@property
def capabilities(self) -> StrategyCapabilities:
return StrategyCapabilities()
def can_handle(self, request: dict[str, Any]) -> float:
return 0.5
def assemble(
self,
fragments: Sequence[ContextFragment],
budget: ContextBudget,
) -> Sequence[ContextFragment]:
# Custom ranking logic
return list(fragments)
def explain(self) -> str:
return "Custom strategy description."
pipeline = ACMSPipeline()
pipeline.register_strategy("custom", MyCustomStrategy())
payload = pipeline.assemble(
plan_id="plan-1",
fragments=fragments,
budget=budget,
strategy="custom",
)
```
## Example Usage
```python
from cleveragents.domain.models.core.context_fragment import (
ContextBudget,
ContextFragment,
ContextPayload,
FragmentProvenance,
)
from cleveragents.application.services.acms_service import ACMSPipeline
# Create fragments
fragments = [
ContextFragment(
uko_node="project://myapp/src/main.py",
content="def main(): ...",
relevance_score=0.9,
token_count=100,
tier="hot",
provenance=FragmentProvenance(resource_uri="project://myapp/src/main.py"),
),
ContextFragment(
uko_node="project://myapp/decisions/001",
content="Use async IO",
relevance_score=0.7,
token_count=50,
tier="warm",
provenance=FragmentProvenance(resource_uri="project://myapp/decisions/001"),
),
ContextFragment(
uko_node="project://myapp/README.md",
content="README.md content",
relevance_score=0.3,
token_count=200,
tier="cold",
provenance=FragmentProvenance(resource_uri="project://myapp/README.md"),
),
]
# Define budget
budget = ContextBudget(max_tokens=2048, reserved_tokens=256)
# Assemble with default relevance strategy
pipeline = ACMSPipeline()
payload = pipeline.assemble(plan_id="plan-1", fragments=fragments, budget=budget)
print(f"Fragments selected: {len(payload.fragments)}")
print(f"Total tokens: {payload.total_tokens}")
print(f"Budget used: {payload.budget_used:.1%}")
print(f"Within budget: {payload.is_within_budget}")
print(f"Remaining: {payload.remaining_tokens}")
print(f"Context hash: {payload.context_hash[:16]}...")
print(f"Strategies: {payload.strategies_used}")
```
features/acms_pipeline.feature
+616
View File
@@ -0,0 +1,616 @@
@phase2 @acms @acms_pipeline
Feature: ACMS v1 Context Assembly Pipeline
As a CleverAgents developer
I want to assemble context fragments into budget-constrained payloads
So that actors receive relevant, prioritized context within token limits
# ---------------------------------------------------------------------------
# ContextFragment — Domain Model
# ---------------------------------------------------------------------------
@acms_fragment
Scenario: Create a context fragment with defaults
Given a context fragment with uko_node "project://app/main.py" and content "hello world" and token_count 10
Then the fragment uko_node should be "project://app/main.py"
And the fragment content should be "hello world"
And the fragment relevance score should be 0.5
And the fragment token count should be 10
And the fragment detail depth should be 0
And the fragment tier should be "warm"
And the fragment metadata should be empty
And the fragment id should be set
And the fragment created_at should be a datetime
And the fragment provenance resource_uri should be set
@acms_fragment
Scenario: Create a fragment with all fields specified
Given a context fragment with all fields:
| field | value |
| uko_node | project://app/io |
| content | Use async IO pattern |
| score | 0.95 |
| tokens | 150 |
| detail_depth | 5 |
| tier | hot |
| meta_key | priority |
| meta_val | high |
| resource_uri | project://app/io |
Then the fragment uko_node should be "project://app/io"
And the fragment content should be "Use async IO pattern"
And the fragment relevance score should be 0.95
And the fragment token count should be 150
And the fragment detail depth should be 5
And the fragment tier should be "hot"
And the fragment metadata key "priority" should be "high"
@acms_fragment @validation
Scenario: Invalid relevance score rejected
When I create a fragment with relevance score 1.5
Then an ACMS validation error should be raised
@acms_fragment @validation
Scenario: Invalid token count rejected
When I create a fragment with token count -1
Then an ACMS validation error should be raised
@acms_fragment
Scenario: Invalid tier rejected
Given the ACMS pipeline modules are available
When I create a fragment with invalid tier "lukewarm"
Then an ACMS validation error should be raised
@acms_fragment @validation
Scenario: Missing uko_node rejected
When I create a fragment with empty uko_node
Then an ACMS validation error should be raised
@acms_fragment @validation
Scenario: Detail depth out of range rejected
When I create a fragment with detail depth 10
Then an ACMS validation error should be raised
# ---------------------------------------------------------------------------
# Structural field assertions — verify payload fields exist
# ---------------------------------------------------------------------------
@acms_fragment @structural
Scenario: ContextFragment has all spec-required fields
Given a context fragment with uko_node "project://app/main.py" and content "x" and token_count 1
Then the fragment should have field "fragment_id"
And the fragment should have field "uko_node"
And the fragment should have field "content"
And the fragment should have field "detail_depth"
And the fragment should have field "token_count"
And the fragment should have field "relevance_score"
And the fragment should have field "provenance"
And the fragment should have field "tier"
And the fragment should have field "metadata"
And the fragment should have field "created_at"
@acms_payload @structural
Scenario: ContextPayload has all spec-required fields
Given the following context fragments:
| uko_node | content | score | tokens |
| project://app/main.py | alpha | 0.9 | 100 |
And a context budget with max_tokens 4096 and reserved_tokens 0
When I assemble with strategy "relevance"
Then the payload should have field "payload_id"
And the payload should have field "plan_id"
And the payload should have field "fragments"
And the payload should have field "total_tokens"
And the payload should have field "budget"
And the payload should have field "budget_used"
And the payload should have field "strategies_used"
And the payload should have field "context_hash"
And the payload should have field "preamble"
And the payload should have field "provenance_map"
And the payload should have field "assembled_at"
# ---------------------------------------------------------------------------
# ContextBudget — Domain Model
# ---------------------------------------------------------------------------
@acms_budget
Scenario: Context budget calculates available tokens
Given a context budget with max_tokens 4096 and reserved_tokens 512
Then the available tokens should be 3584
@acms_budget
Scenario: Context budget with zero reserved tokens
Given a context budget with max_tokens 2048 and reserved_tokens 0
Then the available tokens should be 2048
@acms_budget
Scenario: Budget rejects reserved_tokens >= max_tokens
Given the ACMS pipeline modules are available
When I create a budget with reserved_tokens equal to max_tokens
Then an ACMS validation error should be raised
@acms_budget @validation
Scenario: Budget rejects reserved_tokens greater than max_tokens
When I create a budget with reserved_tokens greater than max_tokens
Then an ACMS validation error should be raised
# ---------------------------------------------------------------------------
# ACMSPipeline — Assembly
# ---------------------------------------------------------------------------
@acms_assemble
Scenario: Assemble with relevance strategy selects highest-relevance fragments
Given the following context fragments:
| uko_node | content | score | tokens |
| project://app/file.py | alpha | 0.9 | 100 |
| project://app/decision/1 | beta | 0.3 | 100 |
| project://app/resource/1 | gamma | 0.7 | 100 |
And a context budget with max_tokens 250 and reserved_tokens 0
When I assemble with strategy "relevance"
Then the payload should contain 2 fragments
And the first fragment content should be "alpha"
And the second fragment content should be "gamma"
@acms_assemble
Scenario: Assemble with recency strategy selects most recent fragments
Given context fragments with different timestamps:
| uko_node | content | tokens | created_at |
| project://app/old.py | old | 100 | 2024-01-01T00:00:00+00:00 |
| project://app/mid.py | mid | 100 | 2024-06-01T00:00:00+00:00 |
| project://app/new.py | new | 100 | 2025-01-01T00:00:00+00:00 |
And a context budget with max_tokens 200 and reserved_tokens 0
When I assemble with strategy "recency"
Then the payload should contain 2 fragments
And the first fragment content should be "new"
And the second fragment content should be "mid"
@acms_assemble
Scenario: Assemble with tiered strategy prioritizes hot tier
Given the following tiered context fragments:
| uko_node | content | score | tokens | tier |
| project://app/cold.py | cold-item | 0.9 | 100 | cold |
| project://app/hot.py | hot-item | 0.5 | 100 | hot |
| project://app/warm.py | warm-item | 0.8 | 100 | warm |
And a context budget with max_tokens 200 and reserved_tokens 0
When I assemble with strategy "tiered"
Then the payload should contain 2 fragments
And the first fragment content should be "hot-item"
And the second fragment content should be "warm-item"
@acms_assemble
Scenario: Assembly respects token budget
Given the following context fragments:
| uko_node | content | score | tokens |
| project://app/a.py | a | 0.9 | 500 |
| project://app/b.py | b | 0.8 | 500 |
| project://app/c.py | c | 0.7 | 500 |
And a context budget with max_tokens 1000 and reserved_tokens 0
When I assemble with strategy "relevance"
Then the payload should contain 2 fragments
And the payload total tokens should be 1000
And the payload should be within budget
@acms_assemble
Scenario: All fragments exceed budget individually
Given the following context fragments:
| uko_node | content | score | tokens |
| project://app/huge.py | huge | 0.9 | 5000 |
And a context budget with max_tokens 100 and reserved_tokens 0
When I assemble with strategy "relevance"
Then the payload should contain 0 fragments
And the payload total tokens should be 0
@acms_assemble
Scenario: Assembly with empty fragments returns empty payload
Given no context fragments
And a context budget with max_tokens 4096 and reserved_tokens 512
When I assemble with strategy "relevance"
Then the payload should contain 0 fragments
And the payload total tokens should be 0
@acms_assemble
Scenario: Assembly with single fragment within budget includes it
Given the following context fragments:
| uko_node | content | score | tokens |
| project://app/only.py | only | 0.5 | 100 |
And a context budget with max_tokens 4096 and reserved_tokens 0
When I assemble with strategy "relevance"
Then the payload should contain 1 fragments
And the first fragment content should be "only"
@acms_assemble
Scenario: Register and use custom context strategy
Given a custom context strategy that reverses fragment order
And the following context fragments:
| uko_node | content | score | tokens |
| project://app/first.py | first | 0.9 | 50 |
| project://app/second.py | second | 0.5 | 50 |
And a context budget with max_tokens 200 and reserved_tokens 0
When I assemble with strategy "reverse"
Then the payload should contain 2 fragments
And the first fragment content should be "second"
@acms_assemble
Scenario: Assemble with default strategy when none specified
Given the following context fragments:
| uko_node | content | score | tokens |
| project://app/alpha.py | alpha | 0.9 | 100 |
| project://app/beta.py | beta | 0.3 | 100 |
And a context budget with max_tokens 250 and reserved_tokens 0
When I assemble without specifying a strategy
Then the payload strategies used should include "relevance"
And the payload should contain 2 fragments
@acms_assemble @validation
Scenario: Non-existent strategy name raises error
Given the following context fragments:
| uko_node | content | score | tokens |
| project://app/alpha.py | alpha | 0.9 | 100 |
And a context budget with max_tokens 4096 and reserved_tokens 0
When I assemble with strategy "nonexistent"
Then an ACMS strategy error should be raised
# ---------------------------------------------------------------------------
# Strategy overwrite test
# ---------------------------------------------------------------------------
@acms_assemble
Scenario: Registering strategy with existing name overwrites it
Given a custom context strategy that reverses fragment order
And the following context fragments:
| uko_node | content | score | tokens |
| project://app/first.py | first | 0.9 | 50 |
| project://app/second.py | second | 0.5 | 50 |
And a context budget with max_tokens 200 and reserved_tokens 0
When I register the reverse strategy as "relevance"
And I assemble with strategy "relevance"
Then the first fragment content should be "second"
# ---------------------------------------------------------------------------
# ContextPayload — Properties
# ---------------------------------------------------------------------------
@acms_payload
Scenario: Payload is_within_budget property is correct
Given the following context fragments:
| uko_node | content | score | tokens |
| project://app/a.py | a | 0.9 | 100 |
And a context budget with max_tokens 200 and reserved_tokens 0
When I assemble with strategy "relevance"
Then the payload should be within budget
@acms_payload
Scenario: Payload remaining_tokens property is correct
Given the following context fragments:
| uko_node | content | score | tokens |
| project://app/a.py | a | 0.9 | 100 |
And a context budget with max_tokens 500 and reserved_tokens 0
When I assemble with strategy "relevance"
Then the payload remaining tokens should be 400
@acms_payload
Scenario: Payload budget_used is correct
Given the following context fragments:
| uko_node | content | score | tokens |
| project://app/a.py | a | 0.9 | 250 |
And a context budget with max_tokens 1000 and reserved_tokens 0
When I assemble with strategy "relevance"
Then the payload budget used should be 0.25
@acms_payload
Scenario: Payload context_hash is non-empty
Given the following context fragments:
| uko_node | content | score | tokens |
| project://app/a.py | a | 0.9 | 100 |
And a context budget with max_tokens 500 and reserved_tokens 0
When I assemble with strategy "relevance"
Then the payload context hash should be non-empty
@acms_payload
Scenario: Payload provenance_map maps fragment IDs
Given the following context fragments:
| uko_node | content | score | tokens |
| project://app/a.py | a | 0.9 | 100 |
And a context budget with max_tokens 500 and reserved_tokens 0
When I assemble with strategy "relevance"
Then the payload provenance map should map each fragment ID
@acms_payload
Scenario: Payload strategies_used lists the strategy name
Given the following context fragments:
| uko_node | content | score | tokens |
| project://app/a.py | a | 0.9 | 100 |
And a context budget with max_tokens 500 and reserved_tokens 0
When I assemble with strategy "tiered"
Then the payload strategies used should include "tiered"
# ---------------------------------------------------------------------------
# Payload structural assertions
# ---------------------------------------------------------------------------
@acms_payload @structural
Scenario: Payload plan_id, payload_id, and assembled_at are populated
Given the following context fragments:
| uko_node | content | score | tokens |
| project://app/a.py | a | 0.9 | 100 |
And a context budget with max_tokens 500 and reserved_tokens 0
When I assemble with strategy "relevance"
Then the payload plan_id should be "01JQTESTPN00000000000000AA"
And the payload payload_id should be a non-empty ULID
And the payload assembled_at should be a recent UTC datetime
# ---------------------------------------------------------------------------
# Budget validation edge case
# ---------------------------------------------------------------------------
@acms_budget @validation
Scenario: Budget rejects reserved_tokens strictly greater than max_tokens
When I create a budget with reserved_tokens 200 and max_tokens 100
Then an ACMS validation error should be raised
# ---------------------------------------------------------------------------
# Pipeline component DI injection (T1)
# ---------------------------------------------------------------------------
@acms_assemble @di
Scenario: Custom strategy selector is invoked during assembly
Given a pipeline with a custom tracking strategy selector
And the following context fragments:
| uko_node | content | score | tokens |
| project://app/a.py | alpha | 0.9 | 100 |
And a context budget with max_tokens 500 and reserved_tokens 0
When I assemble with strategy "relevance"
Then the custom strategy selector should have been called
# v1: strategies_used is hardcoded from the caller's input, not the
# selector output. This assertion validates the v1 pass-through path;
# multi-strategy fusion in a future milestone will make it meaningful.
And the payload strategies used should include "relevance"
@acms_assemble @di
Scenario: Custom budget allocator is invoked during assembly
Given a pipeline with a custom budget allocator that halves the budget
And the following context fragments:
| uko_node | content | score | tokens |
| project://app/a.py | alpha | 0.9 | 100 |
And a context budget with max_tokens 500 and reserved_tokens 0
When I assemble with strategy "relevance"
Then the custom budget allocator should have been called
@acms_assemble @di
Scenario: Custom strategy executor is invoked during assembly
Given a pipeline with a custom strategy executor that reverses fragments
And the following context fragments:
| uko_node | content | score | tokens |
| project://app/a.py | alpha | 0.9 | 50 |
| project://app/b.py | beta | 0.5 | 50 |
And a context budget with max_tokens 500 and reserved_tokens 0
When I assemble with strategy "relevance"
Then the custom strategy executor should have been called
And the first fragment content should be "beta"
@acms_assemble @di
Scenario: Custom deduplicator is invoked during assembly
Given a pipeline with a custom deduplicator that removes duplicates by uko_node
And the following context fragments:
| uko_node | content | score | tokens |
| project://app/a.py | alpha | 0.9 | 100 |
| project://app/a.py | alpha2 | 0.5 | 100 |
And a context budget with max_tokens 500 and reserved_tokens 0
When I assemble with strategy "relevance"
Then the custom deduplicator should have been called
And the payload should contain 1 fragments
@acms_assemble @di
Scenario: Custom preamble generator is invoked during assembly
Given a pipeline with a custom preamble generator
And the following context fragments:
| uko_node | content | score | tokens |
| project://app/a.py | alpha | 0.9 | 100 |
And a context budget with max_tokens 500 and reserved_tokens 0
When I assemble with strategy "relevance"
Then the payload preamble should be "Custom preamble: 1 fragments"
# ---------------------------------------------------------------------------
# SkeletonCompressor (T2)
# ---------------------------------------------------------------------------
@acms_assemble @skeleton
Scenario: DefaultSkeletonCompressor returns fragments unchanged
Given the ACMS pipeline modules are available
When I compress fragments with the default skeleton compressor and budget 100
Then the compressed fragments should equal the original fragments
@acms_assemble @skeleton
Scenario: Custom skeleton compressor can truncate fragments
Given the ACMS pipeline modules are available
When I compress fragments with a truncating skeleton compressor and budget 50
Then the compressed fragments should have reduced content
# ---------------------------------------------------------------------------
# Multi-candidate allocation (T3)
# ---------------------------------------------------------------------------
@acms_assemble @allocation
Scenario: Budget allocator distributes proportionally across multiple candidates
Given the ACMS pipeline modules are available
When I allocate budget 1000 across candidates with confidences 0.8 and 0.2
Then the first candidate should receive approximately 800 tokens
And the second candidate should receive approximately 200 tokens
And the total allocated should not exceed 1000
@acms_assemble @allocation
Scenario: Strategy selector returns all strategies when no hint given
Given the ACMS pipeline modules are available
When I select strategies with no strategy hint from 3 registered strategies
Then all 3 strategies should be returned with confidence scores
# ---------------------------------------------------------------------------
# plan_id ULID validation (SEC-1, SPEC-1, TEST-GAP-1)
# ---------------------------------------------------------------------------
@acms_payload @validation
Scenario: Payload rejects empty plan_id
Given the ACMS pipeline modules are available
When I create a payload with an empty plan_id
Then an ACMS validation error should be raised
@acms_payload @validation
Scenario: Payload accepts valid ULID plan_id
Given the ACMS pipeline modules are available
When I create a payload with plan_id "01JQTESTPN00000000000000AA"
Then the created payload plan_id should be "01JQTESTPN00000000000000AA"
@acms_payload @validation
Scenario: Payload rejects 25-character plan_id (too short for ULID)
Given the ACMS pipeline modules are available
When I create a payload with plan_id "01JQTESTPN0000000000000AA"
Then an ACMS validation error should be raised
@acms_payload @validation
Scenario: Payload rejects 27-character plan_id (too long for ULID)
Given the ACMS pipeline modules are available
When I create a payload with plan_id "01JQTESTPN000000000000000AA"
Then an ACMS validation error should be raised
@acms_payload @validation
Scenario: Payload rejects lowercase plan_id
Given the ACMS pipeline modules are available
When I create a payload with plan_id "01jqtestpn00000000000000aa"
Then an ACMS validation error should be raised
@acms_payload @validation
Scenario Outline: Payload rejects plan_id with excluded Crockford char <char>
Given the ACMS pipeline modules are available
When I create a payload with plan_id "<plan_id>"
Then an ACMS validation error should be raised
Examples:
| char | plan_id |
| I | 01JQTESTPN000000000000I0AA |
| L | 01JQTESTPN000000000000L0AA |
| O | 01JQTESTPN000000000000O0AA |
| U | 01JQTESTPN000000000000U0AA |
@acms_payload @validation @security
Scenario: Payload rejects path traversal in plan_id
Given the ACMS pipeline modules are available
When I create a payload with plan_id "01JQTESTPN0000000/../00AAA"
Then an ACMS validation error should be raised
@acms_payload @validation @security
Scenario: Payload rejects plan_id with slashes
Given the ACMS pipeline modules are available
When I create a payload with plan_id "01JQTESTPN00000/0000000AAA"
Then an ACMS validation error should be raised
@acms_payload @validation
Scenario: Payload rejects plan_id with whitespace
Given the ACMS pipeline modules are available
When I create a payload with plan_id " "
Then an ACMS validation error should be raised
# ---------------------------------------------------------------------------
# Strategy introspection (coverage: capabilities, can_handle, explain)
# ---------------------------------------------------------------------------
@acms_strategy @introspection
Scenario: Each built-in strategy declares distinct capabilities
Given the ACMS pipeline modules are available
When I inspect the capabilities of "relevance"
Then the strategy should declare supports_semantic_search as true
When I inspect the capabilities of "recency"
Then the strategy should declare supports_temporal_archaeology as true
When I inspect the capabilities of "tiered"
Then the strategy should declare supports_semantic_search as false
@acms_strategy @introspection
Scenario: Strategy confidence ranking determines selector priority
Given the ACMS pipeline modules are available
When I query can_handle on all three built-in strategies
Then "relevance" should have the highest confidence
And "tiered" should have higher confidence than "recency"
@acms_strategy @introspection
Scenario: Strategy explain returns strategy-specific descriptions
Given the ACMS pipeline modules are available
When I call explain on each built-in strategy
Then the "relevance" explanation should mention "relevance score"
And the "recency" explanation should mention "creation time"
And the "tiered" explanation should mention "tier priority"
# ---------------------------------------------------------------------------
# Fragment metadata overflow
# ---------------------------------------------------------------------------
@acms_fragment @validation
Scenario: Fragment rejects metadata with more than 64 entries
Given the ACMS pipeline modules are available
When I create a fragment with 65 metadata entries
Then an ACMS validation error should be raised
@acms_fragment @validation
Scenario: Fragment accepts metadata with exactly 64 entries
Given the ACMS pipeline modules are available
When I create a fragment with 64 metadata entries
Then the fragment should have 64 metadata entries
# ---------------------------------------------------------------------------
# Early plan_id rejection in assemble()
# ---------------------------------------------------------------------------
@acms_assemble @validation
Scenario: Pipeline rejects invalid plan_id before executing the pipeline
Given the following context fragments:
| uko_node | content | score | tokens |
| project://app/a.py | alpha | 0.9 | 100 |
And a context budget with max_tokens 500 and reserved_tokens 0
When I assemble with plan_id "not-a-valid-ulid"
Then an ACMS ValueError should be raised mentioning "ULID"
# ---------------------------------------------------------------------------
# Provenance map immutability (TEST-2)
# ---------------------------------------------------------------------------
@acms_payload @immutability
Scenario: Provenance map is immune to external mutation after payload creation
Given the ACMS pipeline modules are available
When I create a payload with a mutable provenance map and mutate the source
Then the payload provenance map should be unchanged
# ---------------------------------------------------------------------------
# Allocator rounding edge cases (TEST-3)
# ---------------------------------------------------------------------------
@acms_assemble @allocation
Scenario: Budget allocator uses full budget with odd split across 3 candidates
Given the ACMS pipeline modules are available
When I allocate budget 999 across 3 candidates with equal confidence
Then the total allocated should equal exactly 999
And each allocation should be 333 or 334
@acms_assemble @allocation
Scenario: Budget allocator handles zero-confidence equal split without token loss
Given the ACMS pipeline modules are available
When I allocate budget 100 across 3 candidates with zero confidence
Then the total allocated should equal exactly 100
@acms_assemble @allocation
Scenario: Budget allocator distributes full budget with unequal confidences
Given the ACMS pipeline modules are available
When I allocate budget 10 across 3 candidates with confidences 0.5 0.3 0.2
Then the total allocated should equal exactly 10
# ---------------------------------------------------------------------------
# Strategy selector fallback (TEST-4)
# ---------------------------------------------------------------------------
@acms_assemble @di
Scenario: Assembly falls back when custom selector omits the requested strategy
Given a pipeline with a custom strategy selector that returns no candidates
And the following context fragments:
| uko_node | content | score | tokens |
| project://app/a.py | alpha | 0.9 | 100 |
| project://app/b.py | beta | 0.5 | 100 |
And a context budget with max_tokens 500 and reserved_tokens 0
When I assemble with strategy "relevance"
Then the empty strategy selector should have been called
And the payload strategies used should include "relevance"
And the payload should contain 2 fragments
And the first fragment content should be "alpha"
features/environment.py
+4
View File
@@ -225,6 +225,10 @@ def before_scenario(context, scenario):
context._cleanup_handlers = []
context.stubbed_clients = {}
# Reset error attributes to prevent stale state leaking between scenarios.
context.acms_error = None
context.assemble_error = None
# Ensure mock AI flag is always set so plan service tests can resolve actors
os.environ["CLEVERAGENTS_TESTING_USE_MOCK_AI"] = "true"
features/steps/acms_pipeline_steps.py
+1080
View File
File diff suppressed because it is too large Load Diff
robot/acms_pipeline.robot
+57
View File
@@ -0,0 +1,57 @@
*** Settings ***
Documentation Integration smoke tests for ACMS v1 context assembly pipeline
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_acms_pipeline.py
*** Test Cases ***
Create Context Fragment
[Documentation] Create a ContextFragment and verify defaults
${result}= Run Process ${PYTHON} ${HELPER} fragment-create cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} acms-fragment-ok
Calculate Budget Available Tokens
[Documentation] Create ContextBudget and verify available_tokens
${result}= Run Process ${PYTHON} ${HELPER} budget-calc cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} acms-budget-ok
Assemble With Relevance Strategy
[Documentation] Assemble fragments with relevance strategy
${result}= Run Process ${PYTHON} ${HELPER} assemble-relevance cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} acms-assemble-relevance-ok
Assemble With Recency Strategy
[Documentation] Assemble fragments with recency strategy
${result}= Run Process ${PYTHON} ${HELPER} assemble-recency cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} acms-assemble-recency-ok
Assemble With Tiered Strategy
[Documentation] Assemble fragments with tiered strategy
${result}= Run Process ${PYTHON} ${HELPER} assemble-tiered cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} acms-assemble-tiered-ok
Verify Payload Budget Check
[Documentation] Verify is_within_budget property on assembled payload
${result}= Run Process ${PYTHON} ${HELPER} payload-budget-check cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} acms-payload-budget-ok
robot/helper_acms_pipeline.py
+293
View File
@@ -0,0 +1,293 @@
"""Robot Framework helper for ACMS v1 pipeline integration tests.
Provides a CLI-style interface for Robot to invoke ACMS pipeline
operations and verify the results. Exit code 0 = success, 1 = failure.
Usage:
python robot/helper_acms_pipeline.py fragment-create
python robot/helper_acms_pipeline.py budget-calc
python robot/helper_acms_pipeline.py assemble-relevance
python robot/helper_acms_pipeline.py assemble-recency
python robot/helper_acms_pipeline.py assemble-tiered
python robot/helper_acms_pipeline.py payload-budget-check
"""
from __future__ import annotations
import sys
from collections.abc import Callable
from datetime import UTC, datetime
from pathlib import Path
_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 ACMSPipeline # noqa: E402
from cleveragents.domain.models.core.context_fragment import ( # noqa: E402
ContextBudget,
ContextFragment,
FragmentProvenance,
)
# Default provenance for test fragments.
_DEFAULT_PROV = FragmentProvenance(resource_uri="test://robot")
def _cmd_fragment_create() -> int:
"""Create a ContextFragment and verify defaults."""
frag = ContextFragment(
uko_node="project://app/main.py",
content="hello world",
token_count=10,
provenance=FragmentProvenance(resource_uri="project://app/main.py"),
)
if frag.uko_node != "project://app/main.py":
print(
f"acms-fail: expected uko_node=project://app/main.py, got {frag.uko_node}"
)
return 1
if frag.relevance_score != 0.5:
print(f"acms-fail: expected score=0.5, got {frag.relevance_score}")
return 1
if not frag.fragment_id:
print("acms-fail: fragment_id is empty")
return 1
if not isinstance(frag.created_at, datetime):
print(f"acms-fail: created_at is not datetime, got {type(frag.created_at)}")
return 1
if frag.token_count != 10:
print(f"acms-fail: expected token_count=10, got {frag.token_count}")
return 1
print(f"acms-fragment-ok: id={frag.fragment_id}")
return 0
def _cmd_budget_calc() -> int:
"""Create ContextBudget and verify available_tokens."""
budget = ContextBudget(max_tokens=4096, reserved_tokens=512)
expected = 3584
if budget.available_tokens != expected:
print(
f"acms-fail: expected available={expected}, got {budget.available_tokens}"
)
return 1
print(f"acms-budget-ok: available={budget.available_tokens}")
return 0
def _cmd_assemble_relevance() -> int:
"""Assemble fragments with relevance strategy."""
frags = [
ContextFragment(
uko_node="project://app/high.py",
content="high",
relevance_score=0.9,
token_count=100,
provenance=_DEFAULT_PROV,
),
ContextFragment(
uko_node="project://app/low.py",
content="low",
relevance_score=0.2,
token_count=100,
provenance=_DEFAULT_PROV,
),
ContextFragment(
uko_node="project://app/mid.py",
content="mid",
relevance_score=0.6,
token_count=100,
provenance=_DEFAULT_PROV,
),
]
budget = ContextBudget(max_tokens=200, reserved_tokens=0)
pipeline = ACMSPipeline()
payload = pipeline.assemble(
plan_id="01JQTESTPN00000000000000AA",
fragments=frags,
budget=budget,
strategy="relevance",
)
if len(payload.fragments) != 2:
print(f"acms-fail: expected 2 fragments, got {len(payload.fragments)}")
return 1
if payload.fragments[0].content != "high":
print(f"acms-fail: expected first=high, got {payload.fragments[0].content}")
return 1
if not payload.strategies_used:
print("acms-fail: strategies_used is empty")
return 1
print(f"acms-assemble-relevance-ok: fragments={len(payload.fragments)}")
return 0
def _cmd_assemble_recency() -> int:
"""Assemble fragments with recency strategy."""
frags = [
ContextFragment(
uko_node="project://app/old.py",
content="old",
token_count=100,
created_at=datetime(2024, 1, 1, tzinfo=UTC),
provenance=_DEFAULT_PROV,
),
ContextFragment(
uko_node="project://app/mid.py",
content="mid",
token_count=100,
created_at=datetime(2024, 6, 1, tzinfo=UTC),
provenance=_DEFAULT_PROV,
),
ContextFragment(
uko_node="project://app/new.py",
content="new",
token_count=100,
created_at=datetime(2025, 1, 1, tzinfo=UTC),
provenance=_DEFAULT_PROV,
),
]
budget = ContextBudget(max_tokens=200, reserved_tokens=0)
pipeline = ACMSPipeline()
payload = pipeline.assemble(
plan_id="01JQTESTPN00000000000000AA",
fragments=frags,
budget=budget,
strategy="recency",
)
if len(payload.fragments) != 2:
print(f"acms-fail: expected 2 fragments, got {len(payload.fragments)}")
return 1
if payload.fragments[0].content != "new":
print(f"acms-fail: expected first=new, got {payload.fragments[0].content}")
return 1
if payload.fragments[1].content != "mid":
print(f"acms-fail: expected second=mid, got {payload.fragments[1].content}")
return 1
print(f"acms-assemble-recency-ok: fragments={len(payload.fragments)}")
return 0
def _cmd_assemble_tiered() -> int:
"""Assemble fragments with tiered strategy."""
frags = [
ContextFragment(
uko_node="project://app/cold.py",
content="cold-item",
relevance_score=0.9,
token_count=100,
tier="cold",
provenance=_DEFAULT_PROV,
),
ContextFragment(
uko_node="project://app/hot.py",
content="hot-item",
relevance_score=0.5,
token_count=100,
tier="hot",
provenance=_DEFAULT_PROV,
),
ContextFragment(
uko_node="project://app/warm.py",
content="warm-item",
relevance_score=0.8,
token_count=100,
tier="warm",
provenance=_DEFAULT_PROV,
),
]
budget = ContextBudget(max_tokens=200, reserved_tokens=0)
pipeline = ACMSPipeline()
payload = pipeline.assemble(
plan_id="01JQTESTPN00000000000000AA",
fragments=frags,
budget=budget,
strategy="tiered",
)
if len(payload.fragments) != 2:
print(f"acms-fail: expected 2 fragments, got {len(payload.fragments)}")
return 1
if payload.fragments[0].content != "hot-item":
print(f"acms-fail: expected first=hot-item, got {payload.fragments[0].content}")
return 1
print(f"acms-assemble-tiered-ok: fragments={len(payload.fragments)}")
return 0
def _cmd_payload_budget_check() -> int:
"""Verify is_within_budget property on assembled payload."""
frags = [
ContextFragment(
uko_node="project://app/item.py",
content="item",
relevance_score=0.9,
token_count=100,
provenance=_DEFAULT_PROV,
),
]
budget = ContextBudget(max_tokens=500, reserved_tokens=0)
pipeline = ACMSPipeline()
payload = pipeline.assemble(
plan_id="01JQTESTPN00000000000000AA",
fragments=frags,
budget=budget,
strategy="relevance",
)
if not payload.is_within_budget:
print("acms-fail: expected is_within_budget=True")
return 1
if payload.remaining_tokens != 400:
print(f"acms-fail: expected remaining=400, got {payload.remaining_tokens}")
return 1
if not payload.context_hash:
print("acms-fail: expected non-empty context_hash")
return 1
if len(payload.context_hash) != 64:
print(
f"acms-fail: expected SHA-256 hex (64 chars), "
f"got {len(payload.context_hash)}"
)
return 1
try:
int(payload.context_hash, 16)
except ValueError:
print(f"acms-fail: context_hash is not valid hex: {payload.context_hash!r}")
return 1
print(
f"acms-payload-budget-ok: within_budget={payload.is_within_budget} "
f"remaining={payload.remaining_tokens}"
)
return 0
_COMMANDS: dict[str, Callable[[], int]] = {
"fragment-create": _cmd_fragment_create,
"budget-calc": _cmd_budget_calc,
"assemble-relevance": _cmd_assemble_relevance,
"assemble-recency": _cmd_assemble_recency,
"assemble-tiered": _cmd_assemble_tiered,
"payload-budget-check": _cmd_payload_budget_check,
}
def main() -> int:
"""Entry point called by Robot Framework ``Run Process``."""
if len(sys.argv) < 2:
print(
"Usage: helper_acms_pipeline.py "
"<fragment-create|budget-calc|assemble-relevance"
"|assemble-recency|assemble-tiered|payload-budget-check>"
)
return 1
command = sys.argv[1]
handler = _COMMANDS.get(command)
if handler is None:
print(f"Unknown command: {command}")
return 1
return handler()
if __name__ == "__main__":
sys.exit(main())
src/cleveragents/application/services/acms_service.py
+734
View File
@@ -0,0 +1,734 @@
"""ACMS v1 context assembly pipeline service.
Implements the Adaptive Context Management System pipeline that integrates
UKO (Unified Knowledge Ontology) and CRP (Context Request Protocol)
components with pluggable context strategies. The pipeline assembles
``ContextFragment`` objects into a budget-constrained ``ContextPayload``
for actor consumption.
The spec defines a 10-component pluggable pipeline across 3 phases:
**Strategy Orchestration**: StrategySelector, BudgetAllocator, StrategyExecutor
**Fragment Fusion**: FragmentDeduplicator, DetailDepthResolver, FragmentScorer,
BudgetPacker, FragmentOrderer
**Context Finalization**: PreambleGenerator, SkeletonCompressor
This v1 implementation provides default (pass-through) implementations for
all 10 components. Production replacements will be wired in via the plugin
registry as the pipeline matures.
Based on ``docs/specification.md`` ~line 42615.
"""
from __future__ import annotations
import re
from collections.abc import Sequence
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, ClassVar, Protocol, runtime_checkable
import structlog
if TYPE_CHECKING:
from cleveragents.config.settings import Settings
from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
from cleveragents.domain.models.core.context_fragment import (
ULID_PATTERN,
ContextBudget,
ContextFragment,
ContextPayload,
build_provenance_map,
compute_context_hash,
)
logger = structlog.get_logger(__name__)
# ---------------------------------------------------------------------------
# Strategy capabilities (spec ~line 25167)
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class StrategyCapabilities:
"""Capabilities declared by a context strategy."""
supports_semantic_search: bool = False
supports_graph_navigation: bool = False
supports_temporal_archaeology: bool = False
max_fragments: int | None = None
# ---------------------------------------------------------------------------
# Context strategy protocol (spec ~line 25167)
# ---------------------------------------------------------------------------
@runtime_checkable
class ContextStrategy(Protocol):
"""Protocol for context strategies.
Based on ``docs/specification.md`` ~line 25167. Each strategy declares
a ``name`` and ``capabilities``, can report its confidence for a request
via ``can_handle``, produce fragments via ``assemble``, and explain its
approach via ``explain``.
"""
@property
def name(self) -> str: ...
@property
def capabilities(self) -> StrategyCapabilities: ...
def can_handle(self, request: dict[str, Any]) -> float:
"""Return confidence (0.0-1.0) that this strategy can handle *request*."""
...
def assemble(
self,
fragments: Sequence[ContextFragment],
budget: ContextBudget,
) -> Sequence[ContextFragment]:
"""Rank/filter *fragments* to fit within *budget*."""
...
def explain(self) -> str:
"""Return a human-readable explanation of this strategy."""
...
# ---------------------------------------------------------------------------
# Built-in strategy implementations
# ---------------------------------------------------------------------------
def _pack_budget(
sorted_frags: Sequence[ContextFragment],
budget: ContextBudget,
) -> list[ContextFragment]:
"""Pack fragments into budget with early termination."""
result: list[ContextFragment] = []
total = 0
available = budget.available_tokens
for frag in sorted_frags:
if total >= available:
break
if total + frag.token_count <= available:
result.append(frag)
total += frag.token_count
return result
class RelevanceStrategy:
"""Rank fragments by relevance score (highest first), fitting budget."""
@property
def name(self) -> str:
return "relevance"
@property
def capabilities(self) -> StrategyCapabilities:
return StrategyCapabilities(supports_semantic_search=True)
def can_handle(self, request: dict[str, Any]) -> float:
return 0.8 # general-purpose, high confidence
def assemble(
self,
fragments: Sequence[ContextFragment],
budget: ContextBudget,
) -> Sequence[ContextFragment]:
sorted_frags = sorted(
fragments,
key=lambda f: f.relevance_score,
reverse=True,
)
return _pack_budget(sorted_frags, budget)
def explain(self) -> str:
return (
"Ranks fragments by relevance score (highest first), packs within budget."
)
class RecencyStrategy:
"""Rank fragments by creation time (most recent first), fitting budget."""
@property
def name(self) -> str:
return "recency"
@property
def capabilities(self) -> StrategyCapabilities:
return StrategyCapabilities(supports_temporal_archaeology=True)
def can_handle(self, request: dict[str, Any]) -> float:
return 0.6
def assemble(
self,
fragments: Sequence[ContextFragment],
budget: ContextBudget,
) -> Sequence[ContextFragment]:
# Sort by datetime object (not string comparison)
sorted_frags = sorted(
fragments,
key=lambda f: f.created_at,
reverse=True,
)
return _pack_budget(sorted_frags, budget)
def explain(self) -> str:
return (
"Ranks fragments by creation time (most recent first), packs within budget."
)
_TIER_PRIORITY: dict[str, int] = {"hot": 0, "warm": 1, "cold": 2}
class TieredStrategy:
"""Rank fragments by tier priority (hot > warm > cold), then relevance.
.. note::
In v1 tiers are sort-priority labels, not storage tiers with
retention policies. Full hot/warm/cold storage-tier semantics
(promotion, demotion, eviction) are implemented separately in
``ContextTierService``.
"""
@property
def name(self) -> str:
return "tiered"
@property
def capabilities(self) -> StrategyCapabilities:
return StrategyCapabilities()
def can_handle(self, request: dict[str, Any]) -> float:
return 0.7
def assemble(
self,
fragments: Sequence[ContextFragment],
budget: ContextBudget,
) -> Sequence[ContextFragment]:
sorted_frags = sorted(
fragments,
key=lambda f: (
_TIER_PRIORITY.get(f.tier, 99),
-f.relevance_score,
),
)
return _pack_budget(sorted_frags, budget)
def explain(self) -> str:
return "Ranks by tier priority (hot > warm > cold), then relevance within tier."
# ---------------------------------------------------------------------------
# Phase 1 — Strategy Orchestration protocols (spec §42630-42636)
# ---------------------------------------------------------------------------
class StrategySelector(Protocol):
"""Decide which strategies to invoke for a given request.
Spec: ``StrategySelectorProtocol`` — calls ``can_handle()`` on all
registered strategies; returns ``(strategy, confidence)`` pairs.
"""
def select(
self,
strategies: Sequence[ContextStrategy],
request: dict[str, Any],
) -> list[tuple[ContextStrategy, float]]:
"""Return (strategy, confidence) pairs sorted by priority."""
...
class BudgetAllocator(Protocol):
"""Distribute the token budget across selected strategies.
Spec: ``BudgetAllocatorProtocol`` — returns
``(strategy, confidence, allocated_tokens)`` triples.
"""
def allocate(
self,
candidates: list[tuple[ContextStrategy, float]],
total_budget: int,
) -> list[tuple[ContextStrategy, float, int]]:
"""Return (strategy, confidence, allocated_tokens) triples."""
...
class StrategyExecutor(Protocol):
"""Control how strategies are invoked (parallelism, timeouts).
Spec: ``StrategyExecutorProtocol`` — executes strategies and
collects all resulting fragments.
"""
def execute(
self,
allocations: list[tuple[ContextStrategy, float, int]],
fragments: Sequence[ContextFragment],
budget: ContextBudget,
) -> Sequence[ContextFragment]:
"""Execute strategies and return collected fragments."""
...
# ---------------------------------------------------------------------------
# Phase 2 — Fragment Fusion protocols (spec §42638-42646)
# ---------------------------------------------------------------------------
class FragmentDeduplicator(Protocol):
"""Remove duplicate fragments."""
def deduplicate(
self,
fragments: Sequence[ContextFragment],
) -> Sequence[ContextFragment]: ...
class DetailDepthResolver(Protocol):
"""Resolve detail depth levels."""
def resolve(
self,
fragments: Sequence[ContextFragment],
) -> Sequence[ContextFragment]: ...
class FragmentScorer(Protocol):
"""Score fragments for relevance."""
def score(
self,
fragments: Sequence[ContextFragment],
) -> Sequence[ContextFragment]: ...
class BudgetPacker(Protocol):
"""Pack fragments within budget."""
def pack(
self,
fragments: Sequence[ContextFragment],
budget: ContextBudget,
) -> Sequence[ContextFragment]: ...
class FragmentOrderer(Protocol):
"""Order fragments for final output."""
def order(
self,
fragments: Sequence[ContextFragment],
) -> Sequence[ContextFragment]: ...
# ---------------------------------------------------------------------------
# Phase 3 — Context Finalization protocols (spec §42648-42653)
# ---------------------------------------------------------------------------
class PreambleGenerator(Protocol):
"""Generate a preamble summary.
Spec: ``PreambleGeneratorProtocol`` — generates a structured
preamble listing included resources, strategies used, budget
utilization.
"""
def generate(
self,
fragments: Sequence[ContextFragment],
) -> str | None: ...
class SkeletonCompressor(Protocol):
"""Compress parent context into a skeleton for child plan inheritance.
Spec: ``SkeletonCompressorProtocol`` — re-renders parent context
fragments at reduced depth (0-1) to fit within ``skeleton_budget``.
"""
def compress(
self,
fragments: tuple[ContextFragment, ...],
skeleton_budget: int,
) -> tuple[ContextFragment, ...]:
"""Return compressed fragments fitting within *skeleton_budget*."""
...
# ---------------------------------------------------------------------------
# Default (pass-through) component implementations
# ---------------------------------------------------------------------------
class DefaultStrategySelector:
"""Select strategies based on confidence scores.
Evaluates all strategies via ``can_handle`` and returns those with
confidence > 0.0, sorted by confidence descending. The pipeline
passes all registered strategies so a custom ``StrategySelector``
can implement multi-strategy fusion or confidence-weighted selection.
"""
def select(
self,
strategies: Sequence[ContextStrategy],
request: dict[str, Any],
) -> list[tuple[ContextStrategy, float]]:
scored = [(s, s.can_handle(request)) for s in strategies]
return [
(s, c)
for s, c in sorted(scored, key=lambda x: x[1], reverse=True)
if c > 0.0
]
class DefaultBudgetAllocator:
"""Distribute the token budget proportionally across candidates.
Each candidate receives a share of ``total_budget`` proportional to
its confidence score. When only one candidate is present (v1 norm),
it receives the full budget. With multiple candidates the sum of
allocated tokens never exceeds ``total_budget`` (spec §42689).
"""
def allocate(
self,
candidates: list[tuple[ContextStrategy, float]],
total_budget: int,
) -> list[tuple[ContextStrategy, float, int]]:
if not candidates:
return []
n = len(candidates)
total_confidence = sum(c for _, c in candidates)
if total_confidence <= 0:
# Equal split when all confidences are zero.
share = total_budget // n
remainder = total_budget - share * n
# Distribute leftover tokens one-per-candidate (largest-remainder).
return [
(s, c, share + (1 if i < remainder else 0))
for i, (s, c) in enumerate(candidates)
]
# Proportional allocation with largest-remainder distribution so
# the full budget is used and no tokens are silently lost.
raw = [(total_budget * c / total_confidence) for _, c in candidates]
floors = [int(r) for r in raw]
remainder = total_budget - sum(floors)
# Award one extra token to candidates with the largest fractional parts.
fractions = [
(r - f, i) for i, (r, f) in enumerate(zip(raw, floors, strict=True))
]
fractions.sort(reverse=True)
for _, i in fractions[:remainder]:
floors[i] += 1
return [(s, c, floors[i]) for i, (s, c) in enumerate(candidates)]
class DefaultStrategyExecutor:
"""Execute strategies synchronously using their allocated budgets.
Each strategy receives a ``ContextBudget`` scoped to the tokens
assigned by the ``BudgetAllocator``. Results from all strategies
are concatenated. Strategies with zero-token allocations are
skipped (per spec §42927 ``min_useful_budget`` exclusion rule).
Aggregate budget enforcement is deferred to the downstream
``BudgetPacker`` component (spec §42937). The ``budget`` parameter
is accepted for interface compatibility and future use.
Future versions will invoke strategies in parallel with timeouts
and circuit breaking.
"""
def execute(
self,
allocations: list[tuple[ContextStrategy, float, int]],
fragments: Sequence[ContextFragment],
budget: ContextBudget,
) -> Sequence[ContextFragment]:
if not allocations:
return []
collected: list[ContextFragment] = []
CoreRasurae commented 2026-03-04 23:42:59 +00:00
Outdated
Review
Copy Link
Copy Source

BUG-EDGE-1 [LOW]: The spec (§42927) states the ProportionalBudgetAllocator "Guarantees each strategy receives at least min_useful_budget tokens or is excluded" (default 500). This guard only skips strategies with 0 tokens. A strategy allocated e.g. 100 tokens would still execute, potentially producing low-quality fragments.

Consider adding a log warning or TODO here for when allocated_tokens < min_useful_budget.

**BUG-EDGE-1 [LOW]:** The spec (§42927) states the `ProportionalBudgetAllocator` "Guarantees each strategy receives at least `min_useful_budget` tokens or is excluded" (default 500). This guard only skips strategies with 0 tokens. A strategy allocated e.g. 100 tokens would still execute, potentially producing low-quality fragments. Consider adding a log warning or TODO here for when `allocated_tokens < min_useful_budget`.
for strategy, _confidence, allocated_tokens in allocations:
if allocated_tokens <= 0:
continue
scoped_budget = ContextBudget(
max_tokens=allocated_tokens,
reserved_tokens=0,
)
collected.extend(strategy.assemble(fragments, scoped_budget))
return collected
class DefaultDeduplicator:
"""No-op deduplicator — returns fragments unchanged."""
def deduplicate(
self,
fragments: Sequence[ContextFragment],
) -> Sequence[ContextFragment]:
return fragments
class DefaultDepthResolver:
"""No-op depth resolver — returns fragments unchanged."""
def resolve(
self,
fragments: Sequence[ContextFragment],
) -> Sequence[ContextFragment]:
return fragments
class DefaultScorer:
"""No-op scorer — returns fragments unchanged."""
def score(
self,
fragments: Sequence[ContextFragment],
) -> Sequence[ContextFragment]:
return fragments
class DefaultOrderer:
"""No-op orderer — returns fragments unchanged."""
def order(
self,
fragments: Sequence[ContextFragment],
) -> Sequence[ContextFragment]:
return fragments
class DefaultPreambleGenerator:
"""No-op preamble generator — returns None."""
def generate(
self,
fragments: Sequence[ContextFragment],
) -> str | None:
return None
class DefaultBudgetPacker:
"""No-op budget packer — returns fragments unchanged.
In v1 budget packing is handled within each strategy's ``assemble``
method via ``_pack_budget``. Production implementations will use
greedy knapsack with depth fallback per spec §42937.
"""
def pack(
self,
fragments: Sequence[ContextFragment],
budget: ContextBudget,
) -> Sequence[ContextFragment]:
return fragments
class DefaultSkeletonCompressor:
"""No-op skeleton compressor — returns fragments unchanged.
Production implementations will re-render fragments at depth 0-1
to fit within ``skeleton_budget`` for child plan inheritance.
"""
def compress(
self,
fragments: tuple[ContextFragment, ...],
skeleton_budget: int,
) -> tuple[ContextFragment, ...]:
return fragments
# ---------------------------------------------------------------------------
# Pipeline
# ---------------------------------------------------------------------------
class ACMSPipeline:
"""ACMS v1 context assembly pipeline.
Assembles ``ContextFragment`` objects into a budget-constrained
``ContextPayload`` using pluggable context strategies and the
10-component pipeline architecture defined in the spec.
Example::
pipeline = ACMSPipeline()
payload = pipeline.assemble(
plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV",
fragments=[frag1, frag2],
budget=ContextBudget(max_tokens=2048),
)
"""
BUILTIN_STRATEGIES: ClassVar[dict[str, type[ContextStrategy]]] = {
"relevance": RelevanceStrategy, # type: ignore[dict-item]
"recency": RecencyStrategy, # type: ignore[dict-item]
"tiered": TieredStrategy, # type: ignore[dict-item]
}
def __init__(
self,
default_strategy: str = "relevance",
*,
settings: Settings | None = None,
unit_of_work: UnitOfWork | None = None,
# Phase 1 — Strategy Orchestration
strategy_selector: StrategySelector | None = None,
budget_allocator: BudgetAllocator | None = None,
strategy_executor: StrategyExecutor | None = None,
# Phase 2 — Fragment Fusion
deduplicator: FragmentDeduplicator | None = None,
depth_resolver: DetailDepthResolver | None = None,
scorer: FragmentScorer | None = None,
packer: BudgetPacker | None = None,
orderer: FragmentOrderer | None = None,
# Phase 3 — Context Finalization
preamble_generator: PreambleGenerator | None = None,
skeleton_compressor: SkeletonCompressor | None = None,
) -> None:
# DI placeholders — not referenced in v1 methods. Stored here to
# match the ``DecisionService`` DI pattern so that future milestones
# can wire configuration (e.g. default budget, strategy allow-lists)
# and persistence without changing the constructor signature.
self._settings = settings
self._unit_of_work = unit_of_work
self._strategies: dict[str, ContextStrategy] = {
name: cls() for name, cls in self.BUILTIN_STRATEGIES.items()
}
if default_strategy not in self._strategies:
msg = (
f"Unknown strategy {default_strategy!r}. "
f"Available: {', '.join(sorted(self._strategies))}"
)
raise ValueError(msg)
self._default_strategy = default_strategy
self._logger = logger.bind(service="acms_pipeline")
# All 10 pipeline components (default to pass-through stubs)
self._strategy_selector = strategy_selector or DefaultStrategySelector()
self._budget_allocator = budget_allocator or DefaultBudgetAllocator()
self._strategy_executor = strategy_executor or DefaultStrategyExecutor()
self._deduplicator = deduplicator or DefaultDeduplicator()
self._depth_resolver = depth_resolver or DefaultDepthResolver()
self._scorer = scorer or DefaultScorer()
self._packer = packer or DefaultBudgetPacker()
self._orderer = orderer or DefaultOrderer()
self._preamble_generator = preamble_generator or DefaultPreambleGenerator()
self._skeleton_compressor = skeleton_compressor or DefaultSkeletonCompressor()
def assemble(
self,
plan_id: str,
fragments: Sequence[ContextFragment],
budget: ContextBudget,
strategy: str | None = None,
) -> ContextPayload:
"""Assemble context fragments into a budget-constrained payload."""
if not re.match(ULID_PATTERN, plan_id):
msg = f"plan_id must be a valid ULID, got {plan_id!r}"
raise ValueError(msg)
strategy_name = strategy or self._default_strategy
if strategy_name not in self._strategies:
msg = (
f"Unknown strategy {strategy_name!r}. "
f"Available: {', '.join(sorted(self._strategies))}"
)
raise ValueError(msg)
self._logger.info(
"Assembling context",
plan_id=plan_id,
strategy=strategy_name,
fragment_count=len(fragments),
budget_tokens=budget.available_tokens,
)
# Phase 1: Strategy Orchestration
# Pass the full registry so custom selectors can see all strategies.
# The request dict includes the caller's chosen strategy name so the
# selector (or pipeline) can narrow down when needed.
all_strategies = list(self._strategies.values())
resolved = self._strategies[strategy_name]
candidates = self._strategy_selector.select(
all_strategies,
{"strategy": strategy_name},
)
CoreRasurae commented 2026-03-04 23:42:59 +00:00
Outdated
Review
Copy Link
Copy Source

BUG-EDGE-2 [LOW]: Name-based comparison is correct for wrapped/copied strategies, but introduces a subtle edge: if two different strategy objects share the same name (e.g., a custom selector returns both a built-in and a wrapper with the same name), both would pass this filter. In v1 single-strategy mode this could lead to duplicate execution. Risk is low since register_strategy overwrites the dict entry by name.

**BUG-EDGE-2 [LOW]:** Name-based comparison is correct for wrapped/copied strategies, but introduces a subtle edge: if two different strategy objects share the same `name` (e.g., a custom selector returns both a built-in and a wrapper with the same name), both would pass this filter. In v1 single-strategy mode this could lead to duplicate execution. Risk is low since `register_strategy` overwrites the dict entry by name.
# Ensure the explicitly-requested strategy is among candidates.
# If the selector returned multiple, keep only the requested one
# for v1 single-strategy execution; future multi-strategy support
# will remove this filter. Compare by name rather than identity
# so custom selectors that wrap/copy strategy objects still match.
candidates = [(s, c) for s, c in candidates if s.name == strategy_name] or [
(resolved, 1.0)
]
allocations = self._budget_allocator.allocate(
candidates,
budget.available_tokens,
)
ranked = self._strategy_executor.execute(allocations, fragments, budget)
# Phase 2: Fragment Fusion pipeline
fused = self._deduplicator.deduplicate(ranked)
fused = self._depth_resolver.resolve(fused)
fused = self._scorer.score(fused)
fused = self._packer.pack(fused, budget)
fused = self._orderer.order(fused)
# Phase 3: Context Finalization
preamble = self._preamble_generator.generate(fused)
final_fragments = tuple(fused)
total_tokens = sum(f.token_count for f in final_fragments)
available = budget.available_tokens
budget_used = total_tokens / available if available > 0 else 0.0
context_hash = compute_context_hash(final_fragments)
provenance_map = build_provenance_map(final_fragments)
self._logger.info(
"Context assembled",
plan_id=plan_id,
strategy=strategy_name,
fragments_selected=len(final_fragments),
total_tokens=total_tokens,
budget_used=round(budget_used, 4),
)
return ContextPayload(
plan_id=plan_id,
fragments=final_fragments,
total_tokens=total_tokens,
budget=budget,
budget_used=round(min(budget_used, 1.0), 4),
strategies_used=(strategy_name,),
context_hash=context_hash,
preamble=preamble,
provenance_map=provenance_map,
)
def register_strategy(
self,
name: str,
strategy: ContextStrategy,
) -> None:
"""Register a custom context strategy instance."""
self._strategies[name] = strategy
self._logger.info("Registered strategy", name=name)
src/cleveragents/domain/models/core/__init__.py
+15 -1
View File
@@ -54,7 +54,7 @@ from cleveragents.domain.models.core.checkpoint import (
CheckpointMetadata,
CheckpointRetentionPolicy,
RollbackResult,
)
) # fmt: skip
from cleveragents.domain.models.core.context import (
Context,
ContextFile,
@@ -63,6 +63,14 @@ from cleveragents.domain.models.core.context import (
MaxContextCount,
SummaryForUpdateContextParams,
)
from cleveragents.domain.models.core.context_fragment import (
ContextBudget,
ContextFragment,
ContextPayload,
FragmentProvenance,
build_provenance_map,
compute_context_hash,
)
# Project context policy model
from cleveragents.domain.models.core.context_policy import (
@@ -317,8 +325,11 @@ __all__ = [
"CloudBillingFields",
"ConfidenceFactors",
"Context",
"ContextBudget",
"ContextConfig",
"ContextFile",
"ContextFragment",
"ContextPayload",
"ContextType",
"ContextUpdateResult",
"ContextView",
@@ -348,6 +359,7 @@ __all__ = [
"ErrorRecoveryPolicy",
"EscalationDecision",
"ExecutionEnvironment",
"FragmentProvenance",
"GuardResult",
"GuardrailAuditEntry",
"GuardrailAuditTrail",
@@ -460,9 +472,11 @@ __all__ = [
"User",
"Validation",
"ValidationMode",
"build_provenance_map",
"can_transition",
"can_transition_job",
"classify_error",
"compute_context_hash",
"deserialize_job_payload",
"get_builtin_profile",
"get_recovery_hints",
src/cleveragents/domain/models/core/context_fragment.py
+251
View File
@@ -0,0 +1,251 @@
"""ACMS v1 context fragment domain models.
Defines the core value objects for the Adaptive Context Management System
(ACMS) context assembly pipeline: ``ContextFragment``, ``FragmentProvenance``,
``ContextBudget``, and ``ContextPayload``. All models are frozen Pydantic v2
value objects.
Based on ``docs/specification.md`` ~line 25075 (ContextFragment) and ~line
25092 (AssembledContext).
"""
from __future__ import annotations
import copy
import hashlib
from datetime import UTC, datetime
from typing import Any, Literal
from pydantic import BaseModel, Field, field_validator, model_validator
from ulid import ULID
from cleveragents.domain.models.core.plan import ULID_PATTERN
# Maximum content length (characters). Enforced to prevent unbounded memory
# consumption when callers create fragments from large files.
MAX_CONTENT_LENGTH: int = 1_000_000 # ~1 MB of text
# Maximum number of metadata entries per fragment.
MAX_METADATA_ENTRIES: int = 64
class FragmentProvenance(BaseModel, frozen=True):
"""Provenance trace for a context fragment.
Records the originating resource and location so that fragments can be
traced back to their source for auditability and debugging.
Based on ``docs/specification.md`` ~line 25088.
"""
resource_uri: str = Field(
...,
min_length=1,
description="URI of the originating resource (e.g. 'project://myapp/src/main.py')",
)
location: str = Field(
default="",
description="Location within the resource (e.g. line range, section name)",
)
resource_type: str = Field(
default="unknown",
description=(
"Type of the originating resource (e.g. 'git-checkout', 'fs-directory')"
),
)
class ContextFragment(BaseModel, frozen=True):
"""A single piece of context assembled by the ACMS pipeline.
Based on ``docs/specification.md`` ~line 25081.
"""
fragment_id: str = Field(default_factory=lambda: str(ULID()))
uko_node: str = Field(
...,
min_length=1,
description="UKO URI of the source node",
)
content: str = Field(
...,
max_length=MAX_CONTENT_LENGTH,
description="Rendered text content",
)
detail_depth: int = Field(
default=0,
ge=0,
le=9,
description=(
"Resolved integer depth: 0 (MODULE_LISTING) through 9 (FULL_SOURCE)"
),
)
token_count: int = Field(
...,
ge=0,
description="Actual token count of content",
)
relevance_score: float = Field(
ge=0.0,
le=1.0,
default=0.5,
description="Relevance to the current request (0.0-1.0)",
)
provenance: FragmentProvenance = Field(
...,
description="Trace back to resource and location",
)
tier: Literal["hot", "warm", "cold"] = Field(default="warm")
metadata: dict[str, str] = Field(default_factory=dict)
created_at: datetime = Field(
default_factory=lambda: datetime.now(UTC),
description="When the fragment was created",
)
@field_validator("metadata")
@classmethod
def _validate_metadata_size(cls, v: dict[str, str]) -> dict[str, str]:
if len(v) > MAX_METADATA_ENTRIES:
msg = (
f"metadata must have at most {MAX_METADATA_ENTRIES} entries, "
f"got {len(v)}"
)
raise ValueError(msg)
# Defensive copy so callers cannot mutate the model's internal state.
return dict(v)
class ContextBudget(BaseModel, frozen=True):
"""Token budget for context assembly.
``reserved_tokens`` must be strictly less than ``max_tokens``. The
effective minimum for ``max_tokens`` is therefore
``reserved_tokens + 1``.
"""
max_tokens: int = Field(ge=1, default=4096)
reserved_tokens: int = Field(ge=0, default=512)
@model_validator(mode="after")
def _check_reserved_within_max(self) -> ContextBudget:
if self.reserved_tokens >= self.max_tokens:
msg = (
f"reserved_tokens ({self.reserved_tokens}) must be less than "
f"max_tokens ({self.max_tokens})"
)
raise ValueError(msg)
return self
@property
def available_tokens(self) -> int:
"""Return tokens available after reserving system-prompt space."""
return self.max_tokens - self.reserved_tokens
class ContextPayload(BaseModel, frozen=True):
"""Assembled context payload ready for actor consumption.
Corresponds to the spec's ``AssembledContext`` (~line 25098). Includes
``budget_used`` fraction, ``strategies_used`` list, ``context_hash`` for
snapshot integrity, optional ``preamble``, and ``provenance_map``.
"""
payload_id: str = Field(default_factory=lambda: str(ULID()))
plan_id: str = Field(
...,
pattern=ULID_PATTERN,
CoreRasurae commented 2026-03-04 23:42:59 +00:00
Outdated
Review
Copy Link
Copy Source

SEC-1 [HIGH]: This regex ^[\w.:/\-]+$ permits path traversal sequences. A plan_id of ../../etc/passwd passes this validation.

Every other model in the codebase (PlanIdentity, Decision, Checkpoint) uses the strict ULID pattern ^[0-9A-HJKMNP-TV-Z]{26}$. The spec (specification.md:18158) explicitly types plan_id as ULID.

The spec also mandates filesystem paths like <data-dir>/checkpoints/<plan_id>/ (line 43360), making this a latent path-traversal vulnerability.

Recommendation: Use the ULID pattern ^[0-9A-HJKMNP-TV-Z]{26}$ consistent with other models, or at minimum reject .. sequences.

**SEC-1 [HIGH]:** This regex `^[\w.:/\-]+$` permits path traversal sequences. A `plan_id` of `../../etc/passwd` passes this validation. Every other model in the codebase (`PlanIdentity`, `Decision`, `Checkpoint`) uses the strict ULID pattern `^[0-9A-HJKMNP-TV-Z]{26}$`. The spec (`specification.md:18158`) explicitly types `plan_id` as ULID. The spec also mandates filesystem paths like `<data-dir>/checkpoints/<plan_id>/` (line 43360), making this a latent path-traversal vulnerability. **Recommendation:** Use the ULID pattern `^[0-9A-HJKMNP-TV-Z]{26}$` consistent with other models, or at minimum reject `..` sequences.
description=(
"Plan identifier. Must be a valid ULID (26-char Crockford "
"Base32), consistent with PlanIdentity and other core models."
),
)
fragments: tuple[ContextFragment, ...] = ()
total_tokens: int = 0
budget: ContextBudget = Field(default_factory=ContextBudget)
budget_used: float = Field(
default=0.0,
ge=0.0,
le=1.0,
description="Fraction of budget consumed (0.0-1.0)",
)
strategies_used: tuple[str, ...] = Field(
default=(),
description="Which strategies contributed to this payload",
)
context_hash: str = Field(
default="",
description="SHA-256 hash of assembled content for snapshot integrity",
)
preamble: str | None = Field(
default=None,
description="Optional structure summary prepended to context",
)
provenance_map: dict[str, Any] = Field(
default_factory=dict,
description="Fragment ID -> provenance mapping for traceability",
)
@field_validator("provenance_map")
@classmethod
def _freeze_provenance_map(cls, v: dict[str, Any]) -> dict[str, Any]:
"""Deep defensive copy so callers cannot mutate nested state."""
return copy.deepcopy(v)
CoreRasurae commented 2026-03-04 23:42:59 +00:00
Outdated
Review
Copy Link
Copy Source

PERF-1 [LOW]: copy.deepcopy is ~10-50x slower than the previous shallow-copy approach. This validator runs on the hot path of every context assembly cycle. Correct fix for the nested-dict mutation bug, but worth noting for future profiling if provenance maps grow large.

**PERF-1 [LOW]:** `copy.deepcopy` is ~10-50x slower than the previous shallow-copy approach. This validator runs on the hot path of every context assembly cycle. Correct fix for the nested-dict mutation bug, but worth noting for future profiling if provenance maps grow large.
assembled_at: datetime = Field(
default_factory=lambda: datetime.now(UTC),
description="When the payload was assembled",
)
@property
def remaining_tokens(self) -> int:
"""Return tokens still available in the budget."""
return self.budget.available_tokens - self.total_tokens
@property
def is_within_budget(self) -> bool:
"""Return True if total tokens do not exceed available budget."""
return self.total_tokens <= self.budget.available_tokens
def compute_context_hash(fragments: tuple[ContextFragment, ...]) -> str:
"""Compute a SHA-256 hash over the ordered fragment contents.
Used for snapshot integrity checks and caching. Each fragment's
content is length-prefixed (8-byte big-endian) so that different
fragment boundary splits produce distinct hashes.
"""
hasher = hashlib.sha256()
for frag in fragments:
encoded = frag.content.encode("utf-8")
hasher.update(len(encoded).to_bytes(8, "big"))
hasher.update(encoded)
return hasher.hexdigest()
def build_provenance_map(
fragments: tuple[ContextFragment, ...],
) -> dict[str, Any]:
"""Build a provenance map from fragment IDs to their provenance data."""
return {
frag.fragment_id: {
"resource_uri": frag.provenance.resource_uri,
"location": frag.provenance.location,
"resource_type": frag.provenance.resource_type,
}
for frag in fragments
}
vulture_whitelist.py
+3
View File
@@ -638,6 +638,9 @@ sparql_query # noqa: B018, F821
get_triples # noqa: B018, F821
traverse # noqa: B018, F821
# ACMS Pipeline — SkeletonCompressor protocol parameter (required by interface)
skeleton_budget # noqa: B018, F821
# Context Tiers — public API (M6 ACMS, issue #208)
ContextTier # noqa: B018, F821
ActorRole # noqa: B018, F821