feat(acms): add context request protocol models
CI / benchmark-publish (pull_request) Has been skipped
CI / lint (pull_request) Successful in 14s
CI / build (pull_request) Successful in 14s
CI / quality (pull_request) Successful in 19s
CI / security (pull_request) Successful in 30s
CI / typecheck (pull_request) Successful in 35s
CI / unit_tests (pull_request) Successful in 2m58s
CI / integration_tests (pull_request) Successful in 3m8s
CI / docker (pull_request) Successful in 41s
CI / coverage (pull_request) Successful in 4m19s
CI / lint (push) Successful in 17s
CI / quality (push) Successful in 18s
CI / security (push) Successful in 31s
CI / typecheck (push) Successful in 38s
CI / build (push) Successful in 21s
CI / benchmark-regression (push) Has been skipped
CI / unit_tests (push) Successful in 2m28s
CI / docker (push) Successful in 38s
CI / integration_tests (push) Successful in 3m13s
CI / coverage (push) Successful in 3m38s
CI / benchmark-publish (push) Has been cancelled
CI / benchmark-regression (pull_request) Successful in 26m30s

Add CRP domain models for the Advanced Context Management System:
- ContextRequest: Focus-driven context retrieval with breadth, depth,
  strategy, temporal_scope, and skeleton_ratio parameters
- ContextFragment: Retrieved context with UKO URI, provenance,
  relevance score, token count, and detail level metadata
- ContextBudget: Token budget management with reservation support
- DetailLevel: Five-tier enum (skeleton through full)
- DetailLevelMap: Name-to-integer resolution registry with inheritance

Add builtin/context skill with stubbed tools (request_context,
query_history, get_context_budget) wired to future ACMS pipeline.

ISSUES CLOSED: #190
This commit was merged in pull request #505.
This commit is contained in:
2026-03-02 06:25:46 +00:00
committed by Forgejo
parent 69f5b2e3e6
commit d3b182e10e
10 changed files with 2087 additions and 6 deletions
+137
View File
@@ -0,0 +1,137 @@
"""ASV benchmarks for CRP domain model validation throughput.
Measures the performance of:
- ContextRequest creation and validation
- ContextFragment creation and validation
- ContextBudget creation and validation
- DetailLevelMap resolution (integer and named)
- AssembledContext creation
"""
from __future__ import annotations
import importlib
import sys
from pathlib import Path
# Ensure the local *source* tree is importable even when ASV has an
# older build of the package installed.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
# Force-reload the top-level package so Python picks up the source tree
# version instead of the potentially stale installed copy.
import cleveragents # noqa: E402
importlib.reload(cleveragents)
from cleveragents.domain.models.acms.crp import ( # noqa: E402
AssembledContext,
ContextBudget,
ContextFragment,
ContextRequest,
DetailLevelMap,
FragmentProvenance,
)
class TimeCRPModelCreation:
"""Benchmark CRP model creation throughput."""
timeout = 60
def setup(self) -> None:
"""Prepare reusable inputs."""
self.provenance = FragmentProvenance(
resource_uri="uko-py:module/auth",
location="lines 1-50",
strategy="simple-keyword",
)
self.dlm = DetailLevelMap(
domain="uko-code:",
max_depth=9,
levels={
"MODULE_LISTING": 0,
"MODULE_GRAPH": 1,
"MEMBER_LISTING": 2,
"MEMBER_SUMMARY": 3,
"SIGNATURES": 4,
"SIGNATURES_WITH_DOCS": 5,
"STRUCTURAL_OUTLINE": 6,
"KEY_LOGIC": 7,
"NEAR_COMPLETE": 8,
"FULL_SOURCE": 9,
},
)
def time_context_request_defaults(self) -> None:
"""Create ContextRequest with defaults."""
for _ in range(1000):
ContextRequest(purpose="benchmark")
def time_context_request_full(self) -> None:
"""Create ContextRequest with all fields populated."""
for _ in range(1000):
ContextRequest(
query="auth flow",
entities=["AuthManager"],
uko_types=["uko-py:Class"],
focus=["uko-py:class/AuthManager"],
breadth=3,
depth=4,
depth_gradient=True,
max_tokens=8000,
preferred_strategies=["arce", "semantic-embedding"],
required_backends=["vector", "graph"],
priority=0.8,
purpose="Understand auth flow",
)
def time_context_fragment(self) -> None:
"""Create ContextFragment with provenance."""
for _ in range(1000):
ContextFragment(
uko_node="uko-py:class/AuthManager",
content="class AuthManager: ...",
detail_depth=9,
token_count=800,
relevance_score=0.95,
provenance=self.provenance,
)
def time_context_budget(self) -> None:
"""Create ContextBudget and access available_tokens."""
for _ in range(1000):
b = ContextBudget(max_tokens=8000, reserved_tokens=1000)
_ = b.available_tokens
def time_detail_level_map_resolve_int(self) -> None:
"""Resolve integer depths via DetailLevelMap."""
for i in range(1000):
self.dlm.resolve(i % 15)
def time_detail_level_map_resolve_named(self) -> None:
"""Resolve named levels via DetailLevelMap."""
names: list[str] = list(self.dlm.levels.keys())
for i in range(1000):
self.dlm.resolve(names[i % len(names)])
def time_assembled_context(self) -> None:
"""Create AssembledContext."""
for _ in range(1000):
AssembledContext(
total_tokens=1500,
budget_used=0.75,
context_hash="abc123",
strategies_used=["simple-keyword", "arce"],
)
def time_fragment_provenance(self) -> None:
"""Create FragmentProvenance."""
for _ in range(1000):
FragmentProvenance(
resource_uri="uko-py:module/auth",
location="lines 1-50",
strategy="arce",
)
+181
View File
@@ -0,0 +1,181 @@
# Context Request Protocol (CRP) Models
The CRP provides the structured vocabulary through which actors (via
skills) declare what information they need, at what level of detail,
and with what scope, while they are reasoning.
## Core Data Types
### DetailLevelMap
Maps named detail levels to integer depths for a UKO domain.
Inheritance allows child maps to include all entries from their parent
map and insert additional levels at any integer position.
| Field | Type | Description |
|-------------|-------------------------|-------------------------------------------------|
| `domain` | `str` | UKO namespace (e.g., `"uko-code:"`, `"uko-py:"`)|
| `parent` | `DetailLevelMap \| None`| Parent map to inherit from |
| `levels` | `dict[str, int]` | Named level -> integer depth mapping |
| `max_depth` | `int` | Maximum meaningful depth for this domain (>= 0) |
**Methods:**
- `resolve(depth: int | str) -> int` -- Resolve a named level or integer to
an integer depth. Integer depths are clamped to `max_depth`.
- `register(name: str, value: int) -> None` -- Register a custom detail level.
- `effective_levels() -> dict[str, int]` -- Return the full merged level map
including inherited entries.
### ContextRequest
A structured request for context, issued by an actor or skill via the
CRP. Each request triggers a context assembly cycle in the ACMS
pipeline.
| Field | Type | Default | Description |
|------------------------|------------------|----------------------|---------------------------------------------|
| `query` | `str \| None` | `None` | Natural language query |
| `entities` | `list[str]` | `[]` | Named entities to focus on |
| `uko_types` | `list[str]` | `[]` | UKO types to filter |
| `focus` | `list[str]` | `[]` | URIs or identifiers to focus on |
| `breadth` | `int` | `2` | Dependency hops outward (>= 0) |
| `depth` | `int \| str` | `3` | Detail depth -- integer or named level |
| `depth_gradient` | `bool` | `True` | Items closer to focus get more detail |
| `temporal` | `TemporalScope` | `CURRENT` | Temporal scope for retrieval |
| `max_tokens` | `int \| None` | `None` | Maximum token budget (>= 0) |
| `preferred_strategies` | `list[str]` | `[]` | Preferred context strategies |
| `required_backends` | `list[str]` | `[]` | Required data backends |
| `priority` | `float` | `0.5` | 0.0 = background, 1.0 = critical |
| `purpose` | `str` | `""` | Why this context is needed |
### ContextFragment
The atomic unit of context returned by strategies and consumed by the
Context Assembly Pipeline.
| Field | Type | Description |
|-------------------|----------------------|-------------------------------------|
| `uko_node` | `str` | UKO URI of the source node |
| `content` | `str` | Rendered text content |
| `detail_depth` | `int` | Resolved integer depth (>= 0) |
| `token_count` | `int` | Token count of content (>= 0) |
| `relevance_score` | `float` | 0.0 -- 1.0 relevance to request |
| `provenance` | `FragmentProvenance` | Trace back to resource + location |
| `metadata` | `dict[str, Any]` | Strategy-specific metadata |
### FragmentProvenance
Links a fragment back to the originating resource and location.
| Field | Type | Default | Description |
|----------------|-------|---------|--------------------------------------|
| `resource_uri` | `str` | -- | URI of the originating resource |
| `location` | `str` | `""` | Location within the resource |
| `strategy` | `str` | `""` | Strategy that produced this fragment |
### ContextBudget
Token budget management with reservation support.
| Field | Type | Default | Description |
|-------------------|-------|---------|----------------------------------------|
| `max_tokens` | `int` | -- | Maximum total token budget (>= 0) |
| `reserved_tokens` | `int` | `0` | Reserved tokens (<= `max_tokens`) |
**Properties:**
- `available_tokens: int` -- `max_tokens - reserved_tokens`
### AssembledContext
The fused, budget-respecting context payload delivered to an actor.
| Field | Type | Description |
|-------------------|------------------------|--------------------------------------|
| `fragments` | `list[ContextFragment]`| Ordered context fragments |
| `total_tokens` | `int` | Total token count (>= 0) |
| `budget_used` | `float` | Budget fraction consumed (0.0--1.0) |
| `strategies_used` | `list[str]` | Contributing strategies |
| `context_hash` | `str` | Cryptographic hash for snapshot |
| `preamble` | `str \| None` | Optional structure summary |
| `provenance_map` | `dict[str, Any]` | Fragment -> resource/location map |
## Detail Level Table (Universal Layer 0)
| Depth | Question Answered | What Gets Included |
|-------|--------------------------------|-----------------------------------------|
| 0 | *What exists?* | Name/identifier of the information unit |
| 1 | *How is it organized?* | Names of immediate children |
| 2 | *What are the key relationships?* | Children + dependency/reference edges|
| 3 | *What is each thing's purpose?* | + Short descriptions/summaries |
| 4 | *What is the structural shape?* | + Type info, size/count metadata |
| 5+ | *How does it work?* | Progressively more content |
| max | *Everything.* | Complete content -- nothing omitted |
## Example Usage
```python
from cleveragents.domain.models.acms.crp import (
ContextBudget,
ContextFragment,
ContextRequest,
DetailLevelMap,
FragmentProvenance,
)
# Create a detail level map for source code
code_map = DetailLevelMap(
domain="uko-code:",
max_depth=9,
levels={
"MODULE_LISTING": 0,
"SIGNATURES": 4,
"FULL_SOURCE": 9,
},
)
# Resolve a named level
depth = code_map.resolve("SIGNATURES") # -> 4
# Create a context request
request = ContextRequest(
query="How does authentication work?",
focus=["uko-py:class/AuthManager"],
breadth=2,
depth="SIGNATURES",
purpose="Understand the auth flow for refactoring",
)
# Create a budget
budget = ContextBudget(max_tokens=8000, reserved_tokens=1000)
print(budget.available_tokens) # 7000
# Create a fragment
fragment = ContextFragment(
uko_node="uko-py:class/AuthManager",
content="class AuthManager: ...",
detail_depth=4,
token_count=120,
relevance_score=0.95,
provenance=FragmentProvenance(
resource_uri="uko-py:module/auth",
location="lines 1-50",
strategy="breadth-depth-navigator",
),
)
```
## Built-in Context Skill
The `builtin/context` skill is automatically injected into every
actor's skill set. It provides three tools:
| Tool | Description |
|-------------------------|-----------------------------------------------|
| `request_context` | Request specific context during reasoning |
| `query_history` | Query historical context about past decisions |
| `get_context_budget` | Check remaining context token budget |
These tools are currently stubbed (`NotImplementedError`) pending the
full ACMS Context Assembly Pipeline implementation.
+213
View File
@@ -0,0 +1,213 @@
Feature: CRP (Context Request Protocol) Domain Models
As a developer
I want CRP domain models for the Advanced Context Management System
So that actors can issue structured context requests during reasoning
# ---- DetailLevelMap ----
Scenario: Create a DetailLevelMap with valid levels
Given a DetailLevelMap for domain "uko-code:" with max depth 9
And the map has level "MODULE_LISTING" at depth 0
And the map has level "FULL_SOURCE" at depth 9
Then the map should resolve "MODULE_LISTING" to 0
And the map should resolve "FULL_SOURCE" to 9
Scenario: DetailLevelMap clamps integer depth to max_depth
Given a DetailLevelMap for domain "uko-code:" with max depth 9
Then the map should resolve integer 15 to 9
And the map should resolve integer 0 to 0
And the map should resolve integer 5 to 5
Scenario: DetailLevelMap resolves from parent map
Given a parent DetailLevelMap for domain "uko-code:" with max depth 9
And the parent map has level "MODULE_LISTING" at depth 0
And a child DetailLevelMap for domain "uko-oo:" with max depth 11
And the child map has level "CLASS_HIERARCHY" at depth 3
Then the child map should resolve "MODULE_LISTING" to 0
And the child map should resolve "CLASS_HIERARCHY" to 3
Scenario: DetailLevelMap raises on unknown named level
Given a DetailLevelMap for domain "uko-code:" with max depth 9
Then resolving "NONEXISTENT" should raise ValueError
Scenario: DetailLevelMap register adds custom level
Given a DetailLevelMap for domain "uko-code:" with max depth 9
When I register level "CUSTOM_LEVEL" at depth 7
Then the map should resolve "CUSTOM_LEVEL" to 7
Scenario: DetailLevelMap rejects negative depth in levels
Then creating a DetailLevelMap with level "BAD" at depth -1 should raise ValueError
Scenario: DetailLevelMap register rejects negative depth
Given a DetailLevelMap for domain "uko-code:" with max depth 9
Then registering level "BAD" at depth -1 should raise ValueError
Scenario: DetailLevelMap effective_levels merges parent and child
Given a parent DetailLevelMap for domain "uko-code:" with max depth 9
And the parent map has level "MODULE_LISTING" at depth 0
And the parent map has level "FULL_SOURCE" at depth 9
And a child DetailLevelMap for domain "uko-oo:" with max depth 11
And the child map has level "CLASS_HIERARCHY" at depth 3
Then the child effective_levels should contain "MODULE_LISTING" at 0
And the child effective_levels should contain "FULL_SOURCE" at 9
And the child effective_levels should contain "CLASS_HIERARCHY" at 3
Scenario: DetailLevelMap created with pre-populated levels dict
Given a DetailLevelMap for domain "uko-code:" with max depth 9 and levels "SIGNATURES" at 4
Then the map should resolve "SIGNATURES" to 4
# ---- FragmentProvenance ----
Scenario: Create a valid FragmentProvenance
Given a FragmentProvenance with resource URI "uko-py:module/auth"
Then the provenance resource_uri should be "uko-py:module/auth"
And the provenance location should be empty
And the provenance strategy should be empty
Scenario: FragmentProvenance rejects empty resource_uri
Then creating a FragmentProvenance with empty resource_uri should raise ValueError
# ---- ContextBudget ----
Scenario: Create a valid ContextBudget
Given a ContextBudget with max_tokens 8000 and reserved_tokens 1000
Then the budget available_tokens should be 7000
And the budget max_tokens should be 8000
And the budget reserved_tokens should be 1000
Scenario: ContextBudget with zero reservation
Given a ContextBudget with max_tokens 4000 and reserved_tokens 0
Then the budget available_tokens should be 4000
Scenario: ContextBudget rejects negative max_tokens
Then creating a ContextBudget with max_tokens -1 should raise ValueError
Scenario: ContextBudget rejects negative reserved_tokens
Then creating a ContextBudget with max_tokens 100 and reserved_tokens -5 should raise ValueError
Scenario: ContextBudget rejects reserved exceeding max
Then creating a ContextBudget with max_tokens 100 and reserved_tokens 200 should raise ValueError
# ---- ContextFragment ----
Scenario: Create a valid ContextFragment
Given a ContextFragment with uko_node "uko-py:class/AuthManager" and content "class AuthManager: ..."
And the fragment has detail_depth 9 and token_count 800
And the fragment has relevance_score 0.95
Then the fragment uko_node should be "uko-py:class/AuthManager"
And the fragment token_count should be 800
And the fragment relevance_score should be 0.95
And the fragment detail_depth should be 9
Scenario: ContextFragment rejects negative token_count
Then creating a ContextFragment with token_count -1 should raise ValueError
Scenario: ContextFragment rejects relevance_score above 1.0
Then creating a ContextFragment with relevance_score 1.5 should raise ValueError
Scenario: ContextFragment rejects relevance_score below 0.0
Then creating a ContextFragment with relevance_score -0.1 should raise ValueError
Scenario: ContextFragment rejects negative detail_depth
Then creating a ContextFragment with detail_depth -1 should raise ValueError
Scenario: ContextFragment rejects empty uko_node
Then creating a ContextFragment with empty uko_node should raise ValueError
# ---- ContextRequest ----
Scenario: Create a ContextRequest with defaults
Given a ContextRequest with purpose "Understand the auth module"
Then the request purpose should be "Understand the auth module"
And the request breadth should be 2
And the request depth should be 3
And the request depth_gradient should be true
And the request priority should be 0.5
And the request temporal should be "current"
And the request query should be None
And the request max_tokens should be None
Scenario: Create a ContextRequest with all fields
Given a full ContextRequest with query "auth flow" and focus "uko-py:class/AuthManager"
And the request has breadth 3 and depth "SIGNATURES"
And the request has priority 0.8 and purpose "Review auth"
Then the request query should be "auth flow"
And the request breadth should be 3
And the request depth should be "SIGNATURES"
And the request priority should be 0.8
Scenario: ContextRequest rejects negative breadth
Then creating a ContextRequest with breadth -1 should raise ValueError
Scenario: ContextRequest rejects negative integer depth
Then creating a ContextRequest with depth -1 should raise ValueError
Scenario: ContextRequest rejects priority above 1.0
Then creating a ContextRequest with priority 1.5 should raise ValueError
Scenario: ContextRequest rejects priority below 0.0
Then creating a ContextRequest with priority -0.1 should raise ValueError
Scenario: ContextRequest accepts string depth
Given a ContextRequest with depth "FULL_SOURCE"
Then the request depth should be "FULL_SOURCE"
Scenario: ContextRequest accepts zero breadth
Given a ContextRequest with breadth 0
Then the request breadth should be 0
Scenario: ContextRequest accepts max_tokens
Given a ContextRequest with max_tokens 4096
Then the request max_tokens should be 4096
Scenario: ContextRequest rejects negative max_tokens
Then creating a ContextRequest with max_tokens -100 should raise ValueError
# ---- AssembledContext ----
Scenario: Create a valid AssembledContext
Given an AssembledContext with total_tokens 1500 and budget_used 0.75
And the assembled context has hash "abc123def"
Then the assembled total_tokens should be 1500
And the assembled budget_used should be 0.75
And the assembled context_hash should be "abc123def"
Scenario: AssembledContext rejects budget_used above 1.0
Then creating an AssembledContext with budget_used 1.5 should raise ValueError
Scenario: AssembledContext rejects negative total_tokens
Then creating an AssembledContext with total_tokens -1 should raise ValueError
# ---- Context Skill Tool Stubs ----
Scenario: request_context tool raises NotImplementedError
When I call the request_context tool handler
Then it should raise NotImplementedError with message containing "ACMS pipeline"
Scenario: query_history tool raises NotImplementedError
When I call the query_history tool handler
Then it should raise NotImplementedError with message containing "ACMS pipeline"
Scenario: get_context_budget tool raises NotImplementedError
When I call the get_context_budget tool handler
Then it should raise NotImplementedError with message containing "ACMS pipeline"
Scenario: Context skill tools are registered in the tool registry
Given a CRP tool registry with context tools registered
Then the CRP registry should contain tool "builtin/request-context"
And the CRP registry should contain tool "builtin/query-history"
And the CRP registry should contain tool "builtin/get-context-budget"
Scenario: Context skill tools have read-only capability
Given a CRP tool registry with context tools registered
Then CRP tool "builtin/request-context" should have read_only capability
And CRP tool "builtin/query-history" should have read_only capability
And CRP tool "builtin/get-context-budget" should have read_only capability
Scenario: request_context tool requires purpose in schema
Given a CRP tool registry with context tools registered
Then CRP tool "builtin/request-context" should require "purpose" in input schema
Scenario: query_history tool requires query in schema
Given a CRP tool registry with context tools registered
Then CRP tool "builtin/query-history" should require "query" in input schema
+623
View File
@@ -0,0 +1,623 @@
"""Step definitions for the CRP Domain Models feature."""
from __future__ import annotations
from typing import Any
from behave import given, then, when
from pydantic import ValidationError
from cleveragents.domain.models.acms.crp import (
AssembledContext,
ContextBudget,
ContextFragment,
ContextRequest,
DetailLevelMap,
FragmentProvenance,
)
from cleveragents.domain.models.core.project import TemporalScope
from cleveragents.skills.builtins.context_ops import (
_handle_get_context_budget,
_handle_query_history,
_handle_request_context,
register_skill_context_tools,
)
from cleveragents.tool.registry import ToolRegistry
__all__: list[str] = []
# ---------------------------------------------------------------------------
# Helper
# ---------------------------------------------------------------------------
def _make_provenance(
resource_uri: str = "uko-py:module/test",
) -> FragmentProvenance:
"""Create a simple provenance for test fragments."""
return FragmentProvenance(resource_uri=resource_uri)
# ---------------------------------------------------------------------------
# DetailLevelMap
# ---------------------------------------------------------------------------
@given('a DetailLevelMap for domain "{domain}" with max depth {max_depth:d}')
def step_given_detail_level_map(context: Any, domain: str, max_depth: int) -> None:
context.detail_level_map = DetailLevelMap(domain=domain, max_depth=max_depth)
@given('the map has level "{name}" at depth {depth:d}')
def step_given_map_has_level(context: Any, name: str, depth: int) -> None:
context.detail_level_map.register(name, depth)
@then('the map should resolve "{name}" to {expected:d}')
def step_then_map_resolve_name(context: Any, name: str, expected: int) -> None:
actual: int = context.detail_level_map.resolve(name)
assert actual == expected, f"Expected {expected}, got {actual}"
@then("the map should resolve integer {value:d} to {expected:d}")
def step_then_map_resolve_int(context: Any, value: int, expected: int) -> None:
actual: int = context.detail_level_map.resolve(value)
assert actual == expected, f"Expected {expected}, got {actual}"
@given('a parent DetailLevelMap for domain "{domain}" with max depth {max_depth:d}')
def step_given_parent_map(context: Any, domain: str, max_depth: int) -> None:
context.parent_detail_level_map = DetailLevelMap(domain=domain, max_depth=max_depth)
@given('the parent map has level "{name}" at depth {depth:d}')
def step_given_parent_level(context: Any, name: str, depth: int) -> None:
context.parent_detail_level_map.register(name, depth)
@given('a child DetailLevelMap for domain "{domain}" with max depth {max_depth:d}')
def step_given_child_map(context: Any, domain: str, max_depth: int) -> None:
context.child_detail_level_map = DetailLevelMap(
domain=domain,
max_depth=max_depth,
parent=context.parent_detail_level_map,
)
@given('the child map has level "{name}" at depth {depth:d}')
def step_given_child_level(context: Any, name: str, depth: int) -> None:
context.child_detail_level_map.register(name, depth)
@then('the child map should resolve "{name}" to {expected:d}')
def step_then_child_resolve(context: Any, name: str, expected: int) -> None:
actual: int = context.child_detail_level_map.resolve(name)
assert actual == expected, f"Expected {expected}, got {actual}"
@then('resolving "{name}" should raise ValueError')
def step_then_resolve_raises(context: Any, name: str) -> None:
try:
context.detail_level_map.resolve(name)
raise AssertionError(f"Expected ValueError for '{name}'")
except ValueError:
pass
@when('I register level "{name}" at depth {depth:d}')
def step_when_register_level(context: Any, name: str, depth: int) -> None:
context.detail_level_map.register(name, depth)
@then(
'creating a DetailLevelMap with level "{name}" at depth {depth:d} '
"should raise ValueError"
)
def step_then_bad_level_raises(context: Any, name: str, depth: int) -> None:
try:
DetailLevelMap(domain="test:", max_depth=9, levels={name: depth})
raise AssertionError("Expected ValueError")
except (ValueError, ValidationError):
pass
@then('registering level "{name}" at depth {depth:d} should raise ValueError')
def step_then_register_bad_raises(context: Any, name: str, depth: int) -> None:
try:
context.detail_level_map.register(name, depth)
raise AssertionError("Expected ValueError")
except ValueError:
pass
@given(
'a DetailLevelMap for domain "{domain}" with max depth {max_depth:d} '
'and levels "{name}" at {depth:d}'
)
def step_given_detail_level_map_with_levels(
context: Any, domain: str, max_depth: int, name: str, depth: int
) -> None:
context.detail_level_map = DetailLevelMap(
domain=domain, max_depth=max_depth, levels={name: depth}
)
@then('the child effective_levels should contain "{name}" at {expected:d}')
def step_then_effective_levels(context: Any, name: str, expected: int) -> None:
levels: dict[str, int] = context.child_detail_level_map.effective_levels()
assert name in levels, f"'{name}' not in effective_levels"
assert levels[name] == expected, (
f"Expected {expected} for '{name}', got {levels[name]}"
)
# ---------------------------------------------------------------------------
# FragmentProvenance
# ---------------------------------------------------------------------------
@given('a FragmentProvenance with resource URI "{uri}"')
def step_given_provenance(context: Any, uri: str) -> None:
context.provenance = FragmentProvenance(resource_uri=uri)
@then('the provenance resource_uri should be "{expected}"')
def step_then_provenance_uri(context: Any, expected: str) -> None:
assert context.provenance.resource_uri == expected
@then("the provenance location should be empty")
def step_then_provenance_location_empty(context: Any) -> None:
assert context.provenance.location == ""
@then("the provenance strategy should be empty")
def step_then_provenance_strategy_empty(context: Any) -> None:
assert context.provenance.strategy == ""
@then("creating a FragmentProvenance with empty resource_uri should raise ValueError")
def step_then_provenance_empty_uri(context: Any) -> None:
try:
FragmentProvenance(resource_uri="")
raise AssertionError("Expected ValueError")
except (ValueError, ValidationError):
pass
# ---------------------------------------------------------------------------
# ContextBudget
# ---------------------------------------------------------------------------
@given("a ContextBudget with max_tokens {max_tok:d} and reserved_tokens {res:d}")
def step_given_budget(context: Any, max_tok: int, res: int) -> None:
context.budget = ContextBudget(max_tokens=max_tok, reserved_tokens=res)
@then("the budget available_tokens should be {expected:d}")
def step_then_budget_available(context: Any, expected: int) -> None:
assert context.budget.available_tokens == expected
@then("the budget max_tokens should be {expected:d}")
def step_then_budget_max(context: Any, expected: int) -> None:
assert context.budget.max_tokens == expected
@then("the budget reserved_tokens should be {expected:d}")
def step_then_budget_reserved(context: Any, expected: int) -> None:
assert context.budget.reserved_tokens == expected
@then("creating a ContextBudget with max_tokens {max_tok:d} should raise ValueError")
def step_then_budget_bad_max(context: Any, max_tok: int) -> None:
try:
ContextBudget(max_tokens=max_tok)
raise AssertionError("Expected ValueError")
except (ValueError, ValidationError):
pass
@then(
"creating a ContextBudget with max_tokens {max_tok:d} "
"and reserved_tokens {res:d} should raise ValueError"
)
def step_then_budget_bad_reserved(context: Any, max_tok: int, res: int) -> None:
try:
ContextBudget(max_tokens=max_tok, reserved_tokens=res)
raise AssertionError("Expected ValueError")
except (ValueError, ValidationError):
pass
# ---------------------------------------------------------------------------
# ContextFragment
# ---------------------------------------------------------------------------
@given('a ContextFragment with uko_node "{node}" and content "{content}"')
def step_given_fragment(context: Any, node: str, content: str) -> None:
context.fragment_node = node
context.fragment_content = content
context.fragment_detail_depth = 0
context.fragment_token_count = 0
context.fragment_relevance = 0.5
@given("the fragment has detail_depth {depth:d} and token_count {tokens:d}")
def step_given_fragment_depth_tokens(context: Any, depth: int, tokens: int) -> None:
context.fragment_detail_depth = depth
context.fragment_token_count = tokens
@given("the fragment has relevance_score {score:g}")
def step_given_fragment_relevance(context: Any, score: float) -> None:
context.fragment_relevance = score
context.fragment = ContextFragment(
uko_node=context.fragment_node,
content=context.fragment_content,
detail_depth=context.fragment_detail_depth,
token_count=context.fragment_token_count,
relevance_score=context.fragment_relevance,
provenance=_make_provenance(),
)
@then('the fragment uko_node should be "{expected}"')
def step_then_fragment_node(context: Any, expected: str) -> None:
assert context.fragment.uko_node == expected
@then("the fragment token_count should be {expected:d}")
def step_then_fragment_tokens(context: Any, expected: int) -> None:
assert context.fragment.token_count == expected
@then("the fragment relevance_score should be {expected:g}")
def step_then_fragment_relevance(context: Any, expected: float) -> None:
assert context.fragment.relevance_score == expected
@then("the fragment detail_depth should be {expected:d}")
def step_then_fragment_depth(context: Any, expected: int) -> None:
assert context.fragment.detail_depth == expected
@then("creating a ContextFragment with token_count {count:d} should raise ValueError")
def step_then_fragment_bad_tokens(context: Any, count: int) -> None:
try:
ContextFragment(
uko_node="uko-py:test",
content="test",
detail_depth=0,
token_count=count,
relevance_score=0.5,
provenance=_make_provenance(),
)
raise AssertionError("Expected ValueError")
except (ValueError, ValidationError):
pass
@then(
"creating a ContextFragment with relevance_score {score:g} should raise ValueError"
)
def step_then_fragment_bad_relevance(context: Any, score: float) -> None:
try:
ContextFragment(
uko_node="uko-py:test",
content="test",
detail_depth=0,
token_count=10,
relevance_score=score,
provenance=_make_provenance(),
)
raise AssertionError("Expected ValueError")
except (ValueError, ValidationError):
pass
@then("creating a ContextFragment with detail_depth {depth:d} should raise ValueError")
def step_then_fragment_bad_depth(context: Any, depth: int) -> None:
try:
ContextFragment(
uko_node="uko-py:test",
content="test",
detail_depth=depth,
token_count=10,
relevance_score=0.5,
provenance=_make_provenance(),
)
raise AssertionError("Expected ValueError")
except (ValueError, ValidationError):
pass
@then("creating a ContextFragment with empty uko_node should raise ValueError")
def step_then_fragment_empty_node(context: Any) -> None:
try:
ContextFragment(
uko_node="",
content="test",
detail_depth=0,
token_count=10,
relevance_score=0.5,
provenance=_make_provenance(),
)
raise AssertionError("Expected ValueError")
except (ValueError, ValidationError):
pass
# ---------------------------------------------------------------------------
# ContextRequest
# ---------------------------------------------------------------------------
@given('a ContextRequest with purpose "{purpose}"')
def step_given_request(context: Any, purpose: str) -> None:
context.ctx_request = ContextRequest(purpose=purpose)
@given('a full ContextRequest with query "{query}" and focus "{focus}"')
def step_given_full_request(context: Any, query: str, focus: str) -> None:
context.ctx_request = ContextRequest(
query=query,
focus=[focus],
)
@given('the request has breadth {breadth:d} and depth "{depth}"')
def step_given_request_breadth_depth_str(
context: Any, breadth: int, depth: str
) -> None:
context.ctx_request = context.ctx_request.model_copy(
update={"breadth": breadth, "depth": depth}
)
@given('the request has priority {priority:g} and purpose "{purpose}"')
def step_given_request_priority_purpose(
context: Any, priority: float, purpose: str
) -> None:
context.ctx_request = context.ctx_request.model_copy(
update={"priority": priority, "purpose": purpose}
)
@given('a ContextRequest with depth "{depth}"')
def step_given_request_depth_str(context: Any, depth: str) -> None:
context.ctx_request = ContextRequest(depth=depth)
@given("a ContextRequest with breadth {breadth:d}")
def step_given_request_breadth(context: Any, breadth: int) -> None:
context.ctx_request = ContextRequest(breadth=breadth)
@given("a ContextRequest with max_tokens {max_tok:d}")
def step_given_request_max_tokens(context: Any, max_tok: int) -> None:
context.ctx_request = ContextRequest(max_tokens=max_tok)
@then('the request purpose should be "{expected}"')
def step_then_request_purpose(context: Any, expected: str) -> None:
assert context.ctx_request.purpose == expected
@then("the request breadth should be {expected:d}")
def step_then_request_breadth(context: Any, expected: int) -> None:
assert context.ctx_request.breadth == expected
@then("the request depth should be {expected:d}")
def step_then_request_depth_int(context: Any, expected: int) -> None:
assert context.ctx_request.depth == expected
@then('the request depth should be "{expected}"')
def step_then_request_depth_str(context: Any, expected: str) -> None:
assert context.ctx_request.depth == expected
@then("the request depth_gradient should be true")
def step_then_request_gradient_true(context: Any) -> None:
assert context.ctx_request.depth_gradient is True
@then("the request priority should be {expected:g}")
def step_then_request_priority(context: Any, expected: float) -> None:
assert context.ctx_request.priority == expected
@then('the request temporal should be "{expected}"')
def step_then_request_temporal(context: Any, expected: str) -> None:
assert context.ctx_request.temporal == TemporalScope(expected)
@then("the request query should be None")
def step_then_request_query_none(context: Any) -> None:
assert context.ctx_request.query is None
@then('the request query should be "{expected}"')
def step_then_request_query(context: Any, expected: str) -> None:
assert context.ctx_request.query == expected
@then("the request max_tokens should be None")
def step_then_request_max_tokens_none(context: Any) -> None:
assert context.ctx_request.max_tokens is None
@then("the request max_tokens should be {expected:d}")
def step_then_request_max_tokens(context: Any, expected: int) -> None:
assert context.ctx_request.max_tokens == expected
@then("creating a ContextRequest with breadth {breadth:d} should raise ValueError")
def step_then_request_bad_breadth(context: Any, breadth: int) -> None:
try:
ContextRequest(breadth=breadth)
raise AssertionError("Expected ValueError")
except (ValueError, ValidationError):
pass
@then("creating a ContextRequest with depth {depth:d} should raise ValueError")
def step_then_request_bad_depth(context: Any, depth: int) -> None:
try:
ContextRequest(depth=depth)
raise AssertionError("Expected ValueError")
except (ValueError, ValidationError):
pass
@then("creating a ContextRequest with priority {priority:g} should raise ValueError")
def step_then_request_bad_priority(context: Any, priority: float) -> None:
try:
ContextRequest(priority=priority)
raise AssertionError("Expected ValueError")
except (ValueError, ValidationError):
pass
@then("creating a ContextRequest with max_tokens {max_tok:d} should raise ValueError")
def step_then_request_bad_max_tokens(context: Any, max_tok: int) -> None:
try:
ContextRequest(max_tokens=max_tok)
raise AssertionError("Expected ValueError")
except (ValueError, ValidationError):
pass
# ---------------------------------------------------------------------------
# AssembledContext
# ---------------------------------------------------------------------------
@given("an AssembledContext with total_tokens {total:d} and budget_used {used:g}")
def step_given_assembled(context: Any, total: int, used: float) -> None:
context.assembled_total = total
context.assembled_used = used
@given('the assembled context has hash "{hash_val}"')
def step_given_assembled_hash(context: Any, hash_val: str) -> None:
context.assembled = AssembledContext(
total_tokens=context.assembled_total,
budget_used=context.assembled_used,
context_hash=hash_val,
)
@then("the assembled total_tokens should be {expected:d}")
def step_then_assembled_total(context: Any, expected: int) -> None:
assert context.assembled.total_tokens == expected
@then("the assembled budget_used should be {expected:g}")
def step_then_assembled_used(context: Any, expected: float) -> None:
assert context.assembled.budget_used == expected
@then('the assembled context_hash should be "{expected}"')
def step_then_assembled_hash(context: Any, expected: str) -> None:
assert context.assembled.context_hash == expected
@then("creating an AssembledContext with budget_used {used:g} should raise ValueError")
def step_then_assembled_bad_used(context: Any, used: float) -> None:
try:
AssembledContext(
total_tokens=100,
budget_used=used,
context_hash="test",
)
raise AssertionError("Expected ValueError")
except (ValueError, ValidationError):
pass
@then(
"creating an AssembledContext with total_tokens {total:d} should raise ValueError"
)
def step_then_assembled_bad_total(context: Any, total: int) -> None:
try:
AssembledContext(
total_tokens=total,
budget_used=0.5,
context_hash="test",
)
raise AssertionError("Expected ValueError")
except (ValueError, ValidationError):
pass
# ---------------------------------------------------------------------------
# Context skill tool stubs
# ---------------------------------------------------------------------------
@when("I call the request_context tool handler")
def step_when_call_request_context(context: Any) -> None:
context.tool_error = None
try:
_handle_request_context({"purpose": "test"})
except NotImplementedError as exc:
context.tool_error = exc
@when("I call the query_history tool handler")
def step_when_call_query_history(context: Any) -> None:
context.tool_error = None
try:
_handle_query_history({"query": "test"})
except NotImplementedError as exc:
context.tool_error = exc
@when("I call the get_context_budget tool handler")
def step_when_call_get_budget(context: Any) -> None:
context.tool_error = None
try:
_handle_get_context_budget({})
except NotImplementedError as exc:
context.tool_error = exc
@then('it should raise NotImplementedError with message containing "{text}"')
def step_then_not_impl_error(context: Any, text: str) -> None:
assert context.tool_error is not None, "Expected NotImplementedError"
assert text in str(context.tool_error), (
f"Expected '{text}' in '{context.tool_error}'"
)
@given("a CRP tool registry with context tools registered")
def step_given_crp_registry_with_context(context: Any) -> None:
context.crp_tool_registry = ToolRegistry()
register_skill_context_tools(context.crp_tool_registry)
@then('the CRP registry should contain tool "{name}"')
def step_then_crp_registry_has_tool(context: Any, name: str) -> None:
spec = context.crp_tool_registry.get(name)
assert spec is not None, f"Tool '{name}' not found in CRP registry"
@then('CRP tool "{name}" should have read_only capability')
def step_then_crp_tool_read_only(context: Any, name: str) -> None:
spec = context.crp_tool_registry.get(name)
assert spec is not None, f"Tool '{name}' not found"
assert spec.capabilities.read_only is True, f"Tool '{name}' is not read-only"
@then('CRP tool "{name}" should require "{field}" in input schema')
def step_then_crp_tool_requires_field(context: Any, name: str, field: str) -> None:
spec = context.crp_tool_registry.get(name)
assert spec is not None, f"Tool '{name}' not found"
required: list[str] = spec.input_schema.get("required", [])
assert field in required, f"Field '{field}' not in required fields: {required}"
+73
View File
@@ -0,0 +1,73 @@
*** Settings ***
Documentation Smoke tests for CRP domain model creation and validation
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${HELPER} ${CURDIR}/helper_crp_models.py
*** Test Cases ***
Create Valid ContextRequest
[Documentation] Create a ContextRequest with defaults and verify success
${result}= Run Process ${PYTHON} ${HELPER} create-request cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} crp-request-ok
Create Valid ContextFragment
[Documentation] Create a ContextFragment with valid inputs and verify success
${result}= Run Process ${PYTHON} ${HELPER} create-fragment cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} crp-fragment-ok
Create Valid ContextBudget
[Documentation] Create a ContextBudget and verify available_tokens
${result}= Run Process ${PYTHON} ${HELPER} create-budget cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} crp-budget-ok
Create Valid DetailLevelMap
[Documentation] Create a DetailLevelMap and resolve a named level
${result}= Run Process ${PYTHON} ${HELPER} create-map cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} crp-map-ok
Reject Invalid ContextBudget
[Documentation] Reject a ContextBudget with reserved > max
${result}= Run Process ${PYTHON} ${HELPER} reject-budget cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} crp-reject-budget-ok
Reject Invalid ContextFragment
[Documentation] Reject a ContextFragment with relevance_score > 1.0
${result}= Run Process ${PYTHON} ${HELPER} reject-fragment cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} crp-reject-fragment-ok
Context Skill Tools Registered
[Documentation] Verify context skill tools register into the tool registry
${result}= Run Process ${PYTHON} ${HELPER} register-tools cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} crp-tools-ok
Create Valid AssembledContext
[Documentation] Create an AssembledContext and verify fields
${result}= Run Process ${PYTHON} ${HELPER} create-assembled cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} crp-assembled-ok
+156
View File
@@ -0,0 +1,156 @@
"""Robot Framework helper for CRP domain model validation.
Provides a CLI-style interface for Robot to invoke model creation
and validation. Exit code 0 = success, 1 = failure.
Usage:
python robot/helper_crp_models.py <command>
"""
from __future__ import annotations
import sys
from pathlib import Path
# Ensure the src directory is on the import path.
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
sys.path.insert(0, _SRC)
from cleveragents.domain.models.acms.crp import ( # noqa: E402
AssembledContext,
ContextBudget,
ContextFragment,
ContextRequest,
DetailLevelMap,
FragmentProvenance,
)
from cleveragents.skills.builtins.context_ops import ( # noqa: E402
register_skill_context_tools,
)
from cleveragents.tool.registry import ToolRegistry # noqa: E402
def main() -> int:
"""Entry point called by Robot Framework ``Run Process``."""
if len(sys.argv) < 2:
print("Usage: helper_crp_models.py <command>")
return 1
command: str = sys.argv[1]
if command == "create-request":
try:
req = ContextRequest(purpose="Test context request")
assert req.breadth == 2
assert req.depth == 3
print("crp-request-ok")
return 0
except Exception as exc:
print(f"crp-request-fail: {exc}")
return 1
if command == "create-fragment":
try:
frag = ContextFragment(
uko_node="uko-py:class/TestClass",
content="class TestClass: pass",
detail_depth=4,
token_count=50,
relevance_score=0.85,
provenance=FragmentProvenance(resource_uri="uko-py:module/test"),
)
assert frag.token_count == 50
assert frag.relevance_score == 0.85
print("crp-fragment-ok")
return 0
except Exception as exc:
print(f"crp-fragment-fail: {exc}")
return 1
if command == "create-budget":
try:
budget = ContextBudget(max_tokens=8000, reserved_tokens=1000)
assert budget.available_tokens == 7000
print("crp-budget-ok")
return 0
except Exception as exc:
print(f"crp-budget-fail: {exc}")
return 1
if command == "create-map":
try:
dlm = DetailLevelMap(
domain="uko-code:",
max_depth=9,
levels={"MODULE_LISTING": 0, "FULL_SOURCE": 9},
)
assert dlm.resolve("MODULE_LISTING") == 0
assert dlm.resolve("FULL_SOURCE") == 9
assert dlm.resolve(15) == 9 # clamped
print("crp-map-ok")
return 0
except Exception as exc:
print(f"crp-map-fail: {exc}")
return 1
if command == "reject-budget":
try:
ContextBudget(max_tokens=100, reserved_tokens=200)
print("crp-reject-budget-fail: should have raised")
return 1
except Exception:
print("crp-reject-budget-ok")
return 0
if command == "reject-fragment":
try:
ContextFragment(
uko_node="uko-py:test",
content="test",
detail_depth=0,
token_count=10,
relevance_score=1.5,
provenance=FragmentProvenance(resource_uri="uko-py:module/test"),
)
print("crp-reject-fragment-fail: should have raised")
return 1
except Exception:
print("crp-reject-fragment-ok")
return 0
if command == "register-tools":
try:
registry = ToolRegistry()
register_skill_context_tools(registry)
assert registry.get("builtin/request-context") is not None
assert registry.get("builtin/query-history") is not None
assert registry.get("builtin/get-context-budget") is not None
print("crp-tools-ok")
return 0
except Exception as exc:
print(f"crp-tools-fail: {exc}")
return 1
if command == "create-assembled":
try:
assembled = AssembledContext(
total_tokens=1500,
budget_used=0.75,
context_hash="abc123def",
strategies_used=["simple-keyword"],
)
assert assembled.total_tokens == 1500
assert assembled.budget_used == 0.75
print("crp-assembled-ok")
return 0
except Exception as exc:
print(f"crp-assembled-fail: {exc}")
return 1
print(f"Unknown command: {command}")
return 1
if __name__ == "__main__":
sys.exit(main())
@@ -0,0 +1,34 @@
"""ACMS (Advanced Context Management System) domain models.
Provides the Context Request Protocol (CRP) data types used by actors,
skills, strategies, and the Context Assembly Pipeline:
- ``DetailLevelMap`` -- Named-level-to-integer resolution with inheritance
- ``ContextRequest`` -- Structured request for context via the CRP
- ``ContextFragment`` -- Atomic unit of retrieved context
- ``FragmentProvenance`` -- Trace-back to source resource and location
- ``ContextBudget`` -- Token budget management with reservation support
- ``AssembledContext`` -- Final budget-respecting context payload
Based on ``docs/specification.md`` ACMS / CRP sections and ADR-014.
"""
from __future__ import annotations
from cleveragents.domain.models.acms.crp import (
AssembledContext,
ContextBudget,
ContextFragment,
ContextRequest,
DetailLevelMap,
FragmentProvenance,
)
__all__: list[str] = [
"AssembledContext",
"ContextBudget",
"ContextFragment",
"ContextRequest",
"DetailLevelMap",
"FragmentProvenance",
]
+445
View File
@@ -0,0 +1,445 @@
"""Context Request Protocol (CRP) domain models for CleverAgents.
The CRP is the structured vocabulary through which actors (via skills)
declare what information they need, at what level of detail, and with
what scope, while they are reasoning. These models are consumed by the
Context Assembly Pipeline and produced by context strategies.
## Core Types
| Type | Role |
|---------------------|-----------------------------------------------------|
| ``DetailLevelMap`` | Named-level-to-integer resolution with inheritance |
| ``ContextRequest`` | Actor-issued request for context |
| ``ContextFragment`` | Atomic unit of context returned by strategies |
| ``FragmentProvenance`` | Trace-back to resource + location |
| ``ContextBudget`` | Token budget management with reservation |
| ``AssembledContext`` | Fused, budget-respecting context payload |
## DetailDepth
At the universal level (UKO Layer 0), detail depth is a **non-negative
integer** -- 0 is the most minimal representation and each increment
reveals more. There is no fixed upper bound; the maximum meaningful
depth depends on the domain and the complexity of the information unit.
Each UKO domain extension registers a ``DetailLevelMap`` -- a table of
named labels mapped to specific integer depths. Maps are inherited:
a child map includes all entries from its parent map, and may insert
additional levels at any integer position.
Based on ``docs/specification.md`` ACMS / CRP sections and ADR-014.
"""
from __future__ import annotations
from typing import Any
from pydantic import (
BaseModel,
ConfigDict,
Field,
field_validator,
model_validator,
)
from cleveragents.domain.models.core.project import TemporalScope
# ---------------------------------------------------------------------------
# DetailLevelMap
# ---------------------------------------------------------------------------
class DetailLevelMap(BaseModel):
"""Maps named detail levels to integer depths for a UKO domain.
Inherited: a child map includes all entries from its parent map,
and may insert additional levels at any integer position.
Example::
code_map = DetailLevelMap(
domain="uko-code:",
parent=None,
levels={"MODULE_LISTING": 0, "FULL_SOURCE": 9},
max_depth=9,
)
assert code_map.resolve(4) == 4
assert code_map.resolve("MODULE_LISTING") == 0
"""
domain: str = Field(
...,
min_length=1,
description="UKO namespace (e.g., 'uko-code:', 'uko-py:')",
)
parent: DetailLevelMap | None = Field(
default=None,
description="Parent map to inherit from",
)
levels: dict[str, int] = Field(
default_factory=dict,
description="Named level -> integer depth",
)
max_depth: int = Field(
...,
ge=0,
description="Maximum meaningful depth for this domain",
)
@field_validator("levels")
@classmethod
def validate_levels(
cls: type[DetailLevelMap],
v: dict[str, int],
) -> dict[str, int]:
"""Ensure all level values are non-negative integers."""
for name, depth in v.items():
if depth < 0:
raise ValueError(f"Detail level '{name}' has negative depth {depth}")
return v
def resolve(self, depth: int | str) -> int:
"""Resolve a named level or integer to an integer depth.
Args:
depth: Either a raw integer (clamped to ``max_depth``) or
a named level string looked up in this map, then in
parent maps.
Returns:
Resolved integer depth.
Raises:
ValueError: If *depth* is a string not found in this map
or any ancestor.
"""
if isinstance(depth, int):
return min(depth, self.max_depth)
# Look up named level in this map, then parent maps
if depth in self.levels:
return self.levels[depth]
if self.parent is not None:
return self.parent.resolve(depth)
raise ValueError(f"Unknown detail level: {depth}")
def register(self, name: str, value: int) -> None:
"""Register a custom detail level.
Args:
name: Named level label.
value: Integer depth to map to.
Raises:
ValueError: If *value* is negative.
"""
if value < 0:
raise ValueError(f"Detail level '{name}' has negative depth {value}")
self.levels[name] = value
def effective_levels(self) -> dict[str, int]:
"""Return the full merged level map including inherited entries.
Parent entries are included first; child entries override any
collisions.
"""
merged = self.parent.effective_levels() if self.parent is not None else {}
merged.update(self.levels)
return merged
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
# ---------------------------------------------------------------------------
# FragmentProvenance
# ---------------------------------------------------------------------------
class FragmentProvenance(BaseModel):
"""Provenance trace for a context fragment.
Links a fragment back to the originating resource and location
within that resource.
"""
resource_uri: str = Field(
...,
min_length=1,
description="URI of the originating resource",
)
location: str = Field(
default="",
description="Location within the resource (e.g., line range, section)",
)
strategy: str = Field(
default="",
description="Name of the strategy that produced this fragment",
)
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
# ---------------------------------------------------------------------------
# ContextBudget
# ---------------------------------------------------------------------------
class ContextBudget(BaseModel):
"""Token budget for context assembly.
Tracks the maximum token budget and any reserved portion, providing
a computed ``available_tokens`` property for the remaining budget.
Example::
budget = ContextBudget(max_tokens=8000, reserved_tokens=1000)
assert budget.available_tokens == 7000
"""
max_tokens: int = Field(
...,
ge=0,
description="Maximum total token budget",
)
reserved_tokens: int = Field(
default=0,
ge=0,
description="Tokens reserved (e.g., for skeleton context)",
)
@model_validator(mode="after")
def _validate_reservation(self) -> ContextBudget:
"""Ensure reserved_tokens does not exceed max_tokens."""
if self.reserved_tokens > self.max_tokens:
raise ValueError("reserved_tokens cannot exceed max_tokens")
return self
@property
def available_tokens(self) -> int:
"""Return the number of tokens available after reservations."""
return self.max_tokens - self.reserved_tokens
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
# ---------------------------------------------------------------------------
# ContextFragment
# ---------------------------------------------------------------------------
class ContextFragment(BaseModel):
"""A single piece of context assembled by a strategy.
The atomic unit of context returned by strategies and consumed by
the Context Assembly Pipeline's fusion phase
(FragmentDeduplicator, DetailDepthResolver, FragmentScorer,
BudgetPacker, FragmentOrderer).
"""
uko_node: str = Field(
...,
min_length=1,
description="UKO URI of the source node",
)
content: str = Field(
...,
description="Rendered text content",
)
detail_depth: int = Field(
...,
ge=0,
description="Resolved integer depth of this fragment",
)
token_count: int = Field(
...,
ge=0,
description="Token count of content",
)
relevance_score: float = Field(
...,
ge=0.0,
le=1.0,
description="0.0-1.0 relevance to the request",
)
provenance: FragmentProvenance = Field(
...,
description="Trace back to resource + location",
)
metadata: dict[str, Any] = Field(
default_factory=dict,
description="Arbitrary strategy-specific metadata",
)
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
# ---------------------------------------------------------------------------
# ContextRequest
# ---------------------------------------------------------------------------
class ContextRequest(BaseModel):
"""A structured request for context, issued by an actor or skill.
Actors issue context requests through the ``builtin/context`` skill
which is automatically injected into every actor's skill set. Each
request triggers a context assembly cycle in the ACMS pipeline.
Fields follow the spec (``docs/specification.md`` > ACMS > CRP >
ContextRequest).
"""
# === What to find ===
query: str | None = Field(
default=None,
description="Natural language query",
)
entities: list[str] = Field(
default_factory=list,
description="Named entities to focus on",
)
uko_types: list[str] = Field(
default_factory=list,
description="UKO types to filter",
)
# === Scope control ===
focus: list[str] = Field(
default_factory=list,
description="URIs or identifiers of specific items to focus on",
)
breadth: int = Field(
default=2,
ge=0,
description="How many hops outward in the dependency/reference graph",
)
depth: int | str = Field(
default=3,
description=(
"How much detail to include for each item found. "
"May be a raw integer (0-N) or a named level string "
"(e.g., 'SIGNATURES') resolved via the active DetailLevelMap."
),
)
# === Focus depth gradient ===
depth_gradient: bool = Field(
default=True,
description="When True, items closer to the focus get more detail",
)
# === Temporal scope ===
temporal: TemporalScope = Field(
default=TemporalScope.CURRENT,
description="Temporal scope for context retrieval",
)
# === Budget ===
max_tokens: int | None = Field(
default=None,
ge=0,
description="Maximum token budget for this request",
)
# === Strategy hints ===
preferred_strategies: list[str] = Field(
default_factory=list,
description="Preferred context strategies to use",
)
required_backends: list[str] = Field(
default_factory=list,
description="Required data backends (text, vector, graph)",
)
# === Priority and purpose ===
priority: float = Field(
default=0.5,
ge=0.0,
le=1.0,
description="0.0 = background, 1.0 = critical",
)
purpose: str = Field(
default="",
description="Why is this context needed?",
)
@field_validator("depth")
@classmethod
def validate_depth(
cls: type[ContextRequest],
v: int | str,
) -> int | str:
"""Ensure integer depths are non-negative."""
if isinstance(v, int) and v < 0:
raise ValueError("depth must be non-negative when specified as integer")
return v
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
use_enum_values=False,
)
# ---------------------------------------------------------------------------
# AssembledContext
# ---------------------------------------------------------------------------
class AssembledContext(BaseModel):
"""The fused, budget-respecting context payload.
Output of a context assembly cycle -- the final payload delivered
to an actor after strategy execution, fragment deduplication,
scoring, budget packing, and ordering.
"""
fragments: list[ContextFragment] = Field(
default_factory=list,
description="Ordered context fragments",
)
total_tokens: int = Field(
...,
ge=0,
description="Total token count",
)
budget_used: float = Field(
...,
ge=0.0,
le=1.0,
description="Fraction of budget consumed (0.0-1.0)",
)
strategies_used: list[str] = Field(
default_factory=list,
description="Which strategies contributed",
)
context_hash: str = Field(
...,
min_length=1,
description="Cryptographic hash for snapshot",
)
preamble: str | None = Field(
default=None,
description="Optional structure summary",
)
provenance_map: dict[str, Any] = Field(
default_factory=dict,
description="Fragment -> resource/location mapping",
)
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
+9 -6
View File
@@ -1,25 +1,28 @@
"""Built-in skill tools for CleverAgents.
Provides file operation and search skill tools that integrate with the
skill framework and tool registry.
Provides file operation, search, and context skill tools that integrate
with the skill framework and tool registry.
## Modules
| Module | Description |
|-----------------|------------------------------------------|
| ``file_ops`` | ReadFile, WriteFile, EditFile, DeleteFile |
| ``search_ops`` | ListDir, Glob, Grep |
| Module | Description |
|-----------------|------------------------------------------------------|
| ``file_ops`` | ReadFile, WriteFile, EditFile, DeleteFile |
| ``search_ops`` | ListDir, Glob, Grep |
| ``context_ops`` | request_context, query_history, get_context_budget |
## Registration
```python
from cleveragents.skills.builtins.file_ops import register_skill_file_tools
from cleveragents.skills.builtins.search_ops import register_skill_search_tools
from cleveragents.skills.builtins.context_ops import register_skill_context_tools
from cleveragents.tool.registry import ToolRegistry
registry = ToolRegistry()
register_skill_file_tools(registry)
register_skill_search_tools(registry)
register_skill_context_tools(registry)
```
"""
@@ -0,0 +1,216 @@
"""Built-in context skill tools for CleverAgents (CRP).
Provides the ``builtin/context`` skill tools that actors use to issue
context requests during reasoning via the Context Request Protocol:
- **request_context**: Request specific context to be added to the
conversation. Accepts query, focus, breadth, depth, and purpose
parameters. Stubbed to raise ``NotImplementedError`` until the
ACMS pipeline is wired.
- **query_history**: Query historical context about past decisions
and changes. Stubbed to raise ``NotImplementedError``.
- **get_context_budget**: Check remaining context token budget.
Stubbed to raise ``NotImplementedError``.
## Registration
```python
from cleveragents.skills.builtins.context_ops import register_skill_context_tools
from cleveragents.tool.registry import ToolRegistry
registry = ToolRegistry()
register_skill_context_tools(registry)
```
Based on ``docs/specification.md`` ACMS > CRP > ``builtin/context`` Skill.
"""
from __future__ import annotations
from typing import Any
from cleveragents.domain.models.core.tool import ToolCapability
from cleveragents.tool.registry import ToolRegistry
from cleveragents.tool.runtime import ToolSpec
# ---------------------------------------------------------------------------
# Handlers (stubs -- wired to future ACMS pipeline)
# ---------------------------------------------------------------------------
def _handle_request_context(inputs: dict[str, Any]) -> dict[str, Any]:
"""Request specific context to be added to the conversation.
Stub implementation -- raises ``NotImplementedError`` until the
ACMS Context Assembly Pipeline is available.
"""
raise NotImplementedError(
"request_context is not yet wired to the ACMS pipeline. "
"This tool will be functional once the Context Assembly Pipeline "
"is implemented."
)
def _handle_query_history(inputs: dict[str, Any]) -> dict[str, Any]:
"""Query historical context about past decisions and changes.
Stub implementation -- raises ``NotImplementedError`` until the
ACMS temporal archaeology strategy is available.
"""
raise NotImplementedError(
"query_history is not yet wired to the ACMS pipeline. "
"This tool will be functional once the temporal archaeology "
"strategy is implemented."
)
def _handle_get_context_budget(inputs: dict[str, Any]) -> dict[str, Any]:
"""Check remaining context token budget.
Stub implementation -- raises ``NotImplementedError`` until the
ACMS budget tracking is available.
"""
raise NotImplementedError(
"get_context_budget is not yet wired to the ACMS pipeline. "
"This tool will be functional once context budget tracking "
"is implemented."
)
# ---------------------------------------------------------------------------
# Tool specs
# ---------------------------------------------------------------------------
SKILL_CONTEXT_REQUEST_SPEC = ToolSpec(
name="builtin/request-context",
description="Request specific context to be added to the conversation",
input_schema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "What information do you need?",
},
"focus": {
"type": "array",
"items": {"type": "string"},
"description": ("Specific files, classes, or functions to focus on"),
},
"breadth": {
"type": "integer",
"default": 2,
"description": "How many dependency hops outward (0-5)",
},
"depth": {
"oneOf": [
{"type": "integer", "minimum": 0},
{"type": "string"},
],
"default": 3,
"description": (
"Detail depth -- integer (0-N) or named level from "
"the active DetailLevelMap (e.g., 'SIGNATURES', "
"'FULL_SOURCE')"
),
},
"purpose": {
"type": "string",
"description": "Why do you need this context?",
},
},
"required": ["purpose"],
},
output_schema={
"type": "object",
"properties": {
"fragments": {
"type": "array",
"items": {"type": "object"},
"description": "Assembled context fragments",
},
"total_tokens": {
"type": "integer",
"description": "Total tokens consumed",
},
},
},
capabilities=ToolCapability(read_only=True),
handler=_handle_request_context,
)
SKILL_CONTEXT_QUERY_HISTORY_SPEC = ToolSpec(
name="builtin/query-history",
description="Query historical context about past decisions and changes",
input_schema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "What historical information do you need?",
},
"scope": {
"type": "string",
"enum": ["current_plan", "plan_tree", "all_plans"],
"default": "plan_tree",
"description": "Scope of history to search",
},
},
"required": ["query"],
},
output_schema={
"type": "object",
"properties": {
"entries": {
"type": "array",
"items": {"type": "object"},
"description": "Historical context entries",
},
},
},
capabilities=ToolCapability(read_only=True),
handler=_handle_query_history,
)
SKILL_CONTEXT_BUDGET_SPEC = ToolSpec(
name="builtin/get-context-budget",
description="Check remaining context token budget",
input_schema={
"type": "object",
"properties": {},
},
output_schema={
"type": "object",
"properties": {
"max_tokens": {
"type": "integer",
"description": "Maximum token budget",
},
"reserved_tokens": {
"type": "integer",
"description": "Tokens reserved for skeleton context",
},
"available_tokens": {
"type": "integer",
"description": "Tokens available for use",
},
"used_tokens": {
"type": "integer",
"description": "Tokens already consumed",
},
},
},
capabilities=ToolCapability(read_only=True),
handler=_handle_get_context_budget,
)
ALL_SKILL_CONTEXT_TOOLS: list[ToolSpec] = [
SKILL_CONTEXT_REQUEST_SPEC,
SKILL_CONTEXT_QUERY_HISTORY_SPEC,
SKILL_CONTEXT_BUDGET_SPEC,
]
def register_skill_context_tools(registry: ToolRegistry) -> None:
"""Register all skill-level context tools into *registry*."""
for spec in ALL_SKILL_CONTEXT_TOOLS:
registry.register(spec)