feat(mem0-plugin): add /mem0:stats, /mem0:health, /mem0:digest skills (Tier 6)

Observability tier: - /mem0:stats: session + lifetime memory dashboard with category breakdown - /mem0:health: 5-check diagnostic (API key, identity, MCP, write/read, tracker) - /mem0:digest: weekly memory activity summary with time-scoped searches - session_stats.py: add `peek` command (JSON without clearing), per-category counters
2026-05-21 18:13:44 +05:30
parent 82ab253f2c
commit 81211be473
5 changed files with 318 additions and 4 deletions
@@ -28,7 +28,13 @@ def _load() -> dict:
                return json.load(f)
        except (json.JSONDecodeError, OSError):
            pass
-    return {"adds": 0, "searches": 0, "categories": [], "started": datetime.now().isoformat()}
+    return {
+        "adds": 0,
+        "searches": 0,
+        "categories": [],
+        "category_counts": {},
+        "started": datetime.now().isoformat(),
+    }


 def _save(stats: dict) -> None:
@@ -37,14 +43,23 @@ def _save(stats: dict) -> None:


 def init() -> None:
-    _save({"adds": 0, "searches": 0, "categories": [], "started": datetime.now().isoformat()})
+    _save({
+        "adds": 0,
+        "searches": 0,
+        "categories": [],
+        "category_counts": {},
+        "started": datetime.now().isoformat(),
+    })


 def record_add(category: str = "") -> None:
    stats = _load()
    stats["adds"] = stats.get("adds", 0) + 1
-    if category and category not in stats.get("categories", []):
-        stats.setdefault("categories", []).append(category)
+    if category:
+        if category not in stats.get("categories", []):
+            stats.setdefault("categories", []).append(category)
+        counts = stats.setdefault("category_counts", {})
+        counts[category] = counts.get(category, 0) + 1
    _save(stats)


@@ -54,6 +69,12 @@ def record_search() -> None:
    _save(stats)


+def peek() -> str:
+    """Return current stats as JSON without clearing the file."""
+    stats = _load()
+    return json.dumps(stats)
+
+
 def report() -> str:
    stats = _load()
    adds = stats.get("adds", 0)
@@ -90,6 +111,8 @@ def main() -> int:
        record_add(category)
    elif cmd == "search":
        record_search()
+    elif cmd == "peek":
+        print(peek())
    elif cmd == "report":
        result = report()
        if result:
@@ -0,0 +1,75 @@
+---
+name: mem0-digest
+description: >
+  Summarize recent memory activity for the current project. Shows new memories,
+  categories touched, and growth trends over the past 7 days.
+  TRIGGER: user runs /mem0:digest, or asks "weekly summary", "what's new in memory",
+  "mem0 digest", "memory recap".
+---
+
+# Mem0 Weekly Digest
+
+Summarize recent memory activity for the current project.
+
+## Execution
+
+### Step 1: Fetch recent memories
+
+Call `search_memories` in parallel with different time-scoped queries:
+
+1. `query="decisions made this week"`, `filters={"AND": [{"user_id": "<id>"}, {"app_id": "<pid>"}, {"created_at": {"gte": "<7 days ago YYYY-MM-DD>"}}]}`, `limit=20`
+2. `query="bugs errors fixes"`, same time filter, `limit=20`
+3. `query="patterns conventions learnings"`, same time filter, `limit=20`
+
+Also call `get_memories` with `user_id` + `app_id` to get the full count for comparison.
+
+### Step 2: Deduplicate and analyze
+
+Merge results by memory ID. For each memory, extract:
+- `created_at` date
+- `categories[0]` or `metadata.type`
+- First 100 chars of content
+
+Group into:
+- **New this week** (created in last 7 days)
+- **Older** (created before, but surfaced by search)
+
+Calculate:
+- Total memories in project
+- Memories added in last 7 days
+- Most active categories this week
+- Days with most activity
+
+### Step 3: Display
+
+```
+## mem0 Weekly Digest — <project_id>
+Period: <start_date> to <today>
+
+### New Memories This Week: <N>
+<category>: <count>
+  - <memory summary, 80 chars> (<date>)
+  - ...
+<category>: <count>
+  - ...
+
+### Activity Pattern
+Most active day: <day> (<N> memories)
+Categories touched: <list>
+
+### Project Totals
+Total memories: <N> (up <N> from last week)
+Top categories: <top 3 by count>
+
+### Highlights
+<2-3 sentence summary of the most important decisions, learnings, or patterns stored this week>
+```
+
+### Step 4: Empty state
+
+If no memories in the last 7 days:
+```
+No new memories in the past week for <project_id>.
+Total project memories: <N>.
+Tip: mem0 captures learnings automatically as you work. Start coding!
+```
@@ -0,0 +1,78 @@
+---
+name: mem0-health
+description: >
+  Diagnostic health check for the mem0 plugin. Verifies API key, MCP server
+  connectivity, identity resolution, and memory read/write capability.
+  TRIGGER: user runs /mem0:health, or asks "is mem0 working", "mem0 health",
+  "check mem0 connection", "debug mem0".
+---
+
+# Mem0 Health Check
+
+Run a diagnostic check on the mem0 plugin. Useful for troubleshooting.
+
+## Execution
+
+Run ALL checks, then display a single summary. Do not stop on the first failure.
+
+### Check 1: API key
+
+```bash
+echo "${MEM0_API_KEY:-${CLAUDE_PLUGIN_OPTION_MEM0_API_KEY:-NOT_SET}}"
+```
+
+- If `NOT_SET`: FAIL — "No API key configured"
+- If set: PASS — show first 6 chars + `...` (never print the full key)
+
+### Check 2: Identity resolution
+
+Read the active identity from the SessionStart banner or resolve manually:
+- `user_id`: from `MEM0_RESOLVED_USER_ID` or `$USER`
+- `project_id`: from `MEM0_PROJECT_ID` or current directory name
+- `branch`: from `MEM0_BRANCH` or `git rev-parse --abbrev-ref HEAD`
+
+PASS if all three are non-empty. WARN if any falls back to defaults.
+
+### Check 3: MCP server connectivity
+
+Call `search_memories` with:
+- `query="health check"`, `user_id=<id>`, `limit=1`
+- `filters={"AND": [{"user_id": "<id>"}, {"app_id": "<project_id>"}]}`
+
+- If returns successfully (even empty): PASS
+- If errors: FAIL — show the error message
+
+### Check 4: Memory write capability
+
+Call `add_memory` with:
+- `messages=[{"role": "user", "content": "Health check probe — safe to delete."}]`
+- `user_id=<id>`, `app_id=<project_id>`
+- `metadata={"type": "health_check", "probe": true}`
+
+If it returns a memory ID: PASS — then immediately call `delete_memory` with that ID to clean up.
+If it errors: FAIL — show the error.
+
+### Check 5: Session stats tracker
+
+```bash
+SCRIPT_DIR="${CLAUDE_PLUGIN_ROOT:-${CODEX_PLUGIN_ROOT:-${CURSOR_PLUGIN_ROOT:-}}}/scripts"
+python3 "$SCRIPT_DIR/session_stats.py" peek 2>/dev/null && echo "OK" || echo "FAIL"
+```
+
+### Display
+
+```
+## mem0 Health Check
+
+| Check              | Status | Detail                        |
+|--------------------|--------|-------------------------------|
+| API Key            | PASS   | m0-dVe...                     |
+| Identity           | PASS   | user=kartik, project=mem0     |
+| MCP Connectivity   | PASS   | 142ms round-trip              |
+| Memory Write/Read  | PASS   | write + delete OK             |
+| Session Tracker    | PASS   | stats file active             |
+
+All checks passed. mem0 is healthy.
+```
+
+If any check fails, add a `## Troubleshooting` section with specific fix steps for each failure.
@@ -0,0 +1,76 @@
+---
+name: mem0-stats
+description: >
+  Show memory statistics for the current session and project lifetime.
+  Combines local session counters with API-fetched totals.
+  TRIGGER: user runs /mem0:stats, or asks "how many memories", "mem0 stats",
+  "memory usage", "show memory count".
+---
+
+# Mem0 Stats
+
+Show session and lifetime memory statistics.
+
+## Execution
+
+### Step 1: Gather session stats
+
+Run the session stats reporter:
+
+```bash
+SCRIPT_DIR="${CLAUDE_PLUGIN_ROOT:-${CODEX_PLUGIN_ROOT:-${CURSOR_PLUGIN_ROOT:-}}}/scripts"
+python3 "$SCRIPT_DIR/session_stats.py" peek 2>/dev/null || echo "{}"
+```
+
+The `peek` command returns JSON without clearing the stats file (unlike `report`).
+
+If the script returns empty or errors, note "No session data available" and continue.
+
+### Step 2: Fetch lifetime stats from API
+
+Call `get_memories` with:
+- `user_id=<active_user_id>`
+- `app_id=<active_project_id>`
+- `page_size=100`
+
+Count the returned memories. Group them by:
+1. `categories[0]` (platform-assigned) — primary grouping
+2. `metadata.type` (agent-assigned) — secondary if no categories
+3. `created_at` date — for age analysis
+
+Also run a `search_memories` call with `query="project"`, `limit=1` to measure round-trip latency (time the call).
+
+### Step 3: Display
+
+Print a compact dashboard:
+
+```
+## mem0 Stats
+
+### This Session
+  Memories written:  <N>
+  Searches run:      <N>
+  Categories touched: <list>
+
+### Project Lifetime (<project_id>)
+  Total memories:    <N>
+  By category:
+    Architecture Decisions: <N>
+    Anti-Patterns:          <N>
+    Task Learnings:         <N>
+    Coding Conventions:     <N>
+    User Preferences:       <N>
+    Tooling & Setup:        <N>
+    Session State:          <N>
+    Other:                  <N>
+  Oldest memory:     <date>
+  Newest memory:     <date>
+
+### Health
+  API latency:       <N>ms
+  User:              <user_id>
+  Project:           <project_id>
+  Branch:            <branch>
+```
+
+Skip any section with zero data. Don't show empty categories.
@@ -141,3 +141,65 @@ def test_cli_report_no_data(tmp_path):
        env=env,
    )
    assert result.returncode == 0
+
+
+def test_peek_returns_json_without_clearing(_isolate_stats_file):
+    """peek returns JSON stats without deleting the stats file."""
+    import session_stats
+
+    session_stats.init()
+    session_stats.record_add("decisions")
+    session_stats.record_add("decisions")
+    session_stats.record_search()
+
+    result = session_stats.peek()
+    data = json.loads(result)
+    assert data["adds"] == 2
+    assert data["searches"] == 1
+
+    assert os.path.isfile(_isolate_stats_file)
+
+
+def test_category_counts_tracked(_isolate_stats_file):
+    """category_counts tracks per-category add counts."""
+    import session_stats
+
+    session_stats.init()
+    session_stats.record_add("bug_fixes")
+    session_stats.record_add("bug_fixes")
+    session_stats.record_add("bug_fixes")
+    session_stats.record_add("decisions")
+
+    with open(_isolate_stats_file) as f:
+        data = json.load(f)
+    assert data["category_counts"]["bug_fixes"] == 3
+    assert data["category_counts"]["decisions"] == 1
+
+
+def test_category_counts_empty_category_not_tracked(_isolate_stats_file):
+    """Empty category string doesn't appear in category_counts."""
+    import session_stats
+
+    session_stats.init()
+    session_stats.record_add("")
+    session_stats.record_add()
+
+    with open(_isolate_stats_file) as f:
+        data = json.load(f)
+    assert data["category_counts"] == {}
+
+
+def test_cli_peek(tmp_path):
+    """Test CLI invocation: session_stats.py peek outputs JSON."""
+    env = {**os.environ, "USER": "test"}
+    subprocess.run(
+        [sys.executable, os.path.join(SCRIPTS_DIR, "session_stats.py"), "init"],
+        capture_output=True, text=True, env=env,
+    )
+    result = subprocess.run(
+        [sys.executable, os.path.join(SCRIPTS_DIR, "session_stats.py"), "peek"],
+        capture_output=True, text=True, env=env,
+    )
+    assert result.returncode == 0
+    data = json.loads(result.stdout)
+    assert "adds" in data