refactor(telemetry): tighten alias helpers and gate on MEM0_TELEMETRY

Review pass on the PostHog identity-stitching fix:

- Gate _maybe_alias_anon_to_email and _maybeAliasAnonToEmail on telemetry
  being enabled. Previous version still read ~/.mem0/config.json and
  persisted telemetry.aliased_to even when MEM0_TELEMETRY=false, which
  meant re-enabling telemetry later would skip identity merging forever.
  Python checks client_telemetry.posthog; TS exports a new
  isTelemetryEnabled() getter from telemetry.ts.
- Trim capture_identify properties to {$anon_distinct_id, client_source}.
  The full platform metadata duplicated capture_event's payload and is
  not needed for $identify; PostHog enriches the person profile from
  subsequent events anyway.
- Flatten config.ts: drop the four-helper structure (loadNodeModules /
  configPath / loadRawConfig / read+write) into one getNodeFs() + a
  shared loadConfig(). Same behaviour, ~30 fewer lines, easier to read.
- Drop the tautological Mem0AnonIds shape test and the
  void createMockFetch hack at the end of the TS test file.
- Add tests for the new telemetry-disabled gate (Python + TS).

Logic traced end-to-end: 114 Python tests + 590 TS tests pass. Ruff and
Prettier clean. Zero new typecheck errors.
This commit is contained in:
gabrielstein-mem0
2026-04-30 17:01:58 -07:00
parent 789122bbea
commit 8a9915d3fe
7 changed files with 119 additions and 108 deletions
+17
View File
@@ -297,6 +297,23 @@ class TestMaybeAliasAnonToEmail:
client_main._maybe_alias_anon_to_email("not-an-email")
telemetry.capture_identify.assert_not_called()
def test_skips_when_telemetry_disabled(self):
"""When client_telemetry.posthog is None (MEM0_TELEMETRY=false), do nothing —
no fs read, no fs write, no event. Re-enabling telemetry later must still alias."""
from mem0.client import main as client_main
disabled = MagicMock()
disabled.posthog = None
with (
patch.object(client_main, "client_telemetry", disabled),
patch.object(client_main, "read_anon_ids") as read,
patch.object(client_main, "mark_aliased") as mark,
):
client_main._maybe_alias_anon_to_email("user@example.com")
read.assert_not_called()
mark.assert_not_called()
disabled.capture_identify.assert_not_called()
def test_does_not_raise_on_telemetry_failure(self):
from mem0.client import main as client_main