"""Structured phone-power storage on top of the memory service. Phase-2 "Jarvis runs my phone" needs a handful of small, structured, mutable collections — saved *places*, a *contacts* phonebook, an *SMS watch-list*, and *routines* — that must survive restarts. Rather than stand up a new database we persist each item as a **typed memory record in a DEDICATED group namespace** (``PHONE_STORE_GROUP``, default ``"_phone"``). That gives three things for free: * durability — the memory service is FalkorDB-backed, so records outlive any gateway restart or redeploy; * isolation — a separate ``group`` means these machine records never pollute Ahmed's semantic long-term memory (``memory_search`` / the persona memory index both read the ``ahmed`` group, not ``ahmed_phone``); * deterministic CRUD — ``GET /memories?group=&kind=`` lists newest-first and filters by an exact ``kind``, ``POST /remember`` creates, ``POST /forget`` deletes by id. No semantic ranking is ever involved, so listing/removing an item is exact, not fuzzy. One record per item. ``kind`` names the collection; the item's fields live as a compact JSON object in the record ``text``. Numbers are treated as sensitive — this module never logs a record's contents. All HTTP funnels through :func:`_mem_request` so tests monkeypatch exactly one function and never touch the network. This module deliberately does NOT import ``tools`` (which imports us), keeping its own tiny HTTP helper to avoid a circular import. """ from __future__ import annotations import json import os import urllib.parse # Collection kinds (the memory-record ``kind`` for each) and the field whose # value uniquely identifies an item within that collection. KIND_PLACE = "phone_place" KIND_CONTACT = "phone_contact" KIND_SMS_WATCH = "phone_sms_watch" KIND_ROUTINE = "phone_routine" _KEY_FIELD = { KIND_PLACE: "label", KIND_CONTACT: "name", KIND_SMS_WATCH: "sender", KIND_ROUTINE: "name", } # --- 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 _base_group() -> str: return os.environ.get("MEMORY_GROUP", "ahmed") def phone_group() -> str: """The dedicated namespace for phone-power records (kept OFF the semantic memory group so these never surface in recall or the persona index).""" return os.environ.get("PHONE_STORE_GROUP") or (_base_group() + "_phone") # --- HTTP primitive (monkeypatch this 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 {} # --- normalisation --------------------------------------------------------- def _norm(value: str) -> str: """Case/space-insensitive key for matching. ``"*"`` is preserved verbatim.""" v = (value or "").strip() return v if v == "*" else v.lower() # --- CRUD ------------------------------------------------------------------ async def list_items(kind: str) -> list[dict]: """Every item in a collection, newest-first. Each item is the stored JSON payload plus an ``_id`` (the memory-record id, used for deletes).""" out = await _mem_request("GET", "/memories", params={"group": phone_group(), "kind": kind, "limit": 200}) items: list[dict] = [] for m in out.get("memories", []): raw = m.get("fact") or m.get("text") or "" try: payload = json.loads(raw) except (ValueError, TypeError): continue if isinstance(payload, dict): payload["_id"] = m.get("id") items.append(payload) return items async def get_item(kind: str, key: str) -> dict | None: """Fetch one item by its key field (label/name/sender), or None.""" want = _norm(key) if not want: return None field = _KEY_FIELD[kind] for item in await list_items(kind): if _norm(str(item.get(field, ""))) == want: return item return None async def delete_item(kind: str, key: str) -> int: """Delete every item whose key field matches. Returns how many were removed (0 if none). Idempotent — used both to remove and to make put_item an upsert.""" want = _norm(key) if not want: return 0 field = _KEY_FIELD[kind] removed = 0 for item in await list_items(kind): if _norm(str(item.get(field, ""))) == want and item.get("_id"): await _mem_request("POST", "/forget", body={"id": item["_id"], "group": phone_group()}) removed += 1 return removed async def put_item(kind: str, payload: dict) -> dict: """Upsert an item (replace any existing one with the same key), then store it. Returns the stored payload (without the transient ``_id``).""" field = _KEY_FIELD[kind] key = str(payload.get(field, "")).strip() if not key: raise ValueError(f"{kind}: missing required '{field}'") clean = {k: v for k, v in payload.items() if k != "_id"} await delete_item(kind, key) await _mem_request("POST", "/remember", body={"text": json.dumps(clean, ensure_ascii=False), "group": phone_group(), "kind": kind, "source": "phone", "confidence": 0.9}) return clean # --- SMS watch-list matching (pure) ---------------------------------------- def sms_matches(sender: str, watch: list[str]) -> bool: """Does ``sender`` match the watch-list? ``"*"`` matches everything; a stored entry matches on exact (case-insensitive) equality OR as a substring of the sender (so ``"SNB"`` catches ``"SNB-AlAhli"``).""" s = _norm(sender) if not s: return False for w in watch: wn = _norm(str(w)) if wn == "*": return True if wn and (wn == s or wn in s): return True return False