"""Tool definitions + executors for the gateway brain. Two backends, both reached over plain async HTTP (httpx): * The memory service (Graphiti/FalkorDB on Railway) — same endpoints and bearer-header shape as ``voice/memory_control.py`` / ``voice/tasks_control.py`` (MEMORY_API_URL + MEMORY_API_KEY, group namespace MEMORY_GROUP). * Supabase (the Ultron master DB) via PostgREST RPCs — the exact call shape copied from ``dashboard/server.js``: POST ``{SUPABASE_REST}/rpc/`` with ``{apikey, Authorization: Bearer }`` headers. Tools exposed to the LLM (OpenAI function-calling schema, ``TOOL_SPECS``): memory_search, memory_remember, task_add, tasks_list, task_done, ultron_leads, ultron_team_activity, ultron_stats, ultron_contact_stats. The HTTP is funnelled through ``_mem_request`` and ``_rpc`` so tests can monkeypatch those two functions and never touch the network. """ from __future__ import annotations import json import os import urllib.parse import phone_store # --- config (read at call time so tests/env changes take effect) ----------- def _mem_url() -> str: return os.environ.get("MEMORY_API_URL", "").rstrip("/") def _mem_key() -> str: return os.environ.get("MEMORY_API_KEY", "") def _group() -> str: return os.environ.get("MEMORY_GROUP", "ahmed") def _rest() -> str: return os.environ.get("SUPABASE_REST", "").rstrip("/") def _sup_key() -> str: return os.environ.get("SUPABASE_SECRET", "") # --- HTTP primitives (monkeypatch these in tests) -------------------------- async def _mem_request(method: str, path: str, params: dict | None = None, body: dict | None = None, timeout: float = 20.0) -> dict: """Call the memory service. Returns parsed JSON ({} on empty body).""" import httpx base = _mem_url() if not base: raise RuntimeError("MEMORY_API_URL not set") url = base + path if params: url += "?" + urllib.parse.urlencode(params) headers = {"Content-Type": "application/json"} key = _mem_key() if key: headers["Authorization"] = f"Bearer {key}" async with httpx.AsyncClient(timeout=timeout) as client: resp = await client.request(method, url, headers=headers, json=body if body is not None else None) resp.raise_for_status() raw = resp.content return json.loads(raw) if raw else {} async def _rpc(fn: str, args: dict | None = None, timeout: float = 30.0) -> object: """Call a Supabase PostgREST RPC — same shape as dashboard/server.js.""" import httpx rest = _rest() if not rest: raise RuntimeError("SUPABASE_REST not set") key = _sup_key() headers = {"apikey": key, "Authorization": f"Bearer {key}", "Content-Type": "application/json"} async with httpx.AsyncClient(timeout=timeout) as client: resp = await client.post(f"{rest}/rpc/{fn}", headers=headers, json=args or {}) resp.raise_for_status() return resp.json() # --- salience heuristic (ported from memory_control._confidence) ----------- _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", "locked in") def _confidence(text: str) -> float: 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 # --- relative-time → (when_from, when_to) for temporal /search ------------- # Self-contained copy of voice/memory_control._temporal_range (keep in sync). _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]: """Turn a relative-time phrase in the query into an ISO (when_from, when_to) window for /search. English phrases + ISO numeric dates only; (None, None) if nothing time-like is present. No deps.""" import datetime as _dt import re as _re t = (q or "").lower() today = _dt.date.today() def _iso(d): return d.isoformat() def _month_end(d): 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) 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) 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: 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 # --- routine-change hook --------------------------------------------------- # The scheduler lives in main.py; when a routine is created/cancelled the store # changes and the running schedule must be reloaded. main registers a callback # here at startup so tools.py never imports main (avoids a circular import). _ROUTINE_HOOK = None def set_routine_hook(fn) -> None: """main registers an async callback fired after any routine mutation.""" global _ROUTINE_HOOK _ROUTINE_HOOK = fn async def _notify_routine_change() -> None: if _ROUTINE_HOOK is not None: try: await _ROUTINE_HOOK() except Exception: # noqa: BLE001 — a reschedule must never break a tool pass # --- sensitive-number masking ---------------------------------------------- def mask_number(number: str) -> str: """A phone number rendered for the model/logs so a full number is never echoed or read aloud — only the last three digits survive.""" digits = "".join(ch for ch in str(number or "") if ch.isdigit()) if not digits: return "(no number)" return "ending " + " ".join(digits[-3:]) # =========================================================================== # Session-start context (used by persona.build_system_prompt) # =========================================================================== async def memory_index_text(limit: int = 25) -> str: """A short 'index' of recent long-term memories for the system prompt. The desktop reads a local memory/MEMORY.md; the phone has no filesystem memory, so we summarise the newest facts from the service instead. Never raises — returns "" on any failure.""" try: out = await _mem_request("GET", "/memories", params={"group": _group(), "limit": limit}) except Exception: # noqa: BLE001 return "" lines = [] for m in out.get("memories", []): txt = (m.get("text") or m.get("fact") or "").strip() if txt: lines.append(f"- {txt}") return "\n".join(lines) async def values_text() -> str: """Ahmed's north-star profile from the memory service. "" on failure.""" try: out = await _mem_request("GET", "/values", params={"group": _group()}) except Exception: # noqa: BLE001 return "" return (out.get("text") or "").strip() async def passive_remember(user_text: str) -> None: """Fire-and-forget passive save after an exchange, with a simple v1 salience gate. Mirrors the desktop's auto-remember intent: only durable- looking utterances are stored, as observations. Never raises.""" t = (user_text or "").strip() if not _is_salient(t): return try: # raw=true: this is a whole utterance, not a clean fact — the server # cleans it into atomic facts and dedups. Passive saves are low-priority. await _mem_request("POST", "/remember", body={"text": t, "group": _group(), "source": "gateway", "kind": "observation", "confidence": _confidence(t), "raw": True, "importance": 4}) except Exception: # noqa: BLE001 — a save must never disrupt the chat pass _TRIVIAL = { "hi", "hey", "hello", "yo", "sup", "thanks", "thank you", "ok", "okay", "cool", "nice", "yes", "no", "yeah", "nope", "stop", "wait", "hmm", "good morning", "good night", "goodnight", "bye", "cheers", } def _is_salient(text: str) -> bool: """Cheap v1 gate: keep declarative, substantive utterances; drop greetings, one-liners, and bare questions (those are asks, not facts to store).""" t = text.strip() low = t.lower().rstrip("!.") if len(t) < 24: return False if low in _TRIVIAL: return False # a bare question is usually a request, not a durable fact if t.endswith("?") and " " in t and not any( c in low for c in ("remember", "note that", "for the record")): return False return True # =========================================================================== # Tool schemas (OpenAI function-calling format) # =========================================================================== def _fn(name: str, description: str, properties: dict, required: list[str] | None = None) -> dict: return { "type": "function", "function": { "name": name, "description": description, "parameters": { "type": "object", "properties": properties, "required": required or [], }, }, } TOOL_SPECS: list[dict] = [ _fn("memory_search", "Search Ahmed's long-term memory by meaning and get back connected " "facts. Use whenever the answer might depend on something he told you " "before — a person, client, meeting, number, price, project or plan. " "Read every fact returned; the specific one is often not first.", {"query": {"type": "string", "description": "What to recall, in natural language."}, "k": {"type": "integer", "description": "Max facts to return (default 10)."}}, ["query"]), _fn("memory_remember", "Store a durable new fact Ahmed just told you (a preference, a person, " "a decision, a number). Saving is quick — acknowledge and move on. Do " "NOT use for transient chit-chat.", {"text": {"type": "string", "description": "One clear fact to remember."}, "kind": {"type": "string", "description": "fact|preference|event|observation (default fact)."}}, ["text"]), _fn("task_add", "Add a task or reminder to Ahmed's shared, cross-device list. Use for " "'remind me to…', 'add a task', 'I need to…'.", {"text": {"type": "string", "description": "Short imperative task, e.g. 'Call the supplier'."}, "due": {"type": "string", "description": "ISO local time for a timed reminder, e.g. " "2026-07-08T17:00:00 (Asia/Riyadh). Omit for a " "plain to-do."}}, ["text"]), _fn("tasks_list", "List Ahmed's open tasks and reminders. Use for 'what are my tasks', " "'what do I have to do'.", {}), _fn("task_done", "Mark a task or reminder complete by its id (from tasks_list).", {"id": {"type": "string", "description": "The task id."}}, ["id"]), _fn("ultron_leads", "Search the Ultron lead database (Saudi business leads scraped from " "Google Maps). Use when Ahmed asks about leads, businesses, or " "prospects in a city.", {"query": {"type": "string", "description": "Free-text search (company/category), optional."}, "city": {"type": "string", "description": "City filter, e.g. 'Al Khobar'. Default all."}, "phone_filter": {"type": "string", "enum": ["all", "yes", "mobile"], "description": "'mobile' = has a WhatsApp-able Saudi " "05… number. Default all."}}, []), _fn("ultron_team_activity", "Who on the agency team contacted which leads recently (last 30 days). " "Use for 'what has the team been doing', outreach activity.", {}), _fn("ultron_stats", "The Ultron dashboard funnel/summary numbers (total leads, verified " "emails, phones, recent haul). Use for 'how many leads do we have'.", {}), _fn("ultron_contact_stats", "Outreach contact volume per day over the last N days (per client). " "Use for 'how much outreach did we do this week'.", {"days": {"type": "integer", "description": "Look-back window in days (default 30)."}}, []), # --- Phase-2 phone powers: places ------------------------------------- _fn("save_place", "Remember a place by a short label so Ahmed can later say 'navigate to " "the office'. Use when he names a location — 'this is my gym', 'save " "home as…'. Store an address, or GPS as 'lat,lng' (e.g. from a " "new-location prompt).", {"label": {"type": "string", "description": "Short name, e.g. 'work office', 'home'."}, "address_or_coords": {"type": "string", "description": "A street address, OR 'lat,lng' " "coordinates like '26.30,50.20'."}}, ["label", "address_or_coords"]), _fn("get_place", "Look up a saved place by its label. Use before navigating or when he " "refers to a named place.", {"label": {"type": "string", "description": "The place label."}}, ["label"]), _fn("list_places", "List every place Ahmed has saved (labels + where they are).", {}), # --- contacts (the phonebook Jarvis remembers) ------------------------ _fn("save_contact", "Remember a person's phone number so Ahmed can later say 'call my " "father'. Use when he gives you a name + number. Numbers are private — " "just confirm you saved it, never read the number back.", {"name": {"type": "string", "description": "Who this is, e.g. 'father', 'Sara'."}, "number": {"type": "string", "description": "Their phone number."}, "relation": {"type": "string", "description": "Optional relationship, e.g. 'family', " "'supplier'."}}, ["name", "number"]), _fn("get_contact", "Look up a saved contact by name. Returns the name/relation and a MASKED " "number (last digits only) — to actually dial, use call_contact.", {"name": {"type": "string", "description": "Who to look up."}}, ["name"]), _fn("list_contacts", "List everyone in Ahmed's saved phonebook (names + relations; numbers " "stay masked).", {}), # --- SMS watch-list ---------------------------------------------------- _fn("sms_watch_add", "Add a sender to the SMS watch-list — texts from it get forwarded to you " "to summarise/act on. Senders like 'SNB-AlAhli', 'Barq', or '*' for ALL " "senders.", {"sender": {"type": "string", "description": "Sender id, or '*' for every sender."}}, ["sender"]), _fn("sms_watch_remove", "Remove a sender from the SMS watch-list.", {"sender": {"type": "string", "description": "Sender id to stop watching."}}, ["sender"]), _fn("sms_watch_list", "List the SMS senders currently on the watch-list.", {}), # --- routines (automations) ------------------------------------------- _fn("routine_create", "Create a routine that fires automatically — e.g. 'each morning at seven " "give me a brief of my tasks' (time 07:00 daily → speak_brief tasks) or " "'when I get to the office, read my tasks' (place trigger). Resolve times " "yourself in 24h Asia/Riyadh.", {"name": {"type": "string", "description": "Short unique name, e.g. 'morning brief'."}, "trigger": {"type": "object", "description": "When it fires.", "properties": { "type": {"type": "string", "enum": ["time", "place", "both"]}, "at": {"type": "string", "description": "24h time 'HH:MM' (time/both)."}, "days": {"type": "array", "items": {"type": "string"}, "description": "Weekdays mon..sun; omit or " "['daily'] = every day."}, "place": {"type": "string", "description": "A saved place label " "(place/both)."}}, "required": ["type"]}, "action": {"type": "object", "description": "What Jarvis does when it fires.", "properties": { "type": {"type": "string", "enum": ["speak_brief", "say_text", "run"]}, "text": {"type": "string", "description": "For say_text/run: what to say/do."}, "brief": {"type": "string", "enum": ["tasks"], "description": "For speak_brief: which brief."}}, "required": ["type"]}}, ["name", "trigger", "action"]), _fn("routine_list", "List Ahmed's routines (name, when they fire, what they do).", {}), _fn("routine_cancel", "Cancel/delete a routine by name.", {"name": {"type": "string", "description": "The routine name."}}, ["name"]), ] # =========================================================================== # Executors # =========================================================================== async def _memory_search(args: dict) -> str: q = str(args.get("query", "")).strip() if not q: return "memory_search failed: no query." 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 out = await _mem_request("GET", "/search", params=params) facts = out.get("facts", []) if not facts: return "(nothing relevant in memory yet)" lines = [] for f in facts: tag = " [connected]" if f.get("linked") else "" when = (f.get("refers_to") or "")[:10] lines.append(f"- {f.get('fact', '')}{tag}" + (f" (re: {when})" if when else "") + f" [id {f.get('id', '?')}]") return "\n".join(lines) async def _memory_remember(args: dict) -> str: text = str(args.get("text", "")).strip() if not text: return "memory_remember failed: no text." kind = str(args.get("kind") or "fact").strip() or "fact" # raw=true: the server cleans the utterance into atomic facts and dedups. # importance 6 — an explicit "remember this" outranks passive observations. await _mem_request("POST", "/remember", body={"text": text, "group": _group(), "source": "gateway", "kind": kind, "confidence": _confidence(text), "raw": True, "importance": 6}) return "Remembered." async def _task_add(args: dict) -> str: text = str(args.get("text", "")).strip() if not text: return "task_add failed: no task text." body = {"text": text, "group": _group(), "source": "jarvis"} if args.get("due"): body["due"] = str(args["due"]) await _mem_request("POST", "/task", body=body) return (f"Added: {text}" + (f" (reminder set for {body['due']})" if body.get("due") else "")) async def _tasks_list(_args: dict) -> str: out = await _mem_request("GET", "/tasks", params={"group": _group()}) tasks = out.get("tasks", []) if not tasks: return "No open tasks." lines = [] for t in tasks: due = (f" (due {t['due'][:16].replace('T', ' ')})" if t.get("due") else "") lines.append(f"- {t.get('text', '')}{due} [id {t.get('id', '?')}]") return "\n".join(lines) async def _task_done(args: dict) -> str: tid = str(args.get("id", "")).strip() if not tid: return "task_done failed: no task id." await _mem_request("POST", "/task/done", body={"id": tid, "group": _group()}) return "Marked done." def _rows(result: object) -> list: """RPCs return either a list of rows or a single JSON object. Normalise.""" if isinstance(result, list): return result if isinstance(result, dict): return [result] return [] async def _ultron_leads(args: dict) -> str: q = str(args.get("query", "")).strip() city = str(args.get("city") or "all").strip() or "all" phone = str(args.get("phone_filter") or "all").strip() or "all" res = await _rpc("search_leads", { "q": q, "p_category": "all", "p_city": city, "p_fresh": "all", "p_verified": "all", "p_phone": phone, "p_client": None, "p_assigned_only": False, "lim": 12, "off": 0, }) rows = _rows(res) if not rows: return "No matching leads." lines = [] for r in rows[:12]: name = r.get("company") or r.get("name") or "(unnamed)" parts = [name] if r.get("category"): parts.append(str(r["category"])) if r.get("city"): parts.append(str(r["city"])) if r.get("phone"): parts.append(str(r["phone"])) lines.append("- " + " · ".join(parts)) return f"{len(rows)} leads:\n" + "\n".join(lines) async def _ultron_team_activity(_args: dict) -> str: res = await _rpc("get_team_activity", {"p_range": "30d"}) rows = _rows(res) if not rows: return "No team activity in the last 30 days." return json.dumps(rows[:20], ensure_ascii=False) async def _ultron_stats(_args: dict) -> str: res = await _rpc("get_dashboard_stats", {"p_range": "all"}) return json.dumps(res, ensure_ascii=False)[:1500] async def _ultron_contact_stats(args: dict) -> str: days = int(args.get("days") or 30) res = await _rpc("get_contact_stats", {"p_range": f"{days}d"}) return json.dumps(res, ensure_ascii=False)[:1500] # ---- phone powers: places ------------------------------------------------- def _parse_coords(s: str) -> tuple[float, float] | None: """Parse 'lat,lng' (or 'lat lng') into floats, else None.""" parts = [p for p in str(s or "").replace(" ", ",").split(",") if p] if len(parts) != 2: return None try: lat, lng = float(parts[0]), float(parts[1]) except ValueError: return None if -90 <= lat <= 90 and -180 <= lng <= 180: return lat, lng return None async def _save_place(args: dict) -> str: label = str(args.get("label", "")).strip() where = str(args.get("address_or_coords", "")).strip() if not label or not where: return "save_place failed: need a label and an address or coordinates." payload: dict = {"label": label} coords = _parse_coords(where) if coords: payload["lat"], payload["lng"] = coords else: payload["address"] = where await phone_store.put_item(phone_store.KIND_PLACE, payload) return f"Saved '{label}'." def _fmt_place(p: dict) -> str: if p.get("address"): return f"{p.get('label')}: {p['address']}" if "lat" in p and "lng" in p: return f"{p.get('label')}: {p['lat']}, {p['lng']}" return str(p.get("label", "")) async def _get_place(args: dict) -> str: label = str(args.get("label", "")).strip() p = await phone_store.get_item(phone_store.KIND_PLACE, label) if not p: return f"No place saved as '{label}'." return _fmt_place(p) async def _list_places(_args: dict) -> str: items = await phone_store.list_items(phone_store.KIND_PLACE) if not items: return "No places saved yet." return "\n".join("- " + _fmt_place(p) for p in items) # ---- phone powers: contacts (numbers never echoed in full) ---------------- async def _save_contact(args: dict) -> str: name = str(args.get("name", "")).strip() number = str(args.get("number", "")).strip() if not name or not number: return "save_contact failed: need a name and a number." payload = {"name": name, "number": number} rel = str(args.get("relation", "")).strip() if rel: payload["relation"] = rel await phone_store.put_item(phone_store.KIND_CONTACT, payload) return f"Saved {name}." # never echo the number def _fmt_contact(c: dict) -> str: rel = f" ({c['relation']})" if c.get("relation") else "" return f"{c.get('name')}{rel} — {mask_number(c.get('number', ''))}" async def _get_contact(args: dict) -> str: name = str(args.get("name", "")).strip() c = await phone_store.get_item(phone_store.KIND_CONTACT, name) if not c: return f"No contact saved for '{name}'." return _fmt_contact(c) async def _list_contacts(_args: dict) -> str: items = await phone_store.list_items(phone_store.KIND_CONTACT) if not items: return "No contacts saved yet." return "\n".join("- " + _fmt_contact(c) for c in items) # ---- phone powers: SMS watch-list ----------------------------------------- async def _sms_watch_add(args: dict) -> str: sender = str(args.get("sender", "")).strip() if not sender: return "sms_watch_add failed: no sender." await phone_store.put_item(phone_store.KIND_SMS_WATCH, {"sender": sender}) if sender == "*": return "Watching all SMS senders now." return f"Watching texts from {sender}." async def _sms_watch_remove(args: dict) -> str: sender = str(args.get("sender", "")).strip() if not sender: return "sms_watch_remove failed: no sender." n = await phone_store.delete_item(phone_store.KIND_SMS_WATCH, sender) return f"Stopped watching {sender}." if n else f"{sender} wasn't on the list." async def _sms_watch_list(_args: dict) -> str: items = await phone_store.list_items(phone_store.KIND_SMS_WATCH) if not items: return "The SMS watch-list is empty." return "Watching: " + ", ".join(str(i.get("sender", "")) for i in items) # ---- phone powers: routines ----------------------------------------------- def _fmt_routine(r: dict) -> str: trig = r.get("trigger") or {} act = r.get("action") or {} when = trig.get("type", "?") if trig.get("at"): when += f" {trig['at']}" if trig.get("days"): when += " " + "/".join(trig["days"]) if trig.get("place"): when += f" @{trig['place']}" what = act.get("type", "?") if act.get("brief"): what += f":{act['brief']}" elif act.get("text"): what += f" \"{act['text'][:40]}\"" return f"{r.get('name')} — when {when} → {what}" async def _routine_create(args: dict) -> str: name = str(args.get("name", "")).strip() trigger = args.get("trigger") action = args.get("action") if not name or not isinstance(trigger, dict) or not isinstance(action, dict): return "routine_create failed: need a name, a trigger and an action." if trigger.get("type") not in ("time", "place", "both"): return "routine_create failed: trigger.type must be time, place or both." if action.get("type") not in ("speak_brief", "say_text", "run"): return "routine_create failed: action.type must be speak_brief, say_text or run." if trigger["type"] in ("time", "both") and not str(trigger.get("at", "")).strip(): return "routine_create failed: a time/both trigger needs a time 'at'." if trigger["type"] in ("place", "both") and not str(trigger.get("place", "")).strip(): return "routine_create failed: a place/both trigger needs a place label." await phone_store.put_item(phone_store.KIND_ROUTINE, {"name": name, "trigger": trigger, "action": action}) await _notify_routine_change() return f"Routine '{name}' set." async def _routine_list(_args: dict) -> str: items = await phone_store.list_items(phone_store.KIND_ROUTINE) if not items: return "No routines set." return "\n".join("- " + _fmt_routine(r) for r in items) async def _routine_cancel(args: dict) -> str: name = str(args.get("name", "")).strip() if not name: return "routine_cancel failed: no name." n = await phone_store.delete_item(phone_store.KIND_ROUTINE, name) await _notify_routine_change() return f"Cancelled '{name}'." if n else f"No routine named '{name}'." _EXECUTORS = { "memory_search": _memory_search, "memory_remember": _memory_remember, "task_add": _task_add, "tasks_list": _tasks_list, "task_done": _task_done, "ultron_leads": _ultron_leads, "ultron_team_activity": _ultron_team_activity, "ultron_stats": _ultron_stats, "ultron_contact_stats": _ultron_contact_stats, "save_place": _save_place, "get_place": _get_place, "list_places": _list_places, "save_contact": _save_contact, "get_contact": _get_contact, "list_contacts": _list_contacts, "sms_watch_add": _sms_watch_add, "sms_watch_remove": _sms_watch_remove, "sms_watch_list": _sms_watch_list, "routine_create": _routine_create, "routine_list": _routine_list, "routine_cancel": _routine_cancel, } async def execute(name: str, args: dict) -> str: """Dispatch a tool call. Returns a compact string for the model to read.""" fn = _EXECUTORS.get(name) if fn is None: return f"unknown tool '{name}'" try: return await fn(args or {}) except Exception as e: # noqa: BLE001 — surfaced to the model, never crashes return f"{name} failed: {str(e)[:200]}"