Compare commits

...

1 Commits

Author SHA1 Message Date
HAL9000 e2614d8c6e feat(cli): add ACMS context index commands (list and add)
This adds three new subcommands to 'agents actor context':
- context_index: Lists ACMS-tiered fragments across hot/warm/cold tiers
- context_add_acms: Stores content as TieredFragment in the ContextTierService
- context_remove_acms: Removes a fragment from all tiers

The commands integrate with the existing ACMS infrastructure (ContextTierService, TieredFragment model) and support multiple output formats (--format).
2026-05-17 05:05:52 +00:00
+326
View File
@@ -9,6 +9,7 @@ Deprecated alias: ``agents context <subcommand>`` (emits deprecation warning)
from __future__ import annotations
import hashlib
from pathlib import Path
from typing import TYPE_CHECKING, Annotated, Any
@@ -16,6 +17,7 @@ import typer
from rich.panel import Panel
from rich.table import Table
from cleveragents.cli.formatting import OutputFormat
from cleveragents.cli.renderers import _get_console
from cleveragents.core.exceptions import (
CleverAgentsError,
@@ -25,6 +27,20 @@ from cleveragents.core.exceptions import (
if TYPE_CHECKING:
from cleveragents.domain.models.core import Context
# Console instances for ACMS commands
_err_console: Any = None
def _get_err_console() -> Any:
"""Lazily create an stderr console (for ACMS command output)."""
global _err_console
if _err_console is None:
from rich.console import Console
_err_console = Console(stderr=True)
return _err_console
# Create sub-app for context commands
app = typer.Typer(
help="Actor context management commands (canonical: agents actor context)"
@@ -872,6 +888,316 @@ def context_analyze(
typer.echo(ContextAnalysisEngine.format_text(result))
# ---------------------------------------------------------------------------
# ACMS-aware commands: ACMS context index (tier-service backed)
# ---------------------------------------------------------------------------
def _truncate(value: str, max_width: int) -> str:
"""Truncate a string to *max_width* characters, appending '...'."""
if len(value) <= max_width:
return value
return value[: max_width - 3] + "..."
_datetime_called = False # for testability
def _datetime_now() -> Any:
"""Return the current UTC datetime.
Overridable in tests by patching this function.
"""
global _datetime_called
_datetime_called = True
try:
return datetime.now(tz=UTC)
except Exception:
from datetime import datetime as _dt
return _dt.utcnow()
@app.command("index")
def context_index(
tier: Annotated[
str | None,
typer.Option(
"--tier",
"-t",
help="Filter by tier: hot, warm, cold (default: show all tiers)",
),
] = None,
fmt: Annotated[
str,
typer.Option(
"--format",
"-f",
help="Output format: rich, color, table, plain, json, yaml (default: rich)",
),
] = "rich",
) -> None:
"""List ACMS context index entries (tier-service backed).
Queries the ContextTierService to display all indexed context
fragments across hot/warm/cold tiers. Optionally filter by a
specific tier with ``--tier``.
Examples::
agents actor context index
agents actor context index --tier hot
agents actor context index --format json
"""
try:
from cleveragents.application.services.context_tiers import (
ContextTierService,
)
from cleveragents.domain.models.acms.tiers import ContextTier
tier_service = ContextTierService()
except Exception as exc:
console.print(f"[red]Error:[/red] Unable to init tier service: {exc}")
raise typer.Abort() from exc
if tier is not None and tier.lower() not in ("hot", "warm", "cold"):
_get_err_console().print(
f"[red]Error:[/red] Invalid tier '{tier}'. Must be one of: hot, warm, cold."
)
raise typer.Exit(code=1)
# Fetch fragments
all_frags = tier_service.get_all_fragments()
if tier is not None:
filter_tier = ContextTier(tier.lower())
all_frags = [f for f in all_frags if f.tier == filter_tier]
if not all_frags:
console.print("[yellow]No context index entries found.[/yellow]")
console.print("Use 'agents actor context add-acms' to add new entries.")
return
# Non-rich output delegates to format_output
if fmt != OutputFormat.RICH.value:
entries = [
{
"fragment_id": f.fragment_id,
"tier": f.tier.value,
"resource_id": f.resource_id or "N/A",
"project_name": f.project_name or "N/A",
"token_count": f.token_count,
"access_count": f.access_count,
"last_accessed": f.last_accessed.isoformat(),
}
for f in all_frags
]
console.print(format_output(entries, fmt))
return
# Rich output: table
table = Table(
title=f"ACMS Context Index ({len(all_frags)} entries)"
+ (f" [tier={tier}]" if tier else ""),
expand=False,
)
table.add_column("Fragment ID", style="cyan")
table.add_column("Tier", style="green")
table.add_column("Resource", style="magenta")
table.add_column("Project", style="blue")
table.add_column("Tokens", justify="right")
table.add_column("Accesses", justify="right")
table.add_column("Last Accessed", style="yellow")
for frag in all_frags:
table.add_row(
_truncate(frag.fragment_id, 40),
frag.tier.value,
_truncate(frag.resource_id or "N/A", 30),
_truncate(frag.project_name or "N/A", 25),
str(frag.token_count),
str(frag.access_count),
frag.last_accessed.strftime("%Y-%m-%d %H:%M"),
)
console.print(table)
@app.command("add-acms")
def context_add_acms(
content: Annotated[
str,
typer.Argument(help="Text content to add as a context fragment"),
],
resource_id: Annotated[
str | None,
typer.Option("--resource", "-r", help="Resource ID / URI for the fragment"),
] = None,
project_name: Annotated[
str | None,
typer.Option(
"--project",
"-p",
help="Project name for scoping the fragment",
),
] = None,
tier: Annotated[
str,
typer.Option(
"--tier",
"-t",
help="Target tier: hot, warm, or cold (default: hot)",
),
] = "hot",
token_count: Annotated[
int | None,
typer.Option("--tokens", help="Token count for the fragment"),
] = None,
fmt: Annotated[
str,
typer.Option(
"--format",
"-f",
help="Output format: rich, color, table, plain, json, yaml (default: rich)",
),
] = "rich",
) -> None:
"""Add a fragment to the ACMS context index.
Stores content as a TieredFragment in the ContextTierService. The
fragment is placed in the specified tier (default: hot).
Examples::
agents actor context add-acms 'Core implementation details' \\
--resource file:///src/app/main.py --project my-app \\
--tier warm --tokens 1200
"""
try:
from cleveragents.application.services.context_tiers import (
ContextTierService,
)
from cleveragents.domain.models.acms.tiers import ContextTier, TieredFragment
tier_service = ContextTierService()
except Exception as exc:
console.print(f"[red]Error:[/red] Unable to init tier service: {exc}")
raise typer.Abort() from exc
# Validate tier
try:
target_tier = ContextTier(tier.lower())
except ValueError:
_get_err_console().print(
f"[red]Error:[/red] Invalid tier '{tier}'. Must be one of: hot, warm, cold."
)
raise typer.Exit(code=1)
# Estimate token count if not provided (~4 chars per token)
if token_count is None:
token_count = max(1, len(content) // 4)
# Generate a fragment_id from resource or content hash
if resource_id:
fragment_id = (
f"{resource_id}:frag-{hashlib.sha256(content.encode()).hexdigest()[:8]}"
)
else:
fragment_id = f"frag-{hashlib.sha256(content.encode()).hexdigest()[:12]}"
# Build the TieredFragment
now = _datetime_now()
fragment = TieredFragment(
fragment_id=fragment_id,
content=content,
tier=target_tier,
resource_id=resource_id or "",
project_name=project_name or "",
token_count=token_count,
last_accessed=now,
access_count=0,
created_at=now,
)
# Store the fragment (will replace if fragment_id already exists)
tier_service.store(fragment)
if fmt != OutputFormat.RICH.value:
data = {
"fragment_added": {
"fragment_id": fragment.fragment_id,
"tier": fragment.tier.value,
"resource_id": fragment.resource_id or "N/A",
"project_name": fragment.project_name or "N/A",
"token_count": fragment.token_count,
}
}
console.print(format_output(data, fmt))
else:
console.print(
Panel(
(
f"[bold]Fragment ID:[/bold] {fragment_id}\n"
f"[bold]Tier:[/bold] {target_tier.value}\n"
f"[bold]Resource:[/bold] {resource_id or 'N/A'}\n"
f"[bold]Project:[/bold] {project_name or 'N/A'}\n"
f"[bold]Tokens:[/bold] {token_count}"
),
title="ACMS Fragment Added",
border_style="green",
expand=False,
)
)
console.print("[green]\u2713 OK[/green] Fragment stored in ACMS context index")
return None
@app.command("remove-acms")
def context_remove_acms(
fragment_id: Annotated[
str,
typer.Argument(help="Fragment ID to remove from the ACMS index"),
],
) -> None:
"""Remove a fragment from the ACMS context index.
Removes the specified fragment from all tiers in the ContextTierService.
Examples::
agents actor context remove-acms file:///src/app/main.py:frag-a1b2c3d4
"""
try:
from cleveragents.application.services.context_tiers import (
ContextTierService,
)
tier_service = ContextTierService()
except Exception as exc:
console.print(f"[red]Error:[/red] Unable to init tier service: {exc}")
raise typer.Abort() from exc
# Remove the fragment from all tiers using internal helper
tier_service._remove_from_all(fragment_id) # noqa: SLF001
# Check that it was actually removed
not_removed = (
fragment_id in tier_service._hot # noqa: SLF001
or fragment_id in tier_service._warm # noqa: SLF001
or fragment_id in tier_service._cold # noqa: SLF001
)
if not not_removed:
console.print(
f"[green]\u2713 OK[/green] Fragment '{fragment_id}' removed from ACMS index."
)
else:
console.print(f"[yellow]Fragment ID '{fragment_id}' was already absent.[/yellow]")
return None
# Add a callback for when context command is called without subcommand
@app.callback(invoke_without_command=True)
def context_default(ctx: typer.Context) -> None: