#!/usr/bin/env python3 # Reads a Claude Code SessionEnd hook JSON on stdin and exports that session's # claude-mem learnings to a single Obsidian note. Stdlib only. Fails soft: # any error exits 0 without blocking the session. # # Data interface (discovered live, not guessed): # - SQLite DB: ~/.claude-mem/claude-mem.db (read-only) # - sdk_sessions.content_session_id == Claude Code session_id # - sdk_sessions.memory_session_id -> observations / session_summaries # - observation types seen: discovery, change, feature, decision, bugfix, refactor # - no private/excluded column exists: claude-mem drops content at # capture time so it never reaches the DB. We still scan defensively. # - viewer deep link that resolves: http://localhost:/?observation= import sys, os, re, json, sqlite3, datetime from pathlib import Path HOME = Path.home() DB_PATH = Path(os.environ.get("CLAUDE_MEM_DB", HOME / ".claude-mem" / "claude-mem.db")) VAULT_DIR = Path(os.environ.get("CMEM_OBSIDIAN_DIR", HOME / "Documents" / "Vault" / "90 - Archive" / "Learnings")) WORKER_PORT = os.environ.get("CLAUDE_MEM_WORKER_PORT", "37700") LOG_PATH = HOME / ".claude" / "hooks" / "cmem-to-obsidian.log" # observation type -> note section SECTIONS = [ ("Decisions", {"decision"}), ("Fixes & gotchas", {"bugfix"}), ("Insights", {"discovery", "feature", "change", "refactor"}), ] def log(msg): try: with open(LOG_PATH, "a") as f: f.write(f"{datetime.datetime.now().isoformat()} {msg}\n") except Exception: pass def slug(s): return re.sub(r"[^a-z0-9]+", "-", (s or "").lower()).strip("-") or "unknown" def is_private(obs): blob = " ".join(str(obs.get(k) or "") for k in ("title", "subtitle", "narrative", "text", "facts")) return "" in blob.lower() def connect_ro(path): # read-only so we never lock or mutate the live DB / WAL return sqlite3.connect(f"file:{path}?mode=ro", uri=True, timeout=5) def fetch(session_id): con = connect_ro(DB_PATH) con.row_factory = sqlite3.Row cur = con.cursor() row = cur.execute( "SELECT memory_session_id, project, started_at FROM sdk_sessions " "WHERE content_session_id = ? LIMIT 1", (session_id,) ).fetchone() if not row or not row["memory_session_id"]: con.close() return None mem_id, project, started_at = row["memory_session_id"], row["project"], row["started_at"] obs = [dict(r) for r in cur.execute( "SELECT id, type, title, subtitle, narrative FROM observations " "WHERE memory_session_id = ? ORDER BY created_at_epoch ASC", (mem_id,) ).fetchall()] summ = cur.execute( "SELECT learned, completed, investigated, request FROM session_summaries " "WHERE memory_session_id = ? ORDER BY created_at_epoch DESC LIMIT 1", (mem_id,) ).fetchone() con.close() return { "mem_id": mem_id, "project": project or "unknown", "started_at": started_at, "observations": [o for o in obs if not is_private(o)], "summary": dict(summ) if summ else {}, } def obs_link(oid): return f"http://localhost:{WORKER_PORT}/?observation={oid}" def one_line(obs): title = (obs.get("title") or "").strip() sub = (obs.get("subtitle") or "").strip() text = f"{title} — {sub}" if title and sub else (title or sub or "(untitled observation)") return f"- {text} [cmem #{obs['id']}]({obs_link(obs['id'])})" def render_note(data, session_id): project = data["project"] pslug = slug(project) # date from the session's own start time -> stable across re-runs (idempotent) date = (data["started_at"] or datetime.datetime.now().isoformat())[:10] short = session_id[:8] summary = data["summary"] abstract = (summary.get("learned") or summary.get("completed") or summary.get("investigated") or summary.get("request") or "").strip() abstract = re.sub(r"\s+", " ", abstract) or "No session summary recorded." fm = [ "---", f"date: {date}", f"project: {project}", f"session: {short}", "source: claude-mem", "tags:", " - claude-mem", f" - project/{pslug}", "---", "", f"# {project} — Session Learnings ({date})", "", "> [!abstract] Session summary", f"> {abstract}", "", f"Project: [[{project}]]", "", ] body = [] by_id = {o["id"]: o for o in data["observations"]} placed = set() for heading, types in SECTIONS: items = [o for o in data["observations"] if o["type"] in types] if not items: continue body.append(f"## {heading}") for o in items: placed.add(o["id"]) if heading == "Decisions": t = (o.get("title") or "").strip() s = (o.get("subtitle") or "").strip() body.append(f"- > [!note] {t}" + (f" — {s}" if s else "") + f" [cmem #{o['id']}]({obs_link(o['id'])})") else: body.append(one_line(o)) body.append("") # any type we didn't map -> Insights catch-all leftover = [o for o in data["observations"] if o["id"] not in placed] if leftover: body.append("## Insights") body.extend(one_line(o) for o in leftover) body.append("") return "\n".join(fm + body).rstrip() + "\n", project, pslug, date, short def ensure_project_moc(project): # so [[Project]] wikilink resolves moc = VAULT_DIR / f"{project}.md" try: if not moc.exists(): moc.write_text( f"---\ntags:\n - moc\n - project/{slug(project)}\n---\n\n# {project}\n\n" "Map of content for claude-mem session learnings.\n" ) except Exception as e: log(f"moc write failed: {e}") def write_note(data, session_id): text, project, pslug, date, short = render_note(data, session_id) folder = VAULT_DIR / project folder.mkdir(parents=True, exist_ok=True) ensure_project_moc(project) # deterministic filename keyed to session -> idempotent upsert path = folder / f"{date} {project} {short}.md" path.write_text(text) log(f"wrote {path} ({len(data['observations'])} observations)") return path def main(): try: payload = json.load(sys.stdin) except Exception: return session_id = payload.get("session_id") if not session_id: return try: if not DB_PATH.exists(): return data = fetch(session_id) if not data or not data["observations"]: return write_note(data, session_id) except Exception as e: log(f"export failed for session {session_id}: {e}") return if __name__ == "__main__": main()