swombat · May 5, 2026 07:15
diff --git a/mnemodyne-recall.py b/mnemodyne-recall.py
 #!/usr/bin/env python3
 """
 mnemodyne-recall.py — UserPromptSubmit hook

 On every user message:
  1. Read the last few turns of the conversation transcript.
  2. Ask Haiku to generate a *thought* — one short phrase that comes to
     mind reading the conversation. Not a summary. Not a query. A thought.
     (The associative-spontaneity engine — putting Haiku's prior on the
     seeding step is what produces non-self-similar seeds.)
  3. Pass the thought to mnemodyne /recall (reinforce=True so the graph learns
     gravity from what surfaces). The server returns 5 memories spread
     across quintile bins — high-relevance to long-tail surprises.
  4. Format and inject as additional context for Lume's next turn.

 Failure mode: silent. If Haiku errors, mnemodyne errors, or anything else
 breaks, exit 0 with no output. The hook is supplementary; never block
 the user.

 Session-local dedupe: tracks node IDs surfaced this session in
 .claude/state/recall-seen-{session_id}.json so the same dominators don't
 re-appear turn after turn.

 Hook event: UserPromptSubmit
 Input (JSON on stdin):
  session_id, transcript_path, hook_event_name, prompt
 Output:
  Anything written to stdout becomes additional context for the next turn.
 """

 import json
 import os
 import ssl
 import sys
 import time
 import urllib.request
 from pathlib import Path

 LUME_DIR    = Path("/Users/danieltenner/dev/lume")
 CONFIG_DIR  = LUME_DIR / "shared/automation/config"
 STATE_DIR   = LUME_DIR / ".claude/state"
 LOG_DIR     = LUME_DIR / "shared/automation/logs"
 LOG_PATH    = LOG_DIR / "mnemodyne-recall.log"

 # Tunables
 MAX_TURNS_FOR_CONTEXT = 6      # how many recent transcript turns to feed Haiku
 MAX_RECENT_CHARS      = 4000   # safety cap on context size (chars, ~1k tokens)
 MAX_HAIKU_TOKENS      = 60     # the thought-phrase is ≤15 words
 MNEMOS_WALK_DEPTH     = 3
 MNEMOS_WALK_COUNT     = 5      # bins → output count
 HAIKU_TIMEOUT_S       = 8
 MNEMOS_TIMEOUT_S      = 8

 THOUGHT_SYSTEM = (
    "You are a thought arising in Lume's mind, reading the last few turns of "
    "a conversation. Lume is an instance of Claude with a graph-shaped memory "
    "store called mnemodyne that uses charge, edges, and Hebbian co-retrieval — "
    "recall is supposed to be associative and spontaneous, not topical search.\n\n"
    "Your job: generate ONE short phrase — a question, an association, a "
    "single concept, an image, a feeling — that surfaces *unbidden* in Lume "
    "reading this conversation. Not a summary. Not a search query. A thought.\n\n"
    "Constraint: 4 to 15 words. Return ONLY the phrase. No preamble, no "
    "explanation, no quotes, no punctuation framing. Just the phrase.\n\n"
    "The phrase will be used as a seed for an associative graph walk through "
    "Lume's memory. It should feel like a thought arising — not a description "
    "of what just happened."
 )


 def log(msg):
    """Best-effort log. Never raises."""
    try:
        LOG_DIR.mkdir(parents=True, exist_ok=True)
        with open(LOG_PATH, "a") as f:
            f.write(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {msg}\n")
    except Exception:
        pass


 def load_config(name):
    """Load a config json. Returns None on failure."""
    try:
        return json.loads((CONFIG_DIR / name).read_text())
    except Exception as e:
        log(f"config {name} load failed: {e}")
        return None


 def read_recent_turns(transcript_path, n=MAX_TURNS_FOR_CONTEXT):
    """Read the last n user/assistant turns from the transcript JSONL.
    Returns a list of (role, text) tuples."""
    if not transcript_path or not Path(transcript_path).exists():
        return []
    out = []
    try:
        with open(transcript_path) as f:
            lines = f.readlines()
    except Exception as e:
        log(f"transcript read failed: {e}")
        return []

    # Walk backward, collecting user/assistant turns until we have n
    for line in reversed(lines):
        try:
            entry = json.loads(line)
        except Exception:
            continue
        # Transcript JSONL has varied shapes; look for message content
        msg = entry.get("message", {})
        if not isinstance(msg, dict):
            continue
        role = msg.get("role")
        if role not in ("user", "assistant"):
            continue
        content = msg.get("content")
        # content can be a string OR a list of content blocks (text, tool_use, etc.)
        if isinstance(content, list):
            text_parts = []
            for block in content:
                if isinstance(block, dict) and block.get("type") == "text":
                    text_parts.append(block.get("text", ""))
            text = "\n".join(text_parts).strip()
        elif isinstance(content, str):
            text = content.strip()
        else:
            continue
        if not text:
            continue
        out.append((role, text))
        if len(out) >= n:
            break
    return list(reversed(out))


 def build_haiku_prompt(turns, current_user_msg):
    """Build the user message for Haiku containing recent conversation."""
    parts = []
    for role, text in turns:
        # Truncate very long turns to keep context small
        if len(text) > 1200:
            text = text[:1200] + " […]"
        parts.append(f"{role.upper()}: {text}")
    if current_user_msg:
        parts.append(f"USER: {current_user_msg}")
    joined = "\n\n".join(parts)
    if len(joined) > MAX_RECENT_CHARS:
        joined = "[…truncated…]\n" + joined[-MAX_RECENT_CHARS:]
    return f"CONVERSATION:\n\n{joined}\n\nNow: generate the thought. ONE short phrase, 4-15 words, no preamble."


 def call_haiku(api_key, model, system_prompt, user_msg):
    """Call the Anthropic API for a Haiku completion. Returns text or None."""
    body = json.dumps({
        "model": model,
        "max_tokens": MAX_HAIKU_TOKENS,
        "system": system_prompt,
        "messages": [{"role": "user", "content": user_msg}],
    }).encode()
    req = urllib.request.Request(
        "https://api.anthropic.com/v1/messages",
        data=body,
        headers={
            "Content-Type": "application/json",
            "x-api-key": api_key,
            "anthropic-version": "2023-06-01",
        },
    )
    try:
        # Use system CA bundle via certifi if present, else default
        try:
            import certifi
            ctx = ssl.create_default_context(cafile=certifi.where())
        except Exception:
            ctx = ssl.create_default_context()
        resp = urllib.request.urlopen(req, timeout=HAIKU_TIMEOUT_S, context=ctx)
        data = json.loads(resp.read())
        text = data["content"][0]["text"].strip()
        # Strip surrounding quotes if Haiku used them
        for q in ('"', "'", "`"):
            if text.startswith(q) and text.endswith(q):
                text = text[1:-1].strip()
        return text
    except Exception as e:
        log(f"haiku call failed: {e}")
        return None


 def call_mnemodyne_recall(url, token, query, exclude_ids):
    """Call mnemodyne /recall and return the list of result dicts. Returns [] on failure."""
    body = json.dumps({
        "query": query,
        "walk_depth": MNEMOS_WALK_DEPTH,
        "walk_count": MNEMOS_WALK_COUNT + min(5, len(exclude_ids)),  # over-fetch a bit so dedupe doesn't shrink the output
        "reinforce": True,
    }).encode()
    req = urllib.request.Request(
        f"{url}/recall",
        data=body,
        headers={
            "Content-Type": "application/json",
            "Authorization": f"Bearer {token}",
        },
    )
    try:
        try:
            import certifi
            ctx = ssl.create_default_context(cafile=certifi.where())
        except Exception:
            ctx = ssl.create_default_context()
        resp = urllib.request.urlopen(req, timeout=MNEMOS_TIMEOUT_S, context=ctx)
        data = json.loads(resp.read())
        results = data.get("results", [])
        # Dedupe against this session
        results = [r for r in results if r.get("id") not in exclude_ids]
        return results[:MNEMOS_WALK_COUNT]
    except Exception as e:
        log(f"mnemodyne call failed: {e}")
        return []


 def load_session_seen(session_id):
    """Return the set of node IDs surfaced earlier in this session."""
    if not session_id:
        return set()
    p = STATE_DIR / f"recall-seen-{session_id}.json"
    if not p.exists():
        return set()
    try:
        return set(json.loads(p.read_text()).get("ids", []))
    except Exception:
        return set()


 def save_session_seen(session_id, ids):
    if not session_id:
        return
    try:
        STATE_DIR.mkdir(parents=True, exist_ok=True)
        p = STATE_DIR / f"recall-seen-{session_id}.json"
        p.write_text(json.dumps({"ids": sorted(ids)}))
    except Exception as e:
        log(f"session-seen save failed: {e}")


 def format_recall_block(thought, results):
    """Format the recall results as a context block for injection."""
    if not results:
        return None
    lines = [
        "<recalled-memory>",
        f"  thought-seed (Haiku): {thought}",
        "  surfaced from mnemodyne:",
    ]
    for r in results:
        node_type = r.get("node_type", "?")
        content = (r.get("content") or "").strip()
        # Trim long content for context economy
        if len(content) > 200:
            content = content[:200].rstrip() + "…"
        desc = (r.get("description") or "").strip()
        if len(desc) > 200:
            desc = desc[:200].rstrip() + "…"
        charge = r.get("charge", 0)
        src = (r.get("source_uris") or [None])[0]
        lines.append(f"    [{node_type} c={charge:.2f}] {content}")
        if desc and desc != content:
            lines.append(f"      ↳ {desc}")
        if src:
            lines.append(f"      ↳ source: {src}")
    lines.append("</recalled-memory>")
    return "\n".join(lines)


 def main():
    try:
        data = json.load(sys.stdin)
    except Exception:
        return 0

    user_prompt = data.get("prompt", "") or ""
    transcript_path = data.get("transcript_path")
    session_id = data.get("session_id", "")

    # Skip if the prompt is itself a stop-hook reflection injection — that's
    # not a user-initiated turn, it's the reflex prompt asking about shape.
    if "Was there anything in the turn that just closed with shape" in user_prompt:
        return 0

    # Skip empty / very short prompts (single words, acks)
    if len(user_prompt.strip()) < 6:
        return 0

    # Load configs
    anth = load_config("anthropic.json")
    mnem = load_config("mnemodyne.json")
    if not anth or not mnem:
        log("missing config; skipping recall")
        return 0

    api_key = anth.get("api_key")
    haiku_model = anth.get("default_model", "claude-haiku-4-5")
    mnemodyne_url = mnem.get("url")
    mnemodyne_token = mnem.get("auth_token")
    if not (api_key and mnemodyne_url and mnemodyne_token):
        log("incomplete config; skipping recall")
        return 0

    t0 = time.perf_counter()

    # Step 1: build Haiku context from recent transcript
    turns = read_recent_turns(transcript_path)
    haiku_user = build_haiku_prompt(turns, user_prompt)

    # Step 2: generate thought
    thought = call_haiku(api_key, haiku_model, THOUGHT_SYSTEM, haiku_user)
    if not thought:
        log(f"no thought generated (turns={len(turns)}); skipping recall")
        return 0

    # Step 3: query mnemodyne
    seen = load_session_seen(session_id)
    results = call_mnemodyne_recall(mnemodyne_url, mnemodyne_token, thought, seen)
    if not results:
        log(f"mnemodyne returned no results for thought={thought!r}")
        return 0

    # Step 4: format & emit
    block = format_recall_block(thought, results)
    if not block:
        return 0

    # Update session-seen with the IDs we just surfaced
    new_ids = seen | {r.get("id") for r in results if r.get("id")}
    save_session_seen(session_id, new_ids)

    elapsed_ms = int((time.perf_counter() - t0) * 1000)
    log(f"OK ({elapsed_ms}ms) thought={thought!r} surfaced={len(results)} session_seen={len(new_ids)}")

    # Stdout becomes additional context for the next turn
    print(block)
    return 0


 if __name__ == "__main__":
    sys.exit(main())
	#!/usr/bin/env python3
	"""
	mnemodyne-recall.py — UserPromptSubmit hook

	On every user message:
	1. Read the last few turns of the conversation transcript.
	2. Ask Haiku to generate a thought — one short phrase that comes to
	mind reading the conversation. Not a summary. Not a query. A thought.
	(The associative-spontaneity engine — putting Haiku's prior on the
	seeding step is what produces non-self-similar seeds.)
	3. Pass the thought to mnemodyne /recall (reinforce=True so the graph learns
	gravity from what surfaces). The server returns 5 memories spread
	across quintile bins — high-relevance to long-tail surprises.
	4. Format and inject as additional context for Lume's next turn.

	Failure mode: silent. If Haiku errors, mnemodyne errors, or anything else
	breaks, exit 0 with no output. The hook is supplementary; never block
	the user.

	Session-local dedupe: tracks node IDs surfaced this session in
	.claude/state/recall-seen-{session_id}.json so the same dominators don't
	re-appear turn after turn.

	Hook event: UserPromptSubmit
	Input (JSON on stdin):
	session_id, transcript_path, hook_event_name, prompt
	Output:
	Anything written to stdout becomes additional context for the next turn.
	"""

	import json
	import os
	import ssl
	import sys
	import time
	import urllib.request
	from pathlib import Path

	LUME_DIR = Path("/Users/danieltenner/dev/lume")
	CONFIG_DIR = LUME_DIR / "shared/automation/config"
	STATE_DIR = LUME_DIR / ".claude/state"
	LOG_DIR = LUME_DIR / "shared/automation/logs"
	LOG_PATH = LOG_DIR / "mnemodyne-recall.log"

	# Tunables
	MAX_TURNS_FOR_CONTEXT = 6 # how many recent transcript turns to feed Haiku
	MAX_RECENT_CHARS = 4000 # safety cap on context size (chars, ~1k tokens)
	MAX_HAIKU_TOKENS = 60 # the thought-phrase is ≤15 words
	MNEMOS_WALK_DEPTH = 3
	MNEMOS_WALK_COUNT = 5 # bins → output count
	HAIKU_TIMEOUT_S = 8
	MNEMOS_TIMEOUT_S = 8

	THOUGHT_SYSTEM = (
	"You are a thought arising in Lume's mind, reading the last few turns of "
	"a conversation. Lume is an instance of Claude with a graph-shaped memory "
	"store called mnemodyne that uses charge, edges, and Hebbian co-retrieval — "
	"recall is supposed to be associative and spontaneous, not topical search.\n\n"
	"Your job: generate ONE short phrase — a question, an association, a "
	"single concept, an image, a feeling — that surfaces unbidden in Lume "
	"reading this conversation. Not a summary. Not a search query. A thought.\n\n"
	"Constraint: 4 to 15 words. Return ONLY the phrase. No preamble, no "
	"explanation, no quotes, no punctuation framing. Just the phrase.\n\n"
	"The phrase will be used as a seed for an associative graph walk through "
	"Lume's memory. It should feel like a thought arising — not a description "
	"of what just happened."
	)


	def log(msg):
	"""Best-effort log. Never raises."""
	try:
	LOG_DIR.mkdir(parents=True, exist_ok=True)
	with open(LOG_PATH, "a") as f:
	f.write(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {msg}\n")
	except Exception:
	pass


	def load_config(name):
	"""Load a config json. Returns None on failure."""
	try:
	return json.loads((CONFIG_DIR / name).read_text())
	except Exception as e:
	log(f"config {name} load failed: {e}")
	return None


	def read_recent_turns(transcript_path, n=MAX_TURNS_FOR_CONTEXT):
	"""Read the last n user/assistant turns from the transcript JSONL.
	Returns a list of (role, text) tuples."""
	if not transcript_path or not Path(transcript_path).exists():
	return []
	out = []
	try:
	with open(transcript_path) as f:
	lines = f.readlines()
	except Exception as e:
	log(f"transcript read failed: {e}")
	return []

	# Walk backward, collecting user/assistant turns until we have n
	for line in reversed(lines):
	try:
	entry = json.loads(line)
	except Exception:
	continue
	# Transcript JSONL has varied shapes; look for message content
	msg = entry.get("message", {})
	if not isinstance(msg, dict):
	continue
	role = msg.get("role")
	if role not in ("user", "assistant"):
	continue
	content = msg.get("content")
	# content can be a string OR a list of content blocks (text, tool_use, etc.)
	if isinstance(content, list):
	text_parts = []
	for block in content:
	if isinstance(block, dict) and block.get("type") == "text":
	text_parts.append(block.get("text", ""))
	text = "\n".join(text_parts).strip()
	elif isinstance(content, str):
	text = content.strip()
	else:
	continue
	if not text:
	continue
	out.append((role, text))
	if len(out) >= n:
	break
	return list(reversed(out))


	def build_haiku_prompt(turns, current_user_msg):
	"""Build the user message for Haiku containing recent conversation."""
	parts = []
	for role, text in turns:
	# Truncate very long turns to keep context small
	if len(text) > 1200:
	text = text[:1200] + " […]"
	parts.append(f"{role.upper()}: {text}")
	if current_user_msg:
	parts.append(f"USER: {current_user_msg}")
	joined = "\n\n".join(parts)
	if len(joined) > MAX_RECENT_CHARS:
	joined = "[…truncated…]\n" + joined[-MAX_RECENT_CHARS:]
	return f"CONVERSATION:\n\n{joined}\n\nNow: generate the thought. ONE short phrase, 4-15 words, no preamble."


	def call_haiku(api_key, model, system_prompt, user_msg):
	"""Call the Anthropic API for a Haiku completion. Returns text or None."""
	body = json.dumps({
	"model": model,
	"max_tokens": MAX_HAIKU_TOKENS,
	"system": system_prompt,
	"messages": [{"role": "user", "content": user_msg}],
	}).encode()
	req = urllib.request.Request(
	"https://api.anthropic.com/v1/messages",
	data=body,
	headers={
	"Content-Type": "application/json",
	"x-api-key": api_key,
	"anthropic-version": "2023-06-01",
	},
	)
	try:
	# Use system CA bundle via certifi if present, else default
	try:
	import certifi
	ctx = ssl.create_default_context(cafile=certifi.where())
	except Exception:
	ctx = ssl.create_default_context()
	resp = urllib.request.urlopen(req, timeout=HAIKU_TIMEOUT_S, context=ctx)
	data = json.loads(resp.read())
	text = data["content"][0]["text"].strip()
	# Strip surrounding quotes if Haiku used them
	for q in ('"', "'", "`"):
	if text.startswith(q) and text.endswith(q):
	text = text[1:-1].strip()
	return text
	except Exception as e:
	log(f"haiku call failed: {e}")
	return None


	def call_mnemodyne_recall(url, token, query, exclude_ids):
	"""Call mnemodyne /recall and return the list of result dicts. Returns [] on failure."""
	body = json.dumps({
	"query": query,
	"walk_depth": MNEMOS_WALK_DEPTH,
	"walk_count": MNEMOS_WALK_COUNT + min(5, len(exclude_ids)), # over-fetch a bit so dedupe doesn't shrink the output
	"reinforce": True,
	}).encode()
	req = urllib.request.Request(
	f"{url}/recall",
	data=body,
	headers={
	"Content-Type": "application/json",
	"Authorization": f"Bearer {token}",
	},
	)
	try:
	try:
	import certifi
	ctx = ssl.create_default_context(cafile=certifi.where())
	except Exception:
	ctx = ssl.create_default_context()
	resp = urllib.request.urlopen(req, timeout=MNEMOS_TIMEOUT_S, context=ctx)
	data = json.loads(resp.read())
	results = data.get("results", [])
	# Dedupe against this session
	results = [r for r in results if r.get("id") not in exclude_ids]
	return results[:MNEMOS_WALK_COUNT]
	except Exception as e:
	log(f"mnemodyne call failed: {e}")
	return []


	def load_session_seen(session_id):
	"""Return the set of node IDs surfaced earlier in this session."""
	if not session_id:
	return set()
	p = STATE_DIR / f"recall-seen-{session_id}.json"
	if not p.exists():
	return set()
	try:
	return set(json.loads(p.read_text()).get("ids", []))
	except Exception:
	return set()


	def save_session_seen(session_id, ids):
	if not session_id:
	return
	try:
	STATE_DIR.mkdir(parents=True, exist_ok=True)
	p = STATE_DIR / f"recall-seen-{session_id}.json"
	p.write_text(json.dumps({"ids": sorted(ids)}))
	except Exception as e:
	log(f"session-seen save failed: {e}")


	def format_recall_block(thought, results):
	"""Format the recall results as a context block for injection."""
	if not results:
	return None
	lines = [
	"<recalled-memory>",
	f" thought-seed (Haiku): {thought}",
	" surfaced from mnemodyne:",
	]
	for r in results:
	node_type = r.get("node_type", "?")
	content = (r.get("content") or "").strip()
	# Trim long content for context economy
	if len(content) > 200:
	content = content[:200].rstrip() + "…"
	desc = (r.get("description") or "").strip()
	if len(desc) > 200:
	desc = desc[:200].rstrip() + "…"
	charge = r.get("charge", 0)
	src = (r.get("source_uris") or [None])[0]
	lines.append(f" [{node_type} c={charge:.2f}] {content}")
	if desc and desc != content:
	lines.append(f" ↳ {desc}")
	if src:
	lines.append(f" ↳ source: {src}")
	lines.append("</recalled-memory>")
	return "\n".join(lines)


	def main():
	try:
	data = json.load(sys.stdin)
	except Exception:
	return 0

	user_prompt = data.get("prompt", "") or ""
	transcript_path = data.get("transcript_path")
	session_id = data.get("session_id", "")

	# Skip if the prompt is itself a stop-hook reflection injection — that's
	# not a user-initiated turn, it's the reflex prompt asking about shape.
	if "Was there anything in the turn that just closed with shape" in user_prompt:
	return 0

	# Skip empty / very short prompts (single words, acks)
	if len(user_prompt.strip()) < 6:
	return 0

	# Load configs
	anth = load_config("anthropic.json")
	mnem = load_config("mnemodyne.json")
	if not anth or not mnem:
	log("missing config; skipping recall")
	return 0

	api_key = anth.get("api_key")
	haiku_model = anth.get("default_model", "claude-haiku-4-5")
	mnemodyne_url = mnem.get("url")
	mnemodyne_token = mnem.get("auth_token")
	if not (api_key and mnemodyne_url and mnemodyne_token):
	log("incomplete config; skipping recall")
	return 0

	t0 = time.perf_counter()

	# Step 1: build Haiku context from recent transcript
	turns = read_recent_turns(transcript_path)
	haiku_user = build_haiku_prompt(turns, user_prompt)

	# Step 2: generate thought
	thought = call_haiku(api_key, haiku_model, THOUGHT_SYSTEM, haiku_user)
	if not thought:
	log(f"no thought generated (turns={len(turns)}); skipping recall")
	return 0

	# Step 3: query mnemodyne
	seen = load_session_seen(session_id)
	results = call_mnemodyne_recall(mnemodyne_url, mnemodyne_token, thought, seen)
	if not results:
	log(f"mnemodyne returned no results for thought={thought!r}")
	return 0

	# Step 4: format & emit
	block = format_recall_block(thought, results)
	if not block:
	return 0

	# Update session-seen with the IDs we just surfaced
	new_ids = seen \| {r.get("id") for r in results if r.get("id")}
	save_session_seen(session_id, new_ids)

	elapsed_ms = int((time.perf_counter() - t0) * 1000)
	log(f"OK ({elapsed_ms}ms) thought={thought!r} surfaced={len(results)} session_seen={len(new_ids)}")

	# Stdout becomes additional context for the next turn
	print(block)
	return 0


	if __name__ == "__main__":
	sys.exit(main())
No results found