fix (telemetry): stitch oss and platform telemetry identities for python and typescript sdk

Co-authored-by: Younes Slaoui <younes.slaoui@mem0.ai>

mem0/memory/setup.py

@@ -1,6 +1,8 @@
 import json
+import logging
 import os
 import uuid
+from hashlib import sha256
 
 # Set up the directory path
 VECTOR_ID = str(uuid.uuid4())
@@ -8,28 +10,113 @@ home_dir = os.path.expanduser("~")
 mem0_dir = os.environ.get("MEM0_DIR") or os.path.join(home_dir, ".mem0")
 os.makedirs(mem0_dir, exist_ok=True)
 
+_logger = logging.getLogger(__name__)
+
+
+def _config_path():
+    return os.path.join(mem0_dir, "config.json")
+
+
+def _load_config():
+    """Load ~/.mem0/config.json, returning {} on missing/malformed file."""
+    path = _config_path()
+    if not os.path.exists(path):
+        return {}
+    try:
+        with open(path, "r") as f:
+            data = json.load(f)
+        return data if isinstance(data, dict) else {}
+    except Exception as e:
+        _logger.debug("Failed to load mem0 config %s: %s", path, e)
+        return {}
+
+
+def _write_config(config):
+    """Best-effort write of ~/.mem0/config.json. Never raises."""
+    path = _config_path()
+    try:
+        with open(path, "w") as f:
+            json.dump(config, f, indent=4)
+    except Exception as e:
+        _logger.debug("Failed to write mem0 config %s: %s", path, e)
+
 
 def setup_config():
-    config_path = os.path.join(mem0_dir, "config.json")
-    if not os.path.exists(config_path):
-        user_id = str(uuid.uuid4())
-        config = {"user_id": user_id}
-        with open(config_path, "w") as config_file:
-            json.dump(config, config_file, indent=4)
+    """Ensure ~/.mem0/config.json exists with a top-level user_id.
+
+    Idempotent: backfills user_id for users whose config was written by the
+    CLI (which writes telemetry.anonymous_id but no top-level user_id).
+    Without this, OSS Python telemetry is silently dropped because
+    get_user_id() returns None when user_id is missing.
+    """
+    config = _load_config()
+    if config.get("user_id"):
+        return
+    config["user_id"] = str(uuid.uuid4())
+    _write_config(config)
 
 
 def get_user_id():
-    config_path = os.path.join(mem0_dir, "config.json")
-    if not os.path.exists(config_path):
+    config = _load_config()
+    if not config:
         return "anonymous_user"
+    return config.get("user_id")
 
-    try:
-        with open(config_path, "r") as config_file:
-            config = json.load(config_file)
-            user_id = config.get("user_id")
-            return user_id
-    except Exception:
-        return "anonymous_user"
+
+def read_anon_ids():
+    """Return anon IDs and alias markers from ~/.mem0/config.json.
+
+    Returns a dict with keys "oss", "cli", "aliased_pairs" (IDs may be
+    None). OSS Python writes top-level "user_id"; the CLI writes
+    "telemetry.anonymous_id". They may coexist depending on which surface ran
+    first.
+    """
+    config = _load_config()
+    telemetry = config.get("telemetry") if isinstance(config.get("telemetry"), dict) else {}
+    aliased_pairs = telemetry.get("aliased_pairs")
+    return {
+        "oss": config.get("user_id"),
+        "cli": telemetry.get("anonymous_id"),
+        "aliased_pairs": aliased_pairs if isinstance(aliased_pairs, list) else [],
+    }
+
+
+def _alias_pair_marker(anon_id, email):
+    return sha256(f"{anon_id}\0{email}".encode("utf-8")).hexdigest()
+
+
+def is_aliased(anon_id, email):
+    """Return whether anon_id -> email has already been identified."""
+    if not anon_id or not email:
+        return False
+    config = _load_config()
+    telemetry = config.get("telemetry") if isinstance(config.get("telemetry"), dict) else {}
+    aliased_pairs = telemetry.get("aliased_pairs")
+    if not isinstance(aliased_pairs, list):
+        return False
+    return _alias_pair_marker(anon_id, email) in aliased_pairs
+
+
+def mark_aliased(anon_id, email):
+    """Persist an anon_id -> email alias marker so $identify fires once per pair.
+
+    The marker is hashed to avoid storing platform emails in the local config.
+    """
+    if not anon_id or not email:
+        return
+    config = _load_config()
+    telemetry = config.get("telemetry")
+    if not isinstance(telemetry, dict):
+        telemetry = {}
+    aliased_pairs = telemetry.get("aliased_pairs")
+    if not isinstance(aliased_pairs, list):
+        aliased_pairs = []
+    marker = _alias_pair_marker(anon_id, email)
+    if marker not in aliased_pairs:
+        aliased_pairs.append(marker)
+    telemetry["aliased_pairs"] = aliased_pairs
+    config["telemetry"] = telemetry
+    _write_config(config)
 
 
 def get_or_create_user_id(vector_store=None):

mem0/memory/telemetry.py

@@ -48,7 +48,8 @@ MEM0_TELEMETRY_SAMPLE_RATE = _parse_sample_rate(os.environ.get("MEM0_TELEMETRY_S
 
 # Events that bypass sampling and always fire. Keep this set in sync with the
 # event names passed to capture_event() in mem0/memory/main.py.
-_LIFECYCLE_EVENTS = frozenset({"mem0.init", "mem0.reset", "mem0._create_procedural_memory"})
+# $identify is included so PostHog person-merging is never lost to sampling.
+_LIFECYCLE_EVENTS = frozenset({"mem0.init", "mem0.reset", "mem0._create_procedural_memory", "$identify"})
 
 
 def _sampling_before_send(msg):
@@ -112,6 +113,23 @@ class AnonymousTelemetry:
         except Exception as e:
             _logger.debug("Failed to capture telemetry event %r: %s", event_name, e)
 
+    def capture_identify(self, anon_id, email):
+        """Fire $identify with $anon_distinct_id so PostHog merges anon_id into email."""
+        if self.posthog is None:
+            return False
+        if not anon_id or not email or anon_id == email:
+            return False
+        try:
+            self.posthog.capture(
+                distinct_id=email,
+                event="$identify",
+                properties={"$anon_distinct_id": anon_id, "client_source": "python"},
+            )
+            return True
+        except Exception as e:
+            _logger.debug("Failed to capture $identify for %r: %s", email, e)
+            return False
+
     def close(self):
         if self.posthog is not None:
             self.posthog.shutdown()