feat(cli): add event commands, --json/--agent flag, agent output sanitization, and edge-case hardening for CLI SDKs (#4649)

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

cli/python/src/mem0_cli/commands/events_cmd.py

@@ -0,0 +1,176 @@
+"""Event commands: list and status."""
+
+from __future__ import annotations
+
+import typer
+from rich.console import Console
+from rich.panel import Panel
+from rich.table import Table
+
+from mem0_cli.backend.base import Backend
+from mem0_cli.branding import (
+    ACCENT_COLOR,
+    BRAND_COLOR,
+    DIM_COLOR,
+    ERROR_COLOR,
+    SUCCESS_COLOR,
+    WARNING_COLOR,
+    print_info,
+    timed_status,
+)
+from mem0_cli.output import format_agent_envelope, format_json
+
+console = Console()
+err_console = Console(stderr=True)
+
+_STATUS_STYLE = {
+    "SUCCEEDED": f"[{SUCCESS_COLOR}]SUCCEEDED[/]",
+    "PENDING": f"[{ACCENT_COLOR}]PENDING[/]",
+    "FAILED": f"[{ERROR_COLOR}]FAILED[/]",
+    "PROCESSING": f"[{WARNING_COLOR}]PROCESSING[/]",
+}
+
+
+def _status_styled(status: str) -> str:
+    return _STATUS_STYLE.get(status.upper(), status)
+
+
+def cmd_event_list(backend: Backend, *, output: str = "table") -> None:
+    """List recent background events."""
+    from mem0_cli.state import is_agent_mode, set_current_command
+
+    set_current_command("event list")
+    if is_agent_mode():
+        output = "agent"
+    import time as _time
+
+    _start = _time.perf_counter()
+    with timed_status(err_console, "Fetching events...") as _ts:
+        try:
+            results = backend.list_events()
+        except Exception as e:
+            _ts.error_msg = str(e)
+            raise typer.Exit(1) from None
+
+    _elapsed = _time.perf_counter() - _start
+
+    if output == "agent":
+        format_agent_envelope(
+            console,
+            command="event list",
+            data=results,
+            count=len(results),
+            duration_ms=int(_elapsed * 1000),
+        )
+        return
+
+    if output == "json":
+        format_json(console, results)
+        return
+
+    if not results:
+        console.print()
+        print_info(console, "No events found.")
+        console.print()
+        return
+
+    table = Table(
+        border_style=BRAND_COLOR,
+        header_style=f"bold {ACCENT_COLOR}",
+        row_styles=["", "dim"],
+        padding=(0, 1),
+    )
+    table.add_column("Event ID", style="dim", max_width=10, no_wrap=True)
+    table.add_column("Type", max_width=14)
+    table.add_column("Status", max_width=12)
+    table.add_column("Latency", max_width=10, justify="right")
+    table.add_column("Created", max_width=20)
+
+    for ev in results:
+        ev_id = str(ev.get("id", ""))[:8]
+        ev_type = str(ev.get("event_type", "—"))
+        status = str(ev.get("status", "—"))
+        latency = ev.get("latency")
+        latency_str = f"{latency:.0f}ms" if isinstance(latency, (int, float)) else "—"
+        created = str(ev.get("created_at", "—"))[:19].replace("T", " ")
+        table.add_row(ev_id, ev_type, _status_styled(status), latency_str, created)
+
+    console.print()
+    console.print(table)
+    console.print(f"  [{DIM_COLOR}]{len(results)} event{'s' if len(results) != 1 else ''}[/]")
+    console.print()
+
+
+def cmd_event_status(backend: Backend, event_id: str, *, output: str = "text") -> None:
+    """Get the status of a specific background event."""
+    from mem0_cli.state import is_agent_mode, set_current_command
+
+    set_current_command("event status")
+    if is_agent_mode():
+        output = "agent"
+    import time as _time
+
+    _start = _time.perf_counter()
+    with timed_status(err_console, "Fetching event...") as _ts:
+        try:
+            ev = backend.get_event(event_id)
+        except Exception as e:
+            _ts.error_msg = str(e)
+            raise typer.Exit(1) from None
+
+    _elapsed = _time.perf_counter() - _start
+
+    if output == "agent":
+        format_agent_envelope(
+            console,
+            command="event status",
+            data=ev,
+            duration_ms=int(_elapsed * 1000),
+        )
+        return
+
+    if output == "json":
+        format_json(console, ev)
+        return
+
+    status = str(ev.get("status", "—"))
+    ev_type = str(ev.get("event_type", "—"))
+    latency = ev.get("latency")
+    latency_str = f"{latency:.0f}ms" if isinstance(latency, (int, float)) else "—"
+    created = str(ev.get("created_at", "—"))[:19].replace("T", " ")
+    updated = str(ev.get("updated_at", "—"))[:19].replace("T", " ")
+    results = ev.get("results")
+
+    lines = []
+    lines.append(f"  [{DIM_COLOR}]Event ID:[/]     {event_id}")
+    lines.append(f"  [{DIM_COLOR}]Type:[/]         {ev_type}")
+    lines.append(f"  [{DIM_COLOR}]Status:[/]       {_status_styled(status)}")
+    lines.append(f"  [{DIM_COLOR}]Latency:[/]      {latency_str}")
+    lines.append(f"  [{DIM_COLOR}]Created:[/]      {created}")
+    lines.append(f"  [{DIM_COLOR}]Updated:[/]      {updated}")
+
+    if results:
+        lines.append("")
+        lines.append(f"  [{DIM_COLOR}]Results ({len(results)}):[/]")
+        for r in results:
+            mem_id = str(r.get("id", ""))[:8]
+            data = r.get("data", {})
+            memory = data.get("memory", "") if isinstance(data, dict) else str(data)
+            ev_name = str(r.get("event", ""))
+            user = str(r.get("user_id", ""))
+            detail = f"{ev_name}  {memory}"
+            if user:
+                detail += f"  [{DIM_COLOR}](user_id={user})[/]"
+            lines.append(f"    [{SUCCESS_COLOR}]·[/] {detail}  [{DIM_COLOR}]({mem_id})[/]")
+
+    content = "\n".join(lines)
+    panel = Panel(
+        content,
+        title=f"[{BRAND_COLOR}]Event Status[/]",
+        title_align="left",
+        border_style=BRAND_COLOR,
+        padding=(1, 1),
+    )
+    console.print()
+    console.print(panel)
+    console.print()