dotfiles/dot_claude/hooks/executable_cmem-to-obsidian.py

#!/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 <private> content at
#     capture time so it never reaches the DB. We still scan defensively.
#   - viewer deep link that resolves: http://localhost:<port>/?observation=<id>
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 "<private>" 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()