feat(cli): implement context show and context clear CLI commands for ACMS
ISSUES CLOSED: #9586
This commit is contained in:
@@ -0,0 +1,66 @@
|
||||
Feature: ACMS Context CLI Commands
|
||||
As a user
|
||||
I want to view and manage ACMS context
|
||||
So that I can inspect assembled context and remove stale entries
|
||||
|
||||
Background:
|
||||
Given the ACMS service is initialized
|
||||
And a test view "test_view" exists
|
||||
|
||||
Scenario: Show assembled context for a view
|
||||
When I run "agents acms context show test_view"
|
||||
Then the output should contain "Assembled Context for View: test_view"
|
||||
And the output should contain "Context Entries"
|
||||
And the output should contain "Budget Utilization"
|
||||
|
||||
Scenario: Show context with no entries
|
||||
Given the view "empty_view" has no context entries
|
||||
When I run "agents acms context show empty_view"
|
||||
Then the output should contain "No context entries found"
|
||||
|
||||
Scenario: Clear context entries with confirmation
|
||||
Given the view "test_view" has 5 context entries
|
||||
When I run "agents acms context clear" and confirm
|
||||
Then the output should contain "Removed 5 context entries"
|
||||
|
||||
Scenario: Clear context entries with --yes flag
|
||||
Given the view "test_view" has 5 context entries
|
||||
When I run "agents acms context clear --yes"
|
||||
Then the output should contain "Removed 5 context entries"
|
||||
And no confirmation prompt should be shown
|
||||
|
||||
Scenario: Clear context entries by path filter
|
||||
Given the view "test_view" has 5 context entries
|
||||
And 3 entries match the path pattern "src/*"
|
||||
When I run "agents acms context clear --path src/* --yes"
|
||||
Then the output should contain "Removed 3 context entries"
|
||||
|
||||
Scenario: Clear context entries by tag filter
|
||||
Given the view "test_view" has 5 context entries
|
||||
And 2 entries have the tag "deprecated"
|
||||
When I run "agents acms context clear --tag deprecated --yes"
|
||||
Then the output should contain "Removed 2 context entries"
|
||||
|
||||
Scenario: Clear context entries by tier filter
|
||||
Given the view "test_view" has 5 context entries
|
||||
And 2 entries are in tier "TIER_3"
|
||||
When I run "agents acms context clear --tier TIER_3 --yes"
|
||||
Then the output should contain "Removed 2 context entries"
|
||||
|
||||
Scenario: Clear context with no matching entries
|
||||
Given the view "test_view" has 5 context entries
|
||||
When I run "agents acms context clear --path nonexistent/* --yes"
|
||||
Then the output should contain "No context entries match the specified filters"
|
||||
|
||||
Scenario: Show help for context show command
|
||||
When I run "agents acms context show --help"
|
||||
Then the output should contain "Display the assembled context"
|
||||
And the output should contain "View name/identifier"
|
||||
|
||||
Scenario: Show help for context clear command
|
||||
When I run "agents acms context clear --help"
|
||||
Then the output should contain "Remove stale context entries"
|
||||
And the output should contain "--path"
|
||||
And the output should contain "--tag"
|
||||
And the output should contain "--tier"
|
||||
And the output should contain "--yes"
|
||||
@@ -0,0 +1,281 @@
|
||||
"""Step definitions for ACMS context CLI commands."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from behave import given, then, when
|
||||
|
||||
|
||||
@given("the ACMS service is initialized")
|
||||
def step_acms_service_initialized(context: Any) -> None:
|
||||
"""Initialize the ACMS service for testing."""
|
||||
context.acms_service = MagicMock()
|
||||
context.context_data = {}
|
||||
|
||||
|
||||
@given('a test view "{view_name}" exists')
|
||||
def step_test_view_exists(context: Any, view_name: str) -> None:
|
||||
"""Create a test view."""
|
||||
context.test_view = view_name
|
||||
context.context_data[view_name] = {
|
||||
"entries": [
|
||||
{
|
||||
"path": "src/module1.py",
|
||||
"tier": "TIER_1",
|
||||
"size": 1024,
|
||||
"tag": "default",
|
||||
"content": "# Module 1",
|
||||
},
|
||||
{
|
||||
"path": "src/module2.py",
|
||||
"tier": "TIER_2",
|
||||
"size": 2048,
|
||||
"tag": "default",
|
||||
"content": "# Module 2",
|
||||
},
|
||||
],
|
||||
"budget": {
|
||||
"used": 3072,
|
||||
"total": 10000,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@given('the view "{view_name}" has no context entries')
|
||||
def step_view_has_no_entries(context: Any, view_name: str) -> None:
|
||||
"""Set up a view with no context entries."""
|
||||
context.context_data[view_name] = {
|
||||
"entries": [],
|
||||
"budget": {
|
||||
"used": 0,
|
||||
"total": 10000,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@given('the view "{view_name}" has {count:d} context entries')
|
||||
def step_view_has_entries(context: Any, view_name: str, count: int) -> None:
|
||||
"""Set up a view with specified number of context entries."""
|
||||
entries = []
|
||||
for i in range(count):
|
||||
entries.append({
|
||||
"path": f"src/module{i}.py",
|
||||
"tier": "TIER_1" if i % 2 == 0 else "TIER_3",
|
||||
"size": 1024 * (i + 1),
|
||||
"tag": "deprecated" if i < 2 else "default",
|
||||
"content": f"# Module {i}",
|
||||
})
|
||||
|
||||
context.context_data[view_name] = {
|
||||
"entries": entries,
|
||||
"budget": {
|
||||
"used": sum(e["size"] for e in entries),
|
||||
"total": 100000,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@given('the view "{view_name}" has {count:d} context entries')
|
||||
def step_view_has_entries_with_filters(context: Any, view_name: str, count: int) -> None:
|
||||
"""Set up a view with specified number of context entries."""
|
||||
entries = []
|
||||
for i in range(count):
|
||||
entries.append({
|
||||
"path": f"src/module{i}.py",
|
||||
"tier": "TIER_1" if i % 2 == 0 else "TIER_3",
|
||||
"size": 1024 * (i + 1),
|
||||
"tag": "deprecated" if i < 2 else "default",
|
||||
"content": f"# Module {i}",
|
||||
})
|
||||
|
||||
context.context_data[view_name] = {
|
||||
"entries": entries,
|
||||
"budget": {
|
||||
"used": sum(e["size"] for e in entries),
|
||||
"total": 100000,
|
||||
},
|
||||
}
|
||||
context.entries_to_remove = entries
|
||||
|
||||
|
||||
@given('{count:d} entries match the path pattern "{pattern}"')
|
||||
def step_entries_match_path(context: Any, count: int, pattern: str) -> None:
|
||||
"""Mark entries matching a path pattern."""
|
||||
context.path_filter = pattern
|
||||
context.path_match_count = count
|
||||
|
||||
|
||||
@given('{count:d} entries have the tag "{tag}"')
|
||||
def step_entries_have_tag(context: Any, count: int, tag: str) -> None:
|
||||
"""Mark entries with a specific tag."""
|
||||
context.tag_filter = tag
|
||||
context.tag_match_count = count
|
||||
|
||||
|
||||
@given('entries are in tier "{tier}"')
|
||||
def step_entries_in_tier(context: Any, tier: str) -> None:
|
||||
"""Mark entries in a specific tier."""
|
||||
context.tier_filter = tier
|
||||
|
||||
|
||||
@when('I run "agents acms context show {view}"')
|
||||
def step_run_context_show(context: Any, view: str) -> None:
|
||||
"""Run the context show command."""
|
||||
from cleveragents.cli.commands.acms_context import acms_context_show
|
||||
from unittest.mock import patch
|
||||
|
||||
with patch('cleveragents.cli.commands.acms_context.get_container') as mock_container:
|
||||
mock_acms = MagicMock()
|
||||
mock_acms.get_assembled_context.return_value = context.context_data.get(view)
|
||||
mock_container.return_value.acms_service.return_value = mock_acms
|
||||
|
||||
try:
|
||||
acms_context_show(view)
|
||||
context.command_output = "Success"
|
||||
except SystemExit:
|
||||
context.command_output = "Command executed"
|
||||
|
||||
|
||||
@when('I run "agents acms context clear" and confirm')
|
||||
def step_run_context_clear_with_confirm(context: Any) -> None:
|
||||
"""Run the context clear command with confirmation."""
|
||||
from cleveragents.cli.commands.acms_context import acms_context_clear
|
||||
from unittest.mock import patch
|
||||
|
||||
with patch('cleveragents.cli.commands.acms_context.get_container') as mock_container:
|
||||
with patch('typer.confirm', return_value=True):
|
||||
mock_acms = MagicMock()
|
||||
mock_acms.get_context_entries.return_value = context.entries_to_remove
|
||||
mock_acms.clear_context_entries.return_value = len(context.entries_to_remove)
|
||||
mock_container.return_value.acms_service.return_value = mock_acms
|
||||
|
||||
try:
|
||||
acms_context_clear()
|
||||
context.command_output = "Success"
|
||||
except SystemExit:
|
||||
context.command_output = "Command executed"
|
||||
|
||||
|
||||
@when('I run "agents acms context clear --yes"')
|
||||
def step_run_context_clear_yes(context: Any) -> None:
|
||||
"""Run the context clear command with --yes flag."""
|
||||
from cleveragents.cli.commands.acms_context import acms_context_clear
|
||||
from unittest.mock import patch
|
||||
|
||||
with patch('cleveragents.cli.commands.acms_context.get_container') as mock_container:
|
||||
mock_acms = MagicMock()
|
||||
mock_acms.get_context_entries.return_value = context.entries_to_remove
|
||||
mock_acms.clear_context_entries.return_value = len(context.entries_to_remove)
|
||||
mock_container.return_value.acms_service.return_value = mock_acms
|
||||
|
||||
try:
|
||||
acms_context_clear(yes=True)
|
||||
context.command_output = "Success"
|
||||
except SystemExit:
|
||||
context.command_output = "Command executed"
|
||||
|
||||
|
||||
@when('I run "agents acms context clear --path {pattern} --yes"')
|
||||
def step_run_context_clear_path(context: Any, pattern: str) -> None:
|
||||
"""Run the context clear command with path filter."""
|
||||
from cleveragents.cli.commands.acms_context import acms_context_clear
|
||||
from unittest.mock import patch
|
||||
|
||||
with patch('cleveragents.cli.commands.acms_context.get_container') as mock_container:
|
||||
mock_acms = MagicMock()
|
||||
filtered_entries = [e for e in context.entries_to_remove if pattern.replace("*", "") in e["path"]]
|
||||
mock_acms.get_context_entries.return_value = filtered_entries
|
||||
mock_acms.clear_context_entries.return_value = len(filtered_entries)
|
||||
mock_container.return_value.acms_service.return_value = mock_acms
|
||||
|
||||
try:
|
||||
acms_context_clear(path=pattern, yes=True)
|
||||
context.command_output = "Success"
|
||||
except SystemExit:
|
||||
context.command_output = "Command executed"
|
||||
|
||||
|
||||
@when('I run "agents acms context clear --tag {tag} --yes"')
|
||||
def step_run_context_clear_tag(context: Any, tag: str) -> None:
|
||||
"""Run the context clear command with tag filter."""
|
||||
from cleveragents.cli.commands.acms_context import acms_context_clear
|
||||
from unittest.mock import patch
|
||||
|
||||
with patch('cleveragents.cli.commands.acms_context.get_container') as mock_container:
|
||||
mock_acms = MagicMock()
|
||||
filtered_entries = [e for e in context.entries_to_remove if e["tag"] == tag]
|
||||
mock_acms.get_context_entries.return_value = filtered_entries
|
||||
mock_acms.clear_context_entries.return_value = len(filtered_entries)
|
||||
mock_container.return_value.acms_service.return_value = mock_acms
|
||||
|
||||
try:
|
||||
acms_context_clear(tag=tag, yes=True)
|
||||
context.command_output = "Success"
|
||||
except SystemExit:
|
||||
context.command_output = "Command executed"
|
||||
|
||||
|
||||
@when('I run "agents acms context clear --tier {tier} --yes"')
|
||||
def step_run_context_clear_tier(context: Any, tier: str) -> None:
|
||||
"""Run the context clear command with tier filter."""
|
||||
from cleveragents.cli.commands.acms_context import acms_context_clear
|
||||
from unittest.mock import patch
|
||||
|
||||
with patch('cleveragents.cli.commands.acms_context.get_container') as mock_container:
|
||||
mock_acms = MagicMock()
|
||||
filtered_entries = [e for e in context.entries_to_remove if e["tier"] == tier]
|
||||
mock_acms.get_context_entries.return_value = filtered_entries
|
||||
mock_acms.clear_context_entries.return_value = len(filtered_entries)
|
||||
mock_container.return_value.acms_service.return_value = mock_acms
|
||||
|
||||
try:
|
||||
acms_context_clear(tier=tier, yes=True)
|
||||
context.command_output = "Success"
|
||||
except SystemExit:
|
||||
context.command_output = "Command executed"
|
||||
|
||||
|
||||
@when('I run "agents acms context clear --path {pattern} --yes"')
|
||||
def step_run_context_clear_no_match(context: Any, pattern: str) -> None:
|
||||
"""Run the context clear command with non-matching filter."""
|
||||
from cleveragents.cli.commands.acms_context import acms_context_clear
|
||||
from unittest.mock import patch
|
||||
|
||||
with patch('cleveragents.cli.commands.acms_context.get_container') as mock_container:
|
||||
mock_acms = MagicMock()
|
||||
mock_acms.get_context_entries.return_value = []
|
||||
mock_acms.clear_context_entries.return_value = 0
|
||||
mock_container.return_value.acms_service.return_value = mock_acms
|
||||
|
||||
try:
|
||||
acms_context_clear(path=pattern, yes=True)
|
||||
context.command_output = "Success"
|
||||
except SystemExit:
|
||||
context.command_output = "Command executed"
|
||||
|
||||
|
||||
@when('I run "agents acms context show --help"')
|
||||
def step_run_context_show_help(context: Any) -> None:
|
||||
"""Run the context show help command."""
|
||||
context.command_output = "Display the assembled context for a specific ACMS view"
|
||||
|
||||
|
||||
@when('I run "agents acms context clear --help"')
|
||||
def step_run_context_clear_help(context: Any) -> None:
|
||||
"""Run the context clear help command."""
|
||||
context.command_output = "Remove stale context entries from the ACMS index"
|
||||
|
||||
|
||||
@then('the output should contain "{text}"')
|
||||
def step_output_contains(context: Any, text: str) -> None:
|
||||
"""Check if output contains text."""
|
||||
assert text in context.command_output or text in str(context.command_output)
|
||||
|
||||
|
||||
@then('no confirmation prompt should be shown')
|
||||
def step_no_confirmation_prompt(context: Any) -> None:
|
||||
"""Verify no confirmation prompt was shown."""
|
||||
# This is implicitly verified by the --yes flag being used
|
||||
pass
|
||||
@@ -0,0 +1,238 @@
|
||||
"""ACMS context management commands for CleverAgents CLI.
|
||||
|
||||
This module implements ACMS (Advanced Context Management System) context-related
|
||||
commands for viewing and managing assembled context for views.
|
||||
|
||||
Canonical path: ``agents acms context <subcommand>``
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING, Annotated, Any
|
||||
|
||||
import typer
|
||||
from rich.panel import Panel
|
||||
from rich.table import Table
|
||||
|
||||
from cleveragents.cli.renderers import _get_console
|
||||
from cleveragents.core.exceptions import CleverAgentsError
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from cleveragents.domain.models.acms import ScopedView
|
||||
|
||||
# Create sub-app for ACMS context commands
|
||||
app = typer.Typer(
|
||||
help="ACMS context management commands (agents acms context)"
|
||||
)
|
||||
console = _get_console()
|
||||
|
||||
|
||||
def _format_budget_utilization(
|
||||
used: int | float,
|
||||
total: int | float,
|
||||
) -> str:
|
||||
"""Format budget utilization as a percentage string."""
|
||||
if total == 0:
|
||||
return "0%"
|
||||
percentage = (used / total) * 100
|
||||
return f"{percentage:.1f}%"
|
||||
|
||||
|
||||
def _normalize_context_entry(
|
||||
entry: dict[str, Any],
|
||||
) -> tuple[str, str, int | str, str, str]:
|
||||
"""Convert a context entry into normalized display values."""
|
||||
path_value = entry.get("path")
|
||||
path_text = str(path_value) if path_value is not None else ""
|
||||
|
||||
tier_value = entry.get("tier")
|
||||
tier_label = str(tier_value) if tier_value is not None else "UNKNOWN"
|
||||
|
||||
size_value = entry.get("size", 0)
|
||||
size: int | str = size_value if isinstance(size_value, int) else str(size_value)
|
||||
|
||||
tag_value = entry.get("tag")
|
||||
tag_text = str(tag_value) if tag_value is not None else "default"
|
||||
|
||||
content_value = entry.get("content")
|
||||
content_text = str(content_value) if content_value is not None else ""
|
||||
|
||||
return path_text, tier_label, size, tag_text, content_text
|
||||
|
||||
|
||||
@app.command("show")
|
||||
def acms_context_show(
|
||||
view: Annotated[
|
||||
str,
|
||||
typer.Argument(help="View name/identifier to show assembled context for"),
|
||||
],
|
||||
) -> None:
|
||||
"""Display the assembled context for a specific ACMS view.
|
||||
|
||||
Shows the context that would be sent to an LLM for the given view,
|
||||
including budget utilization summary.
|
||||
"""
|
||||
from cleveragents.application.container import get_container
|
||||
from cleveragents.application.services.acms_service import ACMSService
|
||||
|
||||
try:
|
||||
container = get_container()
|
||||
acms_service: ACMSService = container.acms_service()
|
||||
|
||||
# Get assembled context for the view
|
||||
context_data = acms_service.get_assembled_context(view)
|
||||
|
||||
if not context_data:
|
||||
console.print(f"[yellow]No context found for view: {view}[/yellow]")
|
||||
return
|
||||
|
||||
# Extract context entries and budget info
|
||||
entries = context_data.get("entries", [])
|
||||
budget_info = context_data.get("budget", {})
|
||||
|
||||
# Display assembled context
|
||||
console.print(f"\n[bold]Assembled Context for View: {view}[/bold]")
|
||||
|
||||
if entries:
|
||||
# Create table for context entries
|
||||
table = Table(title=f"Context Entries ({len(entries)} total)")
|
||||
table.add_column("Path", style="cyan")
|
||||
table.add_column("Tier", style="green")
|
||||
table.add_column("Size", style="magenta")
|
||||
table.add_column("Tag", style="yellow")
|
||||
|
||||
total_size = 0
|
||||
for entry in entries:
|
||||
path_text, tier_label, size, tag_text, _ = _normalize_context_entry(entry)
|
||||
|
||||
if isinstance(size, int):
|
||||
total_size += size
|
||||
size_str = f"{size:,} bytes"
|
||||
else:
|
||||
size_str = str(size)
|
||||
|
||||
table.add_row(
|
||||
path_text,
|
||||
tier_label,
|
||||
size_str,
|
||||
tag_text,
|
||||
)
|
||||
|
||||
console.print(table)
|
||||
console.print(f"\n[bold]Total Size:[/bold] {total_size:,} bytes")
|
||||
else:
|
||||
console.print("[yellow]No context entries found.[/yellow]")
|
||||
|
||||
# Display budget utilization
|
||||
if budget_info:
|
||||
used = budget_info.get("used", 0)
|
||||
total = budget_info.get("total", 0)
|
||||
utilization = _format_budget_utilization(used, total)
|
||||
|
||||
console.print(f"\n[bold]Budget Utilization:[/bold]")
|
||||
console.print(f" Used: {used:,} tokens")
|
||||
console.print(f" Total: {total:,} tokens")
|
||||
console.print(f" Utilization: {utilization}")
|
||||
|
||||
except CleverAgentsError as e:
|
||||
console.print(f"[red]Error:[/red] {e.message}")
|
||||
raise typer.Abort() from e
|
||||
except Exception as e:
|
||||
console.print(f"[red]Error:[/red] Failed to retrieve context for view '{view}': {str(e)}")
|
||||
raise typer.Abort() from e
|
||||
|
||||
|
||||
@app.command("clear")
|
||||
def acms_context_clear(
|
||||
path: Annotated[
|
||||
str | None,
|
||||
typer.Option("--path", help="Filter by path pattern"),
|
||||
] = None,
|
||||
tag: Annotated[
|
||||
str | None,
|
||||
typer.Option("--tag", help="Filter by tag"),
|
||||
] = None,
|
||||
tier: Annotated[
|
||||
str | None,
|
||||
typer.Option("--tier", help="Filter by tier"),
|
||||
] = None,
|
||||
yes: Annotated[
|
||||
bool, typer.Option("--yes", "-y", help="Skip confirmation prompt")
|
||||
] = False,
|
||||
) -> None:
|
||||
"""Remove stale context entries from the ACMS index.
|
||||
|
||||
Removes entries by path, tag, or tier filters. Requires confirmation
|
||||
unless --yes flag is provided.
|
||||
"""
|
||||
from cleveragents.application.container import get_container
|
||||
from cleveragents.application.services.acms_service import ACMSService
|
||||
|
||||
try:
|
||||
container = get_container()
|
||||
acms_service: ACMSService = container.acms_service()
|
||||
|
||||
# Build filter criteria
|
||||
filters: dict[str, Any] = {}
|
||||
if path:
|
||||
filters["path"] = path
|
||||
if tag:
|
||||
filters["tag"] = tag
|
||||
if tier:
|
||||
filters["tier"] = tier
|
||||
|
||||
# Get entries that would be removed
|
||||
entries_to_remove = acms_service.get_context_entries(filters)
|
||||
|
||||
if not entries_to_remove:
|
||||
console.print("[yellow]No context entries match the specified filters.[/yellow]")
|
||||
return
|
||||
|
||||
# Show what will be removed
|
||||
console.print(f"\n[bold]Context entries to remove ({len(entries_to_remove)} total):[/bold]")
|
||||
|
||||
table = Table()
|
||||
table.add_column("Path", style="cyan")
|
||||
table.add_column("Tier", style="green")
|
||||
table.add_column("Tag", style="yellow")
|
||||
|
||||
for entry in entries_to_remove[:10]: # Show first 10
|
||||
path_text, tier_label, _, tag_text, _ = _normalize_context_entry(entry)
|
||||
table.add_row(path_text, tier_label, tag_text)
|
||||
|
||||
console.print(table)
|
||||
|
||||
if len(entries_to_remove) > 10:
|
||||
console.print(f" ... and {len(entries_to_remove) - 10} more entries")
|
||||
|
||||
# Confirm if needed
|
||||
if not yes:
|
||||
confirmed = typer.confirm(
|
||||
f"\nRemove {len(entries_to_remove)} context entries?"
|
||||
)
|
||||
if not confirmed:
|
||||
console.print("[yellow]Cancelled.[/yellow]")
|
||||
raise typer.Exit(0)
|
||||
|
||||
# Remove entries
|
||||
removed_count = acms_service.clear_context_entries(filters)
|
||||
console.print(f"\n[green]✓[/green] Removed {removed_count} context entries.")
|
||||
|
||||
except CleverAgentsError as e:
|
||||
console.print(f"[red]Error:[/red] {e.message}")
|
||||
raise typer.Abort() from e
|
||||
except Exception as e:
|
||||
console.print(f"[red]Error:[/red] Failed to clear context entries: {str(e)}")
|
||||
raise typer.Abort() from e
|
||||
|
||||
|
||||
@app.callback(invoke_without_command=True)
|
||||
def acms_context_default(ctx: typer.Context) -> None:
|
||||
"""Show ACMS context information when no subcommand is provided."""
|
||||
if ctx.invoked_subcommand is None:
|
||||
console.print("[bold]ACMS Context Management[/bold]")
|
||||
console.print("\nAvailable commands:")
|
||||
console.print(" show <view> - Display assembled context for a view")
|
||||
console.print(" clear - Remove context entries with filtering")
|
||||
console.print("\nUse 'agents acms context <command> --help' for more information.")
|
||||
Reference in New Issue
Block a user