feat(cli): implement context list and context add CLI commands for ACMS

Add new acms context list and acms context add CLI commands that
interact with the ACMS (Application Context Management System) tier-based
context pipeline. These commands manage context fragments across the hot,
warm, and cold storage tiers via the ContextTierService:

- acms context list: List all ACMS context fragments with tier placement,
  token counts, access frequency, and metrics. Supports --tier, --project,
  --project-filter, and format options (rich/json/yaml/table/plain).

- acms context add: Add resource files/directories to the ACMS tier store
  as TieredFragment entries under a target project scope and tier. Skips
  binary and oversized files automatically. Supports --tier, --recursive,
  --project, and format options. Also includes an reset subcommand to
  clear all fragments for a project.

src/cleveragents/cli/commands/acms.py

@@ -0,0 +1,466 @@
+"""ACMS context CLI commands.
+
+Implements ``agents acms context list`` and ``agents acms context add`` to
+interact with the Application Context Management System (ACMS) tier-based
+context pipeline.
+
+Commands:
+- ``agents acms context list``  - List all ACMS context fragments across
+  hot/warm/cold tiers for a project, showing fragment metadata and tier metrics.
+- ``agents acms context add``   - Add resource files or directories to the
+  ACMS context tier store (hydrate the ContextTierService).
+
+Based on the ACMS architecture (context_tier_hydrator, ContextTierService)
+and the hot/warm/cold tier fragment lifecycle.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+from datetime import UTC
+from pathlib import Path
+from typing import Annotated, Any
+
+import typer
+from rich.console import Console
+from rich.panel import Panel
+from rich.table import Table
+
+from cleveragents.cli.formatting import OutputFormat, format_output
+from cleveragents.domain.models.acms.tiers import ContextTier, TieredFragment
+
+app = typer.Typer(help="Manage ACMS context (tier-based fragment storage)")
+console = Console()
+err_console = Console(stderr=True)
+
+_FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)"
+
+_MAX_FILE_BYTES = 256 * 1024
+_MAX_TOTAL_BYTES = 10 * 1024 * 1024
+
+_SKIP_DIRS = frozenset(
+    {".git", ".hg", ".svn", "__pycache__", "node_modules",
+     ".venv", "venv", ".nox", ".tox", ".mypy_cache",
+     ".pytest_cache", ".ruff_cache", "dist", "build", ".eggs", ".cleveragents"}
+)
+
+_BINARY_EXTS = frozenset(
+    {".pyc", ".pyo", ".so", ".o", ".a", ".dll", ".exe",
+     ".png", ".jpg", ".jpeg", ".gif", ".bmp", ".ico", ".pdf",
+     ".zip", ".tar", ".gz", ".bz2", ".xz", ".whl", ".egg",
+     ".db", ".sqlite", ".sqlite3"}
+)
+
+
+def _format_size(size_bytes: int) -> str:
+    if size_bytes < 1024:
+        return f"{size_bytes} B"
+    elif size_bytes < 1024 * 1024:
+        return f"{size_bytes / 1024:.1f} KB"
+    else:
+        return f"{size_bytes / (1024 * 1024):.1f} MB"
+
+
+def _resolve_project_name(project_arg: str | None) -> Any:
+    from cleveragents.application.container import get_container
+    from cleveragents.application.services.project_service import ProjectService
+
+    container = get_container()
+    project_service: ProjectService = container.project_service()
+
+    if project_arg:
+        ns_repo = container.namespaced_project_repo()
+        try:
+            project = ns_repo.get(project_arg)
+            return project
+        except Exception as exc:
+            err_console.print(f"[red]Project not found:[/red] {project_arg}")
+            raise typer.Exit(1) from exc
+
+    project = project_service.get_current_project()
+    if not project:
+        err_console.print("[red]Error:[/red] No active project. Run 'cleveragents init' first.")
+        raise typer.Exit(1)
+    return project
+
+
+@app.command("list")
+def acms_context_list(
+    project: Annotated[
+        str | None,
+        typer.Option("--project", "-p", help="Project namespaced name (defaults to current project)"),
+    ] = None,
+    tier: Annotated[
+        ContextTier | None,
+        typer.Option("--tier", "-t", help="Filter by specific tier: hot, warm, or cold (default: all)"),
+    ] = None,
+    project_filter: Annotated[
+        list[str] | None,
+        typer.Option("--project-filter", "focus_projects", help="Project names to scope fragments from (repeatable)"),
+    ] = None,
+    fmt: Annotated[str, typer.Option("--format", "-f", help=_FORMAT_HELP)] = "rich",
+) -> None:
+    """List ACMS context fragments across hot/warm/cold tiers.
+
+    Displays all managed context fragments with their tier placement,
+    token counts, access frequency, and source resource information.
+
+    Examples::
+
+        # List all fragments for the current project
+        agents acms context list
+
+        # List only hot-tier fragments
+        agents acms context list --tier hot
+
+        # List fragments for a specific project by name
+        agents acms context list --project local/my-project
+
+        # Export as JSON for scripting
+        agents acms context list --format json
+    """
+    from cleveragents.application.container import get_container
+
+    try:
+        container = get_container()
+        tier_service = container.context_tier_service()
+
+        if project_filter:
+            fragments = tier_service.get_scoped_view(project_filter)
+        else:
+            proj = _resolve_project_name(project)
+            project_name = getattr(proj, "namespaced_name", str(proj))
+            fragments = tier_service.get_scoped_view([project_name])
+
+        if tier is not None:
+            fragments = [f for f in fragments if f.tier == tier]
+
+        metrics = tier_service.get_metrics()
+        budget = tier_service.budget
+
+        target_project = (
+            str(project_filter) if project_filter else getattr(proj, "namespaced_name", str(proj))
+        )
+        result_data: dict[str, Any] = {
+            "project": target_project,
+            "tier_filter": tier.value if tier else "all",
+            "metrics": {
+                "hot_count": metrics.hot_count,
+                "warm_count": metrics.warm_count,
+                "cold_count": metrics.cold_count,
+                "total_fragments": metrics.total_fragments,
+                "hot_hit_rate": f"{metrics.hot_hit_rate:.2%}",
+                "budget": {
+                    "max_tokens_hot": budget.max_tokens_hot,
+                    "max_decisions_warm": budget.max_decisions_warm,
+                    "max_decisions_cold": budget.max_decisions_cold,
+                },
+            },
+            "fragments": [
+                {
+                    "fragment_id": f.fragment_id,
+                    "tier": f.tier.value,
+                    "resource_id": f.resource_id or "",
+                    "token_count": f.token_count,
+                    "access_count": f.access_count,
+                    "last_accessed": f.last_accessed.isoformat(),
+                    "metadata": f.metadata,
+                }
+                for f in fragments
+            ],
+        }
+
+        if fmt.lower() == OutputFormat.RICH:
+            _render_list_rich(fragments, metrics, budget, tier)
+        elif fmt.lower() == "json":
+            import json as _json_mod
+            console.print(_json_mod.dumps(result_data, indent=2))
+        else:
+            console.print(format_output(result_data, fmt))
+
+    except Exception as exc:
+        err_console.print(f"[red]Error listing ACMS context:[/red] {exc}")
+        raise typer.Exit(1) from exc
+
+
+def _render_list_rich(
+    fragments: list[TieredFragment],
+    metrics: "TierMetrics",  # noqa: F821
+    budget: "TierBudget",     # noqa: F821
+    tier_filter: ContextTier | None,
+) -> None:
+    """Render the ACMS context list using rich terminal output."""
+    tier_label = tier_filter.value if tier_filter else "all"
+
+    summary_lines: list[str] = []
+    summary_lines.append(f"[bold]Total fragments:[/bold] {metrics.total_fragments}")
+    summary_lines.append(
+        f"[bold]Hot:[/bold] {metrics.hot_count}  "
+        f"[bold]Warm:[/bold] {metrics.warm_count}  "
+        f"[bold]Cold:[/bold] {metrics.cold_count}"
+    )
+    summary_lines.append(f"[bold]Budget (hot tokens):[/bold] {budget.max_tokens_hot:,}")
+    summary_lines.append(
+        f"[bold]Hot hit rate:[/bold] "
+        f"{metrics.hot_hit_rate:.2%} ({metrics.hot_hit_count} hits / "
+        f"{metrics.hot_hit_count + metrics.hot_miss_count} accesses)"
+    )
+
+    total_tokens_in_fragments = sum(f.token_count for f in fragments)
+    pct_of_budget = (
+        (total_tokens_in_fragments / budget.max_tokens_hot * 100) if budget.max_tokens_hot > 0 else 0
+    )
+    summary_lines.append(f"[bold]Tokens in view:[/bold] {total_tokens_in_fragments:,} ({pct_of_budget:.1%} of hot budget)")
+
+    console.print(Panel("\n".join(summary_lines), title=f"ACMS Context Fragments ({tier_label} tier)", expand=False))
+
+    if not fragments:
+        console.print("[yellow]No context fragments found.[/yellow]")
+        console.print("Use 'agents acms context add <path>' to add files.")
+        return
+
+    table = Table(title=f"Fragments ({len(fragments)})")
+    table.add_column("#", style="dim", justify="right", width=4)
+    table.add_column("Tier", width=6)
+    table.add_column("Resource / ID", overflow="fold")
+    table.add_column("Tokens", justify="right", width=12)
+    table.add_column("Accesses", justify="right", width=8)
+    table.add_column("Last Accessed", width=20)
+
+    tier_order = {ContextTier.HOT: 0, ContextTier.WARM: 1, ContextTier.COLD: 2}
+    sorted_frags = sorted(fragments, key=lambda f: (tier_order.get(f.tier), -f.access_count))
+
+    for idx, frag in enumerate(sorted_frags, 1):
+        tier_style = {"hot": "green", "warm": "yellow", "cold": "blue"}[frag.tier.value]
+        short_id = (frag.fragment_id[:50] + "..." if len(frag.fragment_id) > 50 else frag.fragment_id)
+        table.add_row(
+            str(idx),
+            f"[{tier_style}]{frag.tier.value}[/{tier_style}]",
+            short_id,
+            f"{frag.token_count:,}",
+            str(frag.access_count),
+            frag.last_accessed.strftime("%Y-%m-%d %H:%M"),
+        )
+
+    console.print(table)
+
+
+@app.command("add")
+def acms_context_add(
+    paths: Annotated[list[str], typer.Argument(help="Paths (files or directories) to add to ACMS context")],
+    recursive: Annotated[bool, typer.Option("-r", "--recursive", help="Add directories recursively")] = True,
+    tier: Annotated[ContextTier, typer.Option("--tier", "-t", help="Target tier (default: hot)")] = ContextTier.HOT,
+    project: Annotated[str | None, typer.Option("--project", "-p", help="Project namespaced name for scoping")] = None,
+    fmt: Annotated[str, typer.Option("--format", "-f", help=_FORMAT_HELP)] = "rich",
+) -> None:
+    """Add resource files to the ACMS context tier store.
+
+    Reads files from disk and stores them as TieredFragment entries in
+    the ContextTierService under the specified project scope and target
+    tier. Binary and large files are automatically skipped.
+
+    Examples::
+
+        # Add current directory to hot tier (current project)
+        agents acms context add .
+
+        # Add specific files/dirs with recursive traversal
+        agents acms context add src/ docs/
+
+        # Add to warm tier for a named project
+        agents acms context add ./project-files --tier warm -p local/my-project
+
+        # Non-recursive (single file only)
+        agents acms context add README.md -r false
+    """
+    from cleveragents.application.container import get_container
+
+    try:
+        container = get_container()
+        tier_service = container.context_tier_service()
+
+        proj = _resolve_project_name(project)
+        project_name = getattr(proj, "namespaced_name", str(proj))
+
+        from cleveragents.application.services.resource_registry_service import ResourceRegistryService
+
+        resource_reg: ResourceRegistryService = container.resource_registry()
+        resources_for_project = []
+        try:
+            if hasattr(proj, "project_id"):
+                resources_for_project = resource_reg.list_resources(project=proj.project_id)
+        except Exception:
+            pass
+
+        if resources_for_project and len(resources_for_project) > 0:
+            res_obj = resources_for_project[0]
+            resource_id = str(res_obj.resource_id)
+            resource_location = getattr(res_obj, "location", None) or getattr(proj, "location", None) or "."
+        else:
+            resource_id = f"adhoc/{project_name}"
+            resource_location = str(Path.cwd())
+
+        added_frags: list[TieredFragment] = []
+        already_in_context: list[str] = []
+        skipped_files: list[str] = []
+        total_bytes = 0
+
+        for path_str in paths:
+            path = Path(path_str).resolve()
+
+            if not path.exists():
+                err_console.print(f"[red]Path does not exist:[/red] {path}")
+                continue
+
+            files_to_add: list[Path] = []
+            if path.is_file():
+                files_to_add = [path]
+            elif path.is_dir() and recursive:
+                for dirpath, dirnames, filenames in os.walk(path):
+                    dirnames[:] = [d for d in dirnames if d not in _SKIP_DIRS and not d.startswith(".")]
+                    for fname in filenames:
+                        if fname.startswith("."):
+                            continue
+                        ext = os.path.splitext(fname)[1].lower()
+                        if ext in _BINARY_EXTS:
+                            skipped_files.append(str(Path(dirpath) / fname))
+                            continue
+                        files_to_add.append(Path(dirpath) / fname)
+            elif path.is_dir() and not recursive:
+                raise typer.BadParameter(f"Directory requires --recursive flag. Use 'add <path>' for individual files only.")
+
+            for file_path in files_to_add:
+                try:
+                    size = file_path.stat().st_size
+                except OSError:
+                    continue
+
+                if size > _MAX_FILE_BYTES:
+                    skipped_files.append(str(file_path))
+                    continue
+                if total_bytes + size > _MAX_TOTAL_BYTES:
+                    break
+
+                content = ""
+                try:
+                    content = file_path.read_text(encoding="utf-8")
+                except (UnicodeDecodeError, OSError):
+                    skipped_files.append(str(file_path))
+                    continue
+
+                try:
+                    res_root = Path(resource_location) if isinstance(resource_location, str) else resource_location
+                    rel_path = str(file_path.relative_to(res_root))
+                except (ValueError, TypeError):
+                    rel_path = str(file_path)
+
+                fragment_id = f"{resource_id}:{rel_path}"
+
+                existing = tier_service._find_fragment(fragment_id)
+                if existing is not None:
+                    already_in_context.append(rel_path)
+                    continue
+
+                fragment = TieredFragment(
+                    fragment_id=fragment_id,
+                    content=content,
+                    tier=tier,
+                    resource_id=resource_id,
+                    project_name=project_name,
+                    token_count=len(content) // 4,
+                    metadata={
+                        "path": rel_path,
+                        "detail_depth": "1",
+                        "relevance_score": "0.5",
+                        "source": str(file_path),
+                    },
+                )
+
+                tier_service.store(fragment)
+                added_frags.append(fragment)
+                total_bytes += size
+
+        result_data: dict[str, Any] = {
+            "project": project_name,
+            "resource_id": resource_id,
+            "tier": tier.value,
+            "added_count": len(added_frags),
+            "already_exists_count": len(already_in_context),
+            "skipped_count": len(skipped_files),
+            "total_bytes": total_bytes,
+            "added_files": [f.fragment_id for f in added_frags],
+            "already_in_context": already_in_context,
+            "skipped_files": skipped_files,
+        }
+
+        if fmt.lower() == OutputFormat.RICH:
+            messages: list[tuple[str, str]] = []
+            if added_frags:
+                shown = min(len(added_frags), 10)
+                file_list = "\n".join(f"  {f.fragment_id}" for f in added_frags[:shown])
+                extra = "" if len(added_frags) <= 10 else f"\n  ... and {len(added_frags) - 10} more"
+                messages.append(("Added", f"[green]Added [bold]{len(added_frags)}[/bold] file(s) to [{tier.value}] tier ({_format_size(total_bytes)})\n{file_list}{extra}"))
+            if already_in_context:
+                shown = min(len(already_in_context), 10)
+                messages.append(("Already in context", f"[yellow]{len(already_in_context)}[/yellow] file(s) already stored:\n" + "\n".join(f"  - {f}" for f in already_in_context[:shown]) + (f"\n  ... and {len(already_in_context) - 10} more" if len(already_in_context) > 10 else "")))
+            if skipped_files:
+                messages.append(("Skipped", f"[dim]{len(skipped_files)} file(s) skipped (binary or oversized)[/dim]"))
+            for title, body in messages:
+                console.print(Panel(body, title=title, expand=False))
+        else:
+            console.print(format_output(result_data, fmt))
+
+    except Exception as exc:
+        err_console.print(f"[red]Error adding to ACMS context:[/red] {exc}")
+        raise typer.Exit(1) from exc
+
+
+@app.command("reset")
+def acms_context_reset(
+    project: Annotated[str | None, typer.Option("--project", "-p", help="Project namespaced name (defaults to current project)")] = None,
+    yes: Annotated[bool, typer.Option("--yes", "-y", help="Skip confirmation prompt")] = False,
+    fmt: Annotated[str, typer.Option("--format", "-f", help=_FORMAT_HELP)] = "rich",
+) -> None:
+    """Reset (clear) all ACMS context fragments for a project."""
+    from cleveragents.application.container import get_container
+    from cleveragents.domain.models.acms.tiers import ContextTier
+
+    try:
+        container = get_container()
+        tier_service = container.context_tier_service()
+
+        proj = _resolve_project_name(project)
+        project_name = getattr(proj, "namespaced_name", str(proj))
+
+        all_frags = tier_service.get_scoped_view([project_name])
+        fragment_ids_to_remove = [f.fragment_id for f in all_frags]
+
+        if not fragment_ids_to_remove:
+            console.print("[yellow]No fragments to reset.[/yellow]")
+            return
+
+        if not yes:
+            typer.echo(f"Found {len(fragment_ids_to_remove)} fragment(s) for project '{project_name}'.")
+            if not typer.confirm("Reset (clear) all fragments?"):
+                typer.echo("Reset cancelled.")
+                return
+
+        stores = {ContextTier.HOT: tier_service._hot, ContextTier.WARM: tier_service._warm, ContextTier.COLD: tier_service._cold}
+        removed_count = 0
+        for store in stores.values():
+            to_remove = [fid for fid, frag in store.items() if frag.project_name == project_name]
+            for fid in to_remove:
+                del store[fid]
+                removed_count += 1
+
+        console.print(Panel(
+            f"[green]Reset ACMS context for project '[bold]{project_name}[/bold]'\n"
+            f"[bold]Removed:[/bold] {removed_count} fragment(s)",
+            title="ACMS Context Reset",
+            expand=False,
+        ))
+
+    except Exception as exc:
+        err_console.print(f"[red]Error resetting ACMS context:[/red] {exc}")
+        raise typer.Exit(1) from exc

src/cleveragents/cli/main.py

@@ -98,6 +98,7 @@ def _register_subcommands() -> None:
             tui,
             validation,
         )
+        from cleveragents.cli.commands.acms import app as acms_app
         from cleveragents.cli.commands.auto_debug import app as auto_debug_app
         from cleveragents.cli.commands.db import app as db_app
         from cleveragents.cli.commands.repl import _repl_app
@@ -228,6 +229,11 @@ def _register_subcommands() -> None:
         name="repo",
         help="Repository indexing management",
     )
+    app.add_typer(
+        acms_app,
+        name="acms",
+        help="Application Context Management System (tier-based context)",
+    )
 
     _subcommands_registered = True
 
@@ -684,6 +690,7 @@ def main(args: list[str] | None = None) -> int:
             "tui",  # Textual TUI
             "server",  # Server connection management
             "repo",  # Repository indexing management
+            "acms",  # ACMS - Application Context Management System
             "apply",  # Shortcut for plan apply
             "context-load",  # Shortcut for context add
             "context-add",  # Shortcut
@@ -716,6 +723,7 @@ def main(args: list[str] | None = None) -> int:
                 "apply",
                 "context-load",
                 "context-add",
+                "acms",
             }
         )
         if hasattr(app, "add_typer") and not (