fix(telemetry): stitch PostHog identities on MemoryClient init
The Python sync/async MemoryClient and the TS MemoryClient were silently switching their PostHog distinct_id from an anonymous on-disk UUID to the user's email post-ping, without sending the merge instruction. PostHog ended up tracking each user as 2-3 disconnected personas (OSS UUID, CLI anon UUID, email), and only ~4 $create_alias events fired across ~300K users. The Python CLI does this correctly today; this change ports the same pattern to MemoryClient. - mem0/memory/setup.py: make setup_config() idempotent so it backfills a top-level user_id for CLI-first users (the CLI writes telemetry.anonymous_id, not user_id, which left OSS telemetry silently dropped). Add read_anon_ids() and mark_aliased() helpers. - mem0/memory/telemetry.py: add AnonymousTelemetry.capture_identify() that posts a PostHog $identify event with $anon_distinct_id. Add "$identify" to _LIFECYCLE_EVENTS so it bypasses the 90% sampling drop. - mem0/client/main.py: at the end of __init__ for both MemoryClient and AsyncMemoryClient, fire $identify for each unaliased anon ID and persist telemetry.aliased_to so the alias only happens once per (anon_id, email) pair, ever. All telemetry calls swallow exceptions. - mem0-ts/src/client/telemetry.ts: mirror with captureIdentify(). - mem0-ts/src/client/config.ts (new): node-only readMem0AnonIds() and markMem0Aliased() using async dynamic imports of fs/path/os, gated on process.versions.node so browsers no-op cleanly. - mem0-ts/src/client/mem0.ts: call _maybeAliasAnonToEmail() inside _initializeClient after ping resolves the email. Tests cover the matrix (OSS-only, CLI-only, both, already-aliased) plus failure modes (malformed JSON, read-only fs, posthog errors, browser env). Telemetry never raises.
This commit is contained in:
gabrielstein-mem0
@@ -0,0 +1,118 @@
|
||||
/**
|
||||
* Best-effort read/write of ~/.mem0/config.json from the TS SDK.
|
||||
*
|
||||
* Used to stitch PostHog identities: the OSS Python SDK and the Python CLI
|
||||
* each persist an anonymous distinct_id here, and the TS MemoryClient needs
|
||||
* to read those on init so it can fire $identify and merge them into the
|
||||
* email-based identity.
|
||||
*
|
||||
* Node-only. In browsers (or any environment without `process.versions.node`)
|
||||
* every function returns null/no-ops without attempting `fs` access.
|
||||
*/
|
||||
|
||||
export interface Mem0AnonIds {
|
||||
oss?: string;
|
||||
cli?: string;
|
||||
aliasedTo?: string;
|
||||
}
|
||||
|
||||
function isNode(): boolean {
|
||||
try {
|
||||
return (
|
||||
typeof process !== "undefined" &&
|
||||
!!process.versions &&
|
||||
!!process.versions.node
|
||||
);
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
async function loadNodeModules(): Promise<{
|
||||
fs: typeof import("fs");
|
||||
path: typeof import("path");
|
||||
os: typeof import("os");
|
||||
} | null> {
|
||||
if (!isNode()) return null;
|
||||
try {
|
||||
const [fs, path, os] = await Promise.all([
|
||||
import("fs"),
|
||||
import("path"),
|
||||
import("os"),
|
||||
]);
|
||||
return {
|
||||
fs: fs.default ?? fs,
|
||||
path: path.default ?? path,
|
||||
os: os.default ?? os,
|
||||
};
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
async function configPath(): Promise<{
|
||||
path: string;
|
||||
modules: NonNullable<Awaited<ReturnType<typeof loadNodeModules>>>;
|
||||
} | null> {
|
||||
const modules = await loadNodeModules();
|
||||
if (!modules) return null;
|
||||
try {
|
||||
const dir =
|
||||
process.env.MEM0_DIR || modules.path.join(modules.os.homedir(), ".mem0");
|
||||
return { path: modules.path.join(dir, "config.json"), modules };
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
async function loadRawConfig(): Promise<Record<string, unknown> | null> {
|
||||
const resolved = await configPath();
|
||||
if (!resolved) return null;
|
||||
try {
|
||||
if (!resolved.modules.fs.existsSync(resolved.path)) return null;
|
||||
const raw = resolved.modules.fs.readFileSync(resolved.path, "utf8");
|
||||
const parsed = JSON.parse(raw);
|
||||
return parsed && typeof parsed === "object" ? parsed : null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
export async function readMem0AnonIds(): Promise<Mem0AnonIds | null> {
|
||||
const config = await loadRawConfig();
|
||||
if (!config) return null;
|
||||
const telemetry =
|
||||
config.telemetry && typeof config.telemetry === "object"
|
||||
? (config.telemetry as Record<string, unknown>)
|
||||
: {};
|
||||
const oss = typeof config.user_id === "string" ? config.user_id : undefined;
|
||||
const cli =
|
||||
typeof telemetry.anonymous_id === "string"
|
||||
? telemetry.anonymous_id
|
||||
: undefined;
|
||||
const aliasedTo =
|
||||
typeof telemetry.aliased_to === "string" ? telemetry.aliased_to : undefined;
|
||||
return { oss, cli, aliasedTo };
|
||||
}
|
||||
|
||||
export async function markMem0Aliased(email: string): Promise<void> {
|
||||
const resolved = await configPath();
|
||||
if (!resolved) return;
|
||||
try {
|
||||
const dirname = resolved.modules.path.dirname(resolved.path);
|
||||
resolved.modules.fs.mkdirSync(dirname, { recursive: true });
|
||||
const config = (await loadRawConfig()) ?? {};
|
||||
const telemetry =
|
||||
config.telemetry && typeof config.telemetry === "object"
|
||||
? (config.telemetry as Record<string, unknown>)
|
||||
: {};
|
||||
telemetry.aliased_to = email;
|
||||
config.telemetry = telemetry;
|
||||
resolved.modules.fs.writeFileSync(
|
||||
resolved.path,
|
||||
JSON.stringify(config, null, 4),
|
||||
);
|
||||
} catch {
|
||||
// Read-only filesystem (Lambda, container) — alias is best-effort.
|
||||
}
|
||||
}
|
||||
@@ -20,7 +20,8 @@ import {
|
||||
CreateMemoryExportPayload,
|
||||
GetMemoryExportPayload,
|
||||
} from "./mem0.types";
|
||||
import { captureClientEvent, generateHash } from "./telemetry";
|
||||
import { captureClientEvent, generateHash, telemetry } from "./telemetry";
|
||||
import { readMem0AnonIds, markMem0Aliased } from "./config";
|
||||
import { camelToSnake, camelToSnakeKeys, snakeToCamelKeys } from "./utils";
|
||||
import { createExceptionFromResponse, MemoryError } from "../common/exceptions";
|
||||
|
||||
@@ -118,6 +119,8 @@ export default class MemoryClient {
|
||||
this.telemetryId = generateHash(this.apiKey);
|
||||
}
|
||||
|
||||
await this._maybeAliasAnonToEmail();
|
||||
|
||||
captureClientEvent("init", this, {
|
||||
client_type: "MemoryClient",
|
||||
}).catch((error: any) => {
|
||||
@@ -132,6 +135,25 @@ export default class MemoryClient {
|
||||
}
|
||||
}
|
||||
|
||||
private async _maybeAliasAnonToEmail(): Promise<void> {
|
||||
try {
|
||||
const email = this.telemetryId;
|
||||
if (!email || !email.includes("@")) return;
|
||||
const anonIds = await readMem0AnonIds();
|
||||
if (!anonIds) return;
|
||||
if (anonIds.aliasedTo === email) return;
|
||||
const candidates = [anonIds.oss, anonIds.cli].filter(
|
||||
(id): id is string => !!id && id !== email,
|
||||
);
|
||||
for (const anonId of candidates) {
|
||||
await telemetry.captureIdentify(anonId, email);
|
||||
}
|
||||
await markMem0Aliased(email);
|
||||
} catch (error: any) {
|
||||
console.error("Failed to alias telemetry identity:", error);
|
||||
}
|
||||
}
|
||||
|
||||
private _captureEvent(methodName: string, args: any[]) {
|
||||
captureClientEvent(methodName, this, {
|
||||
success: true,
|
||||
|
||||
@@ -67,6 +67,15 @@ class UnifiedTelemetry implements TelemetryClient {
|
||||
}
|
||||
}
|
||||
|
||||
async captureIdentify(anonId: string, email: string) {
|
||||
if (!MEM0_TELEMETRY) return;
|
||||
if (!anonId || !email || anonId === email) return;
|
||||
await this.captureEvent(email, "$identify", {
|
||||
$anon_distinct_id: anonId,
|
||||
client_source: "typescript",
|
||||
});
|
||||
}
|
||||
|
||||
async shutdown() {
|
||||
// No shutdown needed for direct API calls
|
||||
}
|
||||
|
||||
@@ -0,0 +1,323 @@
|
||||
/**
|
||||
* Tests for PostHog identity stitching in the TS MemoryClient.
|
||||
*
|
||||
* Covers $identify firing, idempotency via aliased_to, 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 { readMem0AnonIds, markMem0Aliased, Mem0AnonIds } from "../config";
|
||||
import { createMockFetch, TEST_API_KEY } from "./helpers";
|
||||
import { setupMockFetch, installConsoleSuppression } from "./setup";
|
||||
|
||||
installConsoleSuppression();
|
||||
|
||||
// ─── 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,
|
||||
aliasedTo: undefined,
|
||||
});
|
||||
});
|
||||
|
||||
test("reads CLI anonymous_id and aliased_to", async () => {
|
||||
fs.writeFileSync(
|
||||
path.join(tmpHome, "config.json"),
|
||||
JSON.stringify({
|
||||
telemetry: { anonymous_id: "cli-anon", aliased_to: "u@x.com" },
|
||||
}),
|
||||
);
|
||||
const ids = await readMem0AnonIds();
|
||||
expect(ids).toEqual({
|
||||
oss: undefined,
|
||||
cli: "cli-anon",
|
||||
aliasedTo: "u@x.com",
|
||||
});
|
||||
});
|
||||
|
||||
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("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_to).toBe("user@example.com");
|
||||
});
|
||||
|
||||
test("markMem0Aliased creates telemetry section when missing", async () => {
|
||||
fs.writeFileSync(
|
||||
path.join(tmpHome, "config.json"),
|
||||
JSON.stringify({ user_id: "oss-uuid" }),
|
||||
);
|
||||
await markMem0Aliased("user@example.com");
|
||||
const written = JSON.parse(
|
||||
fs.readFileSync(path.join(tmpHome, "config.json"), "utf8"),
|
||||
);
|
||||
expect(written.telemetry.aliased_to).toBe("user@example.com");
|
||||
});
|
||||
|
||||
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("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");
|
||||
});
|
||||
|
||||
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 aliased_to", async () => {
|
||||
fs.writeFileSync(
|
||||
path.join(tmpHome, "config.json"),
|
||||
JSON.stringify({ user_id: "oss-uuid" }),
|
||||
);
|
||||
const fetchMock = setupMockFetch();
|
||||
|
||||
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_to).toBe("test@example.com");
|
||||
});
|
||||
|
||||
test("second init does not refire $identify", async () => {
|
||||
fs.writeFileSync(
|
||||
path.join(tmpHome, "config.json"),
|
||||
JSON.stringify({
|
||||
user_id: "oss-uuid",
|
||||
telemetry: { aliased_to: "test@example.com" },
|
||||
}),
|
||||
);
|
||||
const fetchMock = setupMockFetch();
|
||||
|
||||
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 = setupMockFetch();
|
||||
|
||||
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");
|
||||
});
|
||||
|
||||
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();
|
||||
});
|
||||
});
|
||||
|
||||
// ─── Type smoke test ──────────────────────────────────────────
|
||||
|
||||
describe("Mem0AnonIds shape", () => {
|
||||
test("interface accepts partial fields", () => {
|
||||
const a: Mem0AnonIds = { oss: "x" };
|
||||
const b: Mem0AnonIds = { cli: "y", aliasedTo: "u@x.com" };
|
||||
expect(a.oss).toBe("x");
|
||||
expect(b.aliasedTo).toBe("u@x.com");
|
||||
});
|
||||
});
|
||||
|
||||
// ─── 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;
|
||||
// Simulate a browser: no `process` global.
|
||||
// @ts-expect-error force-undefining global
|
||||
delete global.process;
|
||||
try {
|
||||
// Re-import to pick up the missing global.
|
||||
jest.resetModules();
|
||||
const { readMem0AnonIds: browserRead } = await import("../config");
|
||||
expect(await browserRead()).toBeNull();
|
||||
} finally {
|
||||
global.process = originalProcess;
|
||||
jest.resetModules();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
// ─── Suppress unused-import warnings ──────────────────────────
|
||||
// Some helpers above are intentionally re-exported for clarity in tests.
|
||||
void createMockFetch;
|
||||
Reference in New Issue
Block a user