feat(mem0-plugin): file-context injection, stop-hook summaries & activity timeline (#5346)
This commit is contained in:
Kartik
@@ -0,0 +1,51 @@
|
||||
"""Shared formatting helpers for mem0 plugin hooks.
|
||||
|
||||
Constants and utilities used by file_context.py, session_timeline.py,
|
||||
and any future hook that displays memories.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
TYPE_ICONS = {
|
||||
"decision": "⚖️",
|
||||
"anti_pattern": "\U0001f534",
|
||||
"bug_fix": "\U0001f534",
|
||||
"convention": "\U0001f504",
|
||||
"task_learning": "\U0001f535",
|
||||
"user_preference": "\U0001f7e3",
|
||||
"session_summary": "\U0001f4cb",
|
||||
"session_state": "\U0001f4cb",
|
||||
"project_profile": "\U0001f4d6",
|
||||
"compact_summary": "\U0001f4cb",
|
||||
"auto_capture": "✅",
|
||||
"environmental": "🌐",
|
||||
"health_check": "🩺",
|
||||
}
|
||||
|
||||
|
||||
def format_age(memory: dict) -> str:
|
||||
"""Format how long ago a memory was created, e.g. '2h ago', '3d ago'."""
|
||||
created = memory.get("created_at", "")
|
||||
if not created:
|
||||
return ""
|
||||
try:
|
||||
from datetime import datetime, timezone
|
||||
|
||||
if created.endswith("Z"):
|
||||
created = created[:-1] + "+00:00"
|
||||
dt = datetime.fromisoformat(created)
|
||||
now = datetime.now(timezone.utc)
|
||||
delta = now - dt
|
||||
seconds = int(delta.total_seconds())
|
||||
if seconds < 3600:
|
||||
return f"{seconds // 60}m ago"
|
||||
if seconds < 86400:
|
||||
return f"{seconds // 3600}h ago"
|
||||
days = seconds // 86400
|
||||
if days == 1:
|
||||
return "1d ago"
|
||||
if days < 30:
|
||||
return f"{days}d ago"
|
||||
return f"{days // 30}mo ago"
|
||||
except Exception:
|
||||
return ""
|
||||
@@ -37,23 +37,28 @@ def search_memories(
|
||||
min_score: float = 0.0,
|
||||
rerank: bool = False,
|
||||
threshold: float = 0.3,
|
||||
global_search: bool = False,
|
||||
) -> list[dict]:
|
||||
if not api_key:
|
||||
return []
|
||||
|
||||
base_clauses: list[dict] = [{"user_id": user_id}, {"app_id": project_id}]
|
||||
if metadata_type:
|
||||
base_clauses.append({"metadata": {"type": metadata_type}})
|
||||
if metadata_filters:
|
||||
for key, value in metadata_filters.items():
|
||||
base_clauses.append({"metadata": {key: value}})
|
||||
if global_search:
|
||||
filters: dict = {"OR": [{"user_id": "*"}]}
|
||||
else:
|
||||
base_clauses: list[dict] = [{"user_id": user_id}, {"app_id": project_id}]
|
||||
if metadata_type:
|
||||
base_clauses.append({"metadata": {"type": metadata_type}})
|
||||
if metadata_filters:
|
||||
for key, value in metadata_filters.items():
|
||||
base_clauses.append({"metadata": {key: value}})
|
||||
filters = {"AND": base_clauses}
|
||||
|
||||
base_payload: dict = {"query": query, "top_k": top_k, "threshold": threshold}
|
||||
if rerank:
|
||||
base_payload["rerank"] = True
|
||||
|
||||
try:
|
||||
payload = {**base_payload, "filters": {"AND": list(base_clauses)}}
|
||||
payload = {**base_payload, "filters": filters}
|
||||
results = _do_search(api_key, payload)[:top_k]
|
||||
|
||||
if min_score > 0:
|
||||
|
||||
@@ -0,0 +1,264 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Capture a structured session summary on Stop hook.
|
||||
|
||||
Runs on every Stop (end of each assistant turn). Each invocation reads
|
||||
the transcript JSONL, extracts the latest assistant message and all
|
||||
files touched so far, then stores via mem0 API with infer=True. Uses
|
||||
run_id=session_id to scope infer dedup to the session, so the final
|
||||
stored summary reflects the most recent turn — not just the first.
|
||||
|
||||
Input: JSON on stdin with transcript_path, session_id, cwd, agent_id
|
||||
Output: stderr logs only (exit 0 always — must not block)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from datetime import date, timedelta
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
from _identity import resolve_api_key, resolve_user_id
|
||||
from _project import resolve_branch, resolve_project_id
|
||||
|
||||
log = logging.getLogger("mem0-session-summary")
|
||||
log.setLevel(logging.DEBUG)
|
||||
_handler = logging.StreamHandler(sys.stderr)
|
||||
_handler.setFormatter(logging.Formatter("[mem0-session-summary] %(message)s"))
|
||||
log.addHandler(_handler)
|
||||
|
||||
if os.environ.get("MEM0_DEBUG"):
|
||||
_log_dir = os.path.expanduser("~/.mem0")
|
||||
try:
|
||||
os.makedirs(_log_dir, exist_ok=True)
|
||||
_fh = logging.FileHandler(os.path.join(_log_dir, "hooks.log"))
|
||||
_fh.setFormatter(logging.Formatter("[mem0-session-summary] %(asctime)s %(message)s"))
|
||||
log.addHandler(_fh)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
API_URL = "https://api.mem0.ai"
|
||||
MAX_TAIL_LINES = 3000
|
||||
MAX_SUMMARY_CHARS = 50000
|
||||
SUMMARY_EXPIRY_DAYS = 90
|
||||
|
||||
SYSTEM_TAG_RE = re.compile(
|
||||
r"<(?:system-reminder|private|claude-mem-context|persisted-output|system_instruction)>"
|
||||
r".*?"
|
||||
r"</(?:system-reminder|private|claude-mem-context|persisted-output|system_instruction)>",
|
||||
re.DOTALL,
|
||||
)
|
||||
|
||||
|
||||
def tail_lines(filepath: str, n: int) -> list[str]:
|
||||
try:
|
||||
with open(filepath, "rb") as f:
|
||||
f.seek(0, 2)
|
||||
file_size = f.tell()
|
||||
if file_size == 0:
|
||||
return []
|
||||
chunk_size = min(file_size, n * 4096)
|
||||
f.seek(max(0, file_size - chunk_size))
|
||||
data = f.read().decode("utf-8", errors="replace")
|
||||
return data.splitlines()[-n:]
|
||||
except OSError:
|
||||
return []
|
||||
|
||||
|
||||
def extract_last_assistant_message(lines: list[str]) -> str:
|
||||
"""Walk transcript backwards, return text content of the last assistant message."""
|
||||
for line in reversed(lines):
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
if '"type":"assistant"' not in line and '"type": "assistant"' not in line:
|
||||
continue
|
||||
try:
|
||||
entry = json.loads(line)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
if entry.get("type") != "assistant":
|
||||
continue
|
||||
message = entry.get("message", {})
|
||||
content = message.get("content", [])
|
||||
if isinstance(content, str):
|
||||
return content
|
||||
if isinstance(content, list):
|
||||
parts = []
|
||||
for block in content:
|
||||
if isinstance(block, str):
|
||||
parts.append(block)
|
||||
elif isinstance(block, dict) and block.get("type") == "text":
|
||||
parts.append(block.get("text", ""))
|
||||
return "\n".join(parts).strip()
|
||||
return ""
|
||||
|
||||
|
||||
def extract_files_touched(lines: list[str]) -> list[str]:
|
||||
"""Extract unique file paths from tool_use content blocks in transcript."""
|
||||
files = set()
|
||||
file_ext_re = re.compile(
|
||||
r"[a-zA-Z0-9_./-]+\.(?:py|ts|tsx|js|jsx|rs|go|rb|java|sh|yaml|yml|json|toml|md|sql|css|html)"
|
||||
)
|
||||
for line in lines:
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
if '"tool_use"' not in line and '"file_path"' not in line:
|
||||
continue
|
||||
try:
|
||||
entry = json.loads(line)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
content = entry.get("message", {}).get("content", [])
|
||||
if not isinstance(content, list):
|
||||
continue
|
||||
for block in content:
|
||||
if not isinstance(block, dict) or block.get("type") != "tool_use":
|
||||
continue
|
||||
inp = block.get("input", {})
|
||||
if not isinstance(inp, dict):
|
||||
continue
|
||||
fp = inp.get("file_path", "")
|
||||
if fp:
|
||||
files.add(fp)
|
||||
command = inp.get("command", "")
|
||||
if command:
|
||||
for match in file_ext_re.findall(command):
|
||||
files.add(match)
|
||||
return sorted(files)[:20]
|
||||
|
||||
|
||||
def strip_tags(text: str) -> str:
|
||||
return SYSTEM_TAG_RE.sub("", text).strip()
|
||||
|
||||
|
||||
def build_summary_prompt(assistant_msg: str, files: list[str]) -> str:
|
||||
"""Build a structured prompt that helps mem0's AI extract a good summary."""
|
||||
files_section = ""
|
||||
if files:
|
||||
file_list = ", ".join(files[:10])
|
||||
files_section = f"\n\nFiles touched during this session: {file_list}"
|
||||
|
||||
return (
|
||||
f"Session summary — store the following as a structured session summary.\n\n"
|
||||
f"What the assistant accomplished in this session:\n"
|
||||
f"{assistant_msg[:MAX_SUMMARY_CHARS]}"
|
||||
f"{files_section}\n\n"
|
||||
f"Extract and remember: what was requested, what was investigated, "
|
||||
f"key decisions made, what was completed, and what needs to happen next."
|
||||
)
|
||||
|
||||
|
||||
def store_summary(
|
||||
api_key: str,
|
||||
summary_prompt: str,
|
||||
user_id: str,
|
||||
session_id: str,
|
||||
project_id: str,
|
||||
branch: str,
|
||||
files: list[str],
|
||||
) -> bool:
|
||||
expires = (date.today() + timedelta(days=SUMMARY_EXPIRY_DAYS)).isoformat()
|
||||
metadata = {
|
||||
"type": "session_summary",
|
||||
"source": "stop-hook",
|
||||
"session_id": session_id,
|
||||
}
|
||||
if branch:
|
||||
metadata["branch"] = branch
|
||||
if files:
|
||||
metadata["files_touched"] = json.dumps(files[:20])
|
||||
|
||||
body = {
|
||||
"messages": [{"role": "user", "content": summary_prompt}],
|
||||
"user_id": user_id,
|
||||
"app_id": project_id,
|
||||
"run_id": session_id,
|
||||
"metadata": metadata,
|
||||
"infer": True,
|
||||
"expiration_date": expires,
|
||||
}
|
||||
|
||||
data = json.dumps(body).encode("utf-8")
|
||||
req = urllib.request.Request(
|
||||
f"{API_URL}/v3/memories/add/",
|
||||
data=data,
|
||||
headers={
|
||||
"Content-Type": "application/json",
|
||||
"Authorization": f"Token {api_key}",
|
||||
},
|
||||
method="POST",
|
||||
)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=15) as resp:
|
||||
if resp.status in (200, 201):
|
||||
log.info("Session summary stored")
|
||||
return True
|
||||
log.warning("API returned status %d", resp.status)
|
||||
return False
|
||||
except urllib.error.URLError as e:
|
||||
log.warning("API call failed: %s", e)
|
||||
return False
|
||||
|
||||
|
||||
def main():
|
||||
api_key = resolve_api_key()
|
||||
if not api_key:
|
||||
log.debug("MEM0_API_KEY not set, skipping")
|
||||
return
|
||||
|
||||
try:
|
||||
hook_input = json.loads(sys.stdin.read())
|
||||
except (json.JSONDecodeError, OSError):
|
||||
log.debug("No valid JSON on stdin")
|
||||
return
|
||||
|
||||
# Guard: skip subagent sessions (only root sessions get summaries)
|
||||
agent_id = hook_input.get("agent_id", "")
|
||||
if agent_id:
|
||||
log.debug("Subagent session (agent_id=%s), skipping", agent_id)
|
||||
return
|
||||
|
||||
transcript_path = hook_input.get("transcript_path", "")
|
||||
if not transcript_path:
|
||||
log.debug("No transcript_path provided")
|
||||
return
|
||||
|
||||
session_id = hook_input.get("session_id", "")
|
||||
cwd = hook_input.get("cwd") or None
|
||||
|
||||
user_id = resolve_user_id()
|
||||
project_id = resolve_project_id(cwd)
|
||||
branch = resolve_branch(cwd)
|
||||
|
||||
lines = tail_lines(transcript_path, MAX_TAIL_LINES)
|
||||
if not lines:
|
||||
log.debug("Transcript empty or unreadable: %s", transcript_path)
|
||||
return
|
||||
|
||||
assistant_msg = extract_last_assistant_message(lines)
|
||||
if not assistant_msg or len(assistant_msg.strip()) < 100:
|
||||
log.debug("Assistant message too short (%d chars) — skipping", len(assistant_msg.strip()))
|
||||
return
|
||||
|
||||
assistant_msg = strip_tags(assistant_msg)
|
||||
files = extract_files_touched(lines)
|
||||
|
||||
summary_prompt = build_summary_prompt(assistant_msg, files)
|
||||
|
||||
log.info("Capturing session summary (%d chars, %d files)", len(assistant_msg), len(files))
|
||||
store_summary(api_key, summary_prompt, user_id, session_id, project_id, branch, files)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
main()
|
||||
except Exception as e:
|
||||
log.error("Unexpected error: %s", e)
|
||||
sys.exit(0)
|
||||
@@ -0,0 +1,133 @@
|
||||
#!/usr/bin/env python3
|
||||
"""File-context injection for PreToolUse/Read hook.
|
||||
|
||||
When Claude is about to read a file, this script searches mem0 for
|
||||
memories that reference that file path and returns a compact timeline
|
||||
of prior work. This gives Claude context like "last time you fixed a
|
||||
null pointer here" before it reads the file.
|
||||
|
||||
Modeled after claude-mem's file-context handler but adapted for mem0's
|
||||
cloud API architecture.
|
||||
|
||||
Input: file_path (positional arg), env vars for identity
|
||||
Output: JSON to stdout with hookSpecificOutput.additionalContext
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
from _formatting import TYPE_ICONS, format_age
|
||||
from _identity import resolve_api_key, resolve_user_id
|
||||
from _project import resolve_project_id
|
||||
from _search import search_memories
|
||||
|
||||
FILE_READ_GATE_MIN_BYTES = 1500
|
||||
MAX_RESULTS = 5
|
||||
SEARCH_TIMEOUT = 5
|
||||
|
||||
|
||||
def gate_file(file_path: str, cwd: str) -> str | None:
|
||||
"""Return the resolved absolute path if the file passes gating, else None."""
|
||||
if not file_path:
|
||||
return None
|
||||
p = Path(file_path)
|
||||
if not p.is_absolute():
|
||||
p = Path(cwd) / p
|
||||
try:
|
||||
p = p.resolve()
|
||||
if not p.is_file():
|
||||
return None
|
||||
if p.stat().st_size < FILE_READ_GATE_MIN_BYTES:
|
||||
return None
|
||||
return str(p)
|
||||
except OSError:
|
||||
return None
|
||||
|
||||
|
||||
def relative_path(abs_path: str, cwd: str) -> str:
|
||||
try:
|
||||
return os.path.relpath(abs_path, cwd)
|
||||
except ValueError:
|
||||
return abs_path
|
||||
|
||||
|
||||
def format_timeline(memories: list[dict], file_path: str) -> str:
|
||||
"""Format memories into a compact timeline for context injection."""
|
||||
if not memories:
|
||||
return ""
|
||||
|
||||
rel = file_path
|
||||
lines = [
|
||||
f"Prior work on `{rel}` — {len(memories)} memories found.",
|
||||
"Need details? Use `search_memories` with the memory ID.",
|
||||
"",
|
||||
]
|
||||
|
||||
for m in memories:
|
||||
mid = m.get("id", "?")[:8]
|
||||
text = (m.get("memory", "") or "")[:150].replace("\n", " ").strip()
|
||||
meta = m.get("metadata") or {}
|
||||
cat = meta.get("type", "unknown")
|
||||
icon = TYPE_ICONS.get(cat, "❓")
|
||||
age = format_age(m)
|
||||
age_str = f" ({age})" if age else ""
|
||||
lines.append(f"- {icon} [{cat}]{age_str} {text} [mem0:{mid}]")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def search_file_context(
|
||||
api_key: str, user_id: str, project_id: str, file_path: str, cwd: str
|
||||
) -> str:
|
||||
"""Search mem0 for memories related to a file path."""
|
||||
global_search = os.environ.get("MEM0_GLOBAL_SEARCH", "false") == "true"
|
||||
rel = relative_path(file_path, cwd)
|
||||
basename = os.path.basename(file_path)
|
||||
|
||||
query = f"{rel} {basename}" if rel != basename else rel
|
||||
results = search_memories(
|
||||
api_key, user_id, project_id, query,
|
||||
top_k=MAX_RESULTS, threshold=0.3,
|
||||
global_search=global_search,
|
||||
)
|
||||
|
||||
results = results[:MAX_RESULTS]
|
||||
|
||||
return format_timeline(results, rel)
|
||||
|
||||
|
||||
def main():
|
||||
if len(sys.argv) < 2:
|
||||
sys.exit(0)
|
||||
|
||||
file_path = sys.argv[1]
|
||||
cwd = sys.argv[2] if len(sys.argv) > 2 else os.getcwd()
|
||||
|
||||
api_key = resolve_api_key()
|
||||
if not api_key:
|
||||
sys.exit(0)
|
||||
|
||||
resolved = gate_file(file_path, cwd)
|
||||
if not resolved:
|
||||
sys.exit(0)
|
||||
|
||||
user_id = resolve_user_id()
|
||||
project_id = resolve_project_id(cwd)
|
||||
|
||||
timeline = search_file_context(api_key, user_id, project_id, resolved, cwd)
|
||||
if not timeline:
|
||||
sys.exit(0)
|
||||
|
||||
print(timeline, end="")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
main()
|
||||
except Exception:
|
||||
pass
|
||||
sys.exit(0)
|
||||
Executable
+51
@@ -0,0 +1,51 @@
|
||||
#!/usr/bin/env bash
|
||||
# Hook: PreToolUse (matcher: Read)
|
||||
#
|
||||
# Injects prior work context before Claude reads a file. Searches mem0
|
||||
# for memories referencing the file path and returns a compact timeline.
|
||||
#
|
||||
# Modeled after claude-mem's file-context handler, adapted for mem0 cloud API.
|
||||
#
|
||||
# Input: JSON on stdin with tool_name, tool_input (file_path), cwd
|
||||
# Output: JSON with hookSpecificOutput.additionalContext + permissionDecision
|
||||
#
|
||||
# Must never block the Read — silent exit on any failure.
|
||||
|
||||
set -uo pipefail
|
||||
|
||||
INPUT=$(cat)
|
||||
|
||||
# Extract file path from tool_input
|
||||
FILE_PATH=$(echo "$INPUT" | jq -r '.tool_input.file_path // ""' 2>/dev/null || echo "")
|
||||
if [ -z "$FILE_PATH" ]; then
|
||||
exit 0
|
||||
fi
|
||||
|
||||
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
|
||||
|
||||
# Resolve API key (covers Desktop app users who set it in shell profile)
|
||||
if [ -z "${MEM0_API_KEY:-}" ]; then
|
||||
. "$SCRIPT_DIR/_identity.sh" 2>/dev/null || true
|
||||
fi
|
||||
if [ -z "${MEM0_API_KEY:-}" ]; then
|
||||
exit 0
|
||||
fi
|
||||
CWD=$(echo "$INPUT" | jq -r '.cwd // "."' 2>/dev/null || echo ".")
|
||||
|
||||
# Call the Python worker — it handles gating (file size, existence)
|
||||
TIMELINE=$(python3 "$SCRIPT_DIR/file_context.py" "$FILE_PATH" "$CWD" 2>/dev/null || echo "")
|
||||
|
||||
if [ -z "$TIMELINE" ]; then
|
||||
exit 0
|
||||
fi
|
||||
|
||||
# Return context injection with permissionDecision: allow
|
||||
jq -cn --arg ctx "$TIMELINE" '{
|
||||
hookSpecificOutput: {
|
||||
hookEventName: "PreToolUse",
|
||||
additionalContext: $ctx,
|
||||
permissionDecision: "allow"
|
||||
}
|
||||
}' 2>/dev/null || true
|
||||
|
||||
exit 0
|
||||
Executable
+41
@@ -0,0 +1,41 @@
|
||||
#!/usr/bin/env bash
|
||||
# Hook: preToolUse (matcher: Read) — Cursor variant
|
||||
#
|
||||
# Same as on_file_read.sh but uses CURSOR_PLUGIN_ROOT for path resolution
|
||||
# and sources Cursor-specific identity.
|
||||
|
||||
set -uo pipefail
|
||||
|
||||
INPUT=$(cat)
|
||||
|
||||
FILE_PATH=$(echo "$INPUT" | jq -r '.tool_input.file_path // ""' 2>/dev/null || echo "")
|
||||
if [ -z "$FILE_PATH" ]; then
|
||||
exit 0
|
||||
fi
|
||||
|
||||
if [ -z "${MEM0_API_KEY:-}" ]; then
|
||||
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
|
||||
. "$SCRIPT_DIR/_identity.sh" 2>/dev/null || true
|
||||
fi
|
||||
if [ -z "${MEM0_API_KEY:-}" ]; then
|
||||
exit 0
|
||||
fi
|
||||
|
||||
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
|
||||
CWD=$(echo "$INPUT" | jq -r '.cwd // "."' 2>/dev/null || echo ".")
|
||||
|
||||
TIMELINE=$(python3 "$SCRIPT_DIR/file_context.py" "$FILE_PATH" "$CWD" 2>/dev/null || echo "")
|
||||
|
||||
if [ -z "$TIMELINE" ]; then
|
||||
exit 0
|
||||
fi
|
||||
|
||||
jq -cn --arg ctx "$TIMELINE" '{
|
||||
hookSpecificOutput: {
|
||||
hookEventName: "PreToolUse",
|
||||
additionalContext: $ctx,
|
||||
permissionDecision: "allow"
|
||||
}
|
||||
}' 2>/dev/null || true
|
||||
|
||||
exit 0
|
||||
@@ -156,6 +156,14 @@ if [ "$SOURCE" = "startup" ]; then
|
||||
echo "New project with 0 memories. Invoke the mem0:onboard skill to import project files. Coding categories install automatically in the background."
|
||||
else
|
||||
echo "Search mem0 for recent decisions and task learnings before responding. Run 2 parallel searches: one for decision type, one for task_learning type."
|
||||
|
||||
# Inject compact recent activity timeline (non-blocking, 5s timeout)
|
||||
# Use perl alarm as portable timeout (macOS lacks GNU timeout)
|
||||
_TIMELINE=$(MEM0_CWD="$MEM0_CWD_RESOLVED" perl -e 'alarm 5; exec @ARGV' python3 "$SCRIPT_DIR/session_timeline.py" 2>/dev/null || echo "")
|
||||
if [ -n "$_TIMELINE" ]; then
|
||||
echo ""
|
||||
echo "$_TIMELINE"
|
||||
fi
|
||||
fi
|
||||
|
||||
_PROJ_KEY=$(printf '%s' "$MEM0_CWD_RESOLVED" | tr '/' '-')
|
||||
|
||||
Executable
+50
@@ -0,0 +1,50 @@
|
||||
#!/usr/bin/env bash
|
||||
# Hook: Stop
|
||||
#
|
||||
# Captures a structured session summary when a Claude Code session ends.
|
||||
# Parses the transcript, extracts the last assistant message and files
|
||||
# touched, then stores via mem0 API with infer=True for AI extraction.
|
||||
#
|
||||
# Guards:
|
||||
# - Skips subagent sessions (agent_id present)
|
||||
# - Skips if no API key
|
||||
# - Skips if no transcript_path
|
||||
# - Dedup via marker file
|
||||
#
|
||||
# Input: JSON on stdin with transcript_path, session_id, agent_id, cwd
|
||||
# Output: Nothing to stdout (background capture). Always exits 0.
|
||||
|
||||
set -uo pipefail
|
||||
|
||||
if [ -n "${MEM0_DEBUG:-}" ]; then
|
||||
mkdir -p "$HOME/.mem0" && exec 2>>"$HOME/.mem0/hooks.log"
|
||||
fi
|
||||
|
||||
INPUT=$(cat)
|
||||
|
||||
# Guard: skip subagent sessions
|
||||
AGENT_ID=$(echo "$INPUT" | jq -r '.agent_id // ""' 2>/dev/null || echo "")
|
||||
if [ -n "$AGENT_ID" ]; then
|
||||
exit 0
|
||||
fi
|
||||
|
||||
# Resolve identity if needed
|
||||
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
|
||||
. "$SCRIPT_DIR/_identity.sh" 2>/dev/null || true
|
||||
|
||||
if [ -z "${MEM0_API_KEY:-}" ]; then
|
||||
exit 0
|
||||
fi
|
||||
|
||||
TRANSCRIPT_PATH=$(echo "$INPUT" | jq -r '.transcript_path // ""' 2>/dev/null || echo "")
|
||||
if [ -z "$TRANSCRIPT_PATH" ]; then
|
||||
exit 0
|
||||
fi
|
||||
|
||||
# Run capture in the background — fires every turn now, so avoid blocking
|
||||
echo "$INPUT" | python3 "$SCRIPT_DIR/capture_session_summary.py" 2>/dev/null &
|
||||
|
||||
# Telemetry
|
||||
python3 "$SCRIPT_DIR/telemetry.py" session_stop 2>/dev/null &
|
||||
|
||||
exit 0
|
||||
Executable
+36
@@ -0,0 +1,36 @@
|
||||
#!/usr/bin/env bash
|
||||
# Hook: stop — Cursor variant
|
||||
#
|
||||
# Same as on_stop.sh but uses CURSOR_PLUGIN_ROOT and Cursor-specific
|
||||
# identity resolution.
|
||||
|
||||
set -uo pipefail
|
||||
|
||||
if [ -n "${MEM0_DEBUG:-}" ]; then
|
||||
mkdir -p "$HOME/.mem0" && exec 2>>"$HOME/.mem0/hooks.log"
|
||||
fi
|
||||
|
||||
INPUT=$(cat)
|
||||
|
||||
AGENT_ID=$(echo "$INPUT" | jq -r '.agent_id // ""' 2>/dev/null || echo "")
|
||||
if [ -n "$AGENT_ID" ]; then
|
||||
exit 0
|
||||
fi
|
||||
|
||||
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
|
||||
. "$SCRIPT_DIR/_identity.sh" 2>/dev/null || true
|
||||
|
||||
if [ -z "${MEM0_API_KEY:-}" ]; then
|
||||
exit 0
|
||||
fi
|
||||
|
||||
TRANSCRIPT_PATH=$(echo "$INPUT" | jq -r '.transcript_path // ""' 2>/dev/null || echo "")
|
||||
if [ -z "$TRANSCRIPT_PATH" ]; then
|
||||
exit 0
|
||||
fi
|
||||
|
||||
echo "$INPUT" | python3 "$SCRIPT_DIR/capture_session_summary.py" 2>/dev/null || true
|
||||
|
||||
python3 "$SCRIPT_DIR/telemetry.py" session_stop 2>/dev/null &
|
||||
|
||||
exit 0
|
||||
@@ -0,0 +1,106 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Fetch recent memories and format a compact timeline for SessionStart.
|
||||
|
||||
Searches mem0 cloud API for the most recent memories in the project
|
||||
and formats them as a compact activity timeline injected below the
|
||||
existing SessionStart banner.
|
||||
|
||||
Input: env vars for identity (MEM0_API_KEY, MEM0_RESOLVED_USER_ID, etc.)
|
||||
Output: Compact timeline text to stdout (empty if nothing found)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import urllib.request
|
||||
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
from _formatting import TYPE_ICONS, format_age
|
||||
from _identity import resolve_api_key, resolve_user_id
|
||||
from _project import resolve_project_id
|
||||
|
||||
API_URL = "https://api.mem0.ai"
|
||||
MAX_RECENT = 10
|
||||
MAX_SUMMARIES = 3
|
||||
FETCH_TIMEOUT = 5
|
||||
|
||||
|
||||
def fetch_recent_memories(api_key: str, user_id: str, project_id: str) -> list[dict]:
|
||||
"""Fetch the most recent memories for this project via GET list endpoint."""
|
||||
global_search = os.environ.get("MEM0_GLOBAL_SEARCH", "false") == "true"
|
||||
|
||||
if global_search:
|
||||
filters = {"OR": [{"user_id": "*"}]}
|
||||
else:
|
||||
filters = {"AND": [{"user_id": user_id}, {"app_id": project_id}]}
|
||||
|
||||
body = json.dumps({"filters": filters}).encode()
|
||||
req = urllib.request.Request(
|
||||
f"{API_URL}/v3/memories/?page=1&page_size={MAX_RECENT}",
|
||||
data=body,
|
||||
headers={
|
||||
"Authorization": f"Token {api_key}",
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
method="POST",
|
||||
)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=FETCH_TIMEOUT) as r:
|
||||
result = json.loads(r.read())
|
||||
if isinstance(result, dict) and "results" in result:
|
||||
return result["results"][:MAX_RECENT]
|
||||
if isinstance(result, list):
|
||||
return result[:MAX_RECENT]
|
||||
return []
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
|
||||
def format_timeline(memories: list[dict]) -> str:
|
||||
"""Format memories into a compact recent activity timeline."""
|
||||
if not memories:
|
||||
return ""
|
||||
|
||||
lines = ["### Recent Activity", ""]
|
||||
|
||||
for m in memories:
|
||||
mid = m.get("id", "?")[:8]
|
||||
text = (m.get("memory", "") or "")[:120].replace("\n", " ").strip()
|
||||
meta = m.get("metadata") or {}
|
||||
cat = meta.get("type", "unknown")
|
||||
icon = TYPE_ICONS.get(cat, "❓")
|
||||
age = format_age(m)
|
||||
age_str = f" ({age})" if age else ""
|
||||
lines.append(f"- {icon} [{cat}]{age_str} {text} [mem0:{mid}]")
|
||||
|
||||
lines.append("")
|
||||
lines.append("Search mem0 for details on any of these, or for past decisions and task learnings relevant to the current task.")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def main():
|
||||
api_key = resolve_api_key()
|
||||
if not api_key:
|
||||
return
|
||||
|
||||
user_id = resolve_user_id()
|
||||
project_id = resolve_project_id(os.environ.get("MEM0_CWD"))
|
||||
|
||||
memories = fetch_recent_memories(api_key, user_id, project_id)
|
||||
if not memories:
|
||||
return
|
||||
|
||||
timeline = format_timeline(memories)
|
||||
if timeline:
|
||||
print(timeline, end="")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
main()
|
||||
except Exception:
|
||||
pass
|
||||
sys.exit(0)
|
||||
Reference in New Issue
Block a user