cleveragents-core/docs/reference/acms.md

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
CRP base model mutability	FragmentProvenance, ContextFragment, ContextBudget, AssembledContext are frozen=True	CRP models may be mutable	Core types extend CRP bases via Pydantic v2 inheritance; freezing CRP bases ensures consistent immutability across the hierarchy. No CRP consumer mutates these instances.

Extension Points

Register custom context strategies at runtime:

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

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}")