fix: stitch oss and platform telemetry identities python sdk and typescript sdk
This commit is contained in:
@@ -1,9 +1,9 @@
|
||||
/**
|
||||
* 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 reads
|
||||
* those on init to fire $identify and merge them into the email identity.
|
||||
* 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.
|
||||
*/
|
||||
@@ -11,30 +11,34 @@
|
||||
export interface Mem0AnonIds {
|
||||
oss?: string;
|
||||
cli?: string;
|
||||
aliasedTo?: 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] = await Promise.all([
|
||||
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 {
|
||||
@@ -52,6 +56,45 @@ function loadConfig(node: NodeFs): Record<string, any> | 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;
|
||||
@@ -67,26 +110,55 @@ export async function readMem0AnonIds(): Promise<Mem0AnonIds | null> {
|
||||
typeof telemetry.anonymous_id === "string"
|
||||
? telemetry.anonymous_id
|
||||
: undefined,
|
||||
aliasedTo:
|
||||
typeof telemetry.aliased_to === "string"
|
||||
? telemetry.aliased_to
|
||||
: undefined,
|
||||
aliasedPairs: Array.isArray(telemetry.aliased_pairs)
|
||||
? telemetry.aliased_pairs.filter(
|
||||
(item: unknown) => typeof item === "string",
|
||||
)
|
||||
: [],
|
||||
};
|
||||
}
|
||||
|
||||
export async function markMem0Aliased(email: string): Promise<void> {
|
||||
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 {
|
||||
node.fs.mkdirSync(node.path.dirname(node.configPath), { recursive: true });
|
||||
const config = loadConfig(node) ?? {};
|
||||
const telemetry =
|
||||
config.telemetry && typeof config.telemetry === "object"
|
||||
? config.telemetry
|
||||
: {};
|
||||
telemetry.aliased_to = email;
|
||||
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;
|
||||
node.fs.writeFileSync(node.configPath, JSON.stringify(config, null, 4));
|
||||
writeConfig(node, config);
|
||||
} catch {
|
||||
// Best-effort: read-only filesystems and unwritable paths just skip.
|
||||
}
|
||||
|
||||
@@ -26,7 +26,12 @@ import {
|
||||
isTelemetryEnabled,
|
||||
telemetry,
|
||||
} from "./telemetry";
|
||||
import { readMem0AnonIds, markMem0Aliased } from "./config";
|
||||
import {
|
||||
getOrCreateMem0UserId,
|
||||
isMem0Aliased,
|
||||
markMem0Aliased,
|
||||
readMem0AnonIds,
|
||||
} from "./config";
|
||||
import { camelToSnake, camelToSnakeKeys, snakeToCamelKeys } from "./utils";
|
||||
import { createExceptionFromResponse, MemoryError } from "../common/exceptions";
|
||||
|
||||
@@ -145,16 +150,20 @@ export default class MemoryClient {
|
||||
try {
|
||||
const email = this.telemetryId;
|
||||
if (!email || !email.includes("@")) return;
|
||||
const sharedAnonId = await getOrCreateMem0UserId();
|
||||
const anonIds = await readMem0AnonIds();
|
||||
if (!anonIds) return;
|
||||
if (anonIds.aliasedTo === email) return;
|
||||
const candidates = [anonIds.oss, anonIds.cli].filter(
|
||||
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) {
|
||||
await telemetry.captureIdentify(anonId, email);
|
||||
if (seen.has(anonId) || (await isMem0Aliased(anonId, email))) continue;
|
||||
seen.add(anonId);
|
||||
if (await telemetry.captureIdentify(anonId, email)) {
|
||||
await markMem0Aliased(anonId, email);
|
||||
}
|
||||
}
|
||||
await markMem0Aliased(email);
|
||||
} catch (error: any) {
|
||||
console.error("Failed to alias telemetry identity:", 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,
|
||||
@@ -61,19 +65,51 @@ 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) {
|
||||
if (!MEM0_TELEMETRY) return;
|
||||
if (!anonId || !email || anonId === email) return;
|
||||
await this.captureEvent(email, "$identify", {
|
||||
$anon_distinct_id: anonId,
|
||||
client_source: "typescript",
|
||||
});
|
||||
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.error(
|
||||
"Telemetry identify capture failed:",
|
||||
await response.text(),
|
||||
);
|
||||
return false;
|
||||
}
|
||||
return true;
|
||||
} catch (error) {
|
||||
console.error("Telemetry identify capture failed:", error);
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
async shutdown() {
|
||||
|
||||
@@ -3,7 +3,7 @@ export interface TelemetryClient {
|
||||
distinctId: string,
|
||||
eventName: string,
|
||||
properties?: Record<string, any>,
|
||||
): Promise<void>;
|
||||
): Promise<boolean>;
|
||||
shutdown(): Promise<void>;
|
||||
}
|
||||
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
/**
|
||||
* Tests for PostHog identity stitching in the TS MemoryClient.
|
||||
*
|
||||
* Covers $identify firing, idempotency via aliased_to, and the node/browser
|
||||
* 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";
|
||||
@@ -9,12 +9,23 @@ import * as os from "os";
|
||||
import * as path from "path";
|
||||
import { MemoryClient } from "../mem0";
|
||||
import { telemetry } from "../telemetry";
|
||||
import { readMem0AnonIds, markMem0Aliased } from "../config";
|
||||
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", () => {
|
||||
@@ -50,25 +61,36 @@ describe("config.ts — readMem0AnonIds / markMem0Aliased", () => {
|
||||
expect(ids).toEqual({
|
||||
oss: "oss-uuid",
|
||||
cli: undefined,
|
||||
aliasedTo: undefined,
|
||||
aliasedPairs: [],
|
||||
});
|
||||
});
|
||||
|
||||
test("reads CLI anonymous_id and aliased_to", async () => {
|
||||
test("reads CLI anonymous_id and aliased_pairs", async () => {
|
||||
fs.writeFileSync(
|
||||
path.join(tmpHome, "config.json"),
|
||||
JSON.stringify({
|
||||
telemetry: { anonymous_id: "cli-anon", aliased_to: "u@x.com" },
|
||||
telemetry: { anonymous_id: "cli-anon", aliased_pairs: ["pair-marker"] },
|
||||
}),
|
||||
);
|
||||
const ids = await readMem0AnonIds();
|
||||
expect(ids).toEqual({
|
||||
oss: undefined,
|
||||
cli: "cli-anon",
|
||||
aliasedTo: "u@x.com",
|
||||
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();
|
||||
@@ -82,13 +104,14 @@ describe("config.ts — readMem0AnonIds / markMem0Aliased", () => {
|
||||
telemetry: { anonymous_id: "cli-anon" },
|
||||
}),
|
||||
);
|
||||
await markMem0Aliased("user@example.com");
|
||||
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_to).toBe("user@example.com");
|
||||
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 () => {
|
||||
@@ -96,18 +119,31 @@ describe("config.ts — readMem0AnonIds / markMem0Aliased", () => {
|
||||
path.join(tmpHome, "config.json"),
|
||||
JSON.stringify({ user_id: "oss-uuid" }),
|
||||
);
|
||||
await markMem0Aliased("user@example.com");
|
||||
await markMem0Aliased("oss-uuid", "user@example.com");
|
||||
const written = JSON.parse(
|
||||
fs.readFileSync(path.join(tmpHome, "config.json"), "utf8"),
|
||||
);
|
||||
expect(written.telemetry.aliased_to).toBe("user@example.com");
|
||||
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("user@example.com")).resolves.toBeUndefined();
|
||||
await expect(
|
||||
markMem0Aliased("oss-uuid", "user@example.com"),
|
||||
).resolves.toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
@@ -130,6 +166,7 @@ describe("telemetry.captureIdentify", () => {
|
||||
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 () => {
|
||||
@@ -180,12 +217,12 @@ describe("MemoryClient — _maybeAliasAnonToEmail", () => {
|
||||
return client;
|
||||
}
|
||||
|
||||
test("fires $identify on first init and persists aliased_to", async () => {
|
||||
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 = setupMockFetch();
|
||||
const fetchMock = setupMockFetchWithPostHog();
|
||||
|
||||
const client = makeStubClient("test@example.com");
|
||||
await (client as any)._maybeAliasAnonToEmail();
|
||||
@@ -204,7 +241,31 @@ describe("MemoryClient — _maybeAliasAnonToEmail", () => {
|
||||
const written = JSON.parse(
|
||||
fs.readFileSync(path.join(tmpHome, "config.json"), "utf8"),
|
||||
);
|
||||
expect(written.telemetry.aliased_to).toBe("test@example.com");
|
||||
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 () => {
|
||||
@@ -212,10 +273,11 @@ describe("MemoryClient — _maybeAliasAnonToEmail", () => {
|
||||
path.join(tmpHome, "config.json"),
|
||||
JSON.stringify({
|
||||
user_id: "oss-uuid",
|
||||
telemetry: { aliased_to: "test@example.com" },
|
||||
telemetry: {},
|
||||
}),
|
||||
);
|
||||
const fetchMock = setupMockFetch();
|
||||
await markMem0Aliased("oss-uuid", "test@example.com");
|
||||
const fetchMock = setupMockFetchWithPostHog();
|
||||
|
||||
const client = makeStubClient("test@example.com");
|
||||
await (client as any)._maybeAliasAnonToEmail();
|
||||
@@ -237,7 +299,7 @@ describe("MemoryClient — _maybeAliasAnonToEmail", () => {
|
||||
telemetry: { anonymous_id: "cli-anon" },
|
||||
}),
|
||||
);
|
||||
const fetchMock = setupMockFetch();
|
||||
const fetchMock = setupMockFetchWithPostHog();
|
||||
|
||||
const client = makeStubClient("test@example.com");
|
||||
await (client as any)._maybeAliasAnonToEmail();
|
||||
@@ -255,6 +317,11 @@ describe("MemoryClient — _maybeAliasAnonToEmail", () => {
|
||||
);
|
||||
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 () => {
|
||||
@@ -320,7 +387,7 @@ describe("MemoryClient — _maybeAliasAnonToEmail", () => {
|
||||
const written = JSON.parse(
|
||||
fs.readFileSync(path.join(tmpHome, "config.json"), "utf8"),
|
||||
);
|
||||
expect(written.telemetry?.aliased_to).toBeUndefined();
|
||||
expect(written.telemetry?.aliased_pairs).toBeUndefined();
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
@@ -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) {
|
||||
|
||||
Reference in New Issue
Block a user