"""Entity Dossier — put a profile card of a person/company/concept on the HUD. Two ways in: 1. The brain's ``show_entity(name)`` tool (mcp__entity__show_entity) — Jarvis calls it when Ahmed says "show me Saud" / "who is X / tell me about X". 2. AUTO-SHOW — main.py's listen loop calls ``maybe_autoshow(text)`` on every addressed utterance. When Ahmed names exactly ONE known entity (word- boundary match against a name set refreshed ~every 10 min) and that entity's per-entity cooldown is clear, its card pops on its own. The card is delivered to the HUD as an ``entity_profile`` event: emit("entity_profile", profile={...}) The HUD renders the dossier from that JSON and can send back-channel actions (control/entity_action.json → voice/control.py: drill-down, note, summary edit). SERVER CONTRACT (memory service, GET /profile?name=X → dossier JSON with keys entity, summary, read, quick, relations, timeline, facts_by_kind, tasks, counts, also_matched): - a KNOWN-ABSENT entity comes back HTTP 200 with ``{"entity": null, ...}`` — that is "nothing known", NOT an error: we speak "I don't have anything on X yet" and do NOT emit the event. - an AMBIGUOUS name returns the best entity plus ``also_matched: [names]`` — we name the runner-up in the spoken confirmation so Ahmed can drill down. Everything here degrades gracefully. The network may hang or the route may not exist yet — every call has a hard timeout, swallows errors to a log line, and NEVER blocks the voice loop (the name-set refresh and the auto-show fetch both run on daemon threads; the brain tool wraps the fetch in asyncio.to_thread). """ from __future__ import annotations import os import re import threading import time import urllib.parse from voice import memory_control from voice.events import emit # --------------------------------------------------------------------------- # profile fetch + card push # --------------------------------------------------------------------------- def fetch_profile(name: str) -> dict | None: """GET /profile?name=X — the dossier JSON. Hard 3s ceiling. Returns None only on a network/timeout error or when memory isn't configured. A known-absent entity is NOT an error (the server answers 200 with {"entity": null}); that dict is returned as-is and the caller checks the ``entity`` field.""" name = (name or "").strip() if not name or not memory_control.enabled(): return None try: qs = urllib.parse.urlencode( {"name": name, "group": memory_control._GROUP}) # noqa: SLF001 out = memory_control._req("GET", f"/profile?{qs}", timeout=3) # noqa: SLF001 except Exception as e: # noqa: BLE001 — 404/timeout/offline all degrade print(f" [entity] profile fetch failed for {name!r}: {str(e)[:100]}") return None return out if isinstance(out, dict) else None def fetch_composed(name: str) -> dict | None: """POST /dossier {"name","group"} → {"composed": {...}|null, "cached": bool} — the LLM-COMPOSED dossier (a written narrative over the raw profile). Hard 25s ceiling: composition is 5-12s uncached, instant when cached. Returns the composed dict, or None on a 404 (route not deployed yet), timeout, network error, non-dict body, or a null composition — every one degrades to the HUD's raw rendering. Never raises.""" name = (name or "").strip() if not name or not memory_control.enabled(): return None try: out = memory_control._req( # noqa: SLF001 "POST", "/dossier", {"name": name, "group": memory_control._GROUP}, # noqa: SLF001 timeout=25) except Exception as e: # noqa: BLE001 — 404/timeout/offline all degrade print(f" [entity] dossier compose failed for {name!r}: {str(e)[:100]}") return None if not isinstance(out, dict): return None composed = out.get("composed") return composed if isinstance(composed, dict) else None def _compose_and_emit(shown: str, profile: dict) -> None: """STAGE 2 (daemon thread): compose the dossier off the hot path and swap it into the already-showing card with a SECOND ``entity_profile`` event. On success it carries ``composed= composing=False``; on failure/timeout/ null it carries ``composed=None composing=False`` so the HUD drops the "composing…" spinner and falls back to rendering the raw profile. Never raises, never blocks a turn. One RETRY after a short wait: a slow first composition can outlive our timeout while still finishing (and caching) server-side — the second POST then returns the cached result instantly.""" composed = fetch_composed(shown) if composed is None: time.sleep(_COMPOSE_RETRY_S) composed = fetch_composed(shown) try: emit("entity_profile", profile=profile, composed=composed, composing=False) except Exception as e: # noqa: BLE001 — a card push must never crash a turn print(f" [entity] composed emit failed: {str(e)[:100]}") def _show(name: str) -> tuple[str, str, list]: """The TRUTHFUL core every entry point shares. STAGE 1: fetch /profile (≤3s) and emit the card with ``composed=None composing=True`` — it pops instantly. STAGE 2: a daemon thread POSTs /dossier and a second emit swaps the composed narrative in. Returns (status, shown_name, also_matched): 'shown' — the card IS on the HUD 'unknown' — no such entity in memory; NO card was emitted 'error' — memory unreachable / emit failed; NO card on screen Callers must only claim a card is visible when status == 'shown' — the old fire-and-forget version let the brain say "on your HUD, sir" while nothing showed (Ahmed asked for "reval clients", not an entity, and got gaslit).""" name = (name or "").strip() if not name: return ("error", name, []) profile = fetch_profile(name) if profile is None: return ("error", name, []) # entity:null = nothing known — do NOT emit a card if not profile.get("entity"): return ("unknown", name, []) ent = profile.get("entity") or {} shown = str(ent.get("name") or name).strip() or name also = [str(a).strip() for a in (profile.get("also_matched") or []) if str(a).strip()] try: emit("entity_profile", profile=profile, composed=None, composing=True) except Exception as e: # noqa: BLE001 — a card push must never crash a turn print(f" [entity] emit failed: {str(e)[:100]}") return ("error", shown, also) _mark_shown(shown) # start the cooldown so auto-show doesn't re-pop it threading.Thread(target=_compose_and_emit, args=(shown, profile), daemon=True).start() return ("shown", shown, also) def show_entity(name: str) -> str: """Brain-tool / control-file entry: run ``_show`` and return a short SPOKEN confirmation string (plain prose, TTS-safe) the MOMENT the card is up — it NEVER waits for composition. Callers on the event loop wrap it in asyncio.to_thread / a daemon thread. Never raises.""" name = (name or "").strip() if not name: return "I need a name to show, sir." status, shown, also = _show(name) if status == "unknown": return f"I don't have anything on {shown} yet, sir." if status == "error": return f"I couldn't reach memory for {shown}, sir — no card, I'm afraid." if also: extra = also[0] if len(also) == 1 else f"{also[0]} and others" return f"Showing {shown}, sir — I also know a {extra}." return f"Showing {shown}'s profile, sir." # --------------------------------------------------------------------------- # entity-mention watcher (auto-show) # --------------------------------------------------------------------------- _COMPOSE_RETRY_S = float(os.environ.get("ENTITY_COMPOSE_RETRY_S", "8")) _REFRESH_S = 600.0 # rebuild the known-entity set every ~10 min _COOLDOWN_S = 600.0 # per-entity: don't auto-show the same one within 10 min _MIN_LEN = 2 # ignore 1-char "names" _STRONG_LEN = 3 # auto-show needs a name at least this long # Generic words that must never be treated as an entity — an entity whose name # lowercases to one of these is dropped from the matcher (guards against a stray # ":Entity" like "work"/"team"/"it" auto-popping a card mid-sentence). _STOP = frozenset({ "the", "and", "for", "you", "your", "are", "was", "were", "new", "now", "who", "why", "how", "what", "when", "where", "this", "that", "these", "those", "here", "there", "with", "from", "into", "some", "any", "all", "one", "two", "get", "got", "see", "let", "yes", "yeah", "yep", "nope", "okay", "please", "thanks", "today", "tomorrow", "tonight", "work", "home", "team", "task", "email", "phone", "call", "text", "him", "her", "them", "they", "she", "his", "our", "out", "not", "but", "can", "will", "just", "like", "want", "need", "have", "has", "had", "did", "does", "jarvis", "ahmed", "sir", }) _names_lock = threading.Lock() # map: token(lower) -> display name · pattern: compiled alternation · at: refresh ts _NAMES: dict = {"map": {}, "pattern": None, "at": 0.0, "refreshing": False} _COOLDOWN: dict = {} # name(lower) -> last auto-shown (monotonic) def _mark_shown(name: str) -> None: with _names_lock: _COOLDOWN[name.lower()] = time.monotonic() def _refresh_names() -> None: """Rebuild the known-entity matcher from GET /graph — the lightest endpoint that enumerates entities (/export is a full dump). Entity nodes come back as ``{"id": "ent:", "text": , "kind": "entity:"}``. Runs on a daemon thread with a bounded timeout; on any failure the last good matcher stays in place. Never raises.""" try: qs = urllib.parse.urlencode({"group": memory_control._GROUP}) # noqa: SLF001 out = memory_control._req("GET", f"/graph?{qs}", timeout=8) # noqa: SLF001 toks: dict = {} for n in (out.get("nodes") or []): if not str(n.get("kind") or "").startswith("entity:"): continue nm = str(n.get("text") or "").strip() if len(nm) < _MIN_LEN or nm.lower() in _STOP: continue toks[nm.lower()] = nm # token(lower) -> display name pattern = None if toks: # one alternation, longest tokens first so "al temimi" beats "al"; # (? None: """Kick a background refresh if the matcher is stale and none is in flight. Non-blocking — returns at once; the daemon thread does the network.""" if not memory_control.enabled(): return with _names_lock: recent = (time.time() - _NAMES["at"]) < _REFRESH_S if recent or _NAMES["refreshing"]: return _NAMES["refreshing"] = True threading.Thread(target=_refresh_names, daemon=True).start() def mentioned_entities(text: str) -> list[str]: """Known entity display-names named in ``text``, by Unicode word-boundary match against the cached name set (2+ chars, generic words skipped). Does NO network on the hot path — it only kicks a background refresh — and returns [] until the set is first populated. Safe to call every utterance.""" _maybe_refresh() t = (text or "").lower() if not t: return [] with _names_lock: pattern = _NAMES["pattern"] mapping = _NAMES["map"] if pattern is None: return [] hits: list[str] = [] seen: set = set() for m in pattern.finditer(t): disp = mapping.get(m.group(0), m.group(0)) if disp not in seen: seen.add(disp) hits.append(disp) return hits def _agents_busy() -> bool: """True while a background worker is running — auto-show stays out of the way during an active task dispatch (best-effort, never raises).""" try: from voice import agent_pool return bool(getattr(agent_pool, "pool", None) and agent_pool.pool._running) # noqa: SLF001 except Exception: # noqa: BLE001 return False # Explicit profile asks — "pull up X", "who is X", "من هو X". These are # DETERMINISTIC: no LLM judgment, no busy-gate, no 10-min cooldown. Ahmed's # rule (proven again today): enforce behavior in code, not in the prompt — # the brain delegates research asks to workers that write files, so the card # must fire BEFORE the brain ever sees the utterance. _EXPLICIT_RES = [re.compile(p, re.IGNORECASE) for p in ( r"\b(?:pull up|bring up|show me|open up)\s+(?:everything\s+" r"(?:we|you)\s+know\s+about\s+|everything\s+about\s+|the\s+profile\s+" r"(?:of|for|on)\s+|a\s+profile\s+(?:of|for|on)\s+)?(.{2,60}?)[\s.!?]*$", r"\b(?:who\s+is|who's|tell me (?:everything )?about|what do (?:we|you) " r"know about|everything (?:we|you) know about)\s+(.{2,60}?)[\s.!?]*$", )] + [re.compile(p) for p in ( r"(?:من هو|مين هو|وريني|ورني|اعرض)\s+(.{2,60}?)[\s.!?؟]*$", r"(?:كل (?:شي|شيء) (?:نعرفه? )?عن)\s+(.{2,60}?)[\s.!?؟]*$", )] _TAIL_STRIP = re.compile( r"^(?:the|a|his|her|their|this|that|please|profile of|about)\s+|" r"\s+(?:please|now|again)$", re.IGNORECASE) def explicit_ask(text: str) -> tuple | None: """If the utterance is an explicit profile ask, show the card INLINE (≤3s fetch — truth beats speed here) and return (status, name): ('shown', X) — the card IS on the HUD ('unknown', X) — no such entity; nothing was shown ('error', X) — memory unreachable; nothing was shown Returns None when the utterance isn't a profile ask. Prefers a KNOWN entity named in the tail; otherwise sends the tail itself — the server resolves fuzzily.""" try: for rx in _EXPLICIT_RES: m = rx.search(text or "") if not m: continue tail = (m.group(1) or "").strip() prev = None while tail and tail != prev: # strip filler words iteratively prev, tail = tail, _TAIL_STRIP.sub("", tail).strip() if len(tail) < 2: continue known = mentioned_entities(tail) name = max(known, key=len) if known else tail now = time.monotonic() if now - _COOLDOWN.get("q:" + name.lower(), 0.0) < 10.0: return ("shown", name) # already firing for this ask _COOLDOWN["q:" + name.lower()] = now status, shown, _also = _show(name) return (status, shown) except Exception as e: # noqa: BLE001 print(f" [entity] explicit_ask skipped: {str(e)[:100]}") return None def handle_utterance(text: str) -> tuple | None: """main.py's ONE integration line: deterministic explicit-ask intercept first (always wins), mention auto-show as the fallback. Returns (status, name) for an EXPLICIT ask — 'shown' means the card is genuinely on the HUD (brain: acknowledge in one line); 'unknown'/'error' mean NO card is visible (brain: say so, never claim otherwise). Returns None on a non-ask (a passive mention auto-show still pops its card silently and needs no brain note).""" res = explicit_ask(text) if res: return res maybe_autoshow(text) return None def show_for_query(query: str) -> None: """DETERMINISTIC card on an explicit ask. Called by the brain's memory_search tool: when Ahmed asks memory about something that names a known entity ("pull up everything about AIN"), the dossier card MUST appear — regardless of which tool the brain chose to answer with. Picks the LONGEST entity match in the query, fires on a daemon thread, and only guards against immediate double-fires (30s), not the 10-min auto-show cooldown — an explicit ask always re-shows. Never raises, never blocks.""" try: names = mentioned_entities(query) if not names: return name = max(names, key=len) now = time.monotonic() if now - _COOLDOWN.get("q:" + name.lower(), 0.0) < 30.0: return _COOLDOWN["q:" + name.lower()] = now _COOLDOWN[name.lower()] = now # also settle the auto-show cooldown threading.Thread(target=show_entity, args=(name,), daemon=True).start() except Exception as e: # noqa: BLE001 print(f" [entity] show_for_query skipped: {str(e)[:100]}") def maybe_autoshow(text: str) -> None: """AUTO-SHOW entry point for main.py's listen loop — ONE line of integration. If Ahmed named exactly ONE strong, known entity and its 10-min cooldown is clear, pop its card. Gated by env ENTITY_AUTOSHOW (default on) and suppressed while a background agent is running. Never blocks (the fetch runs on a daemon thread) and never raises.""" try: if os.environ.get("ENTITY_AUTOSHOW", "1") == "0": return if _agents_busy(): return strong = [h for h in mentioned_entities(text) if len(h) >= _STRONG_LEN] if len(strong) != 1: # need a single, unambiguous mention return name = strong[0] now = time.monotonic() with _names_lock: if now - _COOLDOWN.get(name.lower(), 0.0) < _COOLDOWN_S: return _COOLDOWN[name.lower()] = now # claim it so we don't double-fire threading.Thread(target=show_entity, args=(name,), daemon=True).start() except Exception as e: # noqa: BLE001 — auto-show must never break the loop print(f" [entity] auto-show skipped: {str(e)[:100]}") # --------------------------------------------------------------------------- # brain MCP tool (mcp__entity__show_entity) # --------------------------------------------------------------------------- def build_server(): """The in-process ``entity`` MCP server — ONE brain tool, show_entity. Follows the memory/agent-pool pattern; dict-shorthand marks ``name`` required.""" import asyncio from claude_agent_sdk import create_sdk_mcp_server, tool def _text(msg: str) -> dict: return {"content": [{"type": "text", "text": msg}]} @tool("show_entity", "Show a full profile card (dossier) of a person/company/project/" "concept on Ahmed's screen: summary, contacts, relations, timeline " "of every interaction. ALWAYS call this when he says 'show me X', " "'pull up X', 'pull up everything about X', 'who is X', or 'tell me " "about X' — the card is the answer; speak only a brief highlight.", {"name": str}) async def show_entity_tool(args: dict) -> dict: name = str(args.get("name", "")).strip() if not name: return _text("show_entity failed: no name given.") try: msg = await asyncio.to_thread(show_entity, name) except Exception as e: # noqa: BLE001 return _text(f"show_entity failed: {str(e)[:150]}") return _text(msg) return create_sdk_mcp_server( name="entity", version="1.0.0", tools=[show_entity_tool], )