feat(cli): add actor context remove, export, and import commands #1190

Merged
freemo merged 1 commits from feature/m4-actor-context-cmds into master 2026-03-30 19:51:53 +00:00
5 changed files with 1125 additions and 0 deletions
+100
View File
@@ -0,0 +1,100 @@
Feature: Actor context remove, export, and import commands
As a CleverAgents user
I want to manage actor contexts via the CLI
So that I can remove, export, and import conversation contexts
Background:
Given a temporary context directory for actor context tests
# ── context remove ─────────────────────────────────────────
Scenario: Remove a named actor context
Given an actor context named "docs" exists
When I run actor context remove "docs" with --yes
Then the actor context remove command should succeed
And the context "docs" should no longer exist
Scenario: Remove all actor contexts
Given an actor context named "docs" exists
And an actor context named "notes" exists
When I run actor context remove --all with --yes
Then the actor context remove command should succeed
And no actor contexts should remain
Scenario: Remove non-existent context fails
When I run actor context remove "nonexistent" with --yes
Then the actor context remove command should fail with exit code 1
Scenario: Remove requires NAME or --all
When I run actor context remove without name or all
Then the actor context remove command should fail with exit code 1
Scenario: Remove rejects NAME with --all
When I run actor context remove "docs" with --all
Then the actor context remove command should fail with exit code 1
Scenario: Remove outputs JSON format
Given an actor context named "docs" exists
When I run actor context remove "docs" with --yes and format "json"
Then the actor context remove command should succeed
And the output should contain valid JSON with key "context_removed"
# ── context export ─────────────────────────────────────────
Scenario: Export a named actor context to JSON
Given an actor context named "docs" exists with messages
When I run actor context export "docs" to a JSON file
Then the actor context export command should succeed
And the exported file should exist and contain valid JSON
And the exported JSON should contain key "messages"
Scenario: Export a named actor context to YAML
Given an actor context named "docs" exists with messages
When I run actor context export "docs" to a YAML file
Then the actor context export command should succeed
And the exported YAML file should exist and be valid
Scenario: Export non-existent context fails
When I run actor context export "nonexistent" to a JSON file
Then the actor context export command should fail with exit code 1
Scenario: Export outputs JSON format metadata
Given an actor context named "docs" exists with messages
When I run actor context export "docs" to a JSON file with format "json"
Then the actor context export command should succeed
And the output should contain valid JSON with key "context_export"
# ── context import ─────────────────────────────────────────
Scenario: Import a context from a JSON file
Given a valid context JSON file named "imported-ctx.json"
When I run actor context import from that file as "imported-ctx"
Then the actor context import command should succeed
And the context "imported-ctx" should exist
Scenario: Import infers name from file metadata
Given a valid context JSON file with context_name "auto-named"
When I run actor context import from that file without a name
Then the actor context import command should succeed
And the context "auto-named" should exist
Scenario: Import refuses to overwrite without --update
Given an actor context named "existing" exists
And a valid context JSON file named "existing.json"
When I run actor context import from that file as "existing" without update
Then the actor context import command should fail with exit code 1
Scenario: Import with --update replaces existing context
Given an actor context named "existing" exists
And a valid context JSON file named "existing.json"
When I run actor context import from that file as "existing" with --update
Then the actor context import command should succeed
And the context "existing" should exist
Scenario: Export then import round-trip preserves data
Given an actor context named "roundtrip" exists with messages
When I export the context "roundtrip" to a JSON file
And I remove the context "roundtrip"
And I import the context from that JSON file as "roundtrip"
Then the context "roundtrip" should exist
And the imported context should have the same messages as the original
+451
View File
@@ -0,0 +1,451 @@
# pyright: reportRedeclaration=false
"""Step definitions for actor context remove/export/import commands."""
from __future__ import annotations
import json
import shutil
import tempfile
from pathlib import Path
from behave import given, then, when
from typer.testing import CliRunner
from cleveragents.cli.commands.actor_context import app as actor_context_app
from cleveragents.reactive.context_manager import ContextManager
# ---------------------------------------------------------------------------
# Background
# ---------------------------------------------------------------------------
@given("a temporary context directory for actor context tests")
def step_temp_context_dir(context):
context.tmp_dir = Path(tempfile.mkdtemp())
context.context_dir = context.tmp_dir / "contexts"
context.context_dir.mkdir(parents=True, exist_ok=True)
context.runner = CliRunner()
context._cleanup_handlers: list = getattr(context, "_cleanup_handlers", [])
context._cleanup_handlers.append(
lambda: shutil.rmtree(context.tmp_dir, ignore_errors=True)
)
# Will store original messages for round-trip checks
context.original_messages = None
context.export_file = None
# ---------------------------------------------------------------------------
# Givens — context creation helpers
# ---------------------------------------------------------------------------
@given('an actor context named "{name}" exists')
def step_context_exists(context, name):
mgr = ContextManager(name, context.context_dir)
mgr.add_message("system", f"Initial message for {name}")
@given('an actor context named "{name}" exists with messages')
def step_context_exists_with_messages(context, name):
mgr = ContextManager(name, context.context_dir)
mgr.add_message("user", "Hello, how are you?")
mgr.add_message("assistant", "I'm doing well, thanks for asking!")
mgr.add_message("user", "Great, let's get started.")
context.original_messages = list(mgr.messages)
@given('a valid context JSON file named "{filename}"')
def step_create_json_file(context, filename):
data = {
"context_name": filename.replace(".json", ""),
"messages": [
{
"role": "user",
"content": "test message",
"timestamp": "2026-01-01T00:00:00",
"metadata": {},
},
],
"metadata": {"context_name": filename.replace(".json", "")},
"state": {},
"global_context": {},
}
context.import_file = context.tmp_dir / filename
context.import_file.write_text(json.dumps(data, indent=2), encoding="utf-8")
@given('a valid context JSON file with context_name "{name}"')
def step_create_json_file_with_name(context, name):
data = {
"context_name": name,
"messages": [
{
"role": "user",
"content": "auto-named test",
"timestamp": "2026-01-01T00:00:00",
"metadata": {},
},
],
"metadata": {"context_name": name},
"state": {},
"global_context": {},
}
context.import_file = context.tmp_dir / f"{name}.json"
context.import_file.write_text(json.dumps(data, indent=2), encoding="utf-8")
# ---------------------------------------------------------------------------
# When — remove
# ---------------------------------------------------------------------------
@when('I run actor context remove "{name}" with --yes')
def step_remove_named_yes(context, name):
context.result = context.runner.invoke(
actor_context_app,
["remove", name, "--yes", "--context-dir", str(context.context_dir)],
)
@when("I run actor context remove --all with --yes")
def step_remove_all_yes(context):
context.result = context.runner.invoke(
actor_context_app,
["remove", "--all", "--yes", "--context-dir", str(context.context_dir)],
)
@when('I run actor context remove "{name}" with --yes and format "{fmt}"')
def step_remove_named_format(context, name, fmt):
context.result = context.runner.invoke(
actor_context_app,
[
"remove",
name,
"--yes",
"--context-dir",
str(context.context_dir),
"--format",
fmt,
],
)
@when("I run actor context remove without name or all")
def step_remove_no_args(context):
context.result = context.runner.invoke(
actor_context_app,
["remove", "--context-dir", str(context.context_dir)],
)
@when('I run actor context remove "{name}" with --all')
def step_remove_name_and_all(context, name):
context.result = context.runner.invoke(
actor_context_app,
["remove", name, "--all", "--context-dir", str(context.context_dir)],
)
# ---------------------------------------------------------------------------
# When — export
# ---------------------------------------------------------------------------
@when('I run actor context export "{name}" to a JSON file')
def step_export_json(context, name):
context.export_file = context.tmp_dir / f"{name}-export.json"
context.result = context.runner.invoke(
actor_context_app,
[
"export",
name,
"--output",
str(context.export_file),
"--context-dir",
str(context.context_dir),
],
)
@when('I run actor context export "{name}" to a YAML file')
def step_export_yaml(context, name):
context.export_file = context.tmp_dir / f"{name}-export.yaml"
context.result = context.runner.invoke(
actor_context_app,
[
"export",
name,
"--output",
str(context.export_file),
"--context-dir",
str(context.context_dir),
],
)
@when('I run actor context export "{name}" to a JSON file with format "{fmt}"')
def step_export_json_format(context, name, fmt):
context.export_file = context.tmp_dir / f"{name}-export.json"
context.result = context.runner.invoke(
actor_context_app,
[
"export",
name,
"--output",
str(context.export_file),
"--context-dir",
str(context.context_dir),
"--format",
fmt,
],
)
# ---------------------------------------------------------------------------
# When — import
# ---------------------------------------------------------------------------
@when('I run actor context import from that file as "{name}"')
def step_import_named(context, name):
context.result = context.runner.invoke(
actor_context_app,
[
"import",
name,
"--input",
str(context.import_file),
"--context-dir",
str(context.context_dir),
],
)
@when("I run actor context import from that file without a name")
def step_import_no_name(context):
context.result = context.runner.invoke(
actor_context_app,
[
"import",
"--input",
str(context.import_file),
"--context-dir",
str(context.context_dir),
],
)
@when('I run actor context import from that file as "{name}" without update')
def step_import_no_update(context, name):
context.result = context.runner.invoke(
actor_context_app,
[
"import",
name,
"--input",
str(context.import_file),
"--context-dir",
str(context.context_dir),
],
)
@when('I run actor context import from that file as "{name}" with --update')
def step_import_with_update(context, name):
context.result = context.runner.invoke(
actor_context_app,
[
"import",
name,
"--input",
str(context.import_file),
"--update",
"--context-dir",
str(context.context_dir),
],
)
# ---------------------------------------------------------------------------
# When — round-trip helpers
# ---------------------------------------------------------------------------
@when('I export the context "{name}" to a JSON file')
def step_roundtrip_export(context, name):
context.export_file = context.tmp_dir / f"{name}-roundtrip.json"
context.result = context.runner.invoke(
actor_context_app,
[
"export",
name,
"--output",
str(context.export_file),
"--context-dir",
str(context.context_dir),
],
)
assert context.result.exit_code == 0, context.result.output
@when('I remove the context "{name}"')
def step_roundtrip_remove(context, name):
context.result = context.runner.invoke(
actor_context_app,
["remove", name, "--yes", "--context-dir", str(context.context_dir)],
)
assert context.result.exit_code == 0, context.result.output
@when('I import the context from that JSON file as "{name}"')
def step_roundtrip_import(context, name):
context.result = context.runner.invoke(
actor_context_app,
[
"import",
name,
"--input",
str(context.export_file),
"--context-dir",
str(context.context_dir),
],
)
assert context.result.exit_code == 0, context.result.output
# ---------------------------------------------------------------------------
# Then — success / failure
# ---------------------------------------------------------------------------
@then("the actor context remove command should succeed")
def step_remove_success(context):
assert context.result.exit_code == 0, (
f"Expected exit 0, got {context.result.exit_code}.\n"
f"stdout: {context.result.output}\n"
f"stderr: {getattr(context.result, 'stderr', '')}"
)
@then("the actor context export command should succeed")
def step_export_success(context):
assert context.result.exit_code == 0, (
f"Expected exit 0, got {context.result.exit_code}.\n"
f"stdout: {context.result.output}\n"
f"stderr: {getattr(context.result, 'stderr', '')}"
)
@then("the actor context import command should succeed")
def step_import_success(context):
assert context.result.exit_code == 0, (
f"Expected exit 0, got {context.result.exit_code}.\n"
f"stdout: {context.result.output}\n"
f"stderr: {getattr(context.result, 'stderr', '')}"
)
@then("the actor context remove command should fail with exit code 1")
def step_remove_fail(context):
assert context.result.exit_code == 1, (
f"Expected exit 1, got {context.result.exit_code}.\n"
f"stdout: {context.result.output}\n"
f"stderr: {getattr(context.result, 'stderr', '')}"
)
@then("the actor context export command should fail with exit code 1")
def step_export_fail(context):
assert context.result.exit_code == 1, (
f"Expected exit 1, got {context.result.exit_code}.\n"
f"stdout: {context.result.output}\n"
f"stderr: {getattr(context.result, 'stderr', '')}"
)
@then("the actor context import command should fail with exit code 1")
def step_import_fail(context):
assert context.result.exit_code == 1, (
f"Expected exit 1, got {context.result.exit_code}.\n"
f"stdout: {context.result.output}\n"
f"stderr: {getattr(context.result, 'stderr', '')}"
)
# ---------------------------------------------------------------------------
# Then — state assertions
# ---------------------------------------------------------------------------
@then('the context "{name}" should no longer exist')
def step_context_gone(context, name):
ctx_path = context.context_dir / name
assert not ctx_path.exists(), f"Context dir {ctx_path} still exists"
@then("no actor contexts should remain")
def step_no_contexts(context):
remaining = [d for d in context.context_dir.iterdir() if d.is_dir()]
assert len(remaining) == 0, (
f"Expected 0 contexts, found {len(remaining)}: {remaining}"
)
@then('the context "{name}" should exist')
def step_context_exists_check(context, name):
mgr = ContextManager(name, context.context_dir)
assert mgr.exists(), f"Context '{name}' does not exist at {mgr.context_dir}"
# ---------------------------------------------------------------------------
# Then — output assertions
# ---------------------------------------------------------------------------
@then('the output should contain valid JSON with key "{key}"')
def step_output_json_key(context, key):
output = context.result.output
data = json.loads(output)
assert key in data, f"Key '{key}' not found in JSON output: {data.keys()}"
@then("the exported file should exist and contain valid JSON")
def step_exported_json_valid(context):
assert context.export_file.exists(), f"Export file {context.export_file} not found"
data = json.loads(context.export_file.read_text(encoding="utf-8"))
assert isinstance(data, dict)
@then('the exported JSON should contain key "{key}"')
def step_exported_json_key(context, key):
data = json.loads(context.export_file.read_text(encoding="utf-8"))
assert key in data, f"Key '{key}' not found in exported JSON: {data.keys()}"
@then("the exported YAML file should exist and be valid")
def step_exported_yaml_valid(context):
import yaml
assert context.export_file.exists(), f"Export file {context.export_file} not found"
data = yaml.safe_load(context.export_file.read_text(encoding="utf-8"))
assert isinstance(data, dict)
@then("the imported context should have the same messages as the original")
def step_roundtrip_messages(context):
assert context.original_messages is not None, "No original messages saved"
mgr = ContextManager("roundtrip", context.context_dir)
imported = mgr.messages
# Compare content and role, ignoring timestamp differences
assert len(imported) == len(context.original_messages), (
f"Message count mismatch: {len(imported)} vs {len(context.original_messages)}"
)
for orig, imp in zip(context.original_messages, imported, strict=True):
assert orig["role"] == imp["role"], (
f"Role mismatch: {orig['role']} vs {imp['role']}"
)
assert orig["content"] == imp["content"], (
f"Content mismatch: {orig['content']!r} vs {imp['content']!r}"
)
+121
View File
@@ -0,0 +1,121 @@
*** Settings ***
Documentation Integration test for actor context export-then-import round-trip
Library Process
Library OperatingSystem
Library String
Library DateTime
Library Collections
Resource ${CURDIR}/common.resource
Suite Setup Setup Test Environment
Suite Teardown Cleanup Test Environment
*** Variables ***
${TEST_CTX_DIR} ${EMPTY}
${EXPORT_FILE} ${EMPTY}
${CONTEXT_NAME} roundtrip-robot
*** Test Cases ***
Export Then Import Round-Trip Preserves Context
[Documentation] Create a context, export it, delete it, import it, and verify data integrity.
# 1. Create a named context by running actor with --context
${ctx_dir} = Set Variable ${TEMP}/actor_ctx
Set Suite Variable ${TEST_CTX_DIR} ${ctx_dir}
Create Directory ${ctx_dir}
# Create context data manually via Python helper
${result} = Run Process ${PYTHON} -c
... from cleveragents.reactive.context_manager import ContextManager; mgr \= ContextManager("${CONTEXT_NAME}", "${ctx_dir}"); mgr.add_message("user", "Hello from Robot"); mgr.add_message("assistant", "Hello Robot!"); print("created")
Log Create stdout: ${result.stdout}
Log Create stderr: ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} created
# 2. Export the context to JSON
${export_path} = Set Variable ${TEMP}/exported-context.json
Set Suite Variable ${EXPORT_FILE} ${export_path}
${result} = Run Process ${PYTHON} -m cleveragents actor context export
... ${CONTEXT_NAME} --output ${export_path} --context-dir ${ctx_dir}
Log Export stdout: ${result.stdout}
Log Export stderr: ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
File Should Exist ${export_path}
# 3. Remove the original context
${result} = Run Process ${PYTHON} -m cleveragents actor context remove
... ${CONTEXT_NAME} --yes --context-dir ${ctx_dir}
Log Remove stdout: ${result.stdout}
Log Remove stderr: ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Directory Should Not Exist ${ctx_dir}/${CONTEXT_NAME}
# 4. Import the context back
${result} = Run Process ${PYTHON} -m cleveragents actor context import
... ${CONTEXT_NAME} --input ${export_path} --context-dir ${ctx_dir}
Log Import stdout: ${result.stdout}
Log Import stderr: ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
# 5. Verify the round-trip preserved messages
${result} = Run Process ${PYTHON} -c
... import json; from cleveragents.reactive.context_manager import ContextManager; mgr \= ContextManager("${CONTEXT_NAME}", "${ctx_dir}"); msgs \= mgr.messages; assert len(msgs) \=\= 2, f"Expected 2 messages, got {len(msgs)}"; assert msgs[0]["content"] \=\= "Hello from Robot"; assert msgs[1]["content"] \=\= "Hello Robot!"; print("verified")
Log Verify stdout: ${result.stdout}
Log Verify stderr: ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} verified
Export With JSON Format Flag Shows Structured Output
[Documentation] Verify --format json produces machine-readable output.
${ctx_dir} = Set Variable ${TEMP}/actor_ctx_fmt
Create Directory ${ctx_dir}
# Create context
${result} = Run Process ${PYTHON} -c
... from cleveragents.reactive.context_manager import ContextManager; mgr \= ContextManager("fmt-test", "${ctx_dir}"); mgr.add_message("user", "test"); print("ok")
Should Be Equal As Integers ${result.rc} 0
# Export with --format json
${export_path} = Set Variable ${TEMP}/fmt-export.json
${result} = Run Process ${PYTHON} -m cleveragents actor context export
... fmt-test --output ${export_path} --context-dir ${ctx_dir} --format json
Log Export stdout: ${result.stdout}
Log Export stderr: ${result.stderr}
Should Be Equal As Integers ${result.rc} 0
Should Contain ${result.stdout} context_export
Import Without Update Fails For Existing Context
[Documentation] Verify importing into an existing context without --update fails.
${ctx_dir} = Set Variable ${TEMP}/actor_ctx_noupdate
Create Directory ${ctx_dir}
# Create existing context
${result} = Run Process ${PYTHON} -c
... from cleveragents.reactive.context_manager import ContextManager; mgr \= ContextManager("existing", "${ctx_dir}"); mgr.add_message("user", "original"); print("ok")
Should Be Equal As Integers ${result.rc} 0
# Create import file
${import_path} = Set Variable ${TEMP}/existing-import.json
${result} = Run Process ${PYTHON} -c
... import json; data \= {"context_name": "existing", "messages": [{"role": "user", "content": "new", "timestamp": "2026-01-01", "metadata": {}}], "metadata": {}, "state": {}, "global_context": {}}; open("${import_path}", "w").write(json.dumps(data)); print("ok")
Should Be Equal As Integers ${result.rc} 0
# Import without --update should fail
${result} = Run Process ${PYTHON} -m cleveragents actor context import
... existing --input ${import_path} --context-dir ${ctx_dir}
Should Be Equal As Integers ${result.rc} 1
*** Keywords ***
Setup Test Environment
[Documentation] Create test environment
common.Setup Test Environment
${temp} = Evaluate tempfile.mkdtemp() modules=tempfile
Set Suite Variable ${TEMP} ${temp}
Create Directory ${TEMP}
Log Test environment created at: ${TEMP}
Cleanup Test Environment
[Documentation] Clean up test environment
Run Keyword If '${TEMP}' != '${EMPTY}' Remove Directory ${TEMP} recursive=True
+7
View File
@@ -18,6 +18,7 @@ from cleveragents.application.container import get_container
from cleveragents.cli.commands._resolve_actor import (
resolve_config_files as _resolve_config_files,
)
from cleveragents.cli.commands.actor_context import app as actor_context_app
from cleveragents.cli.formatting import OutputFormat, format_output
from cleveragents.core.exceptions import (
BusinessRuleViolation,
@@ -706,3 +707,9 @@ def set_default(
except (ValidationError, NotFoundError, BusinessRuleViolation) as exc:
console.print(f"[red]Error:[/red] {exc}")
raise typer.Abort() from exc
# ---------------------------------------------------------------------------
# Sub-command groups
# ---------------------------------------------------------------------------
app.add_typer(actor_context_app, name="context")
@@ -0,0 +1,446 @@
"""Actor-scoped context management commands.
Implements ``agents actor context remove``, ``agents actor context export``,
and ``agents actor context import`` per the v3 specification. These commands
manage named conversation contexts stored under ``~/.cleveragents/context/``
using the :class:`~cleveragents.reactive.context_manager.ContextManager`
persistence layer.
"""
from __future__ import annotations
import hashlib
import json
from pathlib import Path
from typing import Annotated, Any
import typer
import yaml
from rich.console import Console
from rich.panel import Panel
from cleveragents.cli.formatting import OutputFormat, format_output
from cleveragents.reactive.context_manager import ContextManager
app = typer.Typer(
help="Manage manual contexts for actor runs.",
)
console = Console()
_FORMAT_HELP = "Output format: json, yaml, plain, table, rich, or color (default: rich)"
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _default_context_base(context_dir: Path | None) -> Path:
"""Return the base directory where named contexts are stored."""
if context_dir is not None:
return context_dir
return Path.home() / ".cleveragents" / "context"
def _list_context_names(base: Path) -> list[str]:
"""Return sorted list of context names present under *base*."""
if not base.exists():
return []
return sorted(d.name for d in base.iterdir() if d.is_dir())
def _context_size_kb(ctx_mgr: ContextManager) -> float:
"""Estimate total file size of the context directory in KB."""
total = 0
if ctx_mgr.context_dir.exists():
for f in ctx_mgr.context_dir.rglob("*"):
if f.is_file():
total += f.stat().st_size
return round(total / 1024, 1)
def _file_checksum(path: Path) -> str:
"""Return ``sha256:<hex>`` checksum for a file."""
h = hashlib.sha256()
with open(path, "rb") as fh:
for chunk in iter(lambda: fh.read(8192), b""):
h.update(chunk)
digest = h.hexdigest()
return f"sha256:{digest[:4]}...{digest[-4:]}"
def _render_output(
data: dict[str, Any],
fmt: str,
rich_panels: list[tuple[str, str]] | None = None,
ok_message: str = "",
) -> None:
"""Emit command output in the requested format.
For ``rich`` format the *rich_panels* list of ``(title, body)`` pairs
are rendered via :class:`rich.panel.Panel`. For all other formats the
flat *data* dict is passed through :func:`format_output`.
"""
if fmt == OutputFormat.RICH.value and rich_panels:
for title, body in rich_panels:
console.print(Panel(body, title=title, expand=False))
if ok_message:
console.print(f"[green]✓ OK[/green] {ok_message}")
return
# Machine-readable / non-rich
console.print(format_output(data, fmt))
# ---------------------------------------------------------------------------
# Commands
# ---------------------------------------------------------------------------
@app.command("remove")
def context_remove(
name: Annotated[
str | None,
typer.Argument(help="Context name to remove"),
] = None,
all_contexts: Annotated[
bool,
typer.Option("--all", "-a", help="Remove all contexts"),
] = False,
yes: Annotated[
bool,
typer.Option("--yes", "-y", help="Skip confirmation prompt"),
] = False,
context_dir: Annotated[
Path | None,
typer.Option(
"--context-dir",
help="Directory where contexts are stored",
resolve_path=True,
),
] = None,
fmt: Annotated[
str,
typer.Option("--format", "-f", help=_FORMAT_HELP),
] = "rich",
) -> None:
"""Remove a named actor context or all actor contexts.
Context data is **not recoverable** after removal. Use ``--yes`` to
skip the interactive confirmation prompt.
Examples::
agents actor context remove docs
agents actor context remove --all --yes
"""
import shutil
if name and all_contexts:
typer.echo("Error: Cannot specify NAME when using --all", err=True)
raise typer.Exit(code=1)
if not name and not all_contexts:
typer.echo("Error: Must specify NAME or use --all", err=True)
raise typer.Exit(code=1)
base = _default_context_base(context_dir)
if all_contexts:
names = _list_context_names(base)
if not names:
typer.echo("No contexts found to remove.")
return
if not yes:
typer.echo(f"Found {len(names)} context(s) to remove:")
for cname in names:
typer.echo(f" - {cname}")
if not typer.confirm("Remove all?"):
typer.echo("Remove cancelled.")
return
for cname in names:
shutil.rmtree(base / cname)
data: dict[str, Any] = {
"context_removed": {
"context": "all",
"status": "removed",
"count": len(names),
},
"stats": {
"remaining_size_kb": 0,
},
}
panels = [
(
"Context Removed",
(
f"[bold]Context:[/bold] all ({len(names)} removed)\n"
f"[bold]Status:[/bold] removed"
),
),
(
"Stats",
"[bold]Remaining Size:[/bold] 0 KB",
),
]
_render_output(data, fmt, rich_panels=panels, ok_message="Context updated")
return
# Single context removal
assert name is not None
target = base / name
if not target.exists():
typer.echo(f"Error: Context '{name}' does not exist.", err=True)
raise typer.Exit(code=1)
ctx_mgr = ContextManager(name, context_dir)
if not yes and not typer.confirm(f"Remove context '{name}'?"):
typer.echo("Remove cancelled.")
return
ctx_mgr.delete()
# Compute remaining size across all contexts
remaining_total = 0.0
for other_name in _list_context_names(base):
other_mgr = ContextManager(other_name, context_dir)
remaining_total += _context_size_kb(other_mgr)
data = {
"context_removed": {
"context": name,
"status": "removed",
},
"stats": {
"remaining_size_kb": remaining_total,
},
}
panels = [
(
"Context Removed",
(f"[bold]Context:[/bold] {name}\n[bold]Status:[/bold] removed"),
),
(
"Stats",
(f"[bold]Remaining Size:[/bold] {remaining_total} KB"),
),
]
_render_output(data, fmt, rich_panels=panels, ok_message="Context updated")
@app.command("export")
def context_export(
name: Annotated[
str,
typer.Argument(help="Context name to export"),
],
output: Annotated[
Path,
typer.Option(
"--output",
"-o",
help="Output file path (JSON or YAML)",
resolve_path=True,
),
] = ..., # type: ignore[assignment]
context_dir: Annotated[
Path | None,
typer.Option(
"--context-dir",
help="Directory where contexts are stored",
resolve_path=True,
),
] = None,
fmt: Annotated[
str,
typer.Option("--format", "-f", help=_FORMAT_HELP),
] = "rich",
) -> None:
"""Export a named actor context to a JSON or YAML file.
The exported file contains messages, metadata, state, and
global_context and can later be re-imported with
``agents actor context import``.
Examples::
agents actor context export docs --output /tmp/docs-context.json
agents actor context export docs -o ctx.yaml --format json
"""
base = _default_context_base(context_dir)
if not (base / name).exists():
typer.echo(f"Error: Context '{name}' does not exist.", err=True)
raise typer.Exit(code=1)
ctx_mgr = ContextManager(name, context_dir)
# Build export payload
export_data: dict[str, Any] = {
"context_name": ctx_mgr.context_name,
"messages": ctx_mgr.messages,
"metadata": ctx_mgr.metadata,
"state": ctx_mgr.state,
"global_context": ctx_mgr.global_context,
}
# Determine serialisation format from file extension
output.parent.mkdir(parents=True, exist_ok=True)
suffix = output.suffix.lower()
if suffix in (".yaml", ".yml"):
with open(output, "w", encoding="utf-8") as fh:
yaml.dump(export_data, fh, default_flow_style=False, sort_keys=False)
else:
with open(output, "w", encoding="utf-8") as fh:
json.dump(export_data, fh, indent=2)
items = len(ctx_mgr.messages)
size_kb = round(output.stat().st_size / 1024, 1)
checksum = _file_checksum(output)
data: dict[str, Any] = {
"context_export": {
"context": name,
"output": str(output),
"items": items,
"size_kb": size_kb,
},
"integrity": {
"checksum": checksum,
"compressed": False,
},
}
panels = [
(
"Context Export",
(
f"[bold]Context:[/bold] {name}\n"
f"[bold]Output:[/bold] {output}\n"
f"[bold]Items:[/bold] {items}\n"
f"[bold]Size:[/bold] {size_kb} KB"
),
),
(
"Integrity",
(f"[bold]Checksum:[/bold] {checksum}\n[bold]Compressed:[/bold] no"),
),
]
_render_output(data, fmt, rich_panels=panels, ok_message="Export completed")
@app.command("import")
def context_import(
name: Annotated[
str | None,
typer.Argument(
help="Name for the imported context (inferred from file if omitted)"
),
] = None,
input_file: Annotated[
Path,
typer.Option(
"--input",
"-i",
help="Input file path (JSON or YAML)",
exists=True,
file_okay=True,
dir_okay=False,
readable=True,
resolve_path=True,
),
] = ..., # type: ignore[assignment]
update: Annotated[
bool,
typer.Option("--update", help="Replace existing context with same name"),
] = False,
context_dir: Annotated[
Path | None,
typer.Option(
"--context-dir",
help="Directory where contexts are stored",
resolve_path=True,
),
] = None,
fmt: Annotated[
str,
typer.Option("--format", "-f", help=_FORMAT_HELP),
] = "rich",
) -> None:
"""Import a context from a JSON or YAML file.
If ``NAME`` is omitted the context name is inferred from the
``context_name`` key inside the file. Use ``--update`` to replace
an existing context with the same name.
Examples::
agents actor context import docs --input /tmp/docs-context.json
agents actor context import --input ctx.yaml --update
"""
# Parse input file (JSON or YAML)
text = input_file.read_text(encoding="utf-8")
suffix = input_file.suffix.lower()
if suffix in (".yaml", ".yml"):
file_data: dict[str, Any] = yaml.safe_load(text) or {}
else:
try:
file_data = json.loads(text)
except json.JSONDecodeError:
# Fallback to YAML
file_data = yaml.safe_load(text) or {}
if not isinstance(file_data, dict):
typer.echo("Error: Import file must contain a JSON/YAML object.", err=True)
raise typer.Exit(code=1)
# Determine context name
resolved_name = name or file_data.get("context_name")
if not resolved_name:
# Infer from filename (stem)
resolved_name = input_file.stem
ctx_mgr = ContextManager(resolved_name, context_dir)
if ctx_mgr.exists() and not update:
typer.echo(
f"Error: Context '{resolved_name}' already exists. "
"Use --update to replace.",
err=True,
)
raise typer.Exit(code=1)
strategy = "replace" if ctx_mgr.exists() else "create"
# Import via ContextManager
ctx_mgr.import_context(input_file)
items = len(ctx_mgr.messages)
data: dict[str, Any] = {
"context_import": {
"context": resolved_name,
"input": str(input_file),
"items": items,
},
"merge": {
"strategy": strategy,
"conflicts": 0,
},
}
panels = [
(
"Context Import",
(
f"[bold]Context:[/bold] {resolved_name}\n"
f"[bold]Input:[/bold] {input_file}\n"
f"[bold]Items:[/bold] {items}"
),
),
(
"Merge",
(f"[bold]Strategy:[/bold] {strategy}\n[bold]Conflicts:[/bold] 0"),
),
]
_render_output(data, fmt, rich_panels=panels, ok_message="Import completed")