From febea8950f07025f319025fcb4c1efe26482c0f1 Mon Sep 17 00:00:00 2001 From: khyari hamza Date: Thu, 5 Mar 2026 00:53:11 +0000 Subject: [PATCH] feat(acms): add ACMS v1 context pipeline Implement the 10-component pluggable ACMS context assembly pipeline with three built-in strategies (relevance, recency, tiered), DI-based component injection, ULID-validated plan_id, largest-remainder budget allocation, and frozen Pydantic v2 domain models. Closes #188 --- CHANGELOG.md | 3 + benchmarks/acms_pipeline_bench.py | 213 ++++ docs/reference/acms.md | 274 +++++ features/acms_pipeline.feature | 616 ++++++++++ features/environment.py | 4 + features/steps/acms_pipeline_steps.py | 1080 +++++++++++++++++ robot/acms_pipeline.robot | 57 + robot/helper_acms_pipeline.py | 293 +++++ .../application/services/acms_service.py | 734 +++++++++++ .../domain/models/core/__init__.py | 16 +- .../domain/models/core/context_fragment.py | 251 ++++ vulture_whitelist.py | 3 + 12 files changed, 3543 insertions(+), 1 deletion(-) create mode 100644 benchmarks/acms_pipeline_bench.py create mode 100644 docs/reference/acms.md create mode 100644 features/acms_pipeline.feature create mode 100644 features/steps/acms_pipeline_steps.py create mode 100644 robot/acms_pipeline.robot create mode 100644 robot/helper_acms_pipeline.py create mode 100644 src/cleveragents/application/services/acms_service.py create mode 100644 src/cleveragents/domain/models/core/context_fragment.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d789f65c4..e82d3bc38 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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`, diff --git a/benchmarks/acms_pipeline_bench.py b/benchmarks/acms_pipeline_bench.py new file mode 100644 index 000000000..5092dc756 --- /dev/null +++ b/benchmarks/acms_pipeline_bench.py @@ -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) diff --git a/docs/reference/acms.md b/docs/reference/acms.md new file mode 100644 index 000000000..5de78503f --- /dev/null +++ b/docs/reference/acms.md @@ -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}") +``` diff --git a/features/acms_pipeline.feature b/features/acms_pipeline.feature new file mode 100644 index 000000000..0339cf20a --- /dev/null +++ b/features/acms_pipeline.feature @@ -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 + Given the ACMS pipeline modules are available + When I create a payload with 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" diff --git a/features/environment.py b/features/environment.py index 450ae9c03..31c6f9b42 100644 --- a/features/environment.py +++ b/features/environment.py @@ -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" diff --git a/features/steps/acms_pipeline_steps.py b/features/steps/acms_pipeline_steps.py new file mode 100644 index 000000000..3ec9a5ed0 --- /dev/null +++ b/features/steps/acms_pipeline_steps.py @@ -0,0 +1,1080 @@ +"""Step definitions for features/acms_pipeline.feature. + +Tests the ACMS v1 context assembly pipeline domain models and service +directly in-memory -- no database required. +""" + +from __future__ import annotations + +from collections.abc import Sequence +from datetime import UTC, datetime, timedelta +from typing import Any + +from behave import given, then, when +from behave.runner import Context +from pydantic import ValidationError + +from cleveragents.application.services.acms_service import ( + ACMSPipeline, + DefaultBudgetAllocator, + DefaultSkeletonCompressor, + DefaultStrategySelector, + RecencyStrategy, + RelevanceStrategy, + StrategyCapabilities, + TieredStrategy, +) +from cleveragents.domain.models.core.context_fragment import ( + ContextBudget, + ContextFragment, + ContextPayload, + FragmentProvenance, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +# Default provenance for test fragments. +_DEFAULT_PROVENANCE = FragmentProvenance(resource_uri="test://default") + + +def _make_fragment(**kwargs: Any) -> ContextFragment: + """Create a ContextFragment with sensible test defaults. + + Callers can override any field. ``uko_node``, ``token_count``, and + ``provenance`` get defaults so that most test steps don't need to + specify them explicitly. + """ + kwargs.setdefault("uko_node", "test://default") + kwargs.setdefault("token_count", 0) + kwargs.setdefault("provenance", _DEFAULT_PROVENANCE) + return ContextFragment(**kwargs) + + +class _ReverseContextStrategy: + """Test strategy that reverses fragment order and fits to budget.""" + + @property + def name(self) -> str: + return "reverse" + + @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]: + reversed_frags = list(reversed(fragments)) + result: list[ContextFragment] = [] + total = 0 + available = budget.available_tokens + for frag in reversed_frags: + if total >= available: + break + if total + frag.token_count <= available: + result.append(frag) + total += frag.token_count + return result + + def explain(self) -> str: + return "Reverses fragment order for testing." + + +# --------------------------------------------------------------------------- +# Given — ContextFragment +# --------------------------------------------------------------------------- + + +@given( + 'a context fragment with uko_node "{uko_node}" and content "{content}" and token_count {tokens:d}' +) +def step_fragment_with_defaults( + context: Context, uko_node: str, content: str, tokens: int +) -> None: + context.fragment = ContextFragment( + uko_node=uko_node, + content=content, + token_count=tokens, + provenance=FragmentProvenance(resource_uri=uko_node), + ) + + +@given("a context fragment with all fields:") +def step_fragment_with_all_fields(context: Context) -> None: + data: dict[str, str] = {} + for row in context.table: + data[row["field"]] = row["value"] + meta: dict[str, str] = {} + if "meta_key" in data and "meta_val" in data: + meta[data["meta_key"]] = data["meta_val"] + resource_uri = data.get("resource_uri", data["uko_node"]) + context.fragment = ContextFragment( + uko_node=data["uko_node"], + content=data["content"], + relevance_score=float(data["score"]), + token_count=int(data["tokens"]), + detail_depth=int(data.get("detail_depth", "0")), + tier=data["tier"], + metadata=meta, + provenance=FragmentProvenance(resource_uri=resource_uri), + ) + + +# --------------------------------------------------------------------------- +# Given — ContextBudget +# --------------------------------------------------------------------------- + + +@given("a context budget with max_tokens {max_tok:d} and reserved_tokens {res:d}") +def step_budget(context: Context, max_tok: int, res: int) -> None: + context.budget = ContextBudget(max_tokens=max_tok, reserved_tokens=res) + + +# --------------------------------------------------------------------------- +# Given — Fragment collections +# --------------------------------------------------------------------------- + + +@given("the following context fragments:") +def step_fragments_table(context: Context) -> None: + context.fragments = [] + for row in context.table: + uko_node = row["uko_node"] + frag = ContextFragment( + uko_node=uko_node, + content=row["content"], + relevance_score=float(row["score"]), + token_count=int(row["tokens"]), + provenance=FragmentProvenance(resource_uri=uko_node), + ) + context.fragments.append(frag) + + +@given("context fragments with different timestamps:") +def step_fragments_with_timestamps(context: Context) -> None: + context.fragments = [] + for row in context.table: + uko_node = row["uko_node"] + frag = ContextFragment( + uko_node=uko_node, + content=row["content"], + token_count=int(row["tokens"]), + created_at=datetime.fromisoformat(row["created_at"]), + provenance=FragmentProvenance(resource_uri=uko_node), + ) + context.fragments.append(frag) + + +@given("the following tiered context fragments:") +def step_tiered_fragments(context: Context) -> None: + context.fragments = [] + for row in context.table: + uko_node = row["uko_node"] + frag = ContextFragment( + uko_node=uko_node, + content=row["content"], + relevance_score=float(row["score"]), + token_count=int(row["tokens"]), + tier=row["tier"], + provenance=FragmentProvenance(resource_uri=uko_node), + ) + context.fragments.append(frag) + + +@given("no context fragments") +def step_no_fragments(context: Context) -> None: + context.fragments = [] + + +@given("a custom context strategy that reverses fragment order") +def step_register_custom_strategy(context: Context) -> None: + if not hasattr(context, "pipeline"): + context.pipeline = ACMSPipeline() + context.pipeline.register_strategy("reverse", _ReverseContextStrategy()) + + +# --------------------------------------------------------------------------- +# When — Validation +# --------------------------------------------------------------------------- + + +@given("the ACMS pipeline modules are available") +def step_acms_modules_available(context: Context) -> None: + # Just ensures the imports are loadable; nothing to store. + pass + + +@when("I create a budget with reserved_tokens equal to max_tokens") +def step_budget_reserved_equals_max(context: Context) -> None: + context.acms_error = None + try: + ContextBudget(max_tokens=100, reserved_tokens=100) + except ValidationError as exc: + context.acms_error = exc + + +@when("I create a budget with reserved_tokens greater than max_tokens") +def step_budget_reserved_greater_than_max(context: Context) -> None: + context.acms_error = None + try: + ContextBudget(max_tokens=100, reserved_tokens=200) + except ValidationError as exc: + context.acms_error = exc + + +@when("I create a budget with reserved_tokens {res:d} and max_tokens {max_tok:d}") +def step_budget_specific(context: Context, res: int, max_tok: int) -> None: + context.acms_error = None + try: + ContextBudget(max_tokens=max_tok, reserved_tokens=res) + except ValidationError as exc: + context.acms_error = exc + + +@when('I create a fragment with invalid tier "{tier}"') +def step_invalid_tier(context: Context, tier: str) -> None: + context.acms_error = None + try: + _make_fragment(content="test", tier=tier, token_count=1) + except ValidationError as exc: + context.acms_error = exc + + +@when("I create a fragment with relevance score {score}") +def step_create_fragment_bad_score(context: Context, score: str) -> None: + context.acms_error = None + try: + _make_fragment(content="test", relevance_score=float(score), token_count=1) + except ValidationError as exc: + context.acms_error = exc + + +@when("I create a fragment with token count {tokens:d}") +def step_create_fragment_bad_tokens(context: Context, tokens: int) -> None: + context.acms_error = None + try: + _make_fragment(content="test", token_count=tokens) + except ValidationError as exc: + context.acms_error = exc + + +@when("I create a fragment with empty uko_node") +def step_create_fragment_empty_uko_node(context: Context) -> None: + context.acms_error = None + try: + ContextFragment( + uko_node="", + content="test", + token_count=1, + provenance=_DEFAULT_PROVENANCE, + ) + except ValidationError as exc: + context.acms_error = exc + + +@when("I create a fragment with detail depth {depth:d}") +def step_create_fragment_bad_depth(context: Context, depth: int) -> None: + context.acms_error = None + try: + _make_fragment(content="test", detail_depth=depth, token_count=1) + except ValidationError as exc: + context.acms_error = exc + + +# --------------------------------------------------------------------------- +# When — Assembly +# --------------------------------------------------------------------------- + + +@when('I assemble with strategy "{strategy}"') +def step_assemble_with_strategy(context: Context, strategy: str) -> None: + if not hasattr(context, "pipeline"): + context.pipeline = ACMSPipeline() + context.assemble_error = None + try: + context.payload = context.pipeline.assemble( + plan_id="01JQTESTPN00000000000000AA", + fragments=list(context.fragments), + budget=context.budget, + strategy=strategy, + ) + except ValueError as exc: + context.assemble_error = exc + + +@when("I assemble without specifying a strategy") +def step_assemble_default_strategy(context: Context) -> None: + if not hasattr(context, "pipeline"): + context.pipeline = ACMSPipeline() + context.payload = context.pipeline.assemble( + plan_id="01JQTESTPN00000000000000AA", + fragments=list(context.fragments), + budget=context.budget, + ) + + +@when('I register the reverse strategy as "{name}"') +def step_register_reverse_as(context: Context, name: str) -> None: + if not hasattr(context, "pipeline"): + context.pipeline = ACMSPipeline() + context.pipeline.register_strategy(name, _ReverseContextStrategy()) + + +# --------------------------------------------------------------------------- +# Then — Fragment assertions +# --------------------------------------------------------------------------- + + +@then('the fragment content should be "{content}"') +def step_fragment_content(context: Context, content: str) -> None: + assert context.fragment.content == content + + +@then("the fragment relevance score should be {score}") +def step_fragment_score(context: Context, score: str) -> None: + assert context.fragment.relevance_score == float(score) + + +@then("the fragment token count should be {tokens:d}") +def step_fragment_tokens(context: Context, tokens: int) -> None: + assert context.fragment.token_count == tokens + + +@then("the fragment detail depth should be {depth:d}") +def step_fragment_detail_depth(context: Context, depth: int) -> None: + assert context.fragment.detail_depth == depth + + +@then("the fragment metadata should be empty") +def step_fragment_metadata_empty(context: Context) -> None: + assert context.fragment.metadata == {} + + +@then("the fragment id should be set") +def step_fragment_id_set(context: Context) -> None: + assert context.fragment.fragment_id + assert len(context.fragment.fragment_id) > 0 + + +@then("the fragment created_at should be a datetime") +def step_fragment_created_at_is_datetime(context: Context) -> None: + assert isinstance(context.fragment.created_at, datetime) + + +@then("the fragment provenance resource_uri should be set") +def step_fragment_provenance_set(context: Context) -> None: + assert context.fragment.provenance.resource_uri + assert len(context.fragment.provenance.resource_uri) > 0 + + +@then('the fragment metadata key "{key}" should be "{value}"') +def step_fragment_metadata_key(context: Context, key: str, value: str) -> None: + assert context.fragment.metadata[key] == value + + +@then('the fragment should have field "{field_name}"') +def step_fragment_has_field(context: Context, field_name: str) -> None: + assert hasattr(context.fragment, field_name), ( + f"ContextFragment missing field: {field_name}" + ) + + +# --------------------------------------------------------------------------- +# Then — Budget assertions +# --------------------------------------------------------------------------- + + +@then("the available tokens should be {tokens:d}") +def step_available_tokens(context: Context, tokens: int) -> None: + assert context.budget.available_tokens == tokens + + +# --------------------------------------------------------------------------- +# Then — Validation assertions +# --------------------------------------------------------------------------- + + +@then("an ACMS validation error should be raised") +def step_acms_validation_error_raised(context: Context) -> None: + error = getattr(context, "acms_error", None) + assert error is not None, "Expected a validation error but none was raised" + + +@then("an ACMS strategy error should be raised") +def step_acms_strategy_error_raised(context: Context) -> None: + assert context.assemble_error is not None, ( + "Expected a strategy error but none was raised" + ) + + +# --------------------------------------------------------------------------- +# Then — Payload assertions +# --------------------------------------------------------------------------- + + +@then("the payload should contain {count:d} fragments") +def step_payload_fragment_count(context: Context, count: int) -> None: + assert len(context.payload.fragments) == count + + +@then('the first fragment content should be "{content}"') +def step_first_fragment_content(context: Context, content: str) -> None: + assert context.payload.fragments[0].content == content + + +@then('the second fragment content should be "{content}"') +def step_second_fragment_content(context: Context, content: str) -> None: + assert context.payload.fragments[1].content == content + + +@then("the payload total tokens should be {tokens:d}") +def step_payload_total_tokens(context: Context, tokens: int) -> None: + assert context.payload.total_tokens == tokens + + +@then("the payload should be within budget") +def step_payload_within_budget(context: Context) -> None: + assert context.payload.is_within_budget + + +@then('the payload strategies used should include "{strategy}"') +def step_payload_strategies_used(context: Context, strategy: str) -> None: + assert strategy in context.payload.strategies_used, ( + f"Expected {strategy!r} in strategies_used={context.payload.strategies_used}" + ) + + +@then("the payload remaining tokens should be {tokens:d}") +def step_payload_remaining_tokens(context: Context, tokens: int) -> None: + assert context.payload.remaining_tokens == tokens + + +@then("the payload budget used should be {fraction}") +def step_payload_budget_used(context: Context, fraction: str) -> None: + assert abs(context.payload.budget_used - float(fraction)) < 1e-4, ( + f"Expected budget_used={fraction}, got {context.payload.budget_used}" + ) + + +@then("the payload context hash should be non-empty") +def step_payload_context_hash(context: Context) -> None: + assert context.payload.context_hash, "Expected non-empty context_hash" + assert len(context.payload.context_hash) == 64, ( + f"Expected SHA-256 hex (64 chars), got {len(context.payload.context_hash)}" + ) + + +@then("the payload provenance map should map each fragment ID") +def step_payload_provenance_map(context: Context) -> None: + for frag in context.payload.fragments: + assert frag.fragment_id in context.payload.provenance_map, ( + f"Fragment {frag.fragment_id} not in provenance_map" + ) + + +@then('the payload plan_id should be "{plan_id}"') +def step_payload_plan_id(context: Context, plan_id: str) -> None: + assert context.payload.plan_id == plan_id, ( + f"Expected plan_id={plan_id!r}, got {context.payload.plan_id!r}" + ) + + +@then("the payload payload_id should be a non-empty ULID") +def step_payload_payload_id(context: Context) -> None: + pid = context.payload.payload_id + assert pid, "payload_id is empty" + assert len(pid) == 26, f"Expected ULID (26 chars), got {len(pid)} chars: {pid!r}" + + +@then("the payload assembled_at should be a recent UTC datetime") +def step_payload_assembled_at(context: Context) -> None: + assembled = context.payload.assembled_at + assert isinstance(assembled, datetime), ( + f"assembled_at is not datetime, got {type(assembled)}" + ) + # Should be within the last 60 seconds. + now = datetime.now(UTC) + delta = now - assembled + assert delta < timedelta(seconds=60), ( + f"assembled_at is too old: {assembled} (now={now})" + ) + + +@then('the payload should have field "{field_name}"') +def step_payload_has_field(context: Context, field_name: str) -> None: + assert hasattr(context.payload, field_name), ( + f"ContextPayload missing field: {field_name}" + ) + + +# --------------------------------------------------------------------------- +# T1 — DI injection steps for pipeline components +# --------------------------------------------------------------------------- + + +class _TrackingSelector: + """Strategy selector that records whether it was called.""" + + def __init__(self) -> None: + self.called = False + + def select( + self, + strategies: Sequence[Any], + request: dict[str, Any], + ) -> list[tuple[Any, float]]: + self.called = True + # Delegate to default behaviour. + return DefaultStrategySelector().select(strategies, request) + + +class _HalvingAllocator: + """Budget allocator that halves the total budget for each candidate.""" + + def __init__(self) -> None: + self.called = False + + def allocate( + self, + candidates: list[tuple[Any, float]], + total_budget: int, + ) -> list[tuple[Any, float, int]]: + self.called = True + half = total_budget // 2 + return [(s, c, half) for s, c in candidates] + + +class _ReversingExecutor: + """Strategy executor that reverses the fragment order.""" + + def __init__(self) -> None: + self.called = False + + def execute( + self, + allocations: list[tuple[Any, float, int]], + fragments: Sequence[ContextFragment], + budget: ContextBudget, + ) -> Sequence[ContextFragment]: + self.called = True + return list(reversed(fragments)) + + +class _UkoDeduplicator: + """Deduplicator that keeps only the first fragment per uko_node.""" + + def __init__(self) -> None: + self.called = False + + def deduplicate( + self, + fragments: Sequence[ContextFragment], + ) -> Sequence[ContextFragment]: + self.called = True + seen: set[str] = set() + result: list[ContextFragment] = [] + for f in fragments: + if f.uko_node not in seen: + seen.add(f.uko_node) + result.append(f) + return result + + +class _CustomPreamble: + """Preamble generator that produces a known string.""" + + def generate( + self, + fragments: Sequence[ContextFragment], + ) -> str | None: + return f"Custom preamble: {len(fragments)} fragments" + + +@given("a pipeline with a custom tracking strategy selector") +def step_pipeline_custom_selector(context: Context) -> None: + selector = _TrackingSelector() + context.custom_selector = selector + context.pipeline = ACMSPipeline(strategy_selector=selector) + + +@given("a pipeline with a custom budget allocator that halves the budget") +def step_pipeline_custom_allocator(context: Context) -> None: + allocator = _HalvingAllocator() + context.custom_allocator = allocator + context.pipeline = ACMSPipeline(budget_allocator=allocator) + + +@given("a pipeline with a custom strategy executor that reverses fragments") +def step_pipeline_custom_executor(context: Context) -> None: + executor = _ReversingExecutor() + context.custom_executor = executor + context.pipeline = ACMSPipeline(strategy_executor=executor) + + +@given("a pipeline with a custom deduplicator that removes duplicates by uko_node") +def step_pipeline_custom_dedup(context: Context) -> None: + dedup = _UkoDeduplicator() + context.custom_deduplicator = dedup + context.pipeline = ACMSPipeline(deduplicator=dedup) + + +@given("a pipeline with a custom preamble generator") +def step_pipeline_custom_preamble(context: Context) -> None: + context.pipeline = ACMSPipeline(preamble_generator=_CustomPreamble()) + + +@then("the custom strategy selector should have been called") +def step_selector_called(context: Context) -> None: + assert context.custom_selector.called, "Custom selector was not invoked" + + +@then("the custom budget allocator should have been called") +def step_allocator_called(context: Context) -> None: + assert context.custom_allocator.called, "Custom allocator was not invoked" + + +@then("the custom strategy executor should have been called") +def step_executor_called(context: Context) -> None: + assert context.custom_executor.called, "Custom executor was not invoked" + + +@then("the custom deduplicator should have been called") +def step_deduplicator_called(context: Context) -> None: + assert context.custom_deduplicator.called, "Custom deduplicator was not invoked" + + +@then('the payload preamble should be "{expected}"') +def step_preamble_value(context: Context, expected: str) -> None: + assert context.payload.preamble == expected, ( + f"Expected preamble {expected!r}, got {context.payload.preamble!r}" + ) + + +# --------------------------------------------------------------------------- +# T2 — SkeletonCompressor steps +# --------------------------------------------------------------------------- + + +@when("I compress fragments with the default skeleton compressor and budget {budget:d}") +def step_compress_default(context: Context, budget: int) -> None: + frags = ( + _make_fragment(uko_node="test://a", content="hello", token_count=10), + _make_fragment(uko_node="test://b", content="world", token_count=10), + ) + context.original_fragments = frags + compressor = DefaultSkeletonCompressor() + context.compressed_fragments = compressor.compress(frags, budget) + + +@then("the compressed fragments should equal the original fragments") +def step_compressed_equal(context: Context) -> None: + assert context.compressed_fragments == context.original_fragments + + +@when( + "I compress fragments with a truncating skeleton compressor and budget {budget:d}" +) +def step_compress_truncating(context: Context, budget: int) -> None: + frags = ( + _make_fragment(uko_node="test://a", content="a" * 100, token_count=100), + _make_fragment(uko_node="test://b", content="b" * 100, token_count=100), + ) + context.original_fragments = frags + + class _TruncatingCompressor: + def compress( + self, + fragments: tuple[ContextFragment, ...], + skeleton_budget: int, + ) -> tuple[ContextFragment, ...]: + result = [] + remaining = skeleton_budget + for f in fragments: + if remaining <= 0: + break + # Re-create with truncated content. + trimmed = f.content[:remaining] + result.append( + _make_fragment( + uko_node=f.uko_node, + content=trimmed, + token_count=min(f.token_count, remaining), + ) + ) + remaining -= min(f.token_count, remaining) + return tuple(result) + + compressor = _TruncatingCompressor() + context.compressed_fragments = compressor.compress(frags, budget) + + +@then("the compressed fragments should have reduced content") +def step_compressed_reduced(context: Context) -> None: + orig_total = sum(len(f.content) for f in context.original_fragments) + comp_total = sum(len(f.content) for f in context.compressed_fragments) + assert comp_total < orig_total, ( + f"Expected reduced content: original={orig_total}, compressed={comp_total}" + ) + + +# --------------------------------------------------------------------------- +# T3 — Multi-candidate allocation steps +# --------------------------------------------------------------------------- + + +@when( + "I allocate budget {budget:d} across candidates with confidences {c1:g} and {c2:g}" +) +def step_allocate_multi(context: Context, budget: int, c1: float, c2: float) -> None: + from cleveragents.application.services.acms_service import ( + RecencyStrategy, + RelevanceStrategy, + ) + + s1 = RelevanceStrategy() + s2 = RecencyStrategy() + allocator = DefaultBudgetAllocator() + context.allocations = allocator.allocate([(s1, c1), (s2, c2)], budget) + context.alloc_budget = budget + + +@then("the first candidate should receive approximately {tokens:d} tokens") +def step_first_alloc(context: Context, tokens: int) -> None: + actual = context.allocations[0][2] + assert abs(actual - tokens) <= 1, ( + f"First candidate got {actual}, expected ~{tokens}" + ) + + +@then("the second candidate should receive approximately {tokens:d} tokens") +def step_second_alloc(context: Context, tokens: int) -> None: + actual = context.allocations[1][2] + assert abs(actual - tokens) <= 1, ( + f"Second candidate got {actual}, expected ~{tokens}" + ) + + +@then("the total allocated should not exceed {budget:d}") +def step_total_alloc(context: Context, budget: int) -> None: + total = sum(a[2] for a in context.allocations) + assert total <= budget, f"Total allocated {total} exceeds budget {budget}" + + +@when("I select strategies with no strategy hint from {count:d} registered strategies") +def step_select_no_hint(context: Context, count: int) -> None: + from cleveragents.application.services.acms_service import ( + RecencyStrategy, + RelevanceStrategy, + TieredStrategy, + ) + + strategies = [RelevanceStrategy(), RecencyStrategy(), TieredStrategy()] + selector = DefaultStrategySelector() + context.selected = selector.select(strategies[:count], {}) + + +@then("all {count:d} strategies should be returned with confidence scores") +def step_all_selected(context: Context, count: int) -> None: + assert len(context.selected) == count, ( + f"Expected {count} selected strategies, got {len(context.selected)}" + ) + for s, c in context.selected: + assert c > 0.0, f"Strategy {s.name} has zero confidence" + + +# --------------------------------------------------------------------------- +# plan_id ULID validation steps (SEC-1, SPEC-1, TEST-GAP-1) +# --------------------------------------------------------------------------- + + +@when("I create a payload with an empty plan_id") +def step_create_payload_empty_plan_id(context: Context) -> None: + context.acms_error = None + context.test_payload = None + try: + context.test_payload = ContextPayload(plan_id="") + except ValidationError as exc: + context.acms_error = exc + + +@when('I create a payload with plan_id "{plan_id}"') +def step_create_payload_plan_id(context: Context, plan_id: str) -> None: + context.acms_error = None + context.test_payload = None + try: + context.test_payload = ContextPayload(plan_id=plan_id) + except ValidationError as exc: + context.acms_error = exc + + +@then('the created payload plan_id should be "{plan_id}"') +def step_created_payload_plan_id(context: Context, plan_id: str) -> None: + assert context.test_payload is not None, "Payload was not created" + assert context.test_payload.plan_id == plan_id, ( + f"Expected plan_id={plan_id!r}, got {context.test_payload.plan_id!r}" + ) + + +# --------------------------------------------------------------------------- +# Strategy introspection steps (capabilities, can_handle, explain) +# --------------------------------------------------------------------------- + +_BUILTIN_STRATEGIES: dict[str, Any] = { + "relevance": RelevanceStrategy(), + "recency": RecencyStrategy(), + "tiered": TieredStrategy(), +} + + +@when('I inspect the capabilities of "{strategy_name}"') +def step_inspect_capabilities(context: Context, strategy_name: str) -> None: + context.inspected_capabilities = _BUILTIN_STRATEGIES[strategy_name].capabilities + + +@then("the strategy should declare {attr} as {expected}") +def step_capability_attr(context: Context, attr: str, expected: str) -> None: + caps = context.inspected_capabilities + actual = getattr(caps, attr) + expected_val = expected == "true" + assert actual == expected_val, f"Expected {attr}={expected_val}, got {actual}" + + +@when("I query can_handle on all three built-in strategies") +def step_query_can_handle(context: Context) -> None: + request: dict[str, Any] = {"strategy": "relevance"} + context.confidence_map = { + name: s.can_handle(request) for name, s in _BUILTIN_STRATEGIES.items() + } + + +@then('"{a}" should have the highest confidence') +def step_highest_confidence(context: Context, a: str) -> None: + conf_a = context.confidence_map[a] + for name, conf in context.confidence_map.items(): + if name != a: + assert conf_a > conf, f"Expected {a} ({conf_a}) > {name} ({conf})" + + +@then('"{a}" should have higher confidence than "{b}"') +def step_higher_confidence(context: Context, a: str, b: str) -> None: + assert context.confidence_map[a] > context.confidence_map[b], ( + f"Expected {a} ({context.confidence_map[a]}) > " + f"{b} ({context.confidence_map[b]})" + ) + + +@when("I call explain on each built-in strategy") +def step_call_explain(context: Context) -> None: + context.explanations = { + name: s.explain() for name, s in _BUILTIN_STRATEGIES.items() + } + + +@then('the "{strategy}" explanation should mention "{keyword}"') +def step_explain_keyword(context: Context, strategy: str, keyword: str) -> None: + explanation = context.explanations[strategy] + assert keyword.lower() in explanation.lower(), ( + f"Expected '{keyword}' in {strategy} explanation: {explanation!r}" + ) + + +# --------------------------------------------------------------------------- +# Fragment metadata overflow steps +# --------------------------------------------------------------------------- + + +@when("I create a fragment with {count:d} metadata entries") +def step_create_fragment_metadata(context: Context, count: int) -> None: + context.acms_error = None + context.test_fragment = None + metadata = {f"key_{i}": f"val_{i}" for i in range(count)} + try: + context.test_fragment = ContextFragment( + uko_node="project://test/meta.py", + content="test", + relevance_score=0.5, + token_count=10, + tier="hot", + provenance=FragmentProvenance(resource_uri="project://test/meta.py"), + metadata=metadata, + ) + except ValidationError as exc: + context.acms_error = exc + + +@then("the fragment should have {count:d} metadata entries") +def step_fragment_metadata_count(context: Context, count: int) -> None: + assert context.test_fragment is not None, "Fragment was not created" + assert len(context.test_fragment.metadata) == count + + +# --------------------------------------------------------------------------- +# Early plan_id rejection in assemble() +# --------------------------------------------------------------------------- + + +@when('I assemble with plan_id "{plan_id}"') +def step_assemble_with_plan_id(context: Context, plan_id: str) -> None: + if not hasattr(context, "pipeline"): + context.pipeline = ACMSPipeline() + context.assemble_error = None + try: + context.payload = context.pipeline.assemble( + plan_id=plan_id, + fragments=list(context.fragments), + budget=context.budget, + ) + except (ValueError, ValidationError) as exc: + context.assemble_error = exc + + +@then('an ACMS ValueError should be raised mentioning "{keyword}"') +def step_acms_value_error_mentioning(context: Context, keyword: str) -> None: + assert context.assemble_error is not None, "No error was raised" + assert isinstance(context.assemble_error, ValueError), ( + f"Expected ValueError, got {type(context.assemble_error).__name__}" + ) + assert keyword.lower() in str(context.assemble_error).lower(), ( + f"Expected '{keyword}' in error: {context.assemble_error}" + ) + + +# --------------------------------------------------------------------------- +# TEST-2 — Provenance map immutability step +# --------------------------------------------------------------------------- + + +@when("I create a payload with a mutable provenance map and mutate the source") +def step_create_payload_mutable_provenance(context: Context) -> None: + inner = {"strategy": "relevance", "extra": "value"} + source_map = {"frag-1": inner} + context.test_payload = ContextPayload( + plan_id="01JQTESTPN00000000000000AB", + provenance_map=source_map, + ) + # Mutate the source dict after construction. + inner["strategy"] = "TAMPERED" + source_map["frag-2"] = {"injected": "true"} + + +@then("the payload provenance map should be unchanged") +def step_provenance_map_unchanged(context: Context) -> None: + pmap = context.test_payload.provenance_map + assert "frag-1" in pmap, "frag-1 missing from provenance_map" + assert "frag-2" not in pmap, "frag-2 was injected into provenance_map" + assert pmap["frag-1"]["strategy"] == "relevance", ( + f"Expected 'relevance', got {pmap['frag-1']['strategy']!r} — " + "external mutation leaked into frozen model" + ) + + +# --------------------------------------------------------------------------- +# TEST-3 — Allocator rounding edge case steps +# --------------------------------------------------------------------------- + + +@when("I allocate budget {budget:d} across {count:d} candidates with equal confidence") +def step_allocate_equal(context: Context, budget: int, count: int) -> None: + from cleveragents.application.services.acms_service import ( + RecencyStrategy, + RelevanceStrategy, + TieredStrategy, + ) + + strategies = [RelevanceStrategy(), RecencyStrategy(), TieredStrategy()] + allocator = DefaultBudgetAllocator() + candidates = [(s, 0.5) for s in strategies[:count]] + context.allocations = allocator.allocate(candidates, budget) + context.alloc_budget = budget + + +@when("I allocate budget {budget:d} across {count:d} candidates with zero confidence") +def step_allocate_zero_conf(context: Context, budget: int, count: int) -> None: + from cleveragents.application.services.acms_service import ( + RecencyStrategy, + RelevanceStrategy, + TieredStrategy, + ) + + strategies = [RelevanceStrategy(), RecencyStrategy(), TieredStrategy()] + allocator = DefaultBudgetAllocator() + candidates = [(s, 0.0) for s in strategies[:count]] + context.allocations = allocator.allocate(candidates, budget) + context.alloc_budget = budget + + +@when( + "I allocate budget {budget:d} across {count:d} candidates with confidences {conf_str}" +) +def step_allocate_unequal( + context: Context, budget: int, count: int, conf_str: str +) -> None: + from cleveragents.application.services.acms_service import ( + RecencyStrategy, + RelevanceStrategy, + TieredStrategy, + ) + + confidences = [float(c) for c in conf_str.split()] + strategies = [RelevanceStrategy(), RecencyStrategy(), TieredStrategy()] + allocator = DefaultBudgetAllocator() + candidates = [(s, c) for s, c in zip(strategies[:count], confidences, strict=True)] + context.allocations = allocator.allocate(candidates, budget) + context.alloc_budget = budget + + +@then("the total allocated should equal exactly {budget:d}") +def step_total_exact(context: Context, budget: int) -> None: + total = sum(a[2] for a in context.allocations) + assert total == budget, ( + f"Expected total allocation to be exactly {budget}, got {total}" + ) + + +@then("each allocation should be {low:d} or {high:d}") +def step_each_allocation_bounded(context: Context, low: int, high: int) -> None: + for _strategy, _confidence, tokens in context.allocations: + assert tokens in {low, high}, ( + f"Expected each allocation to be {low} or {high}, got {tokens}" + ) + + +# --------------------------------------------------------------------------- +# TEST-4 — Strategy selector fallback step +# --------------------------------------------------------------------------- + + +class _EmptySelector: + """Strategy selector that always returns an empty candidate list.""" + + def __init__(self) -> None: + self.called = False + + def select( + self, + strategies: Sequence[Any], + request: dict[str, Any], + ) -> list[tuple[Any, float]]: + self.called = True + return [] + + +@given("a pipeline with a custom strategy selector that returns no candidates") +def step_pipeline_empty_selector(context: Context) -> None: + selector = _EmptySelector() + context.empty_selector = selector + context.pipeline = ACMSPipeline(strategy_selector=selector) + + +@then("the empty strategy selector should have been called") +def step_empty_selector_called(context: Context) -> None: + assert context.empty_selector.called, ( + "Expected _EmptySelector.select() to be called, but it was not" + ) diff --git a/robot/acms_pipeline.robot b/robot/acms_pipeline.robot new file mode 100644 index 000000000..1a0efa1cd --- /dev/null +++ b/robot/acms_pipeline.robot @@ -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 diff --git a/robot/helper_acms_pipeline.py b/robot/helper_acms_pipeline.py new file mode 100644 index 000000000..d8c2b5072 --- /dev/null +++ b/robot/helper_acms_pipeline.py @@ -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 " + "" + ) + 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()) diff --git a/src/cleveragents/application/services/acms_service.py b/src/cleveragents/application/services/acms_service.py new file mode 100644 index 000000000..bde322ae3 --- /dev/null +++ b/src/cleveragents/application/services/acms_service.py @@ -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] = [] + 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}, + ) + # 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) diff --git a/src/cleveragents/domain/models/core/__init__.py b/src/cleveragents/domain/models/core/__init__.py index 9fefdd1a5..5bdded228 100644 --- a/src/cleveragents/domain/models/core/__init__.py +++ b/src/cleveragents/domain/models/core/__init__.py @@ -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", diff --git a/src/cleveragents/domain/models/core/context_fragment.py b/src/cleveragents/domain/models/core/context_fragment.py new file mode 100644 index 000000000..16634d0a0 --- /dev/null +++ b/src/cleveragents/domain/models/core/context_fragment.py @@ -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, + 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) + + 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 + } diff --git a/vulture_whitelist.py b/vulture_whitelist.py index 36303264d..f7d47163a 100644 --- a/vulture_whitelist.py +++ b/vulture_whitelist.py @@ -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 -- 2.52.0