feat(acms): implement context policy configuration schema, YAML loader, and view-specific settings for ACMS v1
CI / benchmark-publish (pull_request) Has been skipped
CI / helm (pull_request) Successful in 26s
CI / push-validation (pull_request) Successful in 23s
CI / lint (pull_request) Successful in 3m52s
CI / typecheck (pull_request) Successful in 4m23s
CI / security (pull_request) Successful in 4m32s
CI / quality (pull_request) Successful in 4m46s
CI / build (pull_request) Successful in 3m43s
CI / integration_tests (pull_request) Failing after 5m5s
CI / e2e_tests (pull_request) Successful in 7m1s
CI / unit_tests (pull_request) Successful in 9m44s
CI / docker (pull_request) Failing after 46s
CI / coverage (pull_request) Failing after 11m1s
CI / status-check (pull_request) Failing after 1s
CI / benchmark-regression (pull_request) Successful in 1h12m32s

Rebase on current master and implement ContextPolicy Pydantic v2 model,
ContextPolicyLoader with YAML parsing, and CLI commands (list, show, validate).
Adds pyyaml>=6.0.0 dependency and 17 BDD unit test scenarios.

ISSUES CLOSED: #10028
This commit is contained in:
2026-04-22 22:54:23 +00:00
parent 7523a50db8
commit 432689f928
6 changed files with 1039 additions and 0 deletions
@@ -0,0 +1,172 @@
@acms @context_policy
Feature: ACMS v1 Context Policy Configuration Schema and YAML Loader
As a CleverAgents developer
I want to configure context policies via YAML files
So that actors can have view-specific context assembly settings
@context_policy_model
Scenario: Create a ContextPolicy with valid view_name
Given I create a ContextPolicy with view_name "strategy"
Then the ContextPolicy view_name should be "strategy"
And the ContextPolicy scope should be "project"
And the ContextPolicy strategies should be empty
And the ContextPolicy priority_patterns should be empty
@context_policy_model
Scenario: ContextPolicy rejects invalid view_name
When I try to create a ContextPolicy with view_name "invalid"
Then a context policy validation error should be raised
@context_policy_model
Scenario: ContextPolicy rejects non-positive max_file_size
When I try to create a ContextPolicy with max_file_size 0
Then a context policy validation error should be raised
@context_policy_model
Scenario: ContextPolicy accepts all valid view names
Given I create a ContextPolicy with view_name "strategy"
Then the ContextPolicy view_name should be "strategy"
Given I create a ContextPolicy with view_name "execution"
Then the ContextPolicy view_name should be "execution"
Given I create a ContextPolicy with view_name "estimation"
Then the ContextPolicy view_name should be "estimation"
@context_policy_model
Scenario: ContextPolicy with all fields set
Given I create a ContextPolicy with all fields:
| field | value |
| view_name | strategy |
| max_file_size | 102400 |
| max_total_size | 1048576 |
| scope | project |
Then the ContextPolicy max_file_size should be 102400
And the ContextPolicy max_total_size should be 1048576
And the ContextPolicy scope should be "project"
@context_policy_loader
Scenario: ContextPolicyLoader returns None when no project policy file exists
Given a temporary project directory with no policy file
When I load the project policy
Then the loaded policy should be None
@context_policy_loader
Scenario: ContextPolicyLoader loads a valid project policy file
Given a temporary project directory with a valid policy file:
"""
scope: project
views:
strategy:
max_file_size: 102400
strategies:
- relevance
"""
When I load the project policy
Then the loaded policy should not be None
And the loaded policy scope should be "project"
And the loaded policy should have view "strategy"
And the strategy view max_file_size should be 102400
@context_policy_loader
Scenario: ContextPolicyLoader raises FileNotFoundError for missing file
Given a temporary project directory with no policy file
When I try to load a policy from a non-existent path
Then a context policy FileNotFoundError should be raised
@context_policy_loader
Scenario: ContextPolicyLoader raises ValueError for invalid YAML
Given a temporary project directory with an invalid policy file:
"""
not: valid: yaml: [
"""
When I load the project policy
Then a context policy ValueError should be raised
@context_policy_loader
Scenario: ContextPolicyLoader raises ValueError for non-mapping YAML
Given a temporary project directory with a policy file containing:
"""
- item1
- item2
"""
When I load the project policy
Then a context policy ValueError should be raised
@context_policy_loader
Scenario: ContextPolicyLoader load_effective merges global and project policies
Given a temporary project directory with a valid policy file:
"""
scope: project
views:
strategy:
max_file_size: 102400
"""
And a global policy directory with a valid policy file:
"""
scope: global
views:
execution:
max_total_size: 524288
"""
When I load the effective policy
Then the effective policy should have view "strategy"
And the effective policy should have view "execution"
@context_policy_loader
Scenario: ContextPolicyLoader plan policy overrides project policy
Given a temporary project directory with a valid policy file:
"""
scope: project
views:
strategy:
max_file_size: 102400
"""
And a plan directory with a valid policy file:
"""
scope: plan
views:
strategy:
max_file_size: 51200
"""
When I load the effective policy with plan directory
Then the effective policy should have view "strategy"
And the strategy view max_file_size should be 51200
@context_policy_loader
Scenario: ContextPolicyLoader validate_file returns empty list for valid file
Given a temporary project directory with a valid policy file:
"""
scope: project
views:
strategy:
max_file_size: 102400
"""
When I validate the project policy file
Then the context policy validation errors should be empty
@context_policy_loader
Scenario: ContextPolicyLoader validate_file returns errors for invalid file
Given a temporary project directory with an invalid policy file:
"""
not: valid: yaml: [
"""
When I validate the project policy file
Then the context policy validation errors should not be empty
@context_policy_file
Scenario: ContextPolicyFile from_dict parses views correctly
Given a raw policy dict with scope "project" and strategy view
When I parse the dict into a ContextPolicyFile
Then the ContextPolicyFile scope should be "project"
And the ContextPolicyFile should have view "strategy"
@context_policy_file
Scenario: ContextPolicyFile from_dict raises ValueError for non-dict view
Given a raw policy dict with a non-dict view value
When I try to parse the dict into a ContextPolicyFile
Then a context policy ValueError should be raised
@context_policy_loader
Scenario: ContextPolicyLoader policy_path returns correct path
Given a temporary project directory with no policy file
When I get the policy path for the project root
Then the policy path should end with ".cleveragents/context-policy.yaml"
@@ -0,0 +1,374 @@
"""Step definitions for ACMS context policy configuration tests."""
from __future__ import annotations
import tempfile
from pathlib import Path
from typing import Any
from behave import given, then, when
from behave.runner import Context
from pydantic import ValidationError
from cleveragents.acms.context_policy import (
POLICY_DIR_NAME,
POLICY_FILE_NAME,
ContextPolicy,
ContextPolicyFile,
ContextPolicyLoader,
)
def _make_policy_file(base_dir: Path, content: str) -> Path:
"""Create a policy file in the given directory."""
policy_dir = base_dir / POLICY_DIR_NAME
policy_dir.mkdir(parents=True, exist_ok=True)
policy_path = policy_dir / POLICY_FILE_NAME
policy_path.write_text(content)
return policy_path
# ---------------------------------------------------------------------------
# ContextPolicy model steps
# ---------------------------------------------------------------------------
@given('I create a ContextPolicy with view_name "{view_name}"')
def step_create_policy_with_view_name(context: Context, view_name: str) -> None:
context.policy = ContextPolicy(view_name=view_name)
@given("I create a ContextPolicy with all fields:")
def step_create_policy_with_all_fields(context: Context) -> None:
fields: dict[str, Any] = {}
for row in context.table:
key = row["field"]
value = row["value"]
if key in ("max_file_size", "max_total_size"):
fields[key] = int(value)
else:
fields[key] = value
context.policy = ContextPolicy(**fields)
@when('I try to create a ContextPolicy with view_name "{view_name}"')
def step_try_create_policy_invalid_view(context: Context, view_name: str) -> None:
context.policy_error = None
try:
ContextPolicy(view_name=view_name)
except (ValueError, ValidationError) as exc:
context.policy_error = str(exc)
@when("I try to create a ContextPolicy with max_file_size {size:d}")
def step_try_create_policy_invalid_size(context: Context, size: int) -> None:
context.policy_error = None
try:
ContextPolicy(view_name="strategy", max_file_size=size)
except (ValueError, ValidationError) as exc:
context.policy_error = str(exc)
@then('the ContextPolicy view_name should be "{view_name}"')
def step_policy_view_name(context: Context, view_name: str) -> None:
assert context.policy.view_name == view_name
@then('the ContextPolicy scope should be "{scope}"')
def step_policy_scope(context: Context, scope: str) -> None:
assert context.policy.scope == scope
@then("the ContextPolicy strategies should be empty")
def step_policy_strategies_empty(context: Context) -> None:
assert context.policy.strategies == []
@then("the ContextPolicy priority_patterns should be empty")
def step_policy_priority_patterns_empty(context: Context) -> None:
assert context.policy.priority_patterns == []
@then("the ContextPolicy max_file_size should be {size:d}")
def step_policy_max_file_size(context: Context, size: int) -> None:
assert context.policy.max_file_size == size
@then("the ContextPolicy max_total_size should be {size:d}")
def step_policy_max_total_size(context: Context, size: int) -> None:
assert context.policy.max_total_size == size
@then("a ContextPolicy validation error should be raised")
def step_policy_validation_error(context: Context) -> None:
assert context.policy_error is not None, "Expected a validation error"
# ---------------------------------------------------------------------------
# ContextPolicyLoader steps
# ---------------------------------------------------------------------------
@given("a temporary project directory with no policy file")
def step_temp_dir_no_policy(context: Context) -> None:
context.tmp_dir = tempfile.mkdtemp()
context.project_root = Path(context.tmp_dir)
context.global_dir = Path(context.tmp_dir) / "global"
context.plan_dir = None
context.loader = ContextPolicyLoader(
project_root=context.project_root,
global_dir=context.global_dir,
)
context.policy_error = None
context.loaded_policy = None
@given("a temporary project directory with a valid policy file:")
def step_temp_dir_with_valid_policy(context: Context) -> None:
context.tmp_dir = tempfile.mkdtemp()
context.project_root = Path(context.tmp_dir)
context.global_dir = Path(context.tmp_dir) / "global"
context.plan_dir = None
_make_policy_file(context.project_root, context.text)
context.loader = ContextPolicyLoader(
project_root=context.project_root,
global_dir=context.global_dir,
)
context.policy_error = None
context.loaded_policy = None
@given("a temporary project directory with an invalid policy file:")
def step_temp_dir_with_invalid_policy(context: Context) -> None:
context.tmp_dir = tempfile.mkdtemp()
context.project_root = Path(context.tmp_dir)
context.global_dir = Path(context.tmp_dir) / "global"
context.plan_dir = None
_make_policy_file(context.project_root, context.text)
context.loader = ContextPolicyLoader(
project_root=context.project_root,
global_dir=context.global_dir,
)
context.policy_error = None
context.loaded_policy = None
@given("a temporary project directory with a policy file containing:")
def step_temp_dir_with_policy_containing(context: Context) -> None:
context.tmp_dir = tempfile.mkdtemp()
context.project_root = Path(context.tmp_dir)
context.global_dir = Path(context.tmp_dir) / "global"
context.plan_dir = None
_make_policy_file(context.project_root, context.text)
context.loader = ContextPolicyLoader(
project_root=context.project_root,
global_dir=context.global_dir,
)
context.policy_error = None
context.loaded_policy = None
@given("a global policy directory with a valid policy file:")
def step_global_dir_with_valid_policy(context: Context) -> None:
_make_policy_file(context.global_dir, context.text)
context.loader = ContextPolicyLoader(
project_root=context.project_root,
global_dir=context.global_dir,
)
@given("a plan directory with a valid policy file:")
def step_plan_dir_with_valid_policy(context: Context) -> None:
context.plan_dir = Path(context.tmp_dir) / "plan"
_make_policy_file(context.plan_dir, context.text)
context.loader = ContextPolicyLoader(
project_root=context.project_root,
plan_dir=context.plan_dir,
global_dir=context.global_dir,
)
@when("I load the project policy")
def step_load_project_policy(context: Context) -> None:
context.policy_error = None
context.loaded_policy = None
context.policy_error_type = None
try:
context.loaded_policy = context.loader.load_project()
except (ValueError, FileNotFoundError) as exc:
context.policy_error = str(exc)
context.policy_error_type = type(exc).__name__
@when("I try to load a policy from a non-existent path")
def step_load_nonexistent_path(context: Context) -> None:
context.policy_error = None
context.policy_error_type = None
try:
context.loader.load_from_path(
context.project_root / ".cleveragents" / "context-policy.yaml"
)
except FileNotFoundError as exc:
context.policy_error = str(exc)
context.policy_error_type = "FileNotFoundError"
@when("I load the effective policy")
def step_load_effective_policy(context: Context) -> None:
context.policy_error = None
context.loaded_policy = None
context.policy_error_type = None
try:
context.loaded_policy = context.loader.load_effective()
except (ValueError, FileNotFoundError) as exc:
context.policy_error = str(exc)
@when("I load the effective policy with plan directory")
def step_load_effective_with_plan(context: Context) -> None:
context.policy_error = None
context.loaded_policy = None
try:
context.loaded_policy = context.loader.load_effective()
except (ValueError, FileNotFoundError) as exc:
context.policy_error = str(exc)
@when("I validate the project policy file")
def step_validate_project_policy(context: Context) -> None:
path = context.loader.policy_path(context.project_root)
context.validation_errors = context.loader.validate_file(path)
@when("I get the policy path for the project root")
def step_get_policy_path(context: Context) -> None:
context.policy_path = context.loader.policy_path(context.project_root)
@when("I parse the dict into a ContextPolicyFile")
def step_parse_dict_to_policy_file(context: Context) -> None:
context.policy_error = None
context.policy_file = None
try:
context.policy_file = ContextPolicyFile.from_dict(context.raw_dict)
except (ValueError, TypeError) as exc:
context.policy_error = str(exc)
@when("I try to parse the dict into a ContextPolicyFile")
def step_try_parse_dict(context: Context) -> None:
context.policy_error = None
context.policy_file = None
try:
context.policy_file = ContextPolicyFile.from_dict(context.raw_dict)
except (ValueError, TypeError) as exc:
context.policy_error = str(exc)
@given('a raw policy dict with scope "project" and strategy view')
def step_raw_dict_with_strategy(context: Context) -> None:
context.raw_dict = {
"scope": "project",
"views": {
"strategy": {
"max_file_size": 102400,
"strategies": ["relevance"],
}
},
}
@given("a raw policy dict with a non-dict view value")
def step_raw_dict_with_non_dict_view(context: Context) -> None:
context.raw_dict = {
"scope": "project",
"views": {"strategy": "not-a-dict"},
}
# ---------------------------------------------------------------------------
# Then steps
# ---------------------------------------------------------------------------
@then("the loaded policy should be None")
def step_loaded_policy_is_none(context: Context) -> None:
assert context.loaded_policy is None
@then("the loaded policy should not be None")
def step_loaded_policy_not_none(context: Context) -> None:
assert context.loaded_policy is not None
@then('the loaded policy scope should be "{scope}"')
def step_loaded_policy_scope(context: Context, scope: str) -> None:
assert context.loaded_policy is not None
assert context.loaded_policy.scope == scope
@then('the loaded policy should have view "{view_name}"')
def step_loaded_policy_has_view(context: Context, view_name: str) -> None:
assert context.loaded_policy is not None
assert view_name in context.loaded_policy.views
@then("the strategy view max_file_size should be {size:d}")
def step_strategy_view_max_file_size(context: Context, size: int) -> None:
assert context.loaded_policy is not None
strategy_view = context.loaded_policy.views.get("strategy")
assert strategy_view is not None
assert strategy_view.max_file_size == size
@then("a context policy FileNotFoundError should be raised")
def step_file_not_found_error(context: Context) -> None:
assert context.policy_error_type == "FileNotFoundError", (
f"Expected FileNotFoundError, got {context.policy_error_type}: {context.policy_error}"
)
@then("a context policy ValueError should be raised")
def step_value_error_raised(context: Context) -> None:
assert context.policy_error is not None, "Expected a ValueError but none was raised"
@then("the context policy validation errors should be empty")
def step_validation_errors_empty(context: Context) -> None:
assert context.validation_errors == []
@then("the context policy validation errors should not be empty")
def step_validation_errors_not_empty(context: Context) -> None:
assert len(context.validation_errors) > 0
@then('the policy path should end with ".cleveragents/context-policy.yaml"')
def step_policy_path_ends_with(context: Context) -> None:
path_str = str(context.policy_path)
assert path_str.endswith(".cleveragents/context-policy.yaml"), (
f"Expected path to end with .cleveragents/context-policy.yaml, got {path_str}"
)
@then('the effective policy should have view "{view_name}"')
def step_effective_policy_has_view(context: Context, view_name: str) -> None:
assert context.loaded_policy is not None
assert view_name in context.loaded_policy.views
@then('the ContextPolicyFile scope should be "{scope}"')
def step_policy_file_scope(context: Context, scope: str) -> None:
assert context.policy_file is not None
assert context.policy_file.scope == scope
@then('the ContextPolicyFile should have view "{view_name}"')
def step_policy_file_has_view(context: Context, view_name: str) -> None:
assert context.policy_file is not None
assert view_name in context.policy_file.views
@then("a context policy validation error should be raised")
def step_context_policy_validation_error(context: Context) -> None:
assert context.policy_error is not None, "Expected a validation error"
+1
View File
@@ -31,6 +31,7 @@ dependencies = [
"rx>=3.2.0", # Reactive streams for routing
"dependency-injector>=4.41.0", # DI container
"pydantic>=2.7.0",
"pyyaml>=6.0.0",
"pydantic-settings>=2.11.0",
"structlog>=24.4.0",
"langchain>=0.2.14",
+292
View File
@@ -0,0 +1,292 @@
"""ACMS v1 context policy configuration schema, YAML loader, and view-specific settings.
Provides:
- ``ContextPolicy``: Pydantic v2 model for a single named context policy view
with fields ``view_name``, ``max_file_size``, ``max_total_size``,
``strategies``, ``scope``, and ``priority_patterns``.
- ``ContextPolicyLoader``: reads and validates ``.cleveragents/context-policy.yaml``
files from a project root, supporting global/project/plan scopes and
policy inheritance (plan > project > global).
YAML schema example (``.cleveragents/context-policy.yaml``)::
scope: project
views:
strategy:
max_file_size: 102400
max_total_size: 1048576
strategies:
- relevance
- recency
priority_patterns:
- "src/**/*.py"
- "docs/**"
execution:
max_file_size: 51200
strategies:
- tiered
estimation:
max_total_size: 524288
Policy inheritance:
plan-level policy overrides project-level, project overrides global defaults.
``ContextPolicyLoader.load_effective`` merges policies in that order.
Based on ``docs/specification.md`` ACMS Context Policy section and issue #10028.
"""
from __future__ import annotations
from pathlib import Path
from typing import Any, Literal
import yaml
from pydantic import BaseModel, ConfigDict, Field, field_validator
# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------
#: Valid actor-type view names for ACMS v1.
VALID_VIEW_NAMES: frozenset[str] = frozenset({"strategy", "execution", "estimation"})
#: Valid policy scope values.
VALID_SCOPES: frozenset[str] = frozenset({"global", "project", "plan"})
#: Default YAML file name within the ``.cleveragents/`` directory.
POLICY_FILE_NAME: str = "context-policy.yaml"
#: Default directory name within the project root.
POLICY_DIR_NAME: str = ".cleveragents"
# ---------------------------------------------------------------------------
# ContextPolicy model
# ---------------------------------------------------------------------------
class ContextPolicy(BaseModel):
"""A single named context policy view for an ACMS actor type.
Each ``ContextPolicy`` instance describes the context assembly
constraints for one actor type (strategy, execution, or estimation).
Multiple ``ContextPolicy`` objects are combined into a
``ContextPolicyFile`` that is persisted as YAML.
Fields:
view_name: Actor type this policy applies to. One of
``"strategy"``, ``"execution"``, or ``"estimation"``.
max_file_size: Maximum size in bytes for a single context file.
``None`` means no per-file limit.
max_total_size: Maximum cumulative size in bytes for all context
files in this view. ``None`` means no total limit.
strategies: Ordered list of context assembly strategy names to
apply (e.g. ``["relevance", "recency"]``). An empty list
means "use the pipeline default".
scope: Policy scope -- ``"global"``, ``"project"``, or ``"plan"``.
Determines the inheritance level at which this policy applies.
priority_patterns: Glob patterns for files that should be
prioritised during context assembly (e.g. ``["src/**/*.py"]``).
An empty list means no explicit prioritisation.
"""
view_name: str = Field(
...,
description=(
"Actor type this policy applies to: strategy, execution, or estimation"
),
)
max_file_size: int | None = Field(
default=None,
description="Max size in bytes for a single context file (None = no limit)",
)
max_total_size: int | None = Field(
default=None,
description="Max cumulative context size in bytes (None = no limit)",
)
strategies: list[str] = Field(
default_factory=list,
description="Ordered list of context assembly strategy names",
)
scope: Literal["global", "project", "plan"] = Field(
default="project",
description="Policy scope: global, project, or plan",
)
priority_patterns: list[str] = Field(
default_factory=list,
description="Glob patterns for files to prioritise during context assembly",
)
@field_validator("view_name")
@classmethod
def _validate_view_name(cls: type[ContextPolicy], v: str) -> str:
"""Ensure view_name is one of the valid actor types."""
if v not in VALID_VIEW_NAMES:
raise ValueError(
f"Invalid view_name '{v}': must be one of {sorted(VALID_VIEW_NAMES)}"
)
return v
@field_validator("max_file_size", "max_total_size")
@classmethod
def _validate_positive_size(
cls: type[ContextPolicy],
v: int | None,
) -> int | None:
"""Size limits must be positive when set."""
if v is not None and v <= 0:
raise ValueError("Size limit must be a positive integer or None")
return v
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
# ---------------------------------------------------------------------------
# ContextPolicyFile model (the full YAML document)
# ---------------------------------------------------------------------------
class ContextPolicyFile(BaseModel):
"""Represents the full ``.cleveragents/context-policy.yaml`` document."""
scope: Literal["global", "project", "plan"] = Field(
default="project",
description="Scope at which this policy file applies",
)
views: dict[str, ContextPolicy] = Field(
default_factory=dict,
description="Mapping from view name to ContextPolicy",
)
model_config = ConfigDict(
str_strip_whitespace=True,
validate_assignment=True,
)
@classmethod
def from_dict(
cls: type[ContextPolicyFile],
data: dict[str, Any],
) -> ContextPolicyFile:
"""Parse a raw dictionary (e.g. from YAML) into a ``ContextPolicyFile``."""
scope = data.get("scope", "project")
raw_views: dict[str, Any] = data.get("views", {})
views: dict[str, ContextPolicy] = {}
for view_name, view_data in raw_views.items():
if not isinstance(view_data, dict):
raise ValueError(
f"View '{view_name}' must be a mapping, "
f"got {type(view_data).__name__}"
)
policy_data = dict(view_data)
policy_data["view_name"] = view_name
policy_data.setdefault("scope", scope)
views[view_name] = ContextPolicy(**policy_data)
return cls(scope=scope, views=views)
# ---------------------------------------------------------------------------
# ContextPolicyLoader
# ---------------------------------------------------------------------------
class ContextPolicyLoader:
"""Reads and validates ``.cleveragents/context-policy.yaml`` files."""
def __init__(
self,
project_root: Path | None = None,
plan_dir: Path | None = None,
global_dir: Path | None = None,
) -> None:
"""Initialise the loader with optional directory overrides."""
self._project_root: Path = project_root or Path.cwd()
self._plan_dir: Path | None = plan_dir
self._global_dir: Path = global_dir or (Path.home() / POLICY_DIR_NAME)
def policy_path(self, base_dir: Path) -> Path:
"""Return the expected policy file path within *base_dir*."""
return base_dir / POLICY_DIR_NAME / POLICY_FILE_NAME
def load_from_path(self, path: Path) -> ContextPolicyFile:
"""Load and validate a policy file from an explicit path."""
if not path.exists():
raise FileNotFoundError(f"Context policy file not found: {path}")
raw_text = path.read_text(encoding="utf-8")
return self._parse_yaml(raw_text, source=str(path))
def load_global(self) -> ContextPolicyFile | None:
"""Load the global policy file, or ``None`` if it does not exist."""
path = self.policy_path(self._global_dir)
if not path.exists():
return None
return self.load_from_path(path)
def load_project(self) -> ContextPolicyFile | None:
"""Load the project-level policy file, or ``None`` if absent."""
path = self.policy_path(self._project_root)
if not path.exists():
return None
return self.load_from_path(path)
def load_plan(self) -> ContextPolicyFile | None:
"""Load the plan-level policy file, or ``None`` if absent."""
if self._plan_dir is None:
return None
path = self.policy_path(self._plan_dir)
if not path.exists():
return None
return self.load_from_path(path)
def load_effective(self) -> ContextPolicyFile:
"""Merge global, project, and plan policies into an effective policy."""
global_file = self.load_global()
project_file = self.load_project()
plan_file = self.load_plan()
merged_views: dict[str, ContextPolicy] = {}
effective_scope: Literal["global", "project", "plan"] = "global"
for policy_file in (global_file, project_file, plan_file):
if policy_file is None:
continue
merged_views.update(policy_file.views)
effective_scope = policy_file.scope # type: ignore[assignment]
return ContextPolicyFile(scope=effective_scope, views=merged_views)
def validate_file(self, path: Path) -> list[str]:
"""Validate a policy file and return a list of error messages."""
errors: list[str] = []
if not path.exists():
errors.append(f"File not found: {path}")
return errors
try:
self.load_from_path(path)
except (ValueError, yaml.YAMLError) as exc:
errors.append(str(exc))
return errors
def _parse_yaml(self, text: str, source: str = "<string>") -> ContextPolicyFile:
"""Parse YAML text into a ``ContextPolicyFile``."""
try:
data = yaml.safe_load(text)
except yaml.YAMLError as exc:
raise ValueError(f"Invalid YAML in {source}: {exc}") from exc
if data is None:
return ContextPolicyFile()
if not isinstance(data, dict):
raise ValueError(
f"Context policy file {source} must be a YAML mapping, "
f"got {type(data).__name__}"
)
try:
return ContextPolicyFile.from_dict(data)
except (ValueError, TypeError) as exc:
raise ValueError(f"Invalid context policy in {source}: {exc}") from exc
@@ -0,0 +1,193 @@
"""CLI commands for ``agents context policy``.
Implements ``agents context policy list``, ``show``, and ``validate``
for managing YAML-based ACMS context policy files.
"""
from __future__ import annotations
from pathlib import Path
from typing import Annotated, Any
import typer
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from cleveragents.acms.context_policy import (
VALID_VIEW_NAMES,
ContextPolicyLoader,
)
from cleveragents.cli.formatting import OutputFormat, format_output
app = typer.Typer(help="Manage ACMS context policy files")
console = Console()
err_console = Console(stderr=True)
_FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)"
@app.command(name="list")
def policy_list(
project_root: Annotated[
Path,
typer.Option("--root", "-r", help="Project root directory"),
] = Path("."),
output_format: Annotated[
str,
typer.Option("--format", "-f", help=_FORMAT_HELP),
] = "rich",
) -> None:
"""List context policy views defined in the project policy file."""
loader = ContextPolicyLoader(project_root=project_root.resolve())
policy_file = loader.load_project()
if policy_file is None:
policy_path = loader.policy_path(project_root.resolve())
if output_format.lower() == OutputFormat.RICH:
console.print(f"[yellow]No policy file found at {policy_path}[/yellow]")
else:
console.print(format_output({"views": [], "scope": None}, output_format))
return
views_data: list[dict[str, Any]] = []
for view_name, policy in policy_file.views.items():
views_data.append(
{
"view_name": view_name,
"scope": policy.scope,
"max_file_size": policy.max_file_size,
"max_total_size": policy.max_total_size,
"strategies": policy.strategies,
"priority_patterns": policy.priority_patterns,
}
)
if output_format.lower() == OutputFormat.RICH:
table = Table(title="Context Policy Views", expand=False)
table.add_column("View", style="bold")
table.add_column("Scope")
table.add_column("Max File Size")
table.add_column("Max Total Size")
table.add_column("Strategies")
for v in views_data:
table.add_row(
v["view_name"],
v["scope"],
str(v["max_file_size"]) if v["max_file_size"] else "(no limit)",
str(v["max_total_size"]) if v["max_total_size"] else "(no limit)",
", ".join(v["strategies"]) if v["strategies"] else "(default)",
)
console.print(table)
else:
console.print(
format_output(
{"views": views_data, "scope": policy_file.scope}, output_format
)
)
@app.command(name="show")
def policy_show(
view_name: Annotated[
str,
typer.Argument(help="View name: strategy, execution, or estimation"),
],
project_root: Annotated[
Path,
typer.Option("--root", "-r", help="Project root directory"),
] = Path("."),
output_format: Annotated[
str,
typer.Option("--format", "-f", help=_FORMAT_HELP),
] = "rich",
) -> None:
"""Show details of a specific context policy view."""
if view_name not in VALID_VIEW_NAMES:
err_console.print(
f"[red]Invalid view name '{view_name}': "
f"must be one of {sorted(VALID_VIEW_NAMES)}[/red]"
)
raise typer.Exit(1)
loader = ContextPolicyLoader(project_root=project_root.resolve())
effective = loader.load_effective()
policy = effective.views.get(view_name)
if policy is None:
if output_format.lower() == OutputFormat.RICH:
console.print(f"[yellow]No policy defined for view '{view_name}'[/yellow]")
else:
console.print(
format_output({"view_name": view_name, "policy": None}, output_format)
)
return
data: dict[str, Any] = {
"view_name": policy.view_name,
"scope": policy.scope,
"max_file_size": policy.max_file_size,
"max_total_size": policy.max_total_size,
"strategies": policy.strategies,
"priority_patterns": policy.priority_patterns,
}
if output_format.lower() == OutputFormat.RICH:
lines = [
f"[bold]View:[/bold] {policy.view_name}",
f"[bold]Scope:[/bold] {policy.scope}",
f"[bold]Max file size:[/bold] {policy.max_file_size or '(no limit)'}",
f"[bold]Max total size:[/bold] {policy.max_total_size or '(no limit)'}",
f"[bold]Strategies:[/bold] {', '.join(policy.strategies) or '(default)'}",
(
f"[bold]Priority patterns:[/bold] "
f"{', '.join(policy.priority_patterns) or '(none)'}"
),
]
console.print(
Panel("\n".join(lines), title=f"Context Policy: {view_name}", expand=False)
)
else:
console.print(format_output(data, output_format))
@app.command(name="validate")
def policy_validate(
policy_file: Annotated[
Path | None,
typer.Argument(help="Path to policy file (default: project policy)"),
] = None,
project_root: Annotated[
Path,
typer.Option("--root", "-r", help="Project root directory"),
] = Path("."),
output_format: Annotated[
str,
typer.Option("--format", "-f", help=_FORMAT_HELP),
] = "rich",
) -> None:
"""Validate a context policy YAML file."""
loader = ContextPolicyLoader(project_root=project_root.resolve())
if policy_file is None:
path = loader.policy_path(project_root.resolve())
else:
path = policy_file.resolve()
errors = loader.validate_file(path)
if errors:
data_err: dict[str, Any] = {"valid": False, "errors": errors, "path": str(path)}
if output_format.lower() == OutputFormat.RICH:
err_lines = "\n".join(f" - {e}" for e in errors)
console.print(
Panel(
f"[red]Validation failed:[/red]\n{err_lines}",
title=f"Policy Validation: {path.name}",
expand=False,
)
)
else:
console.print(format_output(data_err, output_format))
raise typer.Exit(1)
else:
data_ok: dict[str, Any] = {"valid": True, "errors": [], "path": str(path)}
if output_format.lower() == OutputFormat.RICH:
console.print(
Panel(
"[green]Policy file is valid.[/green]",
title=f"Policy Validation: {path.name}",
expand=False,
)
)
else:
console.print(format_output(data_ok, output_format))
+7
View File
@@ -99,6 +99,7 @@ def _register_subcommands() -> None:
validation,
)
from cleveragents.cli.commands.auto_debug import app as auto_debug_app
from cleveragents.cli.commands.context_policy import app as context_policy_app
from cleveragents.cli.commands.db import app as db_app
from cleveragents.cli.commands.repl import _repl_app
from cleveragents.cli.commands.server import app as server_app
@@ -223,6 +224,11 @@ def _register_subcommands() -> None:
name="server",
help="Server connection management (stub)",
)
app.add_typer(
context_policy_app,
name="context-policy",
help="Manage ACMS context policy files",
)
app.add_typer(
repo.app,
name="repo",
@@ -744,6 +750,7 @@ def main(args: list[str] | None = None) -> int:
"repl", # Interactive REPL
"tui", # Textual TUI
"server", # Server connection management
"context-policy", # ACMS context policy management
"repo", # Repository indexing management
"tell", # Shortcut for plan tell
"build", # Shortcut for plan build