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

feat(cli): wire project context CLI stubs to ACMS pipeline #534

Merged
CoreRasurae merged 1 commits from feature/m5-context-cli-wiring into master 2026-03-03 23:41:23 +00:00
Conversation 2 Commits 1 Files Changed 13
+2082 -106
13 changed files with 2082 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+17
View File
@@ -2,6 +2,23 @@
## Unreleased
- Fixed `context inspect` to display project-scoped tier fragment counts instead of global
counts. Added `ContextTierService.get_scoped_metrics()` which returns fragment population
counts filtered to the target project while keeping hit/miss counters as global service
metrics. (#499)
- Fixed `context simulate --focus` to filter fragments by the specified focus URIs during
dry-run assembly. Previously, focus URIs were passed to `ContextRequest` but not applied
to fragment selection, making `--focus` a no-op. (#499)
- Wired project context CLI stubs (`inspect`, `simulate`, `set`, `show`) to live ACMS pipeline
services. `context inspect` queries `ContextTierService` for tier metrics and per-project
fragments with optional filtering by strategy, focus area, breadth, and depth. `context
simulate` performs dry-run context assembly using CRP models with configurable token budget
and assembly strategies. `context set` gains 12 ACMS pipeline options (`hot_max_tokens`,
`warm_max_decisions`, `cold_max_decisions`, `summary_max_tokens`, `temporal_scope`,
`auto_refresh`, `focus_area`, `breadth`, `depth`, `assembly_strategy`, `retrieval_strategy`,
`summary_strategy`). `context show` displays ACMS pipeline configuration alongside context
policy. Includes 28 Behave BDD scenarios for wiring coverage, updated Robot Framework
integration tests, and reference documentation. (#499)
- Added hot/warm/cold context tiers with `ContextTier`, `ActorRole`, `TieredFragment`,
`TierBudget`, `ActorContextView`, `TierMetrics`, and `ScopedBackendView` models.
`ContextTierService` provides store/get, promotion/demotion with cold-tier summarisation
docs/reference/project_context_cli.md
+154 -15
View File
@@ -1,7 +1,8 @@
# Project Context CLI Reference
The `agents project context` commands manage per-project context policies
that control what resources and files are visible during each ACMS phase.
that control what resources and files are visible during each ACMS phase,
and provide tools to inspect and simulate the ACMS pipeline state.
## Overview
@@ -20,7 +21,7 @@ An empty policy defaults to including everything.
## `agents project context set`
Persist a context policy view for a project.
Persist a context policy view and ACMS pipeline configuration for a project.
### Usage
@@ -36,6 +37,8 @@ agents project context set PROJECT [OPTIONS]
### Options
#### View Options
| Option | Type | Description |
|----------------------|-----------|------------------------------------------|
| `--view`, `-v` | `str` | Phase view: default, strategize, execute, apply (default: default) |
@@ -46,7 +49,29 @@ agents project context set PROJECT [OPTIONS]
| `--max-file-size` | `int` | Max file size in bytes (None = no limit) |
| `--max-total-size` | `int` | Max total context size in bytes |
| `--clear` | `flag` | Clear (reset) the view for the phase |
| `--format`, `-f` | `str` | Output format (json, yaml, plain, table, rich) |
#### ACMS Pipeline Options
| Option | Type | Default | Description |
|-------------------------|---------|----------|------------------------------------------------|
| `--hot-max-tokens` | `int` | 8000 | Max tokens for the hot tier context window |
| `--warm-max-decisions` | `int` | 500 | Max decisions for the warm tier |
| `--cold-max-decisions` | `int` | 5000 | Max decisions for the cold tier |
| `--query-limit` | `int` | None | Max number of fragments per query |
| `--summarize/--no-summarize` | `bool` | True | Enable or disable context summarization |
| `--summary-max-tokens` | `int` | None | Max tokens for context summaries |
| `--strategy` | `str` | (none) | Preferred context strategy (repeatable) |
| `--default-breadth` | `int` | 2 | Default breadth for context retrieval |
| `--default-depth` | `str` | 3 | Default depth (integer or named level) |
| `--skeleton-ratio` | `float` | 0.2 | Skeleton compression ratio (0.0-1.0) |
| `--temporal-scope` | `str` | current | Temporal scope: current, recent, or all |
| `--auto-refresh/--no-auto-refresh` | `bool` | True | Auto-refresh context on resource changes |
#### Output Options
| Option | Type | Description |
|-----------------|-------|------------------------------------------|
| `--format`, `-f`| `str` | Output format (json, yaml, plain, table, rich) |
### Examples
@@ -70,13 +95,31 @@ agents project context set local/my-app \
# Clear execute view (inherit from strategize)
agents project context set local/my-app --view execute --clear
# Configure ACMS pipeline options
agents project context set local/my-app \
--hot-max-tokens 4000 \
--warm-max-decisions 200 \
--temporal-scope recent \
--no-auto-refresh
# Set preferred strategies
agents project context set local/my-app \
--strategy breadth_first \
--strategy depth_first
# Adjust skeleton compression
agents project context set local/my-app \
--skeleton-ratio 0.5 \
--default-breadth 5 \
--default-depth 7
```
---
## `agents project context show`
Display the context policy for a project.
Display the context policy and ACMS pipeline configuration for a project.
### Usage
@@ -97,16 +140,24 @@ agents project context show PROJECT [OPTIONS]
| `--view`, `-v` | `str` | Show resolved view for a specific phase |
| `--format`, `-f`| `str` | Output format (json, yaml, plain, table, rich) |
### Output
The `show` command displays:
- **View configuration** for each phase (or the resolved view for a specific phase)
- **ACMS Pipeline Config** including hot/warm/cold tier budgets, summarization settings,
strategies, temporal scope, and auto-refresh status
### Examples
```bash
# Show all views
# Show all views and ACMS config
agents project context show local/my-app
# Show resolved view for execute phase
agents project context show local/my-app --view execute
# JSON output
# JSON output (includes acms_config key)
agents project context show local/my-app --format json
```
@@ -114,21 +165,109 @@ agents project context show local/my-app --format json
## `agents project context inspect`
> **Stub** — requires ACMS wiring (not yet implemented).
Inspect the effective context state for a project by querying the ACMS
tier service. Displays current fragment distribution across tiers, tier
metrics, budget utilisation, and per-actor visibility.
Inspect the effective context for a project. This command will show
which resources and files would be included/excluded for each phase
based on the current policy and resource registry state.
### Usage
Raises `NotImplementedError` with a descriptive message.
```
agents project context inspect PROJECT [OPTIONS]
```
### Arguments
| Argument | Description |
|-----------|-------------------------|
| `PROJECT` | Project namespaced name |
### Options
| Option | Type | Description |
|-------------------|-------|------------------------------------------------|
| `--view`, `-v` | `str` | Phase view to inspect |
| `--strategy` | `str` | Filter fragments by strategy name |
| `--focus` | `str` | UKO URIs to focus on (repeatable) |
| `--breadth` | `int` | Override breadth for context retrieval |
| `--depth` | `str` | Override depth (integer or named level) |
| `--format`, `-f` | `str` | Output format (json, yaml, plain, table, rich) |
### Output
The `inspect` command displays:
- **Tier Metrics** — project-scoped hot/warm/cold fragment counts, service-level hit/miss rates
- **Tier Budget** — configured max tokens (hot) and max decisions (warm, cold)
- **Project Fragments** — total count and breakdown by tier for the project
- **Actor Visibility** — how many fragments each actor role (strategist, executor, reviewer) can see
- **Fragment Details** — first 20 fragments with IDs, tiers, token counts, and access metadata
- **ACMS Config** — the project's pipeline configuration
### Examples
```bash
# Inspect all tiers for a project
agents project context inspect local/my-app
# Inspect with JSON output
agents project context inspect local/my-app --format json
# Inspect a specific phase view
agents project context inspect local/my-app --view strategize
```
---
## `agents project context simulate`
> **Stub** — requires ACMS wiring (not yet implemented).
Run a dry-run context window assembly for a project. Creates a simulated
`AssembledContext` using the project's ACMS configuration and available
tier fragments.
Simulate the context window for a project. This command will show
the estimated token counts and context composition for each phase.
### Usage
Raises `NotImplementedError` with a descriptive message.
```
agents project context simulate PROJECT [OPTIONS]
```
### Arguments
| Argument | Description |
|-----------|-------------------------|
| `PROJECT` | Project namespaced name |
### Options
| Option | Type | Description |
|-------------------|-------|------------------------------------------------|
| `--view`, `-v` | `str` | Phase view for simulation |
| `--budget` | `int` | Token budget override |
| `--focus` | `str` | UKO URIs to focus on (repeatable) |
| `--strategy` | `str` | Preferred strategies (repeatable) |
| `--format`, `-f` | `str` | Output format (json, yaml, plain, table, rich) |
### Output
The `simulate` command displays:
- **Summary** — total tokens, budget usage percentage, strategies used, fragment count, context hash
- **Fragment Table** — each assembled fragment with UKO node, token count, relevance score, depth, and strategy. When `--focus` is provided, only fragments matching the focus URIs are included.
- **ACMS Config** — the project's pipeline configuration used for the simulation
### Examples
```bash
# Simulate with default settings
agents project context simulate local/my-app
# Simulate with custom budget
agents project context simulate local/my-app --budget 4000
# Simulate for a specific phase with strategy hints
agents project context simulate local/my-app \
--view strategize \
--strategy breadth_first
# JSON output for programmatic consumption
agents project context simulate local/my-app --format json
```
features/context_cli_wiring.feature
+175
View File
@@ -0,0 +1,175 @@
Feature: Context CLI ACMS pipeline wiring (B2.cli)
As a CleverAgents user
I want context inspect and simulate commands wired to the ACMS pipeline
So that I can examine tier state and dry-run context assembly
Background:
Given a context wiring in-memory database is initialized
And a project "local/wire-app" exists for context wiring
# ── context set with ACMS options ──────────────────────────
Scenario: Set ACMS hot_max_tokens option
When I run wired context set on "local/wire-app" with hot-max-tokens 4000
Then the wired context set command should succeed
And the stored ACMS config hot_max_tokens should be 4000
Scenario: Set ACMS warm_max_decisions option
When I run wired context set on "local/wire-app" with warm-max-decisions 200
Then the wired context set command should succeed
And the stored ACMS config warm_max_decisions should be 200
Scenario: Set ACMS cold_max_decisions option
When I run wired context set on "local/wire-app" with cold-max-decisions 3000
Then the wired context set command should succeed
And the stored ACMS config cold_max_decisions should be 3000
Scenario: Set ACMS temporal_scope option
When I run wired context set on "local/wire-app" with temporal-scope "recent"
Then the wired context set command should succeed
And the stored ACMS config temporal_scope should be "recent"
Scenario: Set invalid temporal_scope fails
When I run wired context set on "local/wire-app" with temporal-scope "invalid"
Then the wired context set command should fail
Scenario: Set ACMS auto-refresh off
When I run wired context set on "local/wire-app" with no-auto-refresh
Then the wired context set command should succeed
And the stored ACMS config auto_refresh should be false
Scenario: Set ACMS summarize off
When I run wired context set on "local/wire-app" with no-summarize
Then the wired context set command should succeed
And the stored ACMS config summarize should be false
Scenario: Set ACMS strategy option
When I run wired context set on "local/wire-app" with strategy "breadth_first"
Then the wired context set command should succeed
And the stored ACMS config strategies should contain "breadth_first"
Scenario: Set ACMS skeleton_ratio option
When I run wired context set on "local/wire-app" with skeleton-ratio 0.5
Then the wired context set command should succeed
And the stored ACMS config skeleton_ratio should be 0.5
Scenario: Set ACMS default_breadth option
When I run wired context set on "local/wire-app" with default-breadth 5
Then the wired context set command should succeed
And the stored ACMS config default_breadth should be 5
Scenario: Set ACMS default_depth option
When I run wired context set on "local/wire-app" with default-depth 7
Then the wired context set command should succeed
And the stored ACMS config default_depth should be 7
Scenario: Set ACMS query_limit option
When I run wired context set on "local/wire-app" with query-limit 50
Then the wired context set command should succeed
And the stored ACMS config query_limit should be 50
Scenario: Set ACMS summary_max_tokens option
When I run wired context set on "local/wire-app" with summary-max-tokens 2000
Then the wired context set command should succeed
And the stored ACMS config summary_max_tokens should be 2000
# ── context show with ACMS config ──────────────────────────
Scenario: Show displays ACMS pipeline config
Given I have set ACMS config on "local/wire-app" with hot-max-tokens 6000
When I run wired context show on "local/wire-app" with format "json"
Then the wired context show command should succeed
And the wired output should be valid JSON
And the wired JSON output should contain key "acms_config"
Scenario: Show with view displays ACMS config
Given I have set ACMS config on "local/wire-app" with hot-max-tokens 6000
When I run wired context show on "local/wire-app" with view "default" and format "json"
Then the wired context show command should succeed
And the wired output should be valid JSON
And the wired JSON output should contain key "acms_config"
# ── context inspect wired to tier service ──────────────────
Scenario: Inspect returns tier metrics with no fragments
When I run wired inspect for project "local/wire-app" using defaults
Then the wired context inspect command should succeed
Scenario: Inspect with JSON format returns valid JSON
When I run wired inspect for project "local/wire-app" using format "json"
Then the wired context inspect command should succeed
And the wired output should be valid JSON
And the wired JSON output should contain key "tier_metrics"
And the wired JSON output should contain key "project_fragments"
And the wired JSON output should contain key "actor_visibility"
Scenario: Inspect with pre-populated fragments shows counts
Given the tier service has a hot fragment "frag-1" for project "local/wire-app"
And the tier service has a warm fragment "frag-2" for project "local/wire-app"
When I run wired inspect for project "local/wire-app" using format "json"
Then the wired context inspect command should succeed
And the wired JSON project_fragments total should be 2
Scenario: Inspect shows project-scoped tier metrics not global counts
Given the tier service has a hot fragment "scoped-a" for project "local/wire-app"
And the tier service has a hot fragment "other-b" for project "local/other-proj"
When I run wired inspect for project "local/wire-app" using format "json"
Then the wired context inspect command should succeed
And the wired JSON tier_metrics hot_count should be 1
Scenario: Inspect on nonexistent project fails
When I run wired inspect for project "local/no-such" using defaults
Then the wired context inspect command should fail
Scenario: Inspect with invalid view fails
When I run wired inspect for project "local/wire-app" using view "bogus"
Then the wired context inspect command should fail
# ── context simulate wired to tier service ─────────────────
Scenario: Simulate returns assembly result with no fragments
When I run wired simulate for project "local/wire-app" using defaults
Then the wired context simulate command should succeed
Scenario: Simulate with JSON format returns valid JSON
When I run wired simulate for project "local/wire-app" using format "json"
Then the wired context simulate command should succeed
And the wired output should be valid JSON
And the wired JSON output should contain key "total_tokens"
And the wired JSON output should contain key "budget_used"
And the wired JSON output should contain key "strategies_used"
And the wired JSON output should contain key "context_hash"
Scenario: Simulate with budget override
When I run wired simulate for project "local/wire-app" using budget 2000
Then the wired context simulate command should succeed
Scenario: Simulate with pre-populated fragments assembles them
Given the tier service has a hot fragment "sim-1" for project "local/wire-app" sized at 100 tokens
When I run wired simulate for project "local/wire-app" using format "json"
Then the wired context simulate command should succeed
And the wired JSON total_tokens should be greater than 0
Scenario: Simulate on nonexistent project fails
When I run wired simulate for project "local/no-such" using defaults
Then the wired context simulate command should fail
Scenario: Simulate with invalid view fails
When I run wired simulate for project "local/wire-app" using view "bogus"
Then the wired context simulate command should fail
Scenario: Simulate with strategy hint
When I run wired simulate for project "local/wire-app" using strategy "breadth_first"
Then the wired context simulate command should succeed
Scenario: Simulate with focus filters fragments by resource URI
Given a focused hot fragment "foc-1" with resource "uko:file/main.py" and 100 tokens exists for project "local/wire-app"
And a focused hot fragment "foc-2" with resource "uko:file/test.py" and 80 tokens exists for project "local/wire-app"
When I run wired simulate for project "local/wire-app" focusing on "uko:file/main.py" in format "json"
Then the wired context simulate command should succeed
And the wired JSON fragment_count should be 1
And the wired JSON total_tokens should be 100
Scenario: Simulate with view option
When I run wired simulate for project "local/wire-app" using view "strategize"
Then the wired context simulate command should succeed
features/m5_acms_smoke.feature
+4 -4
View File
@@ -115,13 +115,13 @@ Feature: M5 ACMS pipeline and large-project context smoke tests
Then the m5 smoke project context show should succeed
And the m5 smoke project context output should contain phase views
Scenario: M5 smoke project context inspect is not yet wired
Scenario: M5 smoke project context inspect is wired to ACMS
When I m5 smoke invoke project context inspect
Then a m5 smoke NotImplementedError should be raised mentioning "ACMS"
Then the m5 smoke project context inspect should succeed
Scenario: M5 smoke project context simulate is not yet wired
Scenario: M5 smoke project context simulate is wired to ACMS
When I m5 smoke invoke project context simulate
Then a m5 smoke NotImplementedError should be raised mentioning "ACMS"
Then the m5 smoke project context simulate should succeed
# --- Context analysis agent ---
features/project_context_cli.feature
+6 -6
View File
@@ -76,17 +76,17 @@ Feature: Project context CLI commands (B2.cli)
When I run context show on "local/nonexistent-proj" without options
Then the context show command should fail
# ── inspect raises NotImplementedError ──────────────────────
# ── inspect now wired to ACMS tier service ─────────────────
Scenario: Inspect raises NotImplementedError with clear message
Scenario: Inspect succeeds and shows tier state
When I run context inspect on "local/ctx-app"
Then a NotImplementedError should be raised with message containing "ACMS"
Then the context inspect command should succeed
# ── simulate raises NotImplementedError ─────────────────────
# ── simulate now wired to ACMS tier service ────────────────
Scenario: Simulate raises NotImplementedError with clear message
Scenario: Simulate succeeds and shows assembly result
When I run context simulate on "local/ctx-app"
Then a NotImplementedError should be raised with message containing "ACMS"
Then the context simulate command should succeed
# ── view inheritance ─────────────────────────────────────────
features/steps/context_cli_wiring_steps.py
+646
View File
@@ -0,0 +1,646 @@
"""Step definitions for context_cli_wiring.feature.
Tests the ``agents project context set/show/inspect/simulate``
CLI commands wired to the ACMS pipeline using an in-memory
SQLite database and a real ContextTierService instance.
"""
from __future__ import annotations
import json
from io import StringIO
from typing import Any
from unittest.mock import MagicMock, patch
from behave import given, then, when # type: ignore[import-untyped]
from sqlalchemy import create_engine # type: ignore[import-untyped]
from sqlalchemy.orm import Session, sessionmaker # type: ignore[import-untyped]
from cleveragents.application.services.context_tiers import ContextTierService
from cleveragents.domain.models.acms.tiers import (
ContextTier,
TieredFragment,
)
# ------------------------------------------------------------------
# Session wrapper (no-op close)
# ------------------------------------------------------------------
class _UnclosableSession:
"""Wraps a SQLAlchemy Session; ``close()`` is a no-op."""
def __init__(self, real_session: Session) -> None:
object.__setattr__(self, "_real", real_session)
def close(self) -> None:
"""No-op."""
def __getattr__(self, name: str) -> Any:
return getattr(object.__getattribute__(self, "_real"), name)
def __setattr__(self, name: str, value: Any) -> None:
setattr(object.__getattribute__(self, "_real"), name, value)
def _make_wiring_session_factory(
context: Any,
) -> tuple[Any, Any]:
"""In-memory SQLite with all tables."""
from cleveragents.infrastructure.database.models import Base
engine = create_engine(
"sqlite:///:memory:",
echo=False,
connect_args={"check_same_thread": False},
)
Base.metadata.create_all(engine)
real_session = sessionmaker(
bind=engine,
expire_on_commit=False,
autoflush=True,
autocommit=False,
)()
wrapper = _UnclosableSession(real_session)
def _factory() -> Any:
return wrapper
return engine, _factory
# ------------------------------------------------------------------
# Background
# ------------------------------------------------------------------
@given("a context wiring in-memory database is initialized")
def step_init_wiring_db(context: Any) -> None:
from cleveragents.infrastructure.database.repositories import (
NamespacedProjectRepository,
)
engine, session_factory = _make_wiring_session_factory(context)
context.w_engine = engine
context.w_session_factory = session_factory
context.w_project_repo = NamespacedProjectRepository(
session_factory=session_factory
)
context.w_tier_service = ContextTierService()
context.w_output = ""
context.w_exit_code = 0
context.w_error = None
@given('a project "{name}" exists for context wiring')
def step_create_project_for_wiring(context: Any, name: str) -> None:
from cleveragents.domain.models.core.project import (
NamespacedProject,
parse_namespaced_name,
)
parsed = parse_namespaced_name(name)
proj = NamespacedProject(
name=parsed.name,
namespace=parsed.namespace,
)
context.w_project_repo.create(proj)
# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------
def _mock_wiring_container(context: Any) -> MagicMock:
"""Build a mock container wired to the test DB and tier service."""
mock_container = MagicMock()
mock_container.namespaced_project_repo.return_value = context.w_project_repo
mock_container.session_factory.return_value = context.w_session_factory
mock_container.context_tier_service.return_value = context.w_tier_service
return mock_container
def _run_wired(context: Any, func: Any, *args: Any, **kwargs: Any) -> None:
"""Execute a CLI function with a mocked DI container."""
import typer
from rich.console import Console as RichConsole
mock_cont = _mock_wiring_container(context)
buf = StringIO()
test_console = RichConsole(
file=buf,
no_color=True,
highlight=False,
force_terminal=False,
width=500,
soft_wrap=True,
)
with (
patch(
"cleveragents.application.container.get_container",
return_value=mock_cont,
),
patch(
"cleveragents.cli.commands.project_context.console",
test_console,
),
patch(
"cleveragents.cli.commands.project_context.err_console",
test_console,
),
):
try:
func(*args, **kwargs)
context.w_exit_code = 0
except typer.Exit as exc:
context.w_exit_code = exc.exit_code
except typer.Abort:
context.w_exit_code = 1
context.w_output = buf.getvalue()
# ------------------------------------------------------------------
# context set with ACMS options
# ------------------------------------------------------------------
@when('I run wired context set on "{project}" with hot-max-tokens {val:d}')
def step_wired_set_hot_max_tokens(context: Any, project: str, val: int) -> None:
from cleveragents.cli.commands.project_context import context_set
_run_wired(context, context_set, project=project, hot_max_tokens=val)
@when('I run wired context set on "{project}" with warm-max-decisions {val:d}')
def step_wired_set_warm_max_decisions(context: Any, project: str, val: int) -> None:
from cleveragents.cli.commands.project_context import context_set
_run_wired(context, context_set, project=project, warm_max_decisions=val)
@when('I run wired context set on "{project}" with cold-max-decisions {val:d}')
def step_wired_set_cold_max_decisions(context: Any, project: str, val: int) -> None:
from cleveragents.cli.commands.project_context import context_set
_run_wired(context, context_set, project=project, cold_max_decisions=val)
@when('I run wired context set on "{project}" with temporal-scope "{scope}"')
def step_wired_set_temporal_scope(context: Any, project: str, scope: str) -> None:
from cleveragents.cli.commands.project_context import context_set
_run_wired(context, context_set, project=project, temporal_scope=scope)
@when('I run wired context set on "{project}" with no-auto-refresh')
def step_wired_set_no_auto_refresh(context: Any, project: str) -> None:
from cleveragents.cli.commands.project_context import context_set
_run_wired(context, context_set, project=project, auto_refresh=False)
@when('I run wired context set on "{project}" with no-summarize')
def step_wired_set_no_summarize(context: Any, project: str) -> None:
from cleveragents.cli.commands.project_context import context_set
_run_wired(context, context_set, project=project, summarize=False)
@when('I run wired context set on "{project}" with strategy "{strategy}"')
def step_wired_set_strategy(context: Any, project: str, strategy: str) -> None:
from cleveragents.cli.commands.project_context import context_set
_run_wired(context, context_set, project=project, strategy=[strategy])
@when('I run wired context set on "{project}" with skeleton-ratio {val:g}')
def step_wired_set_skeleton_ratio(context: Any, project: str, val: float) -> None:
from cleveragents.cli.commands.project_context import context_set
_run_wired(context, context_set, project=project, skeleton_ratio=val)
@when('I run wired context set on "{project}" with default-breadth {val:d}')
def step_wired_set_default_breadth(context: Any, project: str, val: int) -> None:
from cleveragents.cli.commands.project_context import context_set
_run_wired(context, context_set, project=project, default_breadth=val)
@when('I run wired context set on "{project}" with default-depth {val:d}')
def step_wired_set_default_depth(context: Any, project: str, val: int) -> None:
from cleveragents.cli.commands.project_context import context_set
_run_wired(context, context_set, project=project, default_depth=str(val))
@when('I run wired context set on "{project}" with query-limit {val:d}')
def step_wired_set_query_limit(context: Any, project: str, val: int) -> None:
from cleveragents.cli.commands.project_context import context_set
_run_wired(context, context_set, project=project, query_limit=val)
@when('I run wired context set on "{project}" with summary-max-tokens {val:d}')
def step_wired_set_summary_max_tokens(context: Any, project: str, val: int) -> None:
from cleveragents.cli.commands.project_context import context_set
_run_wired(context, context_set, project=project, summary_max_tokens=val)
# ------------------------------------------------------------------
# context show with ACMS config
# ------------------------------------------------------------------
@given('I have set ACMS config on "{project}" with hot-max-tokens {val:d}')
def step_given_acms_config(context: Any, project: str, val: int) -> None:
from cleveragents.cli.commands.project_context import context_set
_run_wired(context, context_set, project=project, hot_max_tokens=val)
@when('I run wired context show on "{project}" with format "{fmt}"')
def step_wired_show_format(context: Any, project: str, fmt: str) -> None:
from cleveragents.cli.commands.project_context import context_show
_run_wired(context, context_show, project=project, output_format=fmt)
@when('I run wired context show on "{project}" with view "{view}" and format "{fmt}"')
def step_wired_show_view_format(
context: Any, project: str, view: str, fmt: str
) -> None:
from cleveragents.cli.commands.project_context import context_show
_run_wired(context, context_show, project=project, view=view, output_format=fmt)
# ------------------------------------------------------------------
# context inspect
# ------------------------------------------------------------------
@when('I run wired inspect for project "{project}" using defaults')
def step_wired_inspect(context: Any, project: str) -> None:
from cleveragents.cli.commands.project_context import context_inspect
_run_wired(context, context_inspect, project=project)
@when('I run wired inspect for project "{project}" using format "{fmt}"')
def step_wired_inspect_format(context: Any, project: str, fmt: str) -> None:
from cleveragents.cli.commands.project_context import context_inspect
_run_wired(context, context_inspect, project=project, output_format=fmt)
@when('I run wired inspect for project "{project}" using view "{view}"')
def step_wired_inspect_view(context: Any, project: str, view: str) -> None:
from cleveragents.cli.commands.project_context import context_inspect
_run_wired(context, context_inspect, project=project, view=view)
@given('the tier service has a hot fragment "{fid}" for project "{project}"')
def step_given_hot_fragment(context: Any, fid: str, project: str) -> None:
frag = TieredFragment(
fragment_id=fid,
content="test content for " + fid,
tier=ContextTier.HOT,
project_name=project,
token_count=50,
)
context.w_tier_service.store(frag)
@given('the tier service has a warm fragment "{fid}" for project "{project}"')
def step_given_warm_fragment(context: Any, fid: str, project: str) -> None:
frag = TieredFragment(
fragment_id=fid,
content="test content for " + fid,
tier=ContextTier.WARM,
project_name=project,
token_count=30,
)
context.w_tier_service.store(frag)
@given(
'the tier service has a hot fragment "{fid}" for project'
' "{project}" sized at {tokens:d} tokens'
)
def step_given_hot_fragment_tokens(
context: Any, fid: str, project: str, tokens: int
) -> None:
frag = TieredFragment(
fragment_id=fid,
content="test content for simulation " + fid,
tier=ContextTier.HOT,
project_name=project,
token_count=tokens,
)
context.w_tier_service.store(frag)
@given(
'a focused hot fragment "{fid}" with resource "{resource}"'
" and {tokens:d} tokens exists"
' for project "{project}"'
)
def step_given_focused_hot_fragment(
context: Any, fid: str, resource: str, tokens: int, project: str
) -> None:
frag = TieredFragment(
fragment_id=fid,
content="test content for " + fid,
tier=ContextTier.HOT,
project_name=project,
resource_id=resource,
token_count=tokens,
)
context.w_tier_service.store(frag)
# ------------------------------------------------------------------
# context simulate
# ------------------------------------------------------------------
@when('I run wired simulate for project "{project}" using defaults')
def step_wired_simulate(context: Any, project: str) -> None:
from cleveragents.cli.commands.project_context import context_simulate
_run_wired(context, context_simulate, project=project)
@when('I run wired simulate for project "{project}" using format "{fmt}"')
def step_wired_simulate_format(context: Any, project: str, fmt: str) -> None:
from cleveragents.cli.commands.project_context import context_simulate
_run_wired(context, context_simulate, project=project, output_format=fmt)
@when('I run wired simulate for project "{project}" using budget {val:d}')
def step_wired_simulate_budget(context: Any, project: str, val: int) -> None:
from cleveragents.cli.commands.project_context import context_simulate
_run_wired(context, context_simulate, project=project, budget=val)
@when('I run wired simulate for project "{project}" using view "{view}"')
def step_wired_simulate_view(context: Any, project: str, view: str) -> None:
from cleveragents.cli.commands.project_context import context_simulate
_run_wired(context, context_simulate, project=project, view=view)
@when('I run wired simulate for project "{project}" using strategy "{strat}"')
def step_wired_simulate_strategy(context: Any, project: str, strat: str) -> None:
from cleveragents.cli.commands.project_context import context_simulate
_run_wired(context, context_simulate, project=project, strategy_hint=[strat])
@when(
'I run wired simulate for project "{project}"'
' focusing on "{focus}" in format "{fmt}"'
)
def step_wired_simulate_focus_format(
context: Any, project: str, focus: str, fmt: str
) -> None:
from cleveragents.cli.commands.project_context import context_simulate
_run_wired(
context,
context_simulate,
project=project,
focus=[focus],
output_format=fmt,
)
@then("the wired context set command should succeed")
def step_wired_set_ok(context: Any) -> None:
assert context.w_exit_code == 0, (
f"Expected exit 0, got {context.w_exit_code}. Output: {context.w_output}"
)
@then("the wired context set command should fail")
def step_wired_set_fail(context: Any) -> None:
assert context.w_exit_code != 0, (
f"Expected non-zero exit, got {context.w_exit_code}"
)
@then("the wired context show command should succeed")
def step_wired_show_ok(context: Any) -> None:
assert context.w_exit_code == 0, (
f"Expected exit 0, got {context.w_exit_code}. Output: {context.w_output}"
)
@then("the wired context inspect command should succeed")
def step_wired_inspect_ok(context: Any) -> None:
assert context.w_exit_code == 0, (
f"Expected exit 0, got {context.w_exit_code}. Output: {context.w_output}"
)
@then("the wired context inspect command should fail")
def step_wired_inspect_fail(context: Any) -> None:
assert context.w_exit_code != 0, (
f"Expected non-zero exit, got {context.w_exit_code}"
)
@then("the wired context simulate command should succeed")
def step_wired_simulate_ok(context: Any) -> None:
assert context.w_exit_code == 0, (
f"Expected exit 0, got {context.w_exit_code}. Output: {context.w_output}"
)
@then("the wired context simulate command should fail")
def step_wired_simulate_fail(context: Any) -> None:
assert context.w_exit_code != 0, (
f"Expected non-zero exit, got {context.w_exit_code}"
)
# ------------------------------------------------------------------
# Then assertions — ACMS config
# ------------------------------------------------------------------
def _read_stored_acms_config(context: Any) -> dict:
"""Read the ACMS config from the test DB."""
from cleveragents.cli.commands.project_context import _read_acms_config
return _read_acms_config(context.w_session_factory, "local/wire-app")
@then("the stored ACMS config hot_max_tokens should be {val:d}")
def step_acms_hot_max_tokens(context: Any, val: int) -> None:
acms = _read_stored_acms_config(context)
assert acms["hot_max_tokens"] == val, (
f"Expected {val}, got {acms['hot_max_tokens']}"
)
@then("the stored ACMS config warm_max_decisions should be {val:d}")
def step_acms_warm_max_decisions(context: Any, val: int) -> None:
acms = _read_stored_acms_config(context)
assert acms["warm_max_decisions"] == val, (
f"Expected {val}, got {acms['warm_max_decisions']}"
)
@then("the stored ACMS config cold_max_decisions should be {val:d}")
def step_acms_cold_max_decisions(context: Any, val: int) -> None:
acms = _read_stored_acms_config(context)
assert acms["cold_max_decisions"] == val, (
f"Expected {val}, got {acms['cold_max_decisions']}"
)
@then('the stored ACMS config temporal_scope should be "{val}"')
def step_acms_temporal_scope(context: Any, val: str) -> None:
acms = _read_stored_acms_config(context)
assert acms["temporal_scope"] == val, (
f"Expected {val}, got {acms['temporal_scope']}"
)
@then("the stored ACMS config auto_refresh should be false")
def step_acms_auto_refresh_false(context: Any) -> None:
acms = _read_stored_acms_config(context)
assert acms["auto_refresh"] is False, f"Expected False, got {acms['auto_refresh']}"
@then("the stored ACMS config summarize should be false")
def step_acms_summarize_false(context: Any) -> None:
acms = _read_stored_acms_config(context)
assert acms["summarize"] is False, f"Expected False, got {acms['summarize']}"
@then('the stored ACMS config strategies should contain "{strat}"')
def step_acms_strategies_contain(context: Any, strat: str) -> None:
acms = _read_stored_acms_config(context)
assert strat in acms.get("strategies", []), (
f"Expected {strat} in {acms.get('strategies', [])}"
)
@then("the stored ACMS config skeleton_ratio should be {val:g}")
def step_acms_skeleton_ratio(context: Any, val: float) -> None:
acms = _read_stored_acms_config(context)
assert abs(acms["skeleton_ratio"] - val) < 1e-9, (
f"Expected {val}, got {acms['skeleton_ratio']}"
)
@then("the stored ACMS config default_breadth should be {val:d}")
def step_acms_default_breadth(context: Any, val: int) -> None:
acms = _read_stored_acms_config(context)
assert acms["default_breadth"] == val, (
f"Expected {val}, got {acms['default_breadth']}"
)
@then("the stored ACMS config default_depth should be {val:d}")
def step_acms_default_depth(context: Any, val: int) -> None:
acms = _read_stored_acms_config(context)
assert acms["default_depth"] == val, f"Expected {val}, got {acms['default_depth']}"
@then("the stored ACMS config query_limit should be {val:d}")
def step_acms_query_limit(context: Any, val: int) -> None:
acms = _read_stored_acms_config(context)
assert acms["query_limit"] == val, f"Expected {val}, got {acms['query_limit']}"
@then("the stored ACMS config summary_max_tokens should be {val:d}")
def step_acms_summary_max_tokens(context: Any, val: int) -> None:
acms = _read_stored_acms_config(context)
assert acms["summary_max_tokens"] == val, (
f"Expected {val}, got {acms['summary_max_tokens']}"
)
# ------------------------------------------------------------------
# Then assertions — JSON output
# ------------------------------------------------------------------
@then("the wired output should be valid JSON")
def step_wired_output_valid_json(context: Any) -> None:
try:
context.w_parsed_json = json.loads(context.w_output.strip())
except json.JSONDecodeError as exc:
raise AssertionError(
f"Output is not valid JSON: {exc}\nOutput: {context.w_output!r}"
) from exc
@then('the wired JSON output should contain key "{key}"')
def step_wired_json_has_key(context: Any, key: str) -> None:
data = getattr(context, "w_parsed_json", None)
if data is None:
data = json.loads(context.w_output.strip())
context.w_parsed_json = data
assert key in data, f"Key '{key}' not found in JSON output: {list(data.keys())}"
@then("the wired JSON project_fragments total should be {val:d}")
def step_wired_json_fragments_total(context: Any, val: int) -> None:
data = getattr(context, "w_parsed_json", None)
if data is None:
data = json.loads(context.w_output.strip())
context.w_parsed_json = data
total = data.get("project_fragments", {}).get("total", 0)
assert total == val, f"Expected {val} fragments, got {total}"
@then("the wired JSON total_tokens should be greater than 0")
def step_wired_json_total_tokens_positive(context: Any) -> None:
data = getattr(context, "w_parsed_json", None)
if data is None:
data = json.loads(context.w_output.strip())
context.w_parsed_json = data
total = data.get("total_tokens", 0)
assert total > 0, f"Expected total_tokens > 0, got {total}"
@then("the wired JSON tier_metrics hot_count should be {val:d}")
def step_wired_json_tier_metrics_hot_count(context: Any, val: int) -> None:
data = getattr(context, "w_parsed_json", None)
if data is None:
data = json.loads(context.w_output.strip())
context.w_parsed_json = data
hot_count = data.get("tier_metrics", {}).get("hot_count", -1)
assert hot_count == val, f"Expected tier_metrics.hot_count={val}, got {hot_count}"
@then("the wired JSON fragment_count should be {val:d}")
def step_wired_json_fragment_count(context: Any, val: int) -> None:
data = getattr(context, "w_parsed_json", None)
if data is None:
data = json.loads(context.w_output.strip())
context.w_parsed_json = data
count = data.get("fragment_count", -1)
assert count == val, f"Expected fragment_count={val}, got {count}"
@then("the wired JSON total_tokens should be {val:d}")
def step_wired_json_total_tokens_exact(context: Any, val: int) -> None:
data = getattr(context, "w_parsed_json", None)
if data is None:
data = json.loads(context.w_output.strip())
context.w_parsed_json = data
total = data.get("total_tokens", -1)
assert total == val, f"Expected total_tokens={val}, got {total}"
features/steps/m5_acms_smoke_steps.py
+73 -20
View File
@@ -16,10 +16,6 @@ from pydantic import ValidationError
from typer.testing import CliRunner
from cleveragents.cli.commands.context import app as context_app
from cleveragents.cli.commands.project_context import (
context_inspect,
context_simulate,
)
from cleveragents.domain.models.core.context import Context as ContextModel
from cleveragents.domain.models.core.context_policy import (
ContextView,
@@ -407,6 +403,8 @@ def step_m5_project_with_policy(context: Context) -> None:
@when("I m5 smoke invoke project context show")
def step_m5_invoke_project_context_show(context: Context) -> None:
from cleveragents.cli.commands.project_context import _default_acms_config
mock_repo = MagicMock()
mock_repo.get.return_value = MagicMock() # project exists
@@ -426,6 +424,10 @@ def step_m5_invoke_project_context_show(context: Context) -> None:
"cleveragents.cli.commands.project_context._read_policy",
return_value=context.m5_saved_policy,
),
patch(
"cleveragents.cli.commands.project_context._read_acms_config",
return_value=_default_acms_config(),
),
):
from cleveragents.cli.commands.project_context import app as pc_app
@@ -448,28 +450,79 @@ def step_m5_project_context_has_phases(context: Context) -> None:
@when("I m5 smoke invoke project context inspect")
def step_m5_invoke_context_inspect(context: Context) -> None:
try:
context_inspect(project="local/m5-proj")
context.m5_error = None
except NotImplementedError as exc:
context.m5_error = exc
from cleveragents.application.services.context_tiers import ContextTierService
from cleveragents.cli.commands.project_context import (
_default_acms_config,
)
from cleveragents.cli.commands.project_context import (
app as pc_app,
)
mock_repo = MagicMock()
mock_repo.get.return_value = MagicMock()
mock_container = MagicMock()
mock_container.session_factory.return_value = MagicMock()
mock_container.namespaced_project_repo.return_value = mock_repo
mock_container.context_tier_service.return_value = ContextTierService()
with (
patch(
"cleveragents.application.container.get_container",
return_value=mock_container,
),
patch(
"cleveragents.cli.commands.project_context._read_acms_config",
return_value=_default_acms_config(),
),
):
result = context.runner.invoke(pc_app, ["inspect", "local/m5-proj"])
context.m5_result = result
@when("I m5 smoke invoke project context simulate")
def step_m5_invoke_context_simulate(context: Context) -> None:
try:
context_simulate(project="local/m5-proj")
context.m5_error = None
except NotImplementedError as exc:
context.m5_error = exc
from cleveragents.application.services.context_tiers import ContextTierService
from cleveragents.cli.commands.project_context import (
_default_acms_config,
)
from cleveragents.cli.commands.project_context import (
app as pc_app,
)
mock_repo = MagicMock()
mock_repo.get.return_value = MagicMock()
mock_container = MagicMock()
mock_container.session_factory.return_value = MagicMock()
mock_container.namespaced_project_repo.return_value = mock_repo
mock_container.context_tier_service.return_value = ContextTierService()
with (
patch(
"cleveragents.application.container.get_container",
return_value=mock_container,
),
patch(
"cleveragents.cli.commands.project_context._read_acms_config",
return_value=_default_acms_config(),
),
):
result = context.runner.invoke(pc_app, ["simulate", "local/m5-proj"])
context.m5_result = result
@then('a m5 smoke NotImplementedError should be raised mentioning "{text}"')
def step_m5_not_implemented_mentioning(context: Context, text: str) -> None:
assert context.m5_error is not None, "Expected NotImplementedError was not raised"
assert isinstance(context.m5_error, NotImplementedError)
assert text.lower() in str(context.m5_error).lower(), (
f"Expected '{text}' in: {context.m5_error}"
@then("the m5 smoke project context inspect should succeed")
def step_m5_project_context_inspect_success(context: Context) -> None:
assert context.m5_result.exit_code == 0, (
f"Exit code: {context.m5_result.exit_code}\n{context.m5_result.output}"
)
@then("the m5 smoke project context simulate should succeed")
def step_m5_project_context_simulate_success(context: Context) -> None:
assert context.m5_result.exit_code == 0, (
f"Exit code: {context.m5_result.exit_code}\n{context.m5_result.output}"
)
features/steps/project_context_cli_steps.py
+28 -1
View File
@@ -106,9 +106,17 @@ def step_create_project_for_ctx(context: Any, name: str) -> None:
def _mock_container(context: Any) -> MagicMock:
"""Build a mock container wired to the test DB."""
from cleveragents.application.services.context_tiers import (
ContextTierService,
)
mock_container = MagicMock()
mock_container.namespaced_project_repo.return_value = context.ctx_project_repo
mock_container.session_factory.return_value = context.ctx_session_factory
# Provide a real (in-memory) ContextTierService for inspect/simulate
if not hasattr(context, "ctx_tier_service"):
context.ctx_tier_service = ContextTierService()
mock_container.context_tier_service.return_value = context.ctx_tier_service
return mock_container
@@ -123,7 +131,12 @@ def _run_with_container(context: Any, func: Any, *args: Any, **kwargs: Any) -> N
# ANSI escape codes when FORCE_COLOR is set or a real colour terminal
# is detected.
test_console = RichConsole(
file=buf, no_color=True, highlight=False, force_terminal=False
file=buf,
no_color=True,
highlight=False,
force_terminal=False,
width=500,
soft_wrap=True,
)
with (
@@ -516,6 +529,20 @@ def step_policy_strat_none(context: Any) -> None:
)
@then("the context inspect command should succeed")
def step_ctx_inspect_ok(context: Any) -> None:
assert context.ctx_exit_code == 0, (
f"Expected exit 0, got {context.ctx_exit_code}. Output: {context.ctx_output}"
)
@then("the context simulate command should succeed")
def step_ctx_simulate_ok(context: Any) -> None:
assert context.ctx_exit_code == 0, (
f"Expected exit 0, got {context.ctx_exit_code}. Output: {context.ctx_output}"
)
@then('a NotImplementedError should be raised with message containing "{msg}"')
def step_not_implemented(context: Any, msg: str) -> None:
assert context.ctx_error is not None, "No error was raised"
robot/helper_project_context_cli.py
+226 -25
View File
@@ -5,12 +5,20 @@ Usage: python robot/helper_project_context_cli.py <test-name>
from __future__ import annotations
import json
import sys
from io import StringIO
from typing import Any
from unittest.mock import MagicMock, patch
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from cleveragents.application.services.context_tiers import ContextTierService
from cleveragents.domain.models.acms.tiers import (
ContextTier,
TieredFragment,
)
from cleveragents.domain.models.core.context_policy import (
ContextView,
ProjectContextPolicy,
@@ -64,6 +72,68 @@ def _create_project(
repo.create(proj)
def _make_mock_container(
repo: NamespacedProjectRepository,
session_factory: Any,
tier_service: ContextTierService | None = None,
) -> MagicMock:
"""Build a mock DI container wired to the test fixtures."""
mock_container = MagicMock()
mock_container.namespaced_project_repo.return_value = repo
mock_container.session_factory.return_value = session_factory
if tier_service is None:
tier_service = ContextTierService()
mock_container.context_tier_service.return_value = tier_service
return mock_container
def _run_cli_func(
func: Any,
repo: NamespacedProjectRepository,
session_factory: Any,
tier_service: ContextTierService | None = None,
**kwargs: Any,
) -> tuple[int, str]:
"""Execute a CLI function with mocked DI and capture output."""
import typer
from rich.console import Console as RichConsole
mock_cont = _make_mock_container(repo, session_factory, tier_service)
buf = StringIO()
test_console = RichConsole(
file=buf,
no_color=True,
highlight=False,
force_terminal=False,
width=500,
soft_wrap=True,
)
exit_code = 0
with (
patch(
"cleveragents.application.container.get_container",
return_value=mock_cont,
),
patch(
"cleveragents.cli.commands.project_context.console",
test_console,
),
patch(
"cleveragents.cli.commands.project_context.err_console",
test_console,
),
):
try:
func(**kwargs)
except typer.Exit as exc:
exit_code = exc.exit_code
except typer.Abort:
exit_code = 1
return exit_code, buf.getvalue()
def test_context_set_show() -> None:
"""Set a context policy and then read it back."""
from cleveragents.cli.commands.project_context import (
@@ -108,42 +178,173 @@ def test_context_set_show() -> None:
print("context-set-show-ok")
def test_context_inspect_stub() -> None:
"""Verify inspect raises NotImplementedError."""
from cleveragents.cli.commands.project_context import (
def test_context_inspect_wired() -> None:
"""Verify inspect is wired to ACMS and succeeds."""
from cleveragents.cli.commands.project_context import context_inspect
repo, sf = _setup_db()
_create_project(repo, "ctx-robot")
tier_svc = ContextTierService()
# Pre-populate a fragment
tier_svc.store(
TieredFragment(
fragment_id="robot-f1",
content="test fragment",
tier=ContextTier.HOT,
project_name="local/ctx-robot",
token_count=50,
)
)
exit_code, output = _run_cli_func(
context_inspect,
repo,
sf,
tier_service=tier_svc,
project="local/ctx-robot",
output_format="json",
)
if exit_code != 0:
raise AssertionError(f"inspect failed with exit {exit_code}: {output}")
data = json.loads(output.strip())
if "tier_metrics" not in data:
raise AssertionError(f"Missing tier_metrics in output: {data.keys()}")
if "project_fragments" not in data:
raise AssertionError(f"Missing project_fragments in output: {data.keys()}")
if data["project_fragments"]["total"] != 1:
raise AssertionError(
f"Expected 1 fragment, got {data['project_fragments']['total']}"
)
print("context-inspect-wired-ok")
def test_context_simulate_wired() -> None:
"""Verify simulate is wired to ACMS and succeeds."""
from cleveragents.cli.commands.project_context import context_simulate
repo, sf = _setup_db()
_create_project(repo, "ctx-robot")
tier_svc = ContextTierService()
# Pre-populate a fragment for simulation
tier_svc.store(
TieredFragment(
fragment_id="robot-sim1",
content="simulation test fragment",
tier=ContextTier.HOT,
project_name="local/ctx-robot",
token_count=100,
)
)
try:
context_inspect(project="local/any")
raise AssertionError("Should have raised")
except NotImplementedError as exc:
if "ACMS" not in str(exc):
raise AssertionError(f"Missing ACMS message: {exc}") from exc
print("context-inspect-stub-ok")
def test_context_simulate_stub() -> None:
"""Verify simulate raises NotImplementedError."""
from cleveragents.cli.commands.project_context import (
exit_code, output = _run_cli_func(
context_simulate,
repo,
sf,
tier_service=tier_svc,
project="local/ctx-robot",
output_format="json",
)
if exit_code != 0:
raise AssertionError(f"simulate failed with exit {exit_code}: {output}")
data = json.loads(output.strip())
if "total_tokens" not in data:
raise AssertionError(f"Missing total_tokens in output: {data.keys()}")
if data["total_tokens"] <= 0:
raise AssertionError(
f"Expected positive total_tokens, got {data['total_tokens']}"
)
if "strategies_used" not in data:
raise AssertionError(f"Missing strategies_used in output: {data.keys()}")
print("context-simulate-wired-ok")
def test_context_set_acms_options() -> None:
"""Verify context set persists ACMS pipeline config."""
from cleveragents.cli.commands.project_context import (
_read_acms_config,
context_set,
)
try:
context_simulate(project="local/any")
raise AssertionError("Should have raised")
except NotImplementedError as exc:
if "ACMS" not in str(exc):
raise AssertionError(f"Missing ACMS message: {exc}") from exc
repo, sf = _setup_db()
_create_project(repo, "ctx-robot")
print("context-simulate-stub-ok")
exit_code, output = _run_cli_func(
context_set,
repo,
sf,
project="local/ctx-robot",
hot_max_tokens=4000,
warm_max_decisions=200,
temporal_scope="recent",
auto_refresh=False,
)
if exit_code != 0:
raise AssertionError(f"set failed with exit {exit_code}: {output}")
acms = _read_acms_config(sf, "local/ctx-robot")
if acms["hot_max_tokens"] != 4000:
raise AssertionError(f"hot_max_tokens mismatch: {acms['hot_max_tokens']}")
if acms["warm_max_decisions"] != 200:
raise AssertionError(
f"warm_max_decisions mismatch: {acms['warm_max_decisions']}"
)
if acms["temporal_scope"] != "recent":
raise AssertionError(f"temporal_scope mismatch: {acms['temporal_scope']}")
if acms["auto_refresh"] is not False:
raise AssertionError(f"auto_refresh mismatch: {acms['auto_refresh']}")
print("context-set-acms-ok")
def test_context_show_acms_config() -> None:
"""Verify context show includes ACMS config in JSON output."""
from cleveragents.cli.commands.project_context import context_set, context_show
repo, sf = _setup_db()
_create_project(repo, "ctx-robot")
# First set some ACMS config
_run_cli_func(
context_set,
repo,
sf,
project="local/ctx-robot",
hot_max_tokens=5000,
)
exit_code, output = _run_cli_func(
context_show,
repo,
sf,
project="local/ctx-robot",
output_format="json",
)
if exit_code != 0:
raise AssertionError(f"show failed with exit {exit_code}: {output}")
data = json.loads(output.strip())
if "acms_config" not in data:
raise AssertionError(f"Missing acms_config in output: {data.keys()}")
if data["acms_config"]["hot_max_tokens"] != 5000:
raise AssertionError(
f"hot_max_tokens mismatch: {data['acms_config']['hot_max_tokens']}"
)
print("context-show-acms-ok")
TESTS: dict[str, Any] = {
"set-show": test_context_set_show,
"inspect-stub": test_context_inspect_stub,
"simulate-stub": test_context_simulate_stub,
"inspect-wired": test_context_inspect_wired,
"simulate-wired": test_context_simulate_wired,
"set-acms": test_context_set_acms_options,
"show-acms": test_context_show_acms_config,
}
robot/project_context_cli.robot
+30 -8
View File
@@ -11,17 +11,39 @@ ${HELPER_SCRIPT} robot/helper_project_context_cli.py
Context Policy Set And Show Round Trip
[Documentation] Set a context policy and verify it reads back correctly
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} set-show cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} context-set-show-ok
Context Inspect Raises NotImplementedError
[Documentation] Verify inspect stub raises NotImplementedError
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} inspect-stub cwd=${WORKSPACE}
Context Inspect Wired To ACMS
[Documentation] Verify inspect queries the ACMS tier service and returns tier state
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} inspect-wired cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} context-inspect-stub-ok
Should Contain ${result.stdout} context-inspect-wired-ok
Context Simulate Raises NotImplementedError
[Documentation] Verify simulate stub raises NotImplementedError
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} simulate-stub cwd=${WORKSPACE}
Context Simulate Wired To ACMS
[Documentation] Verify simulate runs dry-run context assembly via ACMS
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} simulate-wired cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} context-simulate-stub-ok
Should Contain ${result.stdout} context-simulate-wired-ok
Context Set ACMS Pipeline Options
[Documentation] Verify context set persists ACMS pipeline configuration
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} set-acms cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} context-set-acms-ok
Context Show ACMS Pipeline Config
[Documentation] Verify context show displays ACMS pipeline configuration
${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} show-acms cwd=${WORKSPACE}
Log ${result.stdout}
Log ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} context-show-acms-ok
src/cleveragents/application/services/context_tiers.py
+27
View File
@@ -266,6 +266,33 @@ class ContextTierService:
warm_miss_count=self._warm_miss,
)
def get_scoped_metrics(self, project_names: list[str]) -> TierMetrics:
"""Return metrics scoped to the given project names.
Fragment counts (``hot_count``, ``warm_count``, ``cold_count``)
reflect only fragments belonging to the specified projects.
Hit/miss counters remain global as they are service-level cache
performance metrics that are not tracked per-project.
Args:
project_names: Project names to scope fragment counts to.
Raises:
ValueError: If *project_names* is empty.
"""
if not project_names:
raise ValueError("project_names must be non-empty")
scoped = ScopedBackendView(allowed_projects=project_names)
return TierMetrics(
hot_count=sum(1 for f in self._hot.values() if scoped.is_visible(f)),
warm_count=sum(1 for f in self._warm.values() if scoped.is_visible(f)),
cold_count=sum(1 for f in self._cold.values() if scoped.is_visible(f)),
hot_hit_count=self._hot_hit,
hot_miss_count=self._hot_miss,
warm_hit_count=self._warm_hit,
warm_miss_count=self._warm_miss,
)
# ------------------------------------------------------------------
# Budget
# ------------------------------------------------------------------
src/cleveragents/cli/commands/project_context.py
+694 -27
View File
@@ -1,13 +1,18 @@
"""Project context policy CLI commands.
Implements ``agents project context set/show/inspect/simulate`` to
manage per-project context policies with view inheritance.
manage per-project context policies with view inheritance and to
query the ACMS pipeline for context state and dry-run assembly.
Commands:
- ``agents project context set`` — persist a context policy view
and ACMS pipeline configuration overrides
- ``agents project context show`` — display the active context policy
- ``agents project context inspect`` — (stub) inspect effective context
- ``agents project context simulate`` — (stub) simulate context window
including effective ACMS pipeline config
- ``agents project context inspect`` — inspect effective context state
from the ACMS tier service for a project
- ``agents project context simulate`` — dry-run context assembly
showing fragments, budget usage, and strategy selection
Based on ADR-014 (Context Management / ACMS) and the
``ProjectContextPolicy`` domain model.
@@ -15,19 +20,34 @@ Based on ADR-014 (Context Management / ACMS) and the
from __future__ import annotations
import hashlib
import json as _json
from typing import Annotated, Any
import typer
from rich.console import Console
from rich.panel import Panel
from rich.table import Table
from cleveragents.cli.formatting import OutputFormat, format_output
from cleveragents.domain.models.acms.crp import (
AssembledContext,
ContextFragment,
ContextRequest,
FragmentProvenance,
)
from cleveragents.domain.models.acms.tiers import (
ActorRole,
ContextTier,
TieredFragment,
TierMetrics,
)
from cleveragents.domain.models.core.context_policy import (
VALID_PHASES,
ContextView,
ProjectContextPolicy,
)
from cleveragents.domain.models.core.project import TemporalScope
app = typer.Typer(help="Manage project context policies")
console = Console()
@@ -35,6 +55,15 @@ err_console = Console(stderr=True)
_FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)"
# Default ACMS pipeline configuration values
_DEFAULT_HOT_MAX_TOKENS = 8000
_DEFAULT_WARM_MAX_DECISIONS = 500
_DEFAULT_COLD_MAX_DECISIONS = 5000
_DEFAULT_BREADTH = 2
_DEFAULT_DEPTH = 3
_DEFAULT_SKELETON_RATIO = 0.2
_DEFAULT_BUDGET_TOKENS = 8000
# ------------------------------------------------------------------
# Internal helpers
@@ -49,6 +78,14 @@ def _get_namespaced_project_repo() -> Any:
return container.namespaced_project_repo()
def _get_context_tier_service() -> Any:
"""Return the ContextTierService from the DI container."""
from cleveragents.application.container import get_container
container = get_container()
return container.context_tier_service()
def _load_policy_json(
session_factory: Any,
namespaced_name: str,
@@ -112,17 +149,57 @@ def _read_policy(
return ProjectContextPolicy()
def _read_acms_config(
session_factory: Any,
namespaced_name: str,
) -> dict[str, Any]:
"""Read the ACMS pipeline config from the stored policy JSON.
The ACMS config is stored as an ``acms_config`` key in the
same JSON blob as the context policy. Returns defaults when
not present.
"""
raw = _load_policy_json(session_factory, namespaced_name)
if raw is None:
return _default_acms_config()
return raw.get("acms_config", _default_acms_config())
def _default_acms_config() -> dict[str, Any]:
"""Return the default ACMS pipeline configuration."""
return {
"hot_max_tokens": _DEFAULT_HOT_MAX_TOKENS,
"warm_max_decisions": _DEFAULT_WARM_MAX_DECISIONS,
"cold_max_decisions": _DEFAULT_COLD_MAX_DECISIONS,
"query_limit": None,
"summarize": True,
"summary_max_tokens": None,
"strategies": [],
"default_breadth": _DEFAULT_BREADTH,
"default_depth": _DEFAULT_DEPTH,
"depth_gradient": {},
"skeleton_ratio": _DEFAULT_SKELETON_RATIO,
"temporal_scope": "current",
"auto_refresh": True,
}
def _write_policy(
session_factory: Any,
namespaced_name: str,
policy: ProjectContextPolicy,
acms_config: dict[str, Any] | None = None,
) -> None:
"""Serialise and persist the policy."""
_save_policy_json(
session_factory,
namespaced_name,
policy.model_dump(mode="json"),
)
"""Serialise and persist the policy with optional ACMS config."""
data = policy.model_dump(mode="json")
if acms_config is not None:
data["acms_config"] = acms_config
else:
# Preserve existing ACMS config
existing = _load_policy_json(session_factory, namespaced_name)
if existing is not None and "acms_config" in existing:
data["acms_config"] = existing["acms_config"]
_save_policy_json(session_factory, namespaced_name, data)
def _policy_to_dict(policy: ProjectContextPolicy) -> dict[str, Any]:
@@ -143,6 +220,160 @@ def _policy_to_dict(policy: ProjectContextPolicy) -> dict[str, Any]:
return result
def _tier_metrics_to_dict(metrics: TierMetrics) -> dict[str, Any]:
"""Convert tier metrics to a display-friendly dict."""
return {
"hot_count": metrics.hot_count,
"warm_count": metrics.warm_count,
"cold_count": metrics.cold_count,
"hot_hit_count": metrics.hot_hit_count,
"hot_miss_count": metrics.hot_miss_count,
"warm_hit_count": metrics.warm_hit_count,
"warm_miss_count": metrics.warm_miss_count,
}
def _fragment_to_dict(fragment: TieredFragment) -> dict[str, Any]:
"""Convert a tiered fragment to a display-friendly dict."""
return {
"fragment_id": fragment.fragment_id,
"tier": fragment.tier.value,
"resource_id": fragment.resource_id,
"project_name": fragment.project_name,
"token_count": fragment.token_count,
"access_count": fragment.access_count,
"last_accessed": fragment.last_accessed.isoformat(),
}
def _assembled_to_dict(assembled: AssembledContext) -> dict[str, Any]:
"""Convert an AssembledContext to a display-friendly dict."""
return {
"total_tokens": assembled.total_tokens,
"budget_used": assembled.budget_used,
"strategies_used": assembled.strategies_used,
"context_hash": assembled.context_hash,
"preamble": assembled.preamble,
"fragment_count": len(assembled.fragments),
"fragments": [
{
"uko_node": f.uko_node,
"detail_depth": f.detail_depth,
"token_count": f.token_count,
"relevance_score": f.relevance_score,
"provenance": {
"resource_uri": f.provenance.resource_uri,
"location": f.provenance.location,
"strategy": f.provenance.strategy,
},
}
for f in assembled.fragments
],
}
def _simulate_context_assembly(
project_name: str,
acms_config: dict[str, Any],
budget_tokens: int | None = None,
focus_uris: list[str] | None = None,
strategy_hints: list[str] | None = None,
view: str = "default",
) -> AssembledContext:
"""Run a dry-run context assembly using the CRP models.
This creates a simulated AssembledContext based on available
backend data and the project's ACMS configuration. When the
full ACMS pipeline is wired, this will delegate to the live
pipeline instead.
"""
effective_budget = budget_tokens or acms_config.get(
"hot_max_tokens", _DEFAULT_BUDGET_TOKENS
)
effective_strategies = strategy_hints or acms_config.get("strategies", [])
default_breadth = acms_config.get("default_breadth", _DEFAULT_BREADTH)
default_depth = acms_config.get("default_depth", _DEFAULT_DEPTH)
temporal_str = acms_config.get("temporal_scope", "current")
# Resolve temporal scope
temporal = TemporalScope.CURRENT
if temporal_str == "recent":
temporal = TemporalScope.RECENT
elif temporal_str == "all":
temporal = TemporalScope.ALL
# Build the context request
_request = ContextRequest(
query=f"Context assembly for project {project_name} (view: {view})",
focus=focus_uris or [],
breadth=default_breadth,
depth=default_depth,
temporal=temporal,
max_tokens=effective_budget,
preferred_strategies=effective_strategies,
purpose=f"dry-run simulate for {project_name}",
)
# Query the tier service for existing fragments
tier_service = _get_context_tier_service()
project_fragments = tier_service.get_scoped_view([project_name])
# Apply focus URI filtering if specified
if focus_uris:
project_fragments = [
f
for f in project_fragments
if f.resource_id in focus_uris
or any(u in (f.resource_id or "") for u in focus_uris)
]
# Build simulated context fragments from tier data
fragments: list[ContextFragment] = []
total_tokens = 0
for tf in project_fragments:
if total_tokens + tf.token_count > effective_budget:
break
fragments.append(
ContextFragment(
uko_node=tf.resource_id or f"uko:{tf.fragment_id}",
content=tf.content[:100] + "..."
if len(tf.content) > 100
else tf.content,
detail_depth=default_depth if isinstance(default_depth, int) else 3,
token_count=tf.token_count,
relevance_score=min(1.0, tf.access_count / 10.0)
if tf.access_count > 0
else 0.5,
provenance=FragmentProvenance(
resource_uri=tf.resource_id or "unknown",
location="",
strategy="tier_retrieval",
),
)
)
total_tokens += tf.token_count
budget_used = total_tokens / effective_budget if effective_budget > 0 else 0.0
content_for_hash = f"{project_name}:{view}:{total_tokens}"
context_hash = hashlib.sha256(content_for_hash.encode()).hexdigest()[:16]
strategies_used = (
effective_strategies if effective_strategies else ["tier_retrieval"]
)
return AssembledContext(
fragments=fragments,
total_tokens=total_tokens,
budget_used=min(budget_used, 1.0),
strategies_used=strategies_used,
context_hash=context_hash,
preamble=(
f"Simulated context for {project_name}"
f" (view: {view}, budget: {effective_budget} tokens)"
),
)
# ------------------------------------------------------------------
# Commands
# ------------------------------------------------------------------
@@ -213,6 +444,90 @@ def context_set(
help="Default execution environment for this project (host or container)",
),
] = None,
hot_max_tokens: Annotated[
int | None,
typer.Option(
"--hot-max-tokens",
help="Max tokens for the hot tier context window",
),
] = None,
warm_max_decisions: Annotated[
int | None,
typer.Option(
"--warm-max-decisions",
help="Max decisions for the warm tier",
),
] = None,
cold_max_decisions: Annotated[
int | None,
typer.Option(
"--cold-max-decisions",
help="Max decisions for the cold tier",
),
] = None,
query_limit: Annotated[
int | None,
typer.Option(
"--query-limit",
help="Max number of fragments per query",
),
] = None,
summarize: Annotated[
bool | None,
typer.Option(
"--summarize/--no-summarize",
help="Enable or disable context summarization",
),
] = None,
summary_max_tokens: Annotated[
int | None,
typer.Option(
"--summary-max-tokens",
help="Max tokens for context summaries",
),
] = None,
strategy: Annotated[
list[str] | None,
typer.Option(
"--strategy",
help="Preferred context strategy (repeatable)",
),
] = None,
default_breadth: Annotated[
int | None,
typer.Option(
"--default-breadth",
help="Default breadth for context retrieval",
),
] = None,
default_depth: Annotated[
str | None,
typer.Option(
"--default-depth",
help="Default depth (integer or named level)",
),
] = None,
skeleton_ratio: Annotated[
float | None,
typer.Option(
"--skeleton-ratio",
help="Skeleton compression ratio (0.0-1.0)",
),
] = None,
temporal_scope: Annotated[
str | None,
typer.Option(
"--temporal-scope",
help="Temporal scope: current, recent, or all",
),
] = None,
auto_refresh: Annotated[
bool | None,
typer.Option(
"--auto-refresh/--no-auto-refresh",
help="Auto-refresh context on resource changes",
),
] = None,
clear: Annotated[
bool,
typer.Option(
@@ -225,7 +540,12 @@ def context_set(
typer.Option("--format", "-f", help=_FORMAT_HELP),
] = "rich",
) -> None:
"""Set context policy for a project phase view."""
"""Set context policy for a project phase view.
Configures both the context view filtering (resource/path
include/exclude) and the ACMS pipeline parameters (tier budgets,
strategies, temporal scope, etc.) for a project.
"""
if view not in VALID_PHASES:
err_console.print(
f"[red]Invalid view '{view}': must be one of {PHASE_NAMES}[/red]"
@@ -246,6 +566,7 @@ def context_set(
raise typer.Exit(1) from exc
policy = _read_policy(session_factory, project)
acms = _read_acms_config(session_factory, project)
if clear:
# Reset the view to None (inherit from parent)
@@ -264,7 +585,43 @@ def context_set(
)
policy = policy.model_copy(update={f"{view}_view": new_view})
_write_policy(session_factory, project, policy)
# Update ACMS config with any provided overrides
if hot_max_tokens is not None:
acms["hot_max_tokens"] = hot_max_tokens
if warm_max_decisions is not None:
acms["warm_max_decisions"] = warm_max_decisions
if cold_max_decisions is not None:
acms["cold_max_decisions"] = cold_max_decisions
if query_limit is not None:
acms["query_limit"] = query_limit
if summarize is not None:
acms["summarize"] = summarize
if summary_max_tokens is not None:
acms["summary_max_tokens"] = summary_max_tokens
if strategy is not None:
acms["strategies"] = strategy
if default_breadth is not None:
acms["default_breadth"] = default_breadth
if default_depth is not None:
# Parse as int if possible, otherwise keep as string
try:
acms["default_depth"] = int(default_depth)
except ValueError:
acms["default_depth"] = default_depth
if skeleton_ratio is not None:
acms["skeleton_ratio"] = skeleton_ratio
if temporal_scope is not None:
if temporal_scope not in ("current", "recent", "all"):
err_console.print(
f"[red]Invalid temporal scope '{temporal_scope}': "
f"must be current, recent, or all[/red]"
)
raise typer.Exit(1)
acms["temporal_scope"] = temporal_scope
if auto_refresh is not None:
acms["auto_refresh"] = auto_refresh
_write_policy(session_factory, project, policy, acms)
# Handle execution_environment at the project level
if execution_environment is not None:
@@ -285,11 +642,13 @@ def context_set(
project,
{
**policy.model_dump(mode="json"),
"acms_config": acms,
"execution_environment": execution_environment.lower(),
},
)
data = _policy_to_dict(policy)
data["acms_config"] = acms
if output_format.lower() == OutputFormat.RICH:
console.print(
Panel(
@@ -323,7 +682,7 @@ def context_show(
typer.Option("--format", "-f", help=_FORMAT_HELP),
] = "rich",
) -> None:
"""Show the context policy for a project."""
"""Show the context policy and ACMS pipeline config for a project."""
from cleveragents.application.container import get_container
container = get_container()
@@ -338,6 +697,7 @@ def context_show(
raise typer.Exit(1) from exc
policy = _read_policy(session_factory, project)
acms = _read_acms_config(session_factory, project)
if view is not None:
if view not in VALID_PHASES:
@@ -349,9 +709,11 @@ def context_show(
data: dict[str, Any] = {
"phase": view,
"resolved_view": resolved.model_dump(mode="json"),
"acms_config": acms,
}
else:
data = _policy_to_dict(policy)
data["acms_config"] = acms
if output_format.lower() == OutputFormat.RICH:
title = f"Context Policy: {project}"
@@ -387,6 +749,34 @@ def context_show(
lines.append(f"[bold]{phase}:[/bold] configured")
else:
lines.append(f"[bold]{phase}:[/bold] (inherits)")
# ACMS pipeline config section
lines.append("")
lines.append("[bold]ACMS Pipeline Config:[/bold]")
lines.append(
f" Hot max tokens: {acms.get('hot_max_tokens', _DEFAULT_HOT_MAX_TOKENS)}"
)
lines.append(
f" Warm max decisions: "
f"{acms.get('warm_max_decisions', _DEFAULT_WARM_MAX_DECISIONS)}"
)
lines.append(
f" Cold max decisions: "
f"{acms.get('cold_max_decisions', _DEFAULT_COLD_MAX_DECISIONS)}"
)
lines.append(f" Summarize: {acms.get('summarize', True)}")
lines.append(f" Temporal scope: {acms.get('temporal_scope', 'current')}")
lines.append(f" Auto-refresh: {acms.get('auto_refresh', True)}")
strategies = acms.get("strategies", [])
lines.append(
f" Strategies: {', '.join(strategies) if strategies else '(default)'}"
)
lines.append(
f" Default breadth: {acms.get('default_breadth', _DEFAULT_BREADTH)}"
)
lines.append(f" Default depth: {acms.get('default_depth', _DEFAULT_DEPTH)}")
lines.append(
f" Skeleton ratio: {acms.get('skeleton_ratio', _DEFAULT_SKELETON_RATIO)}"
)
console.print(
Panel(
"\n".join(lines),
@@ -404,17 +794,192 @@ def context_inspect(
str,
typer.Argument(help="Project namespaced name"),
],
view: Annotated[
str | None,
typer.Option(
"--view",
"-v",
help="Phase view to inspect: default, strategize, execute, or apply",
),
] = None,
strategy_filter: Annotated[
str | None,
typer.Option(
"--strategy",
help="Filter fragments by strategy name",
),
] = None,
focus: Annotated[
list[str] | None,
typer.Option(
"--focus",
help="UKO URIs to focus on (repeatable)",
),
] = None,
breadth: Annotated[
int | None,
typer.Option(
"--breadth",
help="Override breadth for context retrieval",
),
] = None,
depth: Annotated[
str | None,
typer.Option(
"--depth",
help="Override depth (integer or named level)",
),
] = None,
output_format: Annotated[
str,
typer.Option("--format", "-f", help=_FORMAT_HELP),
] = "rich",
) -> None:
"""Inspect effective context for a project.
"""Inspect effective context state for a project.
NOTE: This command requires ACMS wiring and is not yet
implemented.
Queries the ACMS tier service to display the current context
fragments across hot/warm/cold tiers, tier metrics, budget
utilisation, and the effective ACMS pipeline configuration.
"""
raise NotImplementedError(
"'agents project context inspect' requires ACMS "
"wiring and is not yet implemented. "
"See ADR-014 for the planned integration."
)
if view is not None and view not in VALID_PHASES:
err_console.print(
f"[red]Invalid view '{view}': must be one of {PHASE_NAMES}[/red]"
)
raise typer.Exit(1)
from cleveragents.application.container import get_container
container = get_container()
session_factory = container.session_factory()
# Validate project exists
repo = _get_namespaced_project_repo()
try:
repo.get(project)
except Exception as exc:
err_console.print(f"[red]Project not found:[/red] {project}")
raise typer.Exit(1) from exc
# Query the tier service (project-scoped fragment counts)
tier_service = _get_context_tier_service()
metrics: TierMetrics = tier_service.get_scoped_metrics([project])
budget = tier_service.budget
# Get project-scoped fragments
project_fragments: list[TieredFragment] = tier_service.get_scoped_view([project])
# Apply optional filters
if strategy_filter is not None:
# Filter fragments whose metadata or resource_id relates to the strategy
project_fragments = [
f
for f in project_fragments
if f.metadata.get("strategy") == strategy_filter
or strategy_filter in (f.resource_id or "")
]
if focus is not None:
# Filter fragments whose resource_id matches any focus URI
project_fragments = [
f
for f in project_fragments
if f.resource_id in focus or any(u in (f.resource_id or "") for u in focus)
]
# Get actor views for each role
actor_views: dict[str, int] = {}
for role in ActorRole:
role_frags = tier_service.get_for_actor(role, [project])
actor_views[role.value] = len(role_frags)
# Read ACMS config and apply overrides
acms = _read_acms_config(session_factory, project)
if breadth is not None:
acms["default_breadth"] = breadth
if depth is not None:
try:
acms["default_depth"] = int(depth)
except ValueError:
acms["default_depth"] = depth
# Build result
data: dict[str, Any] = {
"project": project,
"view": view or "all",
"tier_metrics": _tier_metrics_to_dict(metrics),
"tier_budget": {
"max_tokens_hot": budget.max_tokens_hot,
"max_decisions_warm": budget.max_decisions_warm,
"max_decisions_cold": budget.max_decisions_cold,
},
"project_fragments": {
"total": len(project_fragments),
"by_tier": {
"hot": len([f for f in project_fragments if f.tier == ContextTier.HOT]),
"warm": len(
[f for f in project_fragments if f.tier == ContextTier.WARM]
),
"cold": len(
[f for f in project_fragments if f.tier == ContextTier.COLD]
),
},
},
"actor_visibility": actor_views,
"acms_config": acms,
}
# Add fragment details (limited to first 20)
fragment_details = [_fragment_to_dict(f) for f in project_fragments[:20]]
data["fragments"] = fragment_details
if output_format.lower() == OutputFormat.RICH:
title = f"Context State: {project}"
if view:
title += f" (phase: {view})"
# Metrics table
metrics_table = Table(title="Tier Metrics", expand=False)
metrics_table.add_column("Metric", style="bold")
metrics_table.add_column("Value", justify="right")
metrics_table.add_row("Hot fragments", str(metrics.hot_count))
metrics_table.add_row("Warm fragments", str(metrics.warm_count))
metrics_table.add_row("Cold fragments", str(metrics.cold_count))
metrics_table.add_row("Hot hits", str(metrics.hot_hit_count))
metrics_table.add_row("Hot misses", str(metrics.hot_miss_count))
metrics_table.add_row("Warm hits", str(metrics.warm_hit_count))
metrics_table.add_row("Warm misses", str(metrics.warm_miss_count))
console.print(Panel(metrics_table, title=title, expand=False))
# Budget info
budget_lines: list[str] = []
budget_lines.append(f"[bold]Hot max tokens:[/bold] {budget.max_tokens_hot}")
budget_lines.append(
f"[bold]Warm max decisions:[/bold] {budget.max_decisions_warm}"
)
budget_lines.append(
f"[bold]Cold max decisions:[/bold] {budget.max_decisions_cold}"
)
console.print(Panel("\n".join(budget_lines), title="Tier Budget", expand=False))
# Project fragments summary
by_tier = data["project_fragments"]["by_tier"]
console.print(
Panel(
f"Total: {data['project_fragments']['total']} | "
f"Hot: {by_tier['hot']} | "
f"Warm: {by_tier['warm']} | "
f"Cold: {by_tier['cold']}",
title=f"Project Fragments: {project}",
expand=False,
)
)
# Actor visibility
av_lines = [f"[bold]{r}:[/bold] {c} fragments" for r, c in actor_views.items()]
console.print(
Panel("\n".join(av_lines), title="Actor Visibility", expand=False)
)
else:
console.print(format_output(data, output_format))
@app.command(name="simulate")
@@ -423,14 +988,116 @@ def context_simulate(
str,
typer.Argument(help="Project namespaced name"),
],
view: Annotated[
str | None,
typer.Option(
"--view",
"-v",
help="Phase view for simulation: default, strategize, execute, or apply",
),
] = None,
budget: Annotated[
int | None,
typer.Option(
"--budget",
help="Token budget for simulation (overrides project config)",
),
] = None,
focus: Annotated[
list[str] | None,
typer.Option(
"--focus",
help="UKO URIs to focus on (repeatable)",
),
] = None,
strategy_hint: Annotated[
list[str] | None,
typer.Option(
"--strategy",
help="Preferred strategies for simulation (repeatable)",
),
] = None,
output_format: Annotated[
str,
typer.Option("--format", "-f", help=_FORMAT_HELP),
] = "rich",
) -> None:
"""Simulate context window for a project.
"""Simulate context window assembly for a project.
NOTE: This command requires ACMS wiring and is not yet
implemented.
Runs a dry-run context assembly using the project's ACMS
configuration and displays the assembled fragments, budget
usage, and strategy selection.
"""
raise NotImplementedError(
"'agents project context simulate' requires ACMS "
"wiring and is not yet implemented. "
"See ADR-014 for the planned integration."
effective_view = view or "default"
if effective_view not in VALID_PHASES:
err_console.print(
f"[red]Invalid view '{effective_view}': must be one of {PHASE_NAMES}[/red]"
)
raise typer.Exit(1)
from cleveragents.application.container import get_container
container = get_container()
session_factory = container.session_factory()
# Validate project exists
repo = _get_namespaced_project_repo()
try:
repo.get(project)
except Exception as exc:
err_console.print(f"[red]Project not found:[/red] {project}")
raise typer.Exit(1) from exc
acms = _read_acms_config(session_factory, project)
assembled = _simulate_context_assembly(
project_name=project,
acms_config=acms,
budget_tokens=budget,
focus_uris=focus,
strategy_hints=strategy_hint,
view=effective_view,
)
data = _assembled_to_dict(assembled)
data["project"] = project
data["view"] = effective_view
data["acms_config"] = acms
if output_format.lower() == OutputFormat.RICH:
title = f"Context Simulation: {project} (view: {effective_view})"
# Summary panel
summary_lines: list[str] = []
summary_lines.append(f"[bold]Total tokens:[/bold] {assembled.total_tokens}")
summary_lines.append(f"[bold]Budget used:[/bold] {assembled.budget_used:.1%}")
summary_lines.append(
f"[bold]Strategies:[/bold] "
f"{', '.join(assembled.strategies_used) or '(none)'}"
)
summary_lines.append(f"[bold]Fragment count:[/bold] {len(assembled.fragments)}")
summary_lines.append(f"[bold]Context hash:[/bold] {assembled.context_hash}")
if assembled.preamble:
summary_lines.append(f"[bold]Preamble:[/bold] {assembled.preamble}")
console.print(Panel("\n".join(summary_lines), title=title, expand=False))
# Fragment table
if assembled.fragments:
frag_table = Table(title="Assembled Fragments", expand=False)
frag_table.add_column("#", style="dim", justify="right")
frag_table.add_column("UKO Node")
frag_table.add_column("Tokens", justify="right")
frag_table.add_column("Relevance", justify="right")
frag_table.add_column("Depth", justify="right")
frag_table.add_column("Strategy")
for idx, frag in enumerate(assembled.fragments, 1):
frag_table.add_row(
str(idx),
frag.uko_node,
str(frag.token_count),
f"{frag.relevance_score:.2f}",
str(frag.detail_depth),
frag.provenance.strategy,
)
console.print(frag_table)
else:
console.print(format_output(data, output_format))
src/cleveragents/shared/redaction.py
+2
View File
@@ -48,6 +48,8 @@ _FALSE_POSITIVE_KEYS: set[str] = {
"prompt_tokens",
"completion_tokens",
"token_estimate",
"hot_max_tokens",
"summary_max_tokens",
"auth_method",
"auth_type",
"auth_enabled",