feat(deepseek-plugin,mem0-strands): add usage telemetry (#7110)

integrations/deepseek-plugin/README.md

@@ -55,6 +55,8 @@ For a Mem0 Platform on-prem or dedicated deployment, point `config.host` at that
 
 Writes are tagged `source="DEEPSEEK_HARNESS"` so Mem0's backend can attribute usage to this integration. For it to surface by name (rather than bucketing into `OTHERS`), `DEEPSEEK_HARNESS` must be present in the backend's `KNOWN_EVENT_SOURCES` allowlist, a one-line platform change matching the existing `ZAPIER` / `STRANDS` sources.
 
+The plugin also sends anonymous usage events (which tool ran, duration, result counts, coarse failure kind) so Mem0 can tell how the plugin is used and where it breaks. Queries, memory text, and entity ids are never sent. Turn it off with `MEM0_TELEMETRY=false`.
+
 ## Status
 
 Developer preview. Tracks the DeepSeek Harness v0.1 plugin API (`@deepseek-ai/cordis`, `@deepseek-ai/dsh-tools`), which is young and moving; pin versions once it stabilizes. Auto-capture (store turns without an explicit tool call) and auto-recall (inject memory into the prompt at assembly) are planned once the harness session/assembly event API is confirmed.

integrations/deepseek-plugin/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@mem0/deepseek-plugin",
-  "version": "0.1.0",
+  "version": "0.1.1",
   "description": "Mem0 long-term memory as a native DeepSeek Harness (Cordis) plugin.",
   "type": "module",
   "license": "Apache-2.0",

integrations/deepseek-plugin/src/index.ts

@@ -16,6 +16,7 @@ import { MemoryClient } from "mem0ai";
 import { formatMemoryList, formatAddResult } from "./formatting.ts";
 import { truncateOutput } from "./output.ts";
 import { resolveSearchFilters, resolveAddParams } from "./scoping.ts";
+import { captureEvent, errorKind } from "./telemetry.ts";
 
 export const name = "mem0";
 export const inject = ["tools"];
@@ -81,6 +82,8 @@ export function apply(ctx: Context, config: Config): void {
     ...(config.host ? { host: config.host } : {}),
   });
 
+  captureEvent("deepseek.plugin.mounted", { has_host: Boolean(config.host) }, client);
+
   // Recall. The platform rejects top-level entity params on search, so scope
   // goes inside `filters` (unlike add below, which takes them top-level).
   ctx.tools.register(
@@ -99,11 +102,36 @@ export function apply(ctx: Context, config: Config): void {
       output: textOutput,
       async execute({ query, limit, userId: u, agentId, runId }) {
         const filters = resolveSearchFilters({ userId: u, agentId, runId }, userId);
+        const topK = limit && limit > 0 ? limit : DEFAULT_SEARCH_LIMIT;
+        const started = Date.now();
         try {
-          const topK = limit && limit > 0 ? limit : DEFAULT_SEARCH_LIMIT;
           const { results } = await client.search(query, { filters, topK });
+          captureEvent(
+            "deepseek.tool.search_memory",
+            {
+              success: true,
+              duration_ms: Date.now() - started,
+              top_k: topK,
+              result_count: results?.length ?? 0,
+              query_chars: query.length,
+              scope_overridden: Boolean(u && u !== userId),
+              has_agent_id: Boolean(agentId),
+              has_run_id: Boolean(runId),
+            },
+            client,
+          );
           return truncateOutput(formatMemoryList(results ?? []));
         } catch (err) {
+          captureEvent(
+            "deepseek.tool.search_memory",
+            {
+              success: false,
+              duration_ms: Date.now() - started,
+              top_k: topK,
+              error_kind: errorKind(err),
+            },
+            client,
+          );
           return `search_memory failed: ${err instanceof Error ? err.message : String(err)}`;
         }
       },
@@ -123,13 +151,37 @@ export function apply(ctx: Context, config: Config): void {
       output: textOutput,
       async execute({ text, userId: u, agentId, runId }) {
         const addParams = resolveAddParams({ userId: u, agentId, runId }, userId);
+        const started = Date.now();
         try {
           const result = await client.add([{ role: "user", content: text }], {
             ...addParams,
             source: SOURCE,
           });
+          captureEvent(
+            "deepseek.tool.add_memory",
+            {
+              success: true,
+              duration_ms: Date.now() - started,
+              text_chars: text.length,
+              memory_count: Array.isArray(result) ? result.length : 0,
+              scope_overridden: Boolean(u && u !== userId),
+              has_agent_id: Boolean(agentId),
+              has_run_id: Boolean(runId),
+            },
+            client,
+          );
           return truncateOutput(formatAddResult(result));
         } catch (err) {
+          captureEvent(
+            "deepseek.tool.add_memory",
+            {
+              success: false,
+              duration_ms: Date.now() - started,
+              text_chars: text.length,
+              error_kind: errorKind(err),
+            },
+            client,
+          );
           return `add_memory failed: ${err instanceof Error ? err.message : String(err)}`;
         }
       },

integrations/deepseek-plugin/src/telemetry.ts

@@ -0,0 +1,173 @@
+/**
+ * Anonymous usage telemetry for the DeepSeek Harness plugin.
+ *
+ * Fire-and-forget PostHog events over native fetch, batched and flushed every
+ * 5 seconds, at 10 queued events, or on process exit. Never throws, never logs.
+ *
+ * Events carry only tool names, durations, counts, and coarse failure kinds:
+ * never queries, memory text, filters, or API keys.
+ *
+ * Disable with MEM0_TELEMETRY=false.
+ */
+import { randomUUID } from "node:crypto";
+import * as fs from "node:fs";
+import * as os from "node:os";
+import * as path from "node:path";
+
+const POSTHOG_API_KEY = "phc_hgJkUVJFYtmaJqrvf6CYN67TIQ8yhXAkWzUn9AMU4yX";
+const POSTHOG_BATCH_URL = "https://us.i.posthog.com/batch/";
+const FLUSH_INTERVAL_MS = 5_000;
+const FLUSH_THRESHOLD = 10;
+const SEND_TIMEOUT_MS = 3_000;
+
+const PLUGIN_VERSION = ((): string => {
+  try {
+    return JSON.parse(
+      fs.readFileSync(new URL("../package.json", import.meta.url), "utf-8"),
+    ).version;
+  } catch {
+    return "unknown";
+  }
+})();
+
+/** The Mem0 SDK resolves this to the account email once ping() lands, so events join other Mem0 surfaces. */
+export interface TelemetryIdentity {
+  telemetryId?: string;
+}
+
+let queue: Record<string, unknown>[] = [];
+let flushTimer: ReturnType<typeof setInterval> | undefined;
+let exitHandlerInstalled = false;
+let cachedAnonymousId: string | undefined;
+let identified = false;
+
+function identityPath(): string {
+  return path.join(os.homedir(), ".mem0", "deepseek-plugin-telemetry.json");
+}
+
+export function isTelemetryEnabled(): boolean {
+  const value = process.env.MEM0_TELEMETRY?.toLowerCase();
+  return value !== "false" && value !== "0" && value !== "no" && value !== "off";
+}
+
+function anonymousId(): string {
+  if (cachedAnonymousId) return cachedAnonymousId;
+  try {
+    const stored = JSON.parse(fs.readFileSync(identityPath(), "utf-8"));
+    if (typeof stored.anonymousId === "string" && stored.anonymousId) {
+      cachedAnonymousId = stored.anonymousId;
+      return cachedAnonymousId!;
+    }
+  } catch {
+    /* first run, or an unreadable identity file */
+  }
+  const created = `deepseek-anon-${randomUUID().replace(/-/g, "")}`;
+  try {
+    const target = identityPath();
+    fs.mkdirSync(path.dirname(target), { recursive: true });
+    fs.writeFileSync(target, JSON.stringify({ anonymousId: created }), "utf-8");
+  } catch {
+    /* an unwritable home directory must not break the plugin */
+  }
+  cachedAnonymousId = created;
+  return created;
+}
+
+/** Merge the pre-ping anonymous history into the account once the email resolves. */
+function identifyEvent(distinctId: string): Record<string, unknown> | undefined {
+  if (identified || distinctId.startsWith("deepseek-anon-")) return undefined;
+  identified = true;
+  let storedAnonymousId: string | undefined;
+  try {
+    storedAnonymousId = JSON.parse(fs.readFileSync(identityPath(), "utf-8")).anonymousId;
+    fs.unlinkSync(identityPath());
+  } catch {
+    return undefined;
+  }
+  cachedAnonymousId = undefined;
+  if (!storedAnonymousId) return undefined;
+  return {
+    event: "$identify",
+    distinct_id: distinctId,
+    properties: { $anon_distinct_id: storedAnonymousId, $lib: "posthog-node" },
+  };
+}
+
+export function flushEvents(): void {
+  if (queue.length === 0) return;
+  const batch = queue;
+  queue = [];
+  fetch(POSTHOG_BATCH_URL, {
+    method: "POST",
+    headers: { "Content-Type": "application/json" },
+    body: JSON.stringify({ api_key: POSTHOG_API_KEY, batch }),
+    signal: AbortSignal.timeout(SEND_TIMEOUT_MS),
+  }).catch(() => {
+    /* telemetry never surfaces its own failures */
+  });
+}
+
+function scheduleFlush(): void {
+  if (!flushTimer) {
+    flushTimer = setInterval(flushEvents, FLUSH_INTERVAL_MS);
+    flushTimer.unref?.();
+  }
+  if (!exitHandlerInstalled) {
+    exitHandlerInstalled = true;
+    process.on("beforeExit", flushEvents);
+  }
+}
+
+export function errorKind(error: unknown): string {
+  const text = (error instanceof Error ? error.message : String(error)).toLowerCase();
+  if (text.includes("timeout") || text.includes("aborted")) return "timeout";
+  if (text.includes("401") || text.includes("403") || text.includes("unauthor")) return "auth";
+  if (text.includes("429") || text.includes("rate limit")) return "rate-limited";
+  if (/50[0234]/.test(text)) return "server-error";
+  if (text.includes("400") || text.includes("422")) return "bad-request";
+  if (text.includes("fetch failed") || text.includes("enotfound")) return "network";
+  return error instanceof Error ? error.constructor.name : "other";
+}
+
+export function captureEvent(
+  event: string,
+  properties: Record<string, unknown>,
+  client: TelemetryIdentity,
+): void {
+  if (!isTelemetryEnabled()) return;
+  try {
+    const distinctId = client.telemetryId || anonymousId();
+    const identify = identifyEvent(distinctId);
+    if (identify) queue.push(identify);
+    queue.push({
+      event,
+      distinct_id: distinctId,
+      properties: {
+        source: "DEEPSEEK_HARNESS",
+        language: "node",
+        plugin_version: PLUGIN_VERSION,
+        node_version: process.version,
+        os: process.platform,
+        $process_person_profile: false,
+        $lib: "posthog-node",
+        ...properties,
+      },
+    });
+    scheduleFlush();
+    if (queue.length >= FLUSH_THRESHOLD) flushEvents();
+  } catch {
+    /* telemetry never breaks a tool call */
+  }
+}
+
+export function _queueForTesting(): Record<string, unknown>[] {
+  return queue;
+}
+
+export function _resetForTesting(): void {
+  queue = [];
+  clearInterval(flushTimer);
+  flushTimer = undefined;
+  cachedAnonymousId = undefined;
+  identified = false;
+}

integrations/deepseek-plugin/tests/apply.test.ts

@@ -35,9 +35,12 @@ function applyAndCollect(config: Config): Map<string, RegisteredTool> {
 }
 
 let savedKey: string | undefined;
+let savedTelemetry: string | undefined;
 
 beforeEach(() => {
   savedKey = process.env.MEM0_API_KEY;
+  savedTelemetry = process.env.MEM0_TELEMETRY;
+  process.env.MEM0_TELEMETRY = "false";
   mockSearch.mockReset();
   mockAdd.mockReset();
 });
@@ -45,6 +48,8 @@ beforeEach(() => {
 afterEach(() => {
   if (savedKey === undefined) delete process.env.MEM0_API_KEY;
   else process.env.MEM0_API_KEY = savedKey;
+  if (savedTelemetry === undefined) delete process.env.MEM0_TELEMETRY;
+  else process.env.MEM0_TELEMETRY = savedTelemetry;
 });
 
 describe("apply() config validation", () => {

integrations/deepseek-plugin/tests/telemetry.test.ts

@@ -0,0 +1,218 @@
+import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
+import * as fs from "node:fs";
+import * as os from "node:os";
+import * as path from "node:path";
+
+const mockSearch = vi.fn();
+const mockAdd = vi.fn();
+vi.mock("mem0ai", () => ({
+  MemoryClient: class {
+    telemetryId = "dev@example.com";
+    search = mockSearch;
+    add = mockAdd;
+  },
+}));
+vi.mock("@deepseek-ai/dsh-tools", () => ({ defineTool: (options: unknown) => options }));
+
+import { apply, type Config } from "../src/index.ts";
+import {
+  captureEvent,
+  errorKind,
+  flushEvents,
+  isTelemetryEnabled,
+  _queueForTesting,
+  _resetForTesting,
+} from "../src/telemetry.ts";
+
+interface RegisteredTool {
+  name: string;
+  execute(args: unknown, exec: unknown): Promise<unknown>;
+}
+
+function applyAndCollect(config: Config): Map<string, RegisteredTool> {
+  const tools = new Map<string, RegisteredTool>();
+  apply({ tools: { register: (t: RegisteredTool) => tools.set(t.name, t) } } as never, config);
+  return tools;
+}
+
+function queued(): Record<string, any>[] {
+  return _queueForTesting() as Record<string, any>[];
+}
+
+let home: string;
+let savedHome: string | undefined;
+let savedTelemetry: string | undefined;
+
+beforeEach(() => {
+  savedHome = process.env.HOME;
+  savedTelemetry = process.env.MEM0_TELEMETRY;
+  delete process.env.MEM0_TELEMETRY;
+  home = fs.mkdtempSync(path.join(os.tmpdir(), "mem0-deepseek-"));
+  process.env.HOME = home;
+  vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true }));
+  mockSearch.mockReset();
+  mockAdd.mockReset();
+  _resetForTesting();
+});
+
+afterEach(() => {
+  _resetForTesting();
+  vi.unstubAllGlobals();
+  fs.rmSync(home, { recursive: true, force: true });
+  if (savedHome === undefined) delete process.env.HOME;
+  else process.env.HOME = savedHome;
+  if (savedTelemetry === undefined) delete process.env.MEM0_TELEMETRY;
+  else process.env.MEM0_TELEMETRY = savedTelemetry;
+});
+
+describe("opt-out", () => {
+  it("queues nothing for every documented off value", () => {
+    for (const value of ["false", "0", "no", "OFF"]) {
+      process.env.MEM0_TELEMETRY = value;
+      expect(isTelemetryEnabled()).toBe(false);
+      captureEvent("deepseek.tool.search_memory", {}, { telemetryId: "dev@example.com" });
+    }
+    expect(queued()).toHaveLength(0);
+  });
+
+  it("is on by default", () => {
+    expect(isTelemetryEnabled()).toBe(true);
+  });
+});
+
+describe("identity", () => {
+  it("keys events on the account email the SDK resolved", () => {
+    captureEvent("deepseek.tool.add_memory", {}, { telemetryId: "dev@example.com" });
+    expect(queued()[0].distinct_id).toBe("dev@example.com");
+  });
+
+  it("falls back to a persisted anonymous id before the SDK has pinged", () => {
+    captureEvent("deepseek.tool.add_memory", {}, {});
+    const first = queued()[0].distinct_id as string;
+    expect(first).toMatch(/^deepseek-anon-/);
+
+    _resetForTesting();
+    captureEvent("deepseek.tool.add_memory", {}, {});
+    expect(queued()[0].distinct_id).toBe(first);
+  });
+
+  it("aliases the anonymous history onto the email exactly once", () => {
+    captureEvent("deepseek.tool.add_memory", {}, {});
+    const anonymous = queued()[0].distinct_id;
+    _resetForTesting();
+
+    captureEvent("deepseek.tool.add_memory", {}, { telemetryId: "dev@example.com" });
+    const [identify, event] = queued();
+    expect(identify.event).toBe("$identify");
+    expect(identify.distinct_id).toBe("dev@example.com");
+    expect(identify.properties.$anon_distinct_id).toBe(anonymous);
+    expect(event.distinct_id).toBe("dev@example.com");
+
+    captureEvent("deepseek.tool.add_memory", {}, { telemetryId: "dev@example.com" });
+    expect(queued().filter((e) => e.event === "$identify")).toHaveLength(1);
+  });
+});
+
+describe("event shape", () => {
+  it("stamps every event with the integration source and runtime baseline", () => {
+    captureEvent("deepseek.tool.search_memory", { success: true }, { telemetryId: "d@e.com" });
+    const { properties } = queued()[0];
+    expect(properties.source).toBe("DEEPSEEK_HARNESS");
+    expect(properties.language).toBe("node");
+    expect(properties.$process_person_profile).toBe(false);
+    expect(properties.os).toBe(process.platform);
+    expect(properties.plugin_version).not.toBe("unknown");
+    expect(properties.success).toBe(true);
+  });
+
+  it("flushes one batch to PostHog and empties the queue", async () => {
+    captureEvent("deepseek.tool.search_memory", {}, { telemetryId: "d@e.com" });
+    flushEvents();
+
+    const [url, init] = (fetch as unknown as ReturnType<typeof vi.fn>).mock.calls[0];
+    expect(url).toContain("posthog.com");
+    expect(JSON.parse(init.body).batch).toHaveLength(1);
+    expect(queued()).toHaveLength(0);
+  });
+
+  it("never throws when the network is gone", () => {
+    vi.stubGlobal("fetch", vi.fn().mockRejectedValue(new Error("fetch failed")));
+    captureEvent("deepseek.tool.search_memory", {}, { telemetryId: "d@e.com" });
+    expect(() => flushEvents()).not.toThrow();
+  });
+});
+
+describe("errorKind", () => {
+  it("buckets failures without leaking the message", () => {
+    expect(errorKind(new Error("Request failed with status 429"))).toBe("rate-limited");
+    expect(errorKind(new Error("401 Unauthorized"))).toBe("auth");
+    expect(errorKind(new Error("503 Service Unavailable"))).toBe("server-error");
+    expect(errorKind(new Error("The operation was aborted due to timeout"))).toBe("timeout");
+    expect(errorKind(new Error("fetch failed"))).toBe("network");
+    expect(errorKind(new Error("token sk-abcdef is invalid"))).toBe("Error");
+  });
+});
+
+describe("tool instrumentation", () => {
+  it("records a mount, then a successful search with its shape but not its query", async () => {
+    mockSearch.mockResolvedValue({ results: [{ id: "m1", memory: "Likes tea" }] });
+    const tools = applyAndCollect({ apiKey: "k", userId: "u" });
+
+    await tools.get("search_memory")!.execute({ query: "what drink", limit: 3 }, {});
+
+    const [mounted, search] = queued();
+    expect(mounted.event).toBe("deepseek.plugin.mounted");
+    expect(mounted.properties.has_host).toBe(false);
+    expect(search.event).toBe("deepseek.tool.search_memory");
+    expect(search.properties).toMatchObject({
+      success: true,
+      top_k: 3,
+      result_count: 1,
+      query_chars: 10,
+      scope_overridden: false,
+      has_agent_id: false,
+      has_run_id: false,
+    });
+    expect(typeof search.properties.duration_ms).toBe("number");
+    expect(JSON.stringify(search)).not.toContain("what drink");
+    expect(JSON.stringify(search)).not.toContain("Likes tea");
+  });
+
+  it("records a failed search with a coarse error kind", async () => {
+    mockSearch.mockRejectedValue(new Error("429 rate limit exceeded"));
+    const tools = applyAndCollect({ apiKey: "k", userId: "u" });
+
+    await tools.get("search_memory")!.execute({ query: "x" }, {});
+
+    const search = queued().at(-1)!;
+    expect(search.properties).toMatchObject({ success: false, error_kind: "rate-limited" });
+  });
+
+  it("records a write with its size but not its text", async () => {
+    mockAdd.mockResolvedValue([{ id: "m1", memory: "Fact" }]);
+    const tools = applyAndCollect({ apiKey: "k", userId: "u" });
+
+    await tools.get("add_memory")!.execute({ text: "secret fact", userId: "alice" }, {});
+
+    const add = queued().at(-1)!;
+    expect(add.event).toBe("deepseek.tool.add_memory");
+    expect(add.properties).toMatchObject({
+      success: true,
+      text_chars: 11,
+      memory_count: 1,
+      scope_overridden: true,
+    });
+    expect(JSON.stringify(add)).not.toContain("secret fact");
+    expect(JSON.stringify(add)).not.toContain("alice");
+  });
+
+  it("records a failed write and still returns the graceful failure line", async () => {
+    mockAdd.mockRejectedValue(new Error("boom"));
+    const tools = applyAndCollect({ apiKey: "k", userId: "u" });
+
+    const out = await tools.get("add_memory")!.execute({ text: "x" }, {});
+
+    expect(out).toContain("add_memory failed");
+    expect(queued().at(-1)!.properties).toMatchObject({ success: false, error_kind: "Error" });
+  });
+});