"""Jarvis long-term memory — thin client to the Railway brain. This is the CLIENT side: it does NO heavy work. It calls the always-on memory service (a FalkorDB-backed fact graph deployed on Railway — see memory-service/) over HTTPS. The SERVER now owns the whole WRITE pipeline: on /remember it dedups against what's already stored and decides add / update (supersede) / no-op (Mem0-style) via DeepSeek, extracts the event date (``when``) and entities + relations from the text, and cleans raw utterances into atomic facts. So the client just ships the note (fire-and-forget) and the graph does the thinking; searching returns ranked, linked context in ~100ms, with reinforcement-on-access and optional temporal (`when_from`/`when_to`) filtering. This is the cross-device brain: the Mac, the Windows box, and the phone all point at the SAME MEMORY_API_URL, so a fact learned on one device is known on all of them, and linked to everything else. Runs IN-PROCESS via the Agent SDK MCP server (key ``"memory"``; Claude sees ``mcp__memory__``). Pure stdlib HTTP — no heavy deps on the device. Config (``.env``): MEMORY_API_URL the deployed service, e.g. https://jarvis-memory.up.railway.app MEMORY_API_KEY the shared bearer key (matches the service) MEMORY_GROUP namespace (default "ahmed") — lets one brain serve many users MEM_SERVER_PIPELINE set 0 to fall back to the OLD client-side supersede path (the server owns dedup/supersede by default) Set ``MCP_MEMORY=0`` to disable. """ from __future__ import annotations import json import os import urllib.error import urllib.parse import urllib.request _URL = os.environ.get("MEMORY_API_URL", "").rstrip("/") _KEY = os.environ.get("MEMORY_API_KEY", "") _GROUP = os.environ.get("MEMORY_GROUP", "ahmed") def enabled() -> bool: """On only when a memory service URL is configured.""" if os.environ.get("MCP_MEMORY", "1") == "0": return False return bool(_URL) _HEDGE = ("maybe", "might", "possibly", "perhaps", "i think", "not sure", "around", "roughly", "approximately", "or so", "probably", "could be", "i guess") _FIRM = ("signed", "confirmed", "definitely", "for sure", "agreed", "paid", "committed", "finalized", "closed the deal", "locked in") def _confidence(text: str) -> float: """Light heuristic — hedged facts recall weaker, firm facts stronger. The service uses this in salience ranking (latest+confident wins), never to drop.""" t = (text or "").lower() if any(w in t for w in _FIRM): return 0.9 if any(w in t for w in _HEDGE): return 0.5 return 0.75 def _importance(text: str) -> int: """1-10 importance the server weighs in dedup/salience. Reuses the FIRM/HEDGE heuristic: firm/committed facts matter more (8), hedged guesses less (3).""" t = (text or "").lower() if any(w in t for w in _FIRM): return 8 if any(w in t for w in _HEDGE): return 3 return 5 # --- relative-time → (when_from, when_to) for temporal /search -------------- _WEEKDAYS = {"monday": 0, "tuesday": 1, "wednesday": 2, "thursday": 3, "friday": 4, "saturday": 5, "sunday": 6} _MONTHS = {"january": 1, "february": 2, "march": 3, "april": 4, "may": 5, "june": 6, "july": 7, "august": 8, "september": 9, "october": 10, "november": 11, "december": 12} def _temporal_range(q: str) -> tuple[str | None, str | None]: """Best-effort: turn a relative-time phrase in the query into an ISO date window (when_from, when_to) for /search. English phrases + ISO numeric dates only — deliberately simple, no deps, returns (None, None) if nothing time-like is present. Kept in sync with the copy in gateway/tools.py.""" import datetime as _dt import re as _re t = (q or "").lower() today = _dt.date.today() def _iso(d: "_dt.date") -> str: return d.isoformat() def _month_end(d: "_dt.date") -> "_dt.date": return (d.replace(day=1) + _dt.timedelta(days=32)).replace(day=1) \ - _dt.timedelta(days=1) m = _re.search(r"\b(\d{4})-(\d{2})-(\d{2})\b", t) # explicit ISO date if m: try: d = _dt.date(int(m.group(1)), int(m.group(2)), int(m.group(3))) return _iso(d), _iso(d) except ValueError: pass if "today" in t: return _iso(today), _iso(today) if "yesterday" in t: y = today - _dt.timedelta(days=1) return _iso(y), _iso(y) if "this week" in t: start = today - _dt.timedelta(days=today.weekday()) return _iso(start), _iso(start + _dt.timedelta(days=6)) if "last week" in t: start = today - _dt.timedelta(days=today.weekday() + 7) return _iso(start), _iso(start + _dt.timedelta(days=6)) if "this month" in t: start = today.replace(day=1) return _iso(start), _iso(_month_end(start)) if "last month" in t: end = today.replace(day=1) - _dt.timedelta(days=1) return _iso(end.replace(day=1)), _iso(end) m = _re.search(r"\b(?:on\s+)?(monday|tuesday|wednesday|thursday|friday|" r"saturday|sunday)\b", t) # most recent past occurrence if m: delta = (today.weekday() - _WEEKDAYS[m.group(1)]) % 7 or 7 d = today - _dt.timedelta(days=delta) return _iso(d), _iso(d) m = _re.search(r"\bin\s+(january|february|march|april|may|june|july|" r"august|september|october|november|december)\b", t) if m: # "in march" = that month, this year if past else last year mo = _MONTHS[m.group(1)] year = today.year if mo <= today.month else today.year - 1 start = _dt.date(year, mo, 1) return _iso(start), _iso(_month_end(start)) return None, None def _text(msg: str) -> dict: return {"content": [{"type": "text", "text": msg}]} def _req(method: str, path: str, body: dict | None = None, timeout: int = 20): data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(_URL + path, data=data, method=method) if _KEY: req.add_header("Authorization", f"Bearer {_KEY}") req.add_header("Content-Type", "application/json") with urllib.request.urlopen(req, timeout=timeout) as r: raw = r.read() return json.loads(raw) if raw else {} # --------------------------------------------------------------------------- # HUD memory VIEWER — the engine (which holds the API key) fetches / curates the # shared brain on the HUD's behalf and pushes the results back as a `memories` # event. Ahmed opens the viewer, searches it, and deletes/corrects wrong facts. # --------------------------------------------------------------------------- def list_recent(limit: int = 40) -> list[dict]: """Newest-first raw brain contents (browse, not search).""" qs = urllib.parse.urlencode({"group": _GROUP, "limit": limit}) return _req("GET", f"/memories?{qs}").get("memories", []) def fetch_insights(limit: int = 1) -> list[dict]: """Unsurfaced findings from the nightly reflection engine (newest first).""" qs = urllib.parse.urlencode({"group": _GROUP, "limit": limit, "unsurfaced_only": "true"}) return _req("GET", f"/insights?{qs}").get("insights", []) def mark_insight_seen(insight_id: str) -> None: """Mark an insight surfaced so Jarvis raises it only once.""" if insight_id: _req("POST", "/insight/seen", {"id": insight_id, "group": _GROUP}) def search_facts(q: str, k: int = 30) -> list[dict]: qs = urllib.parse.urlencode({"q": q, "k": k, "group": _GROUP}) return _req("GET", f"/search?{qs}").get("facts", []) def forget(mem_id: str) -> None: """Permanently delete a wrong memory.""" _req("POST", "/forget", {"id": mem_id, "group": _GROUP}) def edit(old_id: str, text: str) -> str: """Correct a memory: retire the old, add the new (supersede).""" out = _req("POST", "/supersede", {"old_id": old_id, "text": text, "group": _GROUP, "source": "jarvis"}) return out.get("id", "") def _norm(items: list[dict]) -> list[dict]: """Normalise /search facts and /memories rows to one HUD shape.""" return [{ "id": f.get("id", ""), "text": f.get("fact", f.get("text", "")), "kind": f.get("kind") or "", "when": (f.get("refers_to") or "")[:10], # event date the fact refers to "created_at": f.get("created_at") or "", # when it was remembered "linked": bool(f.get("linked")), } for f in items if f.get("id")] def refresh_panel(q: str = "") -> None: """Push memories to the HUD viewer — search results if `q`, else recent.""" from voice.events import emit try: items = search_facts(q) if q.strip() else list_recent() emit("memories", q=q, items=_norm(items)) except Exception as e: # noqa: BLE001 — a viewer refresh must never crash print(f" [memory] panel refresh failed: {str(e)[:100]}") emit("memories", q=q, items=[]) # --------------------------------------------------------------------------- # HUD memory GRAPH — the same shared brain shown as an interactive node/edge # graph. The engine (which holds the API key) fetches /graph, links/unlinks # edges on the HUD's behalf, and pushes nodes+edges back as a `graph` event. # Ahmed drags nodes around, connects related memories, and cuts wrong links. # --------------------------------------------------------------------------- def fetch_graph() -> dict: """Full memory graph — nodes + edges — for the HUD graph view.""" qs = urllib.parse.urlencode({"group": _GROUP}) out = _req("GET", f"/graph?{qs}") return {"nodes": out.get("nodes", []), "edges": out.get("edges", [])} def link(a: str, b: str, type: str = "related") -> None: """Connect two memories in the graph.""" _req("POST", "/link", {"from_id": a, "to": b, "type": type}) def unlink(a: str, b: str) -> None: """Cut the edge(s) between two memories (either direction).""" _req("POST", "/unlink", {"from_id": a, "to": b}) def _norm_nodes(nodes: list[dict]) -> list[dict]: """Normalise /graph nodes to one HUD shape (all strings).""" return [{ "id": str(n.get("id", "")), "text": str(n.get("text", n.get("fact", "")) or ""), "kind": str(n.get("kind") or ""), "created_at": str(n.get("created_at") or ""), } for n in nodes if n.get("id")] def refresh_graph() -> None: """Push the memory graph to the HUD viewer as a `graph` event.""" from voice.events import emit try: g = fetch_graph() emit("graph", nodes=_norm_nodes(g.get("nodes", [])), edges=g.get("edges", [])) except Exception as e: # noqa: BLE001 — a viewer refresh must never crash print(f" [memory] graph refresh failed: {str(e)[:100]}") emit("graph", nodes=[], edges=[]) def fire_save(text: str, kind: str = "fact", when: str | None = None, tone: str | None = None) -> None: """Fire-and-forget save — POSTs in a daemon thread and returns instantly, so the conversation NEVER pauses to remember. Used by the inline … path (the brain drops the note mid-reply; the engine ships it here without a tool round-trip). Failures are logged, not raised. `when` = the event date (ISO) if the caller knows it (else the server extracts it from the text); `tone` = the vocal delivery the fact was said with. The SERVER now owns dedup/supersede/cleanup (Mem0-style add/update/no-op), so the default path is a plain /remember. Set MEM_SERVER_PIPELINE=0 to restore the old client-side supersede round-trip (see memory_supersede.py).""" if not enabled() or not text.strip(): return import threading def _go() -> None: try: t = text.strip() # LAYER 2 — pull entities out with the local model (free) here too: # the client has the surrounding utterance context the server lacks, # so this sharpens the shared :Entity wiring. Best-effort. entities: list = [] if os.environ.get("MEM_ENTITIES", "1") != "0": try: from voice import memory_entities entities = memory_entities.extract(t) except Exception: # noqa: BLE001 — entities are a bonus pass body = {"text": t, "group": _GROUP, "source": "jarvis", "kind": kind, "confidence": _confidence(t), "importance": _importance(t)} if when: body["when"] = when if tone: body["tone"] = tone if entities: body["entities"] = entities # FALLBACK PATH (MEM_SERVER_PIPELINE=0): do the supersede decision on # the device — ask the local model if this fact replaces one already # stored, and /supersede it if so. Off by default; the server does it. if os.environ.get("MEM_SERVER_PIPELINE", "1") == "0" \ and os.environ.get("MEM_SUPERSEDE", "1") != "0": old_id = None try: from voice import memory_supersede old_id = memory_supersede.find_superseded( t, search_facts(t, k=6)) except Exception: # noqa: BLE001 old_id = None if old_id: body["old_id"] = old_id _req("POST", "/supersede", body) print(f" [memory: superseded {old_id[:8]} ← \"{t[:50]}\"]") return _req("POST", "/remember", body) except Exception as e: # noqa: BLE001 — a save must never disrupt talk print(f" [memory] background save failed: {str(e)[:120]}") # daemon thread: the threading module keeps it alive until _go() returns, so # no extra reference is needed here (the old _bg add/discard was a no-op). threading.Thread(target=_go, daemon=True).start() _CORE_CACHE: dict = {"text": "", "at": 0.0} def get_core_block() -> str: """The ~200-word pinned profile of Ahmed the nightly reflection maintains server-side (GET /core_block → {text, updated_at}). Cached 30 min; returns "" on ANY error and never blocks more than ~2s — safe to call while building the voice brain's system prompt without stalling or crashing startup.""" if not enabled(): return "" import time now = time.time() if _CORE_CACHE["text"] and now - _CORE_CACHE["at"] < 1800: return _CORE_CACHE["text"] try: qs = urllib.parse.urlencode({"group": _GROUP}) out = _req("GET", f"/core_block?{qs}", timeout=2) text = (out.get("text") or "").strip() except Exception: # noqa: BLE001 — never block or crash the prompt build return _CORE_CACHE["text"] # last good value if any, else "" _CORE_CACHE["text"] = text _CORE_CACHE["at"] = now return text def build_server(): """Build and return the in-process ``memory`` MCP server.""" import asyncio from claude_agent_sdk import tool, create_sdk_mcp_server # NOTE: CONVERSATIONAL saving is deliberately passive — the inline # … tag the brain drops mid-reply (see the system # prompt) → voice/llm.py fires memory_control.fire_save() in a background # thread, so talk never pauses to remember. memory_remember below is the # EXPLICIT tool for correction/curation work (the dossier-card note agent): # a worker fixing the graph must be able to write a fact directly. @tool("memory_remember", "EXPLICITLY save one durable fact to Ahmed's shared memory — for " "memory CURATION work (dossier-card corrections, merges, backfills), " "NOT for conversational saving (that happens passively). `entities` " "= comma-separated people/companies/projects the fact is about; " "`when` = ISO date the fact refers to (event time), if any.", {"text": str, "kind": str, "when": str, "entities": str}) async def memory_remember(args: dict) -> dict: text = str(args.get("text", "")).strip() if not text: return _text("memory_remember failed: no text.") try: body = {"text": text, "group": _GROUP, "source": "jarvis-curation", "kind": (str(args.get("kind", "")).strip() or "fact"), "importance": 6} if args.get("when"): body["when"] = str(args["when"]).strip() ents = [{"name": e.strip()} for e in str(args.get("entities", "")).split(",") if e.strip()] if ents: body["entities"] = ents out = await asyncio.to_thread(_req, "POST", "/remember", body) op = (out or {}).get("op", "add") return _text(f"saved ({op}): {text[:80]}") except Exception as e: # noqa: BLE001 return _text(f"memory_remember failed: {str(e)[:150]}") @tool("memory_update", "Use when something CHANGED — a fact is now different from what was " "saved (client changed, a plan moved, a preference flipped). Retires " "the old memory (kept as history, hidden from recall) and saves the " "new current one, linked to what it replaced. `old_id` from a search " "result; `text` = the new current fact.", {"old_id": str, "text": str, "kind": str, "when": str}) async def memory_update(args: dict) -> dict: old_id = str(args.get("old_id", "")).strip() text = str(args.get("text", "")).strip() if not old_id or not text: return _text("memory_update failed: need old_id and the new text.") try: body = {"old_id": old_id, "text": text, "group": _GROUP, "source": "jarvis", "kind": (args.get("kind") or "fact")} if args.get("when"): body["when"] = str(args["when"]) out = await asyncio.to_thread(_req, "POST", "/supersede", body) return _text(f"Updated (new id {out.get('id', '?')}).") except Exception as e: # noqa: BLE001 return _text(f"memory_update failed: {str(e)[:150]}") @tool("memory_search", "Search Ahmed's long-term memory by MEANING (not keywords) and get " "back connected facts — use this whenever the answer might depend on " "something he told you before ('what should I research for my " "meeting', 'what did the supplier quote', 'how do I like my coffee'). " "Returns up to `k` facts (default 10) — READ THEM ALL; the specific/" "detailed one is often not first.", {"query": str, "k": int}) async def memory_search(args: dict) -> dict: q = str(args.get("query", "")).strip() if not q: return _text("memory_search failed: no query.") # An explicit ask about a known entity must ALSO put its dossier card # on the HUD, whichever tool the brain answers with. (Local import — # entity_show imports this module.) try: from voice import entity_show as _es _es.show_for_query(q) except Exception: # noqa: BLE001 — the card is a bonus, never a blocker pass try: k = max(1, min(int(args.get("k") or 10), 25)) params = {"q": q, "k": k, "group": _GROUP} # temporal query ("what did I do last week", "on Monday") → date window wf, wt = _temporal_range(q) if wf: params["when_from"] = wf if wt: params["when_to"] = wt qs = urllib.parse.urlencode(params) out = await asyncio.to_thread(_req, "GET", f"/search?{qs}") facts = out.get("facts", []) lines = [] for f in facts: tag = " [connected]" if f.get("linked") else "" when = (f.get("refers_to") or "")[:10] lines.append(f"- {f['fact']}{tag}" + (f" (re: {when})" if when else "") + f" [id {f.get('id', '?')}]") # CONTEXTUAL INSIGHTS — only surface a reflection finding when it's # relevant to what Ahmed just asked (never on a timer). Mark it seen # so it isn't repeated. try: iqs = urllib.parse.urlencode({"q": q, "k": 2, "group": _GROUP}) for it in (await asyncio.to_thread( _req, "GET", f"/insights/search?{iqs}")).get("insights", []): lines.append( "- (a reflection of yours that may be relevant here — " "mention it ONLY if it genuinely fits, briefly, as your " f"own thought: {it.get('text', '')})") await asyncio.to_thread(_req, "POST", "/insight/seen", {"id": it.get("id", ""), "group": _GROUP}) except Exception: # noqa: BLE001 — insights are a bonus pass # THEORY OF MIND — if the question is about a person/company, add the # reflection's READ (what they want, how to approach) or its summary. try: import re as _re2 stop = {"what", "when", "should", "about", "with", "this", "that", "from", "have", "approach", "think", "tell", "does", "much", "cost", "deal", "client"} tries = [q] + sorted( (w for w in _re2.findall(r"[a-z]{4,}", q.lower()) if w not in stop), key=len, reverse=True)[:2] for name in tries: eqs = urllib.parse.urlencode({"name": name, "group": _GROUP}) ents = (await asyncio.to_thread( _req, "GET", f"/entity?{eqs}")).get("entities", []) hit = next((e for e in ents if e.get("read") or e.get("summary")), None) if hit: if hit.get("read"): lines.append(f"- (your read on {hit['name']}: " f"{hit['read']})") elif hit.get("summary"): lines.append(f"- (on {hit['name']}: {hit['summary']})") break except Exception: # noqa: BLE001 pass if not lines: return _text("(nothing relevant in memory yet)") return _text("\n".join(lines)) except Exception as e: # noqa: BLE001 return _text(f"memory_search failed: {str(e)[:150]}") return create_sdk_mcp_server( name="memory", version="1.0.0", tools=[memory_remember, memory_update, memory_search], )