diff --git a/mem0-ts/src/client/config.ts b/mem0-ts/src/client/config.ts new file mode 100644 index 000000000..646360661 --- /dev/null +++ b/mem0-ts/src/client/config.ts @@ -0,0 +1,165 @@ +/** + * Best-effort read/write of ~/.mem0/config.json from the TS SDK. + * + * Used to stitch PostHog identities: SDKs and CLIs persist anonymous + * distinct_id values here, and the TS MemoryClient reads those on init to + * fire $identify and merge them into the email identity. + * + * Node-only. Browsers (no `process.versions.node`) no-op. + */ + +export interface Mem0AnonIds { + oss?: string; + cli?: string; + aliasedPairs: string[]; +} + +interface NodeFs { + fs: typeof import("fs"); + path: typeof import("path"); + crypto: typeof import("crypto"); + configPath: string; +} + +async function getNodeFs(): Promise { + if (typeof process === "undefined" || !process.versions?.node) return null; + try { + const [fs, path, os, crypto] = await Promise.all([ + import("fs"), + import("path"), + import("os"), + import("crypto"), + ]); + const fsMod = (fs as any).default ?? fs; + const pathMod = (path as any).default ?? path; + const osMod = (os as any).default ?? os; + const cryptoMod = (crypto as any).default ?? crypto; + const dir = process.env.MEM0_DIR || pathMod.join(osMod.homedir(), ".mem0"); + return { + fs: fsMod, + path: pathMod, + crypto: cryptoMod, + configPath: pathMod.join(dir, "config.json"), + }; + } catch { + return null; + } +} + +function loadConfig(node: NodeFs): Record | null { + try { + if (!node.fs.existsSync(node.configPath)) return null; + const parsed = JSON.parse(node.fs.readFileSync(node.configPath, "utf8")); + return parsed && typeof parsed === "object" ? parsed : null; + } catch { + return null; + } +} + +function writeConfig(node: NodeFs, config: Record): void { + node.fs.mkdirSync(node.path.dirname(node.configPath), { recursive: true }); + node.fs.writeFileSync(node.configPath, JSON.stringify(config, null, 4)); +} + +function aliasPairMarker(node: NodeFs, anonId: string, email: string): string { + return node.crypto + .createHash("sha256") + .update(`${anonId}\0${email}`, "utf8") + .digest("hex"); +} + +function randomUserId(node: NodeFs): string { + if (typeof node.crypto.randomUUID === "function") { + return node.crypto.randomUUID(); + } + return ( + Math.random().toString(36).substring(2, 15) + + Math.random().toString(36).substring(2, 15) + ); +} + +export async function getOrCreateMem0UserId(): Promise { + const node = await getNodeFs(); + if (!node) return null; + try { + const config = loadConfig(node) ?? {}; + if (typeof config.user_id === "string" && config.user_id) { + return config.user_id; + } + const userId = randomUserId(node); + config.user_id = userId; + writeConfig(node, config); + return userId; + } catch { + return null; + } +} + +export async function readMem0AnonIds(): Promise { + const node = await getNodeFs(); + if (!node) return null; + const config = loadConfig(node); + if (!config) return null; + const telemetry = + config.telemetry && typeof config.telemetry === "object" + ? config.telemetry + : {}; + return { + oss: typeof config.user_id === "string" ? config.user_id : undefined, + cli: + typeof telemetry.anonymous_id === "string" + ? telemetry.anonymous_id + : undefined, + aliasedPairs: Array.isArray(telemetry.aliased_pairs) + ? telemetry.aliased_pairs.filter( + (item: unknown) => typeof item === "string", + ) + : [], + }; +} + +export async function isMem0Aliased( + anonId: string, + email: string, +): Promise { + if (!anonId || !email) return false; + const node = await getNodeFs(); + if (!node) return false; + const config = loadConfig(node); + if (!config) return false; + const telemetry = + config.telemetry && typeof config.telemetry === "object" + ? config.telemetry + : {}; + const aliasedPairs = Array.isArray(telemetry.aliased_pairs) + ? telemetry.aliased_pairs + : []; + return aliasedPairs.includes(aliasPairMarker(node, anonId, email)); +} + +export async function markMem0Aliased( + anonId: string, + email: string, +): Promise { + const node = await getNodeFs(); + if (!node) return; + try { + const config = loadConfig(node) ?? {}; + const telemetry = + config.telemetry && typeof config.telemetry === "object" + ? config.telemetry + : {}; + const aliasedPairs = Array.isArray(telemetry.aliased_pairs) + ? telemetry.aliased_pairs + : []; + const marker = aliasPairMarker(node, anonId, email); + if (!aliasedPairs.includes(marker)) { + aliasedPairs.push(marker); + } + telemetry.aliased_pairs = aliasedPairs; + config.telemetry = telemetry; + writeConfig(node, config); + } catch { + // Best-effort: read-only filesystems and unwritable paths just skip. + } +} diff --git a/mem0-ts/src/client/mem0.ts b/mem0-ts/src/client/mem0.ts index 783d7f925..cac2fe36c 100644 --- a/mem0-ts/src/client/mem0.ts +++ b/mem0-ts/src/client/mem0.ts @@ -20,7 +20,18 @@ import { CreateMemoryExportPayload, GetMemoryExportPayload, } from "./mem0.types"; -import { captureClientEvent, generateHash } from "./telemetry"; +import { + captureClientEvent, + generateHash, + isTelemetryEnabled, + telemetry, +} from "./telemetry"; +import { + getOrCreateMem0UserId, + isMem0Aliased, + markMem0Aliased, + readMem0AnonIds, +} from "./config"; import { camelToSnake, camelToSnakeKeys, snakeToCamelKeys } from "./utils"; import { createExceptionFromResponse, MemoryError } from "../common/exceptions"; @@ -118,6 +129,8 @@ export default class MemoryClient { this.telemetryId = generateHash(this.apiKey); } + await this._maybeAliasAnonToEmail(); + captureClientEvent("init", this, { client_type: "MemoryClient", }).catch((error: any) => { @@ -132,6 +145,30 @@ export default class MemoryClient { } } + private async _maybeAliasAnonToEmail(): Promise { + if (!isTelemetryEnabled()) return; + try { + const email = this.telemetryId; + if (!email || !email.includes("@")) return; + const sharedAnonId = await getOrCreateMem0UserId(); + const anonIds = await readMem0AnonIds(); + if (!anonIds && !sharedAnonId) return; + const candidates = [anonIds?.oss || sharedAnonId, anonIds?.cli].filter( + (id): id is string => !!id && id !== email, + ); + const seen = new Set(); + for (const anonId of candidates) { + if (seen.has(anonId) || (await isMem0Aliased(anonId, email))) continue; + seen.add(anonId); + if (await telemetry.captureIdentify(anonId, email)) { + await markMem0Aliased(anonId, email); + } + } + } catch (error: any) { + console.error("Failed to alias telemetry identity:", error); + } + } + private _captureEvent(methodName: string, args: any[]) { captureClientEvent(methodName, this, { success: true, diff --git a/mem0-ts/src/client/telemetry.ts b/mem0-ts/src/client/telemetry.ts index ab5887820..df3a722d4 100644 --- a/mem0-ts/src/client/telemetry.ts +++ b/mem0-ts/src/client/telemetry.ts @@ -32,8 +32,12 @@ class UnifiedTelemetry implements TelemetryClient { this.host = host; } - async captureEvent(distinctId: string, eventName: string, properties = {}) { - if (!MEM0_TELEMETRY) return; + async captureEvent( + distinctId: string, + eventName: string, + properties = {}, + ): Promise { + if (!MEM0_TELEMETRY) return false; const eventProperties = { client_version: version, @@ -61,9 +65,50 @@ class UnifiedTelemetry implements TelemetryClient { if (!response.ok) { console.error("Telemetry event capture failed:", await response.text()); + return false; } + return true; } catch (error) { console.error("Telemetry event capture failed:", error); + return false; + } + } + + async captureIdentify(anonId: string, email: string): Promise { + if (!MEM0_TELEMETRY) return false; + if (!anonId || !email || anonId === email) return false; + + const payload = { + api_key: this.apiKey, + distinct_id: email, + event: "$identify", + properties: { + $anon_distinct_id: anonId, + client_source: "typescript", + $lib: "posthog-node", + }, + }; + + try { + const response = await fetch(this.host, { + method: "POST", + headers: { + "Content-Type": "application/json", + }, + body: JSON.stringify(payload), + }); + + if (!response.ok) { + console.error( + "Telemetry identify capture failed:", + await response.text(), + ); + return false; + } + return true; + } catch (error) { + console.error("Telemetry identify capture failed:", error); + return false; } } @@ -72,6 +117,10 @@ class UnifiedTelemetry implements TelemetryClient { } } +function isTelemetryEnabled(): boolean { + return MEM0_TELEMETRY; +} + const telemetry = new UnifiedTelemetry(POSTHOG_API_KEY, POSTHOG_HOST); async function captureClientEvent( @@ -101,4 +150,4 @@ async function captureClientEvent( ); } -export { telemetry, captureClientEvent, generateHash }; +export { telemetry, captureClientEvent, generateHash, isTelemetryEnabled }; diff --git a/mem0-ts/src/client/telemetry.types.ts b/mem0-ts/src/client/telemetry.types.ts index 5b307d99d..cca3a58f4 100644 --- a/mem0-ts/src/client/telemetry.types.ts +++ b/mem0-ts/src/client/telemetry.types.ts @@ -3,7 +3,7 @@ export interface TelemetryClient { distinctId: string, eventName: string, properties?: Record, - ): Promise; + ): Promise; shutdown(): Promise; } diff --git a/mem0-ts/src/client/tests/telemetry-aliasing.test.ts b/mem0-ts/src/client/tests/telemetry-aliasing.test.ts new file mode 100644 index 000000000..a0351bbf1 --- /dev/null +++ b/mem0-ts/src/client/tests/telemetry-aliasing.test.ts @@ -0,0 +1,410 @@ +/** + * Tests for PostHog identity stitching in the TS MemoryClient. + * + * Covers $identify firing, idempotency via pair markers, and the node/browser + * gate. Mocks fs and fetch; never touches the real ~/.mem0/config.json. + */ +import * as fs from "fs"; +import * as os from "os"; +import * as path from "path"; +import { MemoryClient } from "../mem0"; +import { telemetry } from "../telemetry"; +import { + getOrCreateMem0UserId, + isMem0Aliased, + markMem0Aliased, + readMem0AnonIds, +} from "../config"; +import { TEST_API_KEY } from "./helpers"; +import { setupMockFetch, installConsoleSuppression } from "./setup"; + +installConsoleSuppression(); + +function setupMockFetchWithPostHog(): jest.Mock { + return setupMockFetch( + new Map([["us.i.posthog.com", { status: 200, body: "ok" }]]), + ); +} + +// ─── config.ts (node-only fs read/write) ────────────────────── + +describe("config.ts — readMem0AnonIds / markMem0Aliased", () => { + let tmpHome: string; + const originalMem0Dir = process.env.MEM0_DIR; + + beforeEach(() => { + tmpHome = fs.mkdtempSync(path.join(os.tmpdir(), "mem0-ts-test-")); + process.env.MEM0_DIR = tmpHome; + }); + + afterEach(() => { + if (fs.existsSync(tmpHome)) { + fs.rmSync(tmpHome, { recursive: true, force: true }); + } + if (originalMem0Dir === undefined) { + delete process.env.MEM0_DIR; + } else { + process.env.MEM0_DIR = originalMem0Dir; + } + }); + + test("returns null when config file does not exist", async () => { + expect(await readMem0AnonIds()).toBeNull(); + }); + + test("reads OSS user_id only", async () => { + fs.writeFileSync( + path.join(tmpHome, "config.json"), + JSON.stringify({ user_id: "oss-uuid" }), + ); + const ids = await readMem0AnonIds(); + expect(ids).toEqual({ + oss: "oss-uuid", + cli: undefined, + aliasedPairs: [], + }); + }); + + test("reads CLI anonymous_id and aliased_pairs", async () => { + fs.writeFileSync( + path.join(tmpHome, "config.json"), + JSON.stringify({ + telemetry: { anonymous_id: "cli-anon", aliased_pairs: ["pair-marker"] }, + }), + ); + const ids = await readMem0AnonIds(); + expect(ids).toEqual({ + oss: undefined, + cli: "cli-anon", + aliasedPairs: ["pair-marker"], + }); + }); + + test("getOrCreateMem0UserId creates and reuses shared SDK user_id", async () => { + const first = await getOrCreateMem0UserId(); + const second = await getOrCreateMem0UserId(); + expect(first).toBeTruthy(); + expect(second).toBe(first); + const written = JSON.parse( + fs.readFileSync(path.join(tmpHome, "config.json"), "utf8"), + ); + expect(written.user_id).toBe(first); + }); + + test("returns null on malformed JSON", async () => { + fs.writeFileSync(path.join(tmpHome, "config.json"), "{not json"); + expect(await readMem0AnonIds()).toBeNull(); + }); + + test("markMem0Aliased preserves other fields", async () => { + fs.writeFileSync( + path.join(tmpHome, "config.json"), + JSON.stringify({ + user_id: "oss-uuid", + telemetry: { anonymous_id: "cli-anon" }, + }), + ); + await markMem0Aliased("oss-uuid", "user@example.com"); + const written = JSON.parse( + fs.readFileSync(path.join(tmpHome, "config.json"), "utf8"), + ); + expect(written.user_id).toBe("oss-uuid"); + expect(written.telemetry.anonymous_id).toBe("cli-anon"); + expect(written.telemetry.aliased_pairs).toHaveLength(1); + expect(await isMem0Aliased("oss-uuid", "user@example.com")).toBe(true); + }); + + test("markMem0Aliased creates telemetry section when missing", async () => { + fs.writeFileSync( + path.join(tmpHome, "config.json"), + JSON.stringify({ user_id: "oss-uuid" }), + ); + await markMem0Aliased("oss-uuid", "user@example.com"); + const written = JSON.parse( + fs.readFileSync(path.join(tmpHome, "config.json"), "utf8"), + ); + expect(written.telemetry.aliased_pairs).toHaveLength(1); + }); + + test("markMem0Aliased tracks each pair independently", async () => { + fs.writeFileSync( + path.join(tmpHome, "config.json"), + JSON.stringify({ user_id: "oss-uuid" }), + ); + await markMem0Aliased("oss-uuid", "user@example.com"); + expect(await isMem0Aliased("oss-uuid", "user@example.com")).toBe(true); + expect(await isMem0Aliased("other-uuid", "user@example.com")).toBe(false); + expect(await isMem0Aliased("oss-uuid", "other@example.com")).toBe(false); + }); + + test("markMem0Aliased does not throw when target dir is unwritable", async () => { + // Point at a path that cannot be written to (a file-as-dir collision). + fs.writeFileSync(path.join(tmpHome, "blocker"), "x"); + process.env.MEM0_DIR = path.join(tmpHome, "blocker"); // file used as dir + await expect( + markMem0Aliased("oss-uuid", "user@example.com"), + ).resolves.toBeUndefined(); + }); +}); + +// ─── telemetry.captureIdentify ─────────────────────────────── + +describe("telemetry.captureIdentify", () => { + test("fires $identify with $anon_distinct_id", async () => { + const fetchMock = jest.fn(async () => ({ + ok: true, + status: 200, + text: async () => "ok", + })) as unknown as typeof fetch; + global.fetch = fetchMock as any; + + await telemetry.captureIdentify("anon-uuid", "user@example.com"); + + expect(fetchMock).toHaveBeenCalledTimes(1); + const [, init] = (fetchMock as jest.Mock).mock.calls[0]; + const payload = JSON.parse(init.body); + expect(payload.event).toBe("$identify"); + expect(payload.distinct_id).toBe("user@example.com"); + expect(payload.properties.$anon_distinct_id).toBe("anon-uuid"); + expect(payload.properties.$process_person_profile).toBeUndefined(); + }); + + test("skips when anon equals email", async () => { + const fetchMock = jest.fn() as unknown as typeof fetch; + global.fetch = fetchMock as any; + await telemetry.captureIdentify("user@example.com", "user@example.com"); + expect(fetchMock).not.toHaveBeenCalled(); + }); + + test("skips when either input is empty", async () => { + const fetchMock = jest.fn() as unknown as typeof fetch; + global.fetch = fetchMock as any; + await telemetry.captureIdentify("", "user@example.com"); + await telemetry.captureIdentify("anon", ""); + expect(fetchMock).not.toHaveBeenCalled(); + }); +}); + +// ─── MemoryClient init aliasing ────────────────────────────── + +describe("MemoryClient — _maybeAliasAnonToEmail", () => { + let tmpHome: string; + const originalMem0Dir = process.env.MEM0_DIR; + + beforeEach(() => { + tmpHome = fs.mkdtempSync(path.join(os.tmpdir(), "mem0-ts-init-")); + process.env.MEM0_DIR = tmpHome; + }); + + afterEach(() => { + if (fs.existsSync(tmpHome)) { + fs.rmSync(tmpHome, { recursive: true, force: true }); + } + if (originalMem0Dir === undefined) { + delete process.env.MEM0_DIR; + } else { + process.env.MEM0_DIR = originalMem0Dir; + } + }); + + // Construct a non-initialised client so we can call _maybeAliasAnonToEmail + // in isolation (the real constructor's _initializeClient also fires it). + function makeStubClient(telemetryId: string): MemoryClient { + const client = Object.create(MemoryClient.prototype) as MemoryClient; + (client as any).apiKey = TEST_API_KEY; + (client as any).host = "https://api.mem0.ai"; + (client as any).telemetryId = telemetryId; + return client; + } + + test("fires $identify on first init and persists pair marker", async () => { + fs.writeFileSync( + path.join(tmpHome, "config.json"), + JSON.stringify({ user_id: "oss-uuid" }), + ); + const fetchMock = setupMockFetchWithPostHog(); + + const client = makeStubClient("test@example.com"); + await (client as any)._maybeAliasAnonToEmail(); + + const identifyCalls = (fetchMock.mock.calls as any[]).filter( + ([, init]: [string, RequestInit]) => { + if (!init?.body) return false; + return JSON.parse(init.body as string).event === "$identify"; + }, + ); + expect(identifyCalls.length).toBe(1); + const body = JSON.parse(identifyCalls[0][1].body); + expect(body.distinct_id).toBe("test@example.com"); + expect(body.properties.$anon_distinct_id).toBe("oss-uuid"); + + const written = JSON.parse( + fs.readFileSync(path.join(tmpHome, "config.json"), "utf8"), + ); + expect(written.telemetry.aliased_pairs).toHaveLength(1); + }); + + test("platform-first init creates shared anon ID and identifies it", async () => { + const fetchMock = setupMockFetchWithPostHog(); + + const client = makeStubClient("test@example.com"); + await (client as any)._maybeAliasAnonToEmail(); + + const written = JSON.parse( + fs.readFileSync(path.join(tmpHome, "config.json"), "utf8"), + ); + expect(written.user_id).toBeTruthy(); + expect(written.telemetry.aliased_pairs).toHaveLength(1); + + const identifyCalls = (fetchMock.mock.calls as any[]).filter( + ([, init]: [string, RequestInit]) => { + if (!init?.body) return false; + return JSON.parse(init.body as string).event === "$identify"; + }, + ); + expect(identifyCalls.length).toBe(1); + const body = JSON.parse(identifyCalls[0][1].body); + expect(body.distinct_id).toBe("test@example.com"); + expect(body.properties.$anon_distinct_id).toBe(written.user_id); + }); + + test("second init does not refire $identify", async () => { + fs.writeFileSync( + path.join(tmpHome, "config.json"), + JSON.stringify({ + user_id: "oss-uuid", + telemetry: {}, + }), + ); + await markMem0Aliased("oss-uuid", "test@example.com"); + const fetchMock = setupMockFetchWithPostHog(); + + const client = makeStubClient("test@example.com"); + await (client as any)._maybeAliasAnonToEmail(); + + const identifyCalls = (fetchMock.mock.calls as any[]).filter( + ([, init]: [string, RequestInit]) => { + if (!init?.body) return false; + return JSON.parse(init.body as string).event === "$identify"; + }, + ); + expect(identifyCalls.length).toBe(0); + }); + + test("fires $identify for both OSS and CLI anon ids", async () => { + fs.writeFileSync( + path.join(tmpHome, "config.json"), + JSON.stringify({ + user_id: "oss-uuid", + telemetry: { anonymous_id: "cli-anon" }, + }), + ); + const fetchMock = setupMockFetchWithPostHog(); + + const client = makeStubClient("test@example.com"); + await (client as any)._maybeAliasAnonToEmail(); + + const identifyCalls = (fetchMock.mock.calls as any[]).filter( + ([, init]: [string, RequestInit]) => { + if (!init?.body) return false; + return JSON.parse(init.body as string).event === "$identify"; + }, + ); + expect(identifyCalls.length).toBe(2); + const anonIds = identifyCalls.map( + (c: [string, RequestInit]) => + JSON.parse(c[1].body as string).properties.$anon_distinct_id, + ); + expect(anonIds).toContain("oss-uuid"); + expect(anonIds).toContain("cli-anon"); + + const written = JSON.parse( + fs.readFileSync(path.join(tmpHome, "config.json"), "utf8"), + ); + expect(written.telemetry.aliased_pairs).toHaveLength(2); + }); + + test("noop when telemetryId is not an email", async () => { + fs.writeFileSync( + path.join(tmpHome, "config.json"), + JSON.stringify({ user_id: "oss-uuid" }), + ); + const fetchMock = setupMockFetch(); + + const client = makeStubClient("not-an-email"); + await (client as any)._maybeAliasAnonToEmail(); + + const identifyCalls = (fetchMock.mock.calls as any[]).filter( + ([, init]: [string, RequestInit]) => { + if (!init?.body) return false; + return JSON.parse(init.body as string).event === "$identify"; + }, + ); + expect(identifyCalls.length).toBe(0); + }); + + test("does not throw when config read fails", async () => { + fs.writeFileSync(path.join(tmpHome, "config.json"), "{not json"); + setupMockFetch(); + + const client = makeStubClient("test@example.com"); + await expect( + (client as any)._maybeAliasAnonToEmail(), + ).resolves.toBeUndefined(); + }); + + test("noop when telemetry disabled — no fs read, no fs write, no events", async () => { + fs.writeFileSync( + path.join(tmpHome, "config.json"), + JSON.stringify({ user_id: "oss-uuid" }), + ); + const fetchMock = setupMockFetch(); + + jest.resetModules(); + const original = process.env.MEM0_TELEMETRY; + process.env.MEM0_TELEMETRY = "false"; + try { + const { MemoryClient: ColdClient } = await import("../mem0"); + const client = Object.create(ColdClient.prototype); + client.apiKey = TEST_API_KEY; + client.host = "https://api.mem0.ai"; + client.telemetryId = "test@example.com"; + await client._maybeAliasAnonToEmail(); + } finally { + if (original === undefined) delete process.env.MEM0_TELEMETRY; + else process.env.MEM0_TELEMETRY = original; + jest.resetModules(); + } + + const identifyCalls = (fetchMock.mock.calls as any[]).filter( + ([, init]: [string, RequestInit]) => { + if (!init?.body) return false; + return JSON.parse(init.body as string).event === "$identify"; + }, + ); + expect(identifyCalls.length).toBe(0); + + const written = JSON.parse( + fs.readFileSync(path.join(tmpHome, "config.json"), "utf8"), + ); + expect(written.telemetry?.aliased_pairs).toBeUndefined(); + }); +}); + +// ─── Browser env path (no process.versions.node) ───────────── + +describe("config.ts in browser-like environment", () => { + test("readMem0AnonIds returns null when not Node", async () => { + const originalProcess = global.process; + // @ts-expect-error force-undefining global to simulate a browser + delete global.process; + try { + jest.resetModules(); + const { readMem0AnonIds: browserRead } = await import("../config"); + expect(await browserRead()).toBeNull(); + } finally { + global.process = originalProcess; + jest.resetModules(); + } + }); +}); diff --git a/mem0-ts/src/oss/src/memory/index.ts b/mem0-ts/src/oss/src/memory/index.ts index f2856c493..e175e71ff 100644 --- a/mem0-ts/src/oss/src/memory/index.ts +++ b/mem0-ts/src/oss/src/memory/index.ts @@ -53,6 +53,7 @@ import { ScoredResult, } from "../utils/scoring"; import { getDefaultVectorStoreDbPath } from "../utils/sqlite"; +import { getOrCreateMem0UserId } from "../../../client/config"; // Entity params that must be passed via filters - check both snake_case and camelCase const ENTITY_PARAMS = [ @@ -466,7 +467,12 @@ export class Memory { this.telemetryId === "anonymous" || this.telemetryId === "anonymous-supabase" ) { - this.telemetryId = await this.vectorStore.getUserId(); + this.telemetryId = + (await getOrCreateMem0UserId()) || + (await this.vectorStore.getUserId()); + try { + await this.vectorStore.setUserId(this.telemetryId); + } catch {} } return this.telemetryId; } catch (error) { diff --git a/mem0/client/main.py b/mem0/client/main.py index 34db1c782..69ff77bd8 100644 --- a/mem0/client/main.py +++ b/mem0/client/main.py @@ -19,8 +19,8 @@ from mem0.client.types import ( from mem0.client.utils import api_error_handler # Exception classes are referenced in docstrings only -from mem0.memory.setup import get_user_id, setup_config -from mem0.memory.telemetry import capture_client_event +from mem0.memory.setup import get_user_id, is_aliased, mark_aliased, read_anon_ids, setup_config +from mem0.memory.telemetry import capture_client_event, client_telemetry logger = logging.getLogger(__name__) @@ -33,6 +33,32 @@ setup_config() ENTITY_PARAMS = frozenset({"user_id", "agent_id", "app_id", "run_id"}) +def _maybe_alias_anon_to_email(user_email): + """Fire $identify per prior anon ID so PostHog merges them into email. + + Idempotent via telemetry.aliased_pairs: only writes markers when + telemetry is actually enabled, so disabling/re-enabling MEM0_TELEMETRY still works. + Best-effort: never raises. + """ + if client_telemetry.posthog is None: + return + if not user_email or "@" not in user_email: + return + try: + anon_ids = read_anon_ids() + seen = set() + for anon_id in (anon_ids.get("oss"), anon_ids.get("cli")): + if not anon_id or anon_id == user_email or anon_id in seen: + continue + seen.add(anon_id) + if is_aliased(anon_id, user_email): + continue + if client_telemetry.capture_identify(anon_id, user_email): + mark_aliased(anon_id, user_email) + except Exception as e: + logger.debug("Failed to alias anon telemetry to %r: %s", user_email, e) + + class MemoryClient: """Client for interacting with the Mem0 API. @@ -108,6 +134,7 @@ class MemoryClient: user_email=self.user_email, ) + _maybe_alias_anon_to_email(self.user_email) capture_client_event("client.init", self, {"sync_type": "sync"}) def _validate_api_key(self): @@ -985,6 +1012,7 @@ class AsyncMemoryClient: user_email=self.user_email, ) + _maybe_alias_anon_to_email(self.user_email) capture_client_event("client.init", self, {"sync_type": "async"}) def _validate_api_key(self): diff --git a/mem0/memory/setup.py b/mem0/memory/setup.py index 3dbd70667..7d98bcae4 100644 --- a/mem0/memory/setup.py +++ b/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): diff --git a/mem0/memory/telemetry.py b/mem0/memory/telemetry.py index f0c46cbb2..0c2c544ad 100644 --- a/mem0/memory/telemetry.py +++ b/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() diff --git a/tests/test_telemetry_aliasing.py b/tests/test_telemetry_aliasing.py new file mode 100644 index 000000000..bd6fcf0ee --- /dev/null +++ b/tests/test_telemetry_aliasing.py @@ -0,0 +1,411 @@ +"""Tests for PostHog identity stitching: anon → email alias on MemoryClient init. + +Covers the four matrix cases (OSS-only, CLI-only, both, already-aliased) plus +failure modes: missing config, malformed JSON, read-only filesystem, +broken posthog client. Telemetry must never raise. +""" + +import importlib +import json +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + + +@pytest.fixture +def tmp_mem0_dir(tmp_path, monkeypatch): + """Point the mem0 setup module at a tempdir for the duration of the test.""" + monkeypatch.setenv("MEM0_DIR", str(tmp_path)) + # Reload setup so module-level mem0_dir picks up the env var. + import mem0.memory.setup as setup_module + + importlib.reload(setup_module) + yield tmp_path + # Restore default state. + monkeypatch.delenv("MEM0_DIR", raising=False) + importlib.reload(setup_module) + + +def _write_config(tmp_path: Path, payload: dict) -> Path: + config_path = tmp_path / "config.json" + config_path.write_text(json.dumps(payload)) + return config_path + + +# ─── setup_config idempotency ──────────────────────────────────────────────── + + +class TestSetupConfigIdempotent: + def test_creates_config_when_missing(self, tmp_mem0_dir): + import mem0.memory.setup as setup_module + + setup_module.setup_config() + config = json.loads((tmp_mem0_dir / "config.json").read_text()) + assert "user_id" in config and config["user_id"] + + def test_backfills_user_id_when_only_telemetry_present(self, tmp_mem0_dir): + import mem0.memory.setup as setup_module + + _write_config( + tmp_mem0_dir, + {"telemetry": {"anonymous_id": "cli-anon-abc123"}}, + ) + setup_module.setup_config() + config = json.loads((tmp_mem0_dir / "config.json").read_text()) + assert config.get("user_id"), "user_id must be backfilled for CLI-first users" + assert config["telemetry"]["anonymous_id"] == "cli-anon-abc123" + + def test_does_not_overwrite_existing_user_id(self, tmp_mem0_dir): + import mem0.memory.setup as setup_module + + _write_config(tmp_mem0_dir, {"user_id": "existing-uuid"}) + setup_module.setup_config() + config = json.loads((tmp_mem0_dir / "config.json").read_text()) + assert config["user_id"] == "existing-uuid" + + def test_handles_malformed_json(self, tmp_mem0_dir): + import mem0.memory.setup as setup_module + + (tmp_mem0_dir / "config.json").write_text("{not json") + setup_module.setup_config() # must not raise + config = json.loads((tmp_mem0_dir / "config.json").read_text()) + assert "user_id" in config + + +# ─── read_anon_ids ─────────────────────────────────────────────────────────── + + +class TestReadAnonIds: + def test_returns_oss_only(self, tmp_mem0_dir): + import mem0.memory.setup as setup_module + + _write_config(tmp_mem0_dir, {"user_id": "oss-uuid"}) + anon = setup_module.read_anon_ids() + assert anon == {"oss": "oss-uuid", "cli": None, "aliased_pairs": []} + + def test_returns_cli_only(self, tmp_mem0_dir): + import mem0.memory.setup as setup_module + + _write_config( + tmp_mem0_dir, + {"telemetry": {"anonymous_id": "cli-anon-123"}}, + ) + anon = setup_module.read_anon_ids() + assert anon == {"oss": None, "cli": "cli-anon-123", "aliased_pairs": []} + + def test_returns_both(self, tmp_mem0_dir): + import mem0.memory.setup as setup_module + + _write_config( + tmp_mem0_dir, + { + "user_id": "oss-uuid", + "telemetry": {"anonymous_id": "cli-anon-123", "aliased_pairs": ["pair-marker"]}, + }, + ) + anon = setup_module.read_anon_ids() + assert anon == { + "oss": "oss-uuid", + "cli": "cli-anon-123", + "aliased_pairs": ["pair-marker"], + } + + def test_no_config_returns_all_none(self, tmp_mem0_dir): + import mem0.memory.setup as setup_module + + anon = setup_module.read_anon_ids() + assert anon == {"oss": None, "cli": None, "aliased_pairs": []} + + def test_malformed_json_does_not_raise(self, tmp_mem0_dir): + import mem0.memory.setup as setup_module + + (tmp_mem0_dir / "config.json").write_text("{not json") + anon = setup_module.read_anon_ids() + assert anon == {"oss": None, "cli": None, "aliased_pairs": []} + + +# ─── mark_aliased ──────────────────────────────────────────────────────────── + + +class TestMarkAliased: + def test_writes_aliased_pair_preserving_other_fields(self, tmp_mem0_dir): + import mem0.memory.setup as setup_module + + _write_config( + tmp_mem0_dir, + { + "user_id": "oss-uuid", + "telemetry": {"anonymous_id": "cli-anon-123"}, + }, + ) + setup_module.mark_aliased("oss-uuid", "user@example.com") + config = json.loads((tmp_mem0_dir / "config.json").read_text()) + assert config["user_id"] == "oss-uuid" + assert config["telemetry"]["anonymous_id"] == "cli-anon-123" + assert len(config["telemetry"]["aliased_pairs"]) == 1 + assert setup_module.is_aliased("oss-uuid", "user@example.com") + + def test_creates_telemetry_section_when_missing(self, tmp_mem0_dir): + import mem0.memory.setup as setup_module + + _write_config(tmp_mem0_dir, {"user_id": "oss-uuid"}) + setup_module.mark_aliased("oss-uuid", "user@example.com") + config = json.loads((tmp_mem0_dir / "config.json").read_text()) + assert len(config["telemetry"]["aliased_pairs"]) == 1 + + def test_tracks_each_pair_independently(self, tmp_mem0_dir): + import mem0.memory.setup as setup_module + + _write_config(tmp_mem0_dir, {"user_id": "oss-uuid"}) + setup_module.mark_aliased("oss-uuid", "user@example.com") + assert setup_module.is_aliased("oss-uuid", "user@example.com") + assert not setup_module.is_aliased("new-uuid", "user@example.com") + assert not setup_module.is_aliased("oss-uuid", "other@example.com") + + +# ─── capture_identify ──────────────────────────────────────────────────────── + + +class TestCaptureIdentify: + def test_fires_identify_with_anon_distinct_id(self): + import mem0.memory.telemetry as telemetry_module + + with patch.object(telemetry_module, "MEM0_TELEMETRY", True): + with patch("mem0.memory.telemetry.Posthog") as mock_posthog_cls: + at = telemetry_module.AnonymousTelemetry() + at.capture_identify("anon-123", "user@example.com") + mock_ph = mock_posthog_cls.return_value + mock_ph.capture.assert_called_once() + _, kwargs = mock_ph.capture.call_args + assert kwargs["distinct_id"] == "user@example.com" + assert kwargs["event"] == "$identify" + assert kwargs["properties"]["$anon_distinct_id"] == "anon-123" + + def test_skips_when_anon_equals_email(self): + import mem0.memory.telemetry as telemetry_module + + with patch.object(telemetry_module, "MEM0_TELEMETRY", True): + with patch("mem0.memory.telemetry.Posthog") as mock_posthog_cls: + at = telemetry_module.AnonymousTelemetry() + at.capture_identify("user@example.com", "user@example.com") + mock_posthog_cls.return_value.capture.assert_not_called() + + def test_skips_when_inputs_empty(self): + import mem0.memory.telemetry as telemetry_module + + with patch.object(telemetry_module, "MEM0_TELEMETRY", True): + with patch("mem0.memory.telemetry.Posthog") as mock_posthog_cls: + at = telemetry_module.AnonymousTelemetry() + at.capture_identify("", "user@example.com") + at.capture_identify("anon-123", "") + mock_posthog_cls.return_value.capture.assert_not_called() + + def test_noop_when_telemetry_disabled(self): + import mem0.memory.telemetry as telemetry_module + + with patch.object(telemetry_module, "MEM0_TELEMETRY", False): + at = telemetry_module.AnonymousTelemetry() + at.capture_identify("anon-123", "user@example.com") # must not raise + assert at.posthog is None + + def test_does_not_raise_on_posthog_error(self): + import mem0.memory.telemetry as telemetry_module + + with patch.object(telemetry_module, "MEM0_TELEMETRY", True): + with patch("mem0.memory.telemetry.Posthog") as mock_posthog_cls: + mock_posthog_cls.return_value.capture.side_effect = RuntimeError("boom") + at = telemetry_module.AnonymousTelemetry() + at.capture_identify("anon-123", "user@example.com") # must not raise + + def test_identify_is_in_lifecycle_events(self): + """$identify must bypass the 90% sampling drop.""" + import mem0.memory.telemetry as telemetry_module + + assert "$identify" in telemetry_module._LIFECYCLE_EVENTS + + +# ─── _maybe_alias_anon_to_email integration ────────────────────────────────── + + +class TestMaybeAliasAnonToEmail: + """Test the alias helper in isolation by mocking out the config readers + and the telemetry client, since module-level setup_config() side effects + make end-to-end fixturing awkward.""" + + def test_fires_identify_for_oss_uuid(self): + from mem0.client import main as client_main + + with ( + patch.object( + client_main, + "read_anon_ids", + return_value={"oss": "oss-uuid", "cli": None, "aliased_pairs": []}, + ), + patch.object(client_main, "is_aliased", return_value=False), + patch.object(client_main, "mark_aliased") as mark, + patch.object(client_main, "client_telemetry") as telemetry, + ): + telemetry.capture_identify.return_value = True + client_main._maybe_alias_anon_to_email("user@example.com") + telemetry.capture_identify.assert_called_once_with("oss-uuid", "user@example.com") + mark.assert_called_once_with("oss-uuid", "user@example.com") + + def test_fires_identify_for_cli_anon(self): + from mem0.client import main as client_main + + with ( + patch.object( + client_main, + "read_anon_ids", + return_value={"oss": None, "cli": "cli-anon-xyz", "aliased_pairs": []}, + ), + patch.object(client_main, "is_aliased", return_value=False), + patch.object(client_main, "mark_aliased"), + patch.object(client_main, "client_telemetry") as telemetry, + ): + telemetry.capture_identify.return_value = True + client_main._maybe_alias_anon_to_email("user@example.com") + telemetry.capture_identify.assert_called_once_with("cli-anon-xyz", "user@example.com") + + def test_fires_identify_for_both_anon_ids(self): + from mem0.client import main as client_main + + with ( + patch.object( + client_main, + "read_anon_ids", + return_value={"oss": "oss-uuid", "cli": "cli-anon", "aliased_pairs": []}, + ), + patch.object(client_main, "is_aliased", return_value=False), + patch.object(client_main, "mark_aliased"), + patch.object(client_main, "client_telemetry") as telemetry, + ): + telemetry.capture_identify.return_value = True + client_main._maybe_alias_anon_to_email("user@example.com") + assert telemetry.capture_identify.call_count == 2 + calls = {c.args for c in telemetry.capture_identify.call_args_list} + assert ("oss-uuid", "user@example.com") in calls + assert ("cli-anon", "user@example.com") in calls + + def test_skips_when_pair_already_aliased(self): + from mem0.client import main as client_main + + with ( + patch.object( + client_main, + "read_anon_ids", + return_value={"oss": "oss-uuid", "cli": None, "aliased_pairs": ["pair-marker"]}, + ), + patch.object(client_main, "is_aliased", return_value=True), + patch.object(client_main, "mark_aliased") as mark, + patch.object(client_main, "client_telemetry") as telemetry, + ): + client_main._maybe_alias_anon_to_email("user@example.com") + telemetry.capture_identify.assert_not_called() + mark.assert_not_called() + + def test_skips_when_email_invalid(self): + from mem0.client import main as client_main + + with patch.object(client_main, "client_telemetry") as telemetry: + client_main._maybe_alias_anon_to_email(None) + client_main._maybe_alias_anon_to_email("") + 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 + + mock_telemetry = MagicMock() + mock_telemetry.capture_identify.side_effect = RuntimeError("boom") + with ( + patch.object( + client_main, + "read_anon_ids", + return_value={"oss": "oss-uuid", "cli": None, "aliased_pairs": []}, + ), + patch.object(client_main, "is_aliased", return_value=False), + patch.object(client_main, "mark_aliased") as mark, + patch.object(client_main, "client_telemetry", mock_telemetry), + ): + client_main._maybe_alias_anon_to_email("user@example.com") # must not raise + mark.assert_not_called() + + def test_skips_anon_id_equal_to_email(self): + """Defensive: if the anon_id somehow already is the email, don't self-alias.""" + from mem0.client import main as client_main + + with ( + patch.object( + client_main, + "read_anon_ids", + return_value={"oss": "user@example.com", "cli": None, "aliased_pairs": []}, + ), + patch.object(client_main, "is_aliased", return_value=False), + patch.object(client_main, "mark_aliased"), + patch.object(client_main, "client_telemetry") as telemetry, + ): + client_main._maybe_alias_anon_to_email("user@example.com") + telemetry.capture_identify.assert_not_called() + + def test_does_not_raise_on_read_failure(self): + """If read_anon_ids itself raises (e.g. IO error), helper must swallow it.""" + from mem0.client import main as client_main + + with ( + patch.object(client_main, "read_anon_ids", side_effect=OSError("fs broken")), + patch.object(client_main, "client_telemetry") as telemetry, + ): + client_main._maybe_alias_anon_to_email("user@example.com") # must not raise + telemetry.capture_identify.assert_not_called() + + +# ─── End-to-end idempotency through real config ────────────────────────────── + + +class TestEndToEndIdempotency: + """Verify the real config flow: two consecutive _maybe_alias_anon_to_email + calls fire $identify exactly once thanks to the persisted pair marker.""" + + def test_second_call_is_noop_after_pair_marker_persisted(self, tmp_mem0_dir): + # Pre-populate config with an OSS user_id only. + _write_config(tmp_mem0_dir, {"user_id": "oss-uuid"}) + # Reload setup so it uses the tempdir, then reload client.main so it + # picks up the freshly-loaded read_anon_ids/mark_aliased bindings. + import mem0.memory.setup as setup_module + + importlib.reload(setup_module) + from mem0.client import main as client_main + + importlib.reload(client_main) + + with patch.object(client_main, "client_telemetry") as telemetry: + telemetry.capture_identify.return_value = True + client_main._maybe_alias_anon_to_email("user@example.com") + first_call_count = telemetry.capture_identify.call_count + assert first_call_count >= 1 + + # Second call should hit the aliased_pairs short-circuit. + client_main._maybe_alias_anon_to_email("user@example.com") + assert telemetry.capture_identify.call_count == first_call_count + + config = json.loads((tmp_mem0_dir / "config.json").read_text()) + assert len(config["telemetry"]["aliased_pairs"]) == 1