Compare commits
9 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 88e9f3973e | |||
| 4176293cc9 | |||
| 64487cbf31 | |||
| d97b7e9131 | |||
| c5bf46c44c | |||
| ec5baaba20 | |||
| 12bb2588f0 | |||
| 8a9915d3fe | |||
| 789122bbea |
@@ -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<NodeFs | null> {
|
||||
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<string, any> | 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<string, any>): 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<string | null> {
|
||||
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<Mem0AnonIds | null> {
|
||||
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<boolean> {
|
||||
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<void> {
|
||||
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.
|
||||
}
|
||||
}
|
||||
@@ -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,10 +129,12 @@ export default class MemoryClient {
|
||||
this.telemetryId = generateHash(this.apiKey);
|
||||
}
|
||||
|
||||
await this._maybeAliasAnonToEmail();
|
||||
|
||||
captureClientEvent("init", this, {
|
||||
client_type: "MemoryClient",
|
||||
}).catch((error: any) => {
|
||||
console.error("Failed to capture event:", error);
|
||||
console.debug("Failed to capture event:", error);
|
||||
});
|
||||
} catch (error: any) {
|
||||
console.error("Failed to initialize client:", error);
|
||||
@@ -132,13 +145,37 @@ export default class MemoryClient {
|
||||
}
|
||||
}
|
||||
|
||||
private async _maybeAliasAnonToEmail(): Promise<void> {
|
||||
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<string>();
|
||||
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.debug("Failed to alias telemetry identity:", error);
|
||||
}
|
||||
}
|
||||
|
||||
private _captureEvent(methodName: string, args: any[]) {
|
||||
captureClientEvent(methodName, this, {
|
||||
success: true,
|
||||
args_count: args.length,
|
||||
keys: args.length > 0 ? args[0] : [],
|
||||
}).catch((error: any) => {
|
||||
console.error("Failed to capture event:", error);
|
||||
console.debug("Failed to capture event:", error);
|
||||
});
|
||||
}
|
||||
|
||||
|
||||
@@ -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<boolean> {
|
||||
if (!MEM0_TELEMETRY) return false;
|
||||
|
||||
const eventProperties = {
|
||||
client_version: version,
|
||||
@@ -60,10 +64,52 @@ class UnifiedTelemetry implements TelemetryClient {
|
||||
});
|
||||
|
||||
if (!response.ok) {
|
||||
console.error("Telemetry event capture failed:", await response.text());
|
||||
console.debug("Telemetry event capture failed:", response.status);
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
} catch (error) {
|
||||
console.error("Telemetry event capture failed:", error);
|
||||
console.debug("Telemetry event capture failed:", error);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
async captureIdentify(anonId: string, email: string): Promise<boolean> {
|
||||
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.debug("Telemetry identify capture failed:", response.status);
|
||||
}
|
||||
// Any HTTP response (2xx / 4xx / 5xx) marks the pair aliased.
|
||||
// PostHog person-merge is idempotent at the person level, so a
|
||||
// duplicate $identify on a recovered server is harmless. Only a
|
||||
// genuine network failure (no response) leaves the marker absent
|
||||
// so the next init can retry.
|
||||
return true;
|
||||
} catch (error) {
|
||||
console.debug("Telemetry identify capture failed:", error);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -72,6 +118,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 +151,4 @@ async function captureClientEvent(
|
||||
);
|
||||
}
|
||||
|
||||
export { telemetry, captureClientEvent, generateHash };
|
||||
export { telemetry, captureClientEvent, generateHash, isTelemetryEnabled };
|
||||
|
||||
@@ -3,7 +3,7 @@ export interface TelemetryClient {
|
||||
distinctId: string,
|
||||
eventName: string,
|
||||
properties?: Record<string, any>,
|
||||
): Promise<void>;
|
||||
): Promise<boolean>;
|
||||
shutdown(): Promise<void>;
|
||||
}
|
||||
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
});
|
||||
});
|
||||
@@ -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,18 @@ export class Memory {
|
||||
this.telemetryId === "anonymous" ||
|
||||
this.telemetryId === "anonymous-supabase"
|
||||
) {
|
||||
this.telemetryId = await this.vectorStore.getUserId();
|
||||
const existing = await this.vectorStore.getUserId();
|
||||
if (existing && !existing.startsWith("anonymous")) {
|
||||
// Preserve a real vector-store UUID so prior PostHog events stay
|
||||
// tied to this user's identity. Cross-surface stitching with
|
||||
// CLI/Python OSS for these existing users is deferred.
|
||||
this.telemetryId = existing;
|
||||
} else {
|
||||
this.telemetryId = (await getOrCreateMem0UserId()) || existing;
|
||||
try {
|
||||
await this.vectorStore.setUserId(this.telemetryId);
|
||||
} catch {}
|
||||
}
|
||||
}
|
||||
return this.telemetryId;
|
||||
} catch (error) {
|
||||
|
||||
+30
-2
@@ -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: %s", 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):
|
||||
|
||||
+102
-15
@@ -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):
|
||||
|
||||
@@ -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,32 @@ 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 synchronously so the bool return reflects delivery.
|
||||
|
||||
Returns True only when the SDK returned a non-None UUID. False
|
||||
when telemetry is disabled, when the sync client cannot be
|
||||
constructed, when the SDK returns None (e.g. @no_throw caught
|
||||
an HTTP error), or when capture() raises.
|
||||
"""
|
||||
if self.posthog is None:
|
||||
return False
|
||||
if not anon_id or not email or anon_id == email:
|
||||
return False
|
||||
sync_client = _get_sync_posthog()
|
||||
if sync_client is None:
|
||||
return False
|
||||
try:
|
||||
result = sync_client.capture(
|
||||
distinct_id=email,
|
||||
event="$identify",
|
||||
properties={"$anon_distinct_id": anon_id, "client_source": "python"},
|
||||
)
|
||||
return result is not None
|
||||
except Exception as e:
|
||||
_logger.debug("Failed to capture $identify: %s", e)
|
||||
return False
|
||||
|
||||
def close(self):
|
||||
if self.posthog is not None:
|
||||
self.posthog.shutdown()
|
||||
@@ -159,6 +186,41 @@ def _shutdown_oss_telemetry():
|
||||
_oss_telemetry_instance = None
|
||||
|
||||
|
||||
# ─── Sync PostHog client for $identify only ──────────────────────────────
|
||||
# $identify must give an honest delivery signal so we only persist an alias
|
||||
# marker when PostHog actually accepted the event. The default async client
|
||||
# enqueues on a background thread, so capture() returns a UUID immediately
|
||||
# regardless of HTTP outcome. sync_mode=True bypasses the queue and either
|
||||
# raises (posthog<7) or returns None (posthog>=7 via @no_throw) on failure.
|
||||
_sync_posthog_instance = None
|
||||
_sync_posthog_lock = threading.Lock()
|
||||
|
||||
|
||||
def _get_sync_posthog():
|
||||
"""Lazy-init a sync-mode PostHog client used only for $identify.
|
||||
|
||||
Returns None if telemetry is disabled or the constructor raises.
|
||||
"""
|
||||
global _sync_posthog_instance
|
||||
if not MEM0_TELEMETRY:
|
||||
return None
|
||||
if _sync_posthog_instance is not None:
|
||||
return _sync_posthog_instance
|
||||
with _sync_posthog_lock:
|
||||
if _sync_posthog_instance is not None:
|
||||
return _sync_posthog_instance
|
||||
try:
|
||||
_sync_posthog_instance = Posthog(
|
||||
project_api_key=PROJECT_API_KEY,
|
||||
host=HOST,
|
||||
sync_mode=True,
|
||||
)
|
||||
except Exception as e:
|
||||
_logger.debug("Failed to init sync PostHog client: %s", e)
|
||||
_sync_posthog_instance = None
|
||||
return _sync_posthog_instance
|
||||
|
||||
|
||||
# Module-level client telemetry singleton (used by capture_client_event).
|
||||
# No before_send — hosted MemoryClient traffic must never be sampled.
|
||||
client_telemetry = AnonymousTelemetry()
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user