From c90cbc75a27dd76786e27071f5455ee2951dc68f Mon Sep 17 00:00:00 2001 From: Chaithanya Kumar Date: Fri, 8 May 2026 01:21:33 +0530 Subject: [PATCH 1/4] =?UTF-8?q?docs:=20memory=20decay=20v0.5=20=E2=80=94?= =?UTF-8?q?=20platform=20feature=20page=20+=20API=20reference=20(#5056)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Claude Opus 4.7 (1M context) --- docs/api-reference/organizations-projects.mdx | 13 ++ docs/changelog/platform.mdx | 7 + docs/docs.json | 3 +- docs/llms.txt | 1 + docs/platform/features/memory-decay.mdx | 193 ++++++++++++++++++ 5 files changed, 216 insertions(+), 1 deletion(-) create mode 100644 docs/platform/features/memory-decay.mdx diff --git a/docs/api-reference/organizations-projects.mdx b/docs/api-reference/organizations-projects.mdx index cbdf1f162..d4958b260 100644 --- a/docs/api-reference/organizations-projects.mdx +++ b/docs/api-reference/organizations-projects.mdx @@ -109,6 +109,19 @@ client.project.update( ) ``` +#### Toggle Memory Decay + +`decay` is a per-project boolean that turns on [Memory Decay](/platform/features/memory-decay) — a search-time ranking bias that reinforces recently-accessed memories and gently dampens stale ones. The flag is `false` by default; set it via the same project-update endpoint: + +```bash cURL +curl -X PATCH https://api.mem0.ai/api/v1/orgs/organizations/$ORG_ID/projects/$PROJECT_ID/ \ + -H "Authorization: Token $MEM0_API_KEY" \ + -H "Content-Type: application/json" \ + -d '{"decay": true}' +``` + +The current state is returned on every project read (and supports `?fields=decay` for a minimal response). Toggling has no effect on stored memories, only on how v3 search ranks them. + ### Delete Project diff --git a/docs/changelog/platform.mdx b/docs/changelog/platform.mdx index 985e9e2c4..996aa20c4 100644 --- a/docs/changelog/platform.mdx +++ b/docs/changelog/platform.mdx @@ -4,6 +4,13 @@ description: "Release notes for the Mem0 hosted platform — backend, dashboard, mode: "wide" --- + + +**New Features:** +- **Memory Decay:** Per-project search-time ranking bias that boosts recently-used memories and gently dampens stale ones. Opt-in via `decay` on the project endpoint; off by default. The scaling factor stays in `0.3×–1.5×`, the public `score` remains clamped to `[0, 1]`, and the bias never filters a candidate out. See [Memory Decay docs](/platform/features/memory-decay). + + + **Improvements:** diff --git a/docs/docs.json b/docs/docs.json index f77d05f26..8e0bf8056 100644 --- a/docs/docs.json +++ b/docs/docs.json @@ -83,7 +83,8 @@ "platform/advanced-memory-operations", "platform/features/criteria-retrieval", "platform/features/contextual-add", - "platform/features/custom-instructions" + "platform/features/custom-instructions", + "platform/features/memory-decay" ] }, { diff --git a/docs/llms.txt b/docs/llms.txt index 1725f45ab..2b1d7e0b8 100644 --- a/docs/llms.txt +++ b/docs/llms.txt @@ -187,6 +187,7 @@ If the user is on a pre-current major (Python < 2, TS < 3, or Platform `output_f - [Criteria-Based Retrieval](https://docs.mem0.ai/platform/features/criteria-retrieval) [Platform]: Use when targeting memories by custom criteria, not just semantic similarity. - [Contextual Add](https://docs.mem0.ai/platform/features/contextual-add) [Platform]: Use when `add()` should consider the surrounding conversation, not just the latest turn. - [Custom Instructions](https://docs.mem0.ai/platform/features/custom-instructions) [Platform]: Use when tailoring what Mem0 extracts and stores on Platform. +- [Memory Decay](https://docs.mem0.ai/platform/features/memory-decay) [Platform]: Use when search results should boost recently-reinforced memories and dampen stale ones — opt-in per project, search-time only, never filters candidates out. - [Advanced Memory Operations](https://docs.mem0.ai/platform/advanced-memory-operations) [Platform]: Use when basic CRUD is not enough - batch ops, complex filters, workflows. ### Features - Data Management diff --git a/docs/platform/features/memory-decay.mdx b/docs/platform/features/memory-decay.mdx new file mode 100644 index 000000000..612219e63 --- /dev/null +++ b/docs/platform/features/memory-decay.mdx @@ -0,0 +1,193 @@ +--- +title: Memory Decay +description: "Boost recently-used memories and gently dampen stale ones at search time, without filtering anything out." +--- + +# Memory Decay + +Older memories drift in relevance at different speeds. A user's coffee order matters every morning; a one-off project name from last quarter rarely matters again. Memory Decay makes that intuition explicit at search time: every time a memory is returned in a search it gets a small reinforcement, and memories that haven't been touched in a while have their ranking score gently dampened. + +It is **a soft ranking bias, never a filter.** Decay never zeroes a candidate out — at worst it scales its score by `0.3×`. Anything that would have surfaced without decay can still surface with decay on, just with a different ranking among similarly-scored results. + + + **Use Memory Decay when…** + - Search results are crowded with old facts the user no longer cares about. + - You want recently-used memories to drift to the top automatically — without writing custom scoring logic. + - You want this preference applied per project so cohorts can be compared side-by-side. + + + + Memory Decay is **opt-in per project** and **off by default**. Search behavior is bit-identical to today until you turn it on. The toggle applies to v3 search only. + + +## How it works + +Every memory carries a small piece of bookkeeping: when was it last retrieved, and how often. Memory Decay turns that history into a *scaling factor* in the range `0.3×` to `1.5×` and multiplies it into the ranking score at search time. + +| Memory state | Scaling factor | Ranking effect | +|---|---|---| +| Just accessed | ≈ **1.5×** | Strong boost | +| Touched today | 1.2 – 1.4× | Mild boost | +| Idle for a few days | 0.6 – 1.0× | Mild dampening | +| Idle for weeks | 0.4 – 0.6× | Stronger dampening | +| Idle for many months / years | ≈ **0.3×** | Floor — never lower | + +The bounds matter: `0.3` is the floor and `1.5` is the ceiling, so decay can meaningfully reorder candidates without ever dominating the underlying relevance score. + +At search time the pipeline: + +1. Widens the candidate pool (`top_k × 3`, with a floor of 50) so reordering has room. +2. Multiplies each candidate's score by its scaling factor. +3. Sorts on the unclamped product so the full `0.3×–1.5×` range can rearrange candidates. +4. Returns the public `score` clamped to `[0, 1]` so the API contract is preserved. +5. Truncates to the `top_k` you requested. +6. Records a fire-and-forget reinforcement against each returned memory — its access history grows by one, capped at the most recent 20 touches. + +Memories created before decay was enabled don't yet have an access history. They use a sensible fallback: their `updated_at` is treated as a single past touch, so the same scale above applies based on how stale that update is — a recently-updated legacy memory enters near the neutral band, a long-stale one sits closer to the floor. Once surfaced in a search after decay is on, they accumulate access history naturally and behave like any other memory. + +## Configure access + +- Set `MEM0_API_KEY` in your environment, or pass it to the SDK constructor. +- Initialize the client with the organization and project you want to scope to. + +The toggle lives on the project. You enable decay by patching the project's `decay` field; everything else — your `add` calls, your `search` calls, your application code — stays exactly the same. + +## Enable decay for a project + +### 1. Turn the flag on + +The toggle is exposed on the standard project-update endpoint, the same place where `multilingual` and `custom_categories` live. + + +```bash cURL +curl -X PATCH https://api.mem0.ai/api/v1/orgs/organizations/$ORG_ID/projects/$PROJECT_ID/ \ + -H "Authorization: Token $MEM0_API_KEY" \ + -H "Content-Type: application/json" \ + -d '{"decay": true}' +``` + +```python Python +import os +import requests + +org_id = os.environ["MEM0_ORG_ID"] +project_id = os.environ["MEM0_PROJECT_ID"] + +requests.patch( + f"https://api.mem0.ai/api/v1/orgs/organizations/{org_id}/projects/{project_id}/", + headers={"Authorization": f"Token {os.environ['MEM0_API_KEY']}"}, + json={"decay": True}, +) +``` + +```javascript Node.js +const res = await fetch( + `https://api.mem0.ai/api/v1/orgs/organizations/${process.env.MEM0_ORG_ID}/projects/${process.env.MEM0_PROJECT_ID}/`, + { + method: "PATCH", + headers: { + Authorization: `Token ${process.env.MEM0_API_KEY}`, + "Content-Type": "application/json", + }, + body: JSON.stringify({ decay: true }), + }, +); +``` + +```json Response +{ "message": "Updated decay" } +``` + + +### 2. Confirm the state + +`decay` is returned on every project read. To fetch only this field, use `?fields=decay`. + + +```bash cURL +curl "https://api.mem0.ai/api/v1/orgs/organizations/$ORG_ID/projects/$PROJECT_ID/?fields=decay" \ + -H "Authorization: Token $MEM0_API_KEY" +``` + +```json Response +{ "decay": true } +``` + + +### 3. Turn it back off + +The toggle is fully reversible. Setting it to `false` immediately restores the pre-decay ranking; nothing about your stored memories is modified or lost. + + +```bash cURL +curl -X PATCH https://api.mem0.ai/api/v1/orgs/organizations/$ORG_ID/projects/$PROJECT_ID/ \ + -H "Authorization: Token $MEM0_API_KEY" \ + -H "Content-Type: application/json" \ + -d '{"decay": false}' +``` + + + + The toggle is idempotent. Re-applying the same value is a no-op, and access history accumulated while decay was on is preserved if you flip it back on later. + + +## What changes when decay is on + +- **Search ranking reorders.** A relevant memory you reinforced an hour ago will tend to outrank an equally-relevant memory that was last touched a month ago. +- **The candidate pool over-fetches** to give the scaling factor room to reorder. You still get exactly the `top_k` you requested, but the items returned can come from a deeper slice of the pre-decay ranking than before. +- **The public `score` field stays in `[0, 1]`.** Even when the internal product exceeds 1, the field returned to the client is clamped, so existing assertions and downstream UI logic continue to work. + +## What stays the same + +- **Public API shape** — every endpoint accepts the same parameters and returns the same fields. You don't touch your client code. +- **Threshold semantics on the request side** — your `threshold` is still applied during candidate selection. +- **Memory creation and storage** — every new memory still lands the same way. Decay is a search-time concern. +- **Per-memory data** — categories, metadata, timestamps, embeddings: untouched. + + + Because the scaling factor is applied *after* the threshold filter has already run, an item that passed the request `threshold` can come back with a public `score` slightly below it (a stale candidate dampened by `0.3×`). This is intentional — decay is a soft bias, not a filter. If you require a hard `score >= threshold` invariant on the response, filter client-side after the call. + + +## Lifecycle of a memory under decay + +| Stage | Scaling factor | Effect | +|---|---|---| +| Just added | ≈ 1.5× | Strong boost — fresh facts surface easily. | +| Reinforced on a recent search | 1.2 – 1.5× | Sustains its boost for the next several searches. | +| Idle for a few days | 0.6 – 1.0× | Falls back into the neutral band. | +| Idle for weeks | 0.4 – 0.6× | Mild dampening — can still surface for strong matches. | +| Pre-decay legacy memory (no access history) | 0.3 – 1.0× | Falls back to `updated_at`: recently-updated entries land near 1.0×, long-stale entries approach the 0.3× floor. | + +The reinforcement is bounded: each memory tracks at most the last 20 access timestamps, so the boost stays well-behaved no matter how many times a memory is retrieved. + +## FAQ + +**Will decay ever drop a result that would otherwise surface?** +No. The floor is `0.3×` — the scaling factor can dampen a score, never zero it. Threshold filtering happens *before* decay, so any candidate that cleared the threshold is in the pool decay reorders. + +**Why is the public score sometimes below my requested threshold?** +The threshold is applied to the candidate pool pre-decay; the scaling factor then reshapes scores in the `0.3×–1.5×` band. A stale-but-relevant candidate can come back with a final score slightly under your threshold by design — the candidate stays visible but visibly dampened. Filter client-side if you need a hard floor on the response. + +**Does decay change how I add memories?** +No. The `client.add(...)` path is unchanged. Decay is a search-time ranking adjustment. + +**What if I had memories before turning decay on?** +They use a fallback: the memory's `updated_at` is treated as a single historical touch, so the same scaling applies based on how stale that update is — a recently-updated legacy memory enters near the neutral band (~1.0×), a long-stale one closer to the floor (~0.3×). Once retrieved they accumulate access history and behave like any other memory. + +**Can I tune how aggressively decay scales scores?** +Not in this version. The current scaling is calibrated to be conservative — wide enough to meaningfully reorder candidates, narrow enough to never dominate the underlying relevance score. Per-project tuning is on the roadmap. + +**Can I see the scaling factor per result?** +Internal scoring details are persisted on the search Event for support and debugging. They aren't exposed in the public response by design — the response surface stays a single `score` field. + +**Does decay interact with reranking?** +Yes — they layer cleanly. The reranker produces a richer relevance score; decay then biases that score by reinforcement history before final truncation to `top_k`. + +## What's next + +This release is deliberately the simplest version of decay we could ship — every memory contributes to ranking through its access history alone, so the signal can be evaluated in isolation. On the roadmap: + +- **Category-aware weighting.** A fact tagged `health` will be able to carry more weight than a passing observation tagged `misc`, so important categories don't get dampened the same way as noise. +- **Auto-tuning per project.** Project-scoped automatic adjustment of how aggressively decay scales scores, based on observed access patterns — replacing the fixed scaling band with one that fits your workload. + +Both extensions are forward-compatible — no migration on your side will be needed when they ship. From 9043fbf61e60c9e2f2e60ddddc849adebc273608 Mon Sep 17 00:00:00 2001 From: Mragank Shekhar Date: Fri, 8 May 2026 01:27:23 +0530 Subject: [PATCH 2/4] chore(release): bump mem0ai to 2.0.2 (py) and 3.0.3 (ts) (#5078) --- docs/changelog/highlights.mdx | 13 +++++++++++++ docs/changelog/sdk.mdx | 26 ++++++++++++++++++++++++++ mem0-ts/package.json | 2 +- pyproject.toml | 2 +- 4 files changed, 41 insertions(+), 2 deletions(-) diff --git a/docs/changelog/highlights.mdx b/docs/changelog/highlights.mdx index a20a2e17e..71530a1b2 100644 --- a/docs/changelog/highlights.mdx +++ b/docs/changelog/highlights.mdx @@ -4,6 +4,19 @@ description: "Major product launches, headline features, and milestones for Mem0 mode: "wide" --- + + +**Memory Decay — Recently-Used Memories Surface Higher, Automatically** + +Per-project search-time ranking bias that boosts recently-touched memories and gently dampens stale ones. Off by default; opt in per project via the `decay` field on the project endpoint, or via `client.project.update(decay=True)` in the SDKs (Python `v2.0.2` / TypeScript `v3.0.3`). + +- **Soft bias, never a filter.** The scaling factor stays in `0.3×–1.5×`. Decay can reorder candidates but never zeros them out — anything that surfaced before decay can still surface after. +- **Reinforcement loop.** Every memory returned in a search has its access history updated, so frequently-used facts naturally float to the top over time. +- **Public score still clamped to `[0, 1]`.** Existing API contract preserved; no client-side changes needed. +- **v3 search only**, fully reversible. See [Memory Decay docs](/platform/features/memory-decay). + + + **New Memory Algorithm — State-of-the-Art Accuracy at ~3-4x Lower Cost** diff --git a/docs/changelog/sdk.mdx b/docs/changelog/sdk.mdx index d20702f98..9025b5e42 100644 --- a/docs/changelog/sdk.mdx +++ b/docs/changelog/sdk.mdx @@ -7,6 +7,20 @@ mode: "wide" + + +**Bug Fixes:** +- **Telemetry:** Stitch OSS and platform PostHog identities on `MemoryClient` init so `$identify` events fire and a single user is no longer tracked as two or three disconnected personas ([#5040](https://github.com/mem0ai/mem0/pull/5040)) +- **Security:** Harden against SQL injection and prompt injection ([#4997](https://github.com/mem0ai/mem0/pull/4997)) + +**New Features:** +- **SDK:** Expose `decay` on `project.update` ([#5062](https://github.com/mem0ai/mem0/pull/5062)) + +**Improvements:** +- **Plugin:** Hand `mem0` search decisions to the agent ([#4992](https://github.com/mem0ai/mem0/pull/4992)) + + + **Bug Fixes:** @@ -910,6 +924,18 @@ See the [OSS v1 to v2 migration guide](https://docs.mem0.ai/migration/oss-v1-to- + + +**Bug Fixes:** +- **Telemetry:** Stitch OSS and platform PostHog identities on `MemoryClient` init so `$identify` events fire and a single user is no longer tracked as two or three disconnected personas ([#5040](https://github.com/mem0ai/mem0/pull/5040)) +- **Vector Stores:** Fix inverted vector distance in PGVector implementation ([#4944](https://github.com/mem0ai/mem0/pull/4944)) +- **Security:** Harden against SQL injection and prompt injection ([#4997](https://github.com/mem0ai/mem0/pull/4997)) + +**New Features:** +- **SDK:** Expose `decay` on `project.update` ([#5062](https://github.com/mem0ai/mem0/pull/5062)) + + + **Bug Fixes:** diff --git a/mem0-ts/package.json b/mem0-ts/package.json index ec7235bac..b83f33c4f 100644 --- a/mem0-ts/package.json +++ b/mem0-ts/package.json @@ -1,6 +1,6 @@ { "name": "mem0ai", - "version": "3.0.2", + "version": "3.0.3", "description": "The Memory Layer For Your AI Apps", "main": "./dist/index.js", "module": "./dist/index.mjs", diff --git a/pyproject.toml b/pyproject.toml index c5bf151c4..e905f57d2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "mem0ai" -version = "2.0.1" +version = "2.0.2" description = "Long-term memory for AI Agents" authors = [ { name = "Mem0", email = "support@mem0.ai" } From 92491c00c2a098be686ad3d1d0e5ab749cf1d159 Mon Sep 17 00:00:00 2001 From: Chaithanya Kumar Date: Fri, 8 May 2026 01:53:55 +0530 Subject: [PATCH 3/4] docs(memory-decay): use SDK calls in code samples (#5079) Co-authored-by: Claude Opus 4.7 (1M context) --- docs/platform/features/memory-decay.mdx | 54 ++++++++++++------------- 1 file changed, 26 insertions(+), 28 deletions(-) diff --git a/docs/platform/features/memory-decay.mdx b/docs/platform/features/memory-decay.mdx index 612219e63..4c4cfd37d 100644 --- a/docs/platform/features/memory-decay.mdx +++ b/docs/platform/features/memory-decay.mdx @@ -59,6 +59,14 @@ The toggle lives on the project. You enable decay by patching the project's `dec The toggle is exposed on the standard project-update endpoint, the same place where `multilingual` and `custom_categories` live. +```python Python +client.project.update(decay=True) +``` + +```javascript JavaScript +await client.project.update({ decay: true }); +``` + ```bash cURL curl -X PATCH https://api.mem0.ai/api/v1/orgs/organizations/$ORG_ID/projects/$PROJECT_ID/ \ -H "Authorization: Token $MEM0_API_KEY" \ @@ -66,34 +74,6 @@ curl -X PATCH https://api.mem0.ai/api/v1/orgs/organizations/$ORG_ID/projects/$PR -d '{"decay": true}' ``` -```python Python -import os -import requests - -org_id = os.environ["MEM0_ORG_ID"] -project_id = os.environ["MEM0_PROJECT_ID"] - -requests.patch( - f"https://api.mem0.ai/api/v1/orgs/organizations/{org_id}/projects/{project_id}/", - headers={"Authorization": f"Token {os.environ['MEM0_API_KEY']}"}, - json={"decay": True}, -) -``` - -```javascript Node.js -const res = await fetch( - `https://api.mem0.ai/api/v1/orgs/organizations/${process.env.MEM0_ORG_ID}/projects/${process.env.MEM0_PROJECT_ID}/`, - { - method: "PATCH", - headers: { - Authorization: `Token ${process.env.MEM0_API_KEY}`, - "Content-Type": "application/json", - }, - body: JSON.stringify({ decay: true }), - }, -); -``` - ```json Response { "message": "Updated decay" } ``` @@ -104,6 +84,16 @@ const res = await fetch( `decay` is returned on every project read. To fetch only this field, use `?fields=decay`. +```python Python +response = client.project.get(fields=["decay"]) +print(response["decay"]) +``` + +```javascript JavaScript +const response = await client.project.get({ fields: ["decay"] }); +console.log(response.decay); +``` + ```bash cURL curl "https://api.mem0.ai/api/v1/orgs/organizations/$ORG_ID/projects/$PROJECT_ID/?fields=decay" \ -H "Authorization: Token $MEM0_API_KEY" @@ -119,6 +109,14 @@ curl "https://api.mem0.ai/api/v1/orgs/organizations/$ORG_ID/projects/$PROJECT_ID The toggle is fully reversible. Setting it to `false` immediately restores the pre-decay ranking; nothing about your stored memories is modified or lost. +```python Python +client.project.update(decay=False) +``` + +```javascript JavaScript +await client.project.update({ decay: false }); +``` + ```bash cURL curl -X PATCH https://api.mem0.ai/api/v1/orgs/organizations/$ORG_ID/projects/$PROJECT_ID/ \ -H "Authorization: Token $MEM0_API_KEY" \ From a623cfaf76ae7379a58be1e837f8a88a9b15a184 Mon Sep 17 00:00:00 2001 From: youneshima Date: Thu, 7 May 2026 19:34:09 -0700 Subject: [PATCH 4/4] Oss qdrant hosted memories to platform migration (#5080) --- scripts/oss-to-platform-migrate.sh | 1195 +++++++++++++++++++++++++ tests/test_oss_to_platform_migrate.py | 838 +++++++++++++++++ 2 files changed, 2033 insertions(+) create mode 100755 scripts/oss-to-platform-migrate.sh create mode 100644 tests/test_oss_to_platform_migrate.py diff --git a/scripts/oss-to-platform-migrate.sh b/scripts/oss-to-platform-migrate.sh new file mode 100755 index 000000000..b3a984e82 --- /dev/null +++ b/scripts/oss-to-platform-migrate.sh @@ -0,0 +1,1195 @@ +#!/usr/bin/env bash +set -euo pipefail + +if ! command -v python3 >/dev/null 2>&1; then + printf '%s\n' "Error: python3 is required to run the Mem0 migration. Install Python 3, then rerun this command." >&2 + exit 1 +fi + +exec python3 - "$@" <<'PY' +from __future__ import annotations + +import argparse +from collections import Counter +import getpass +import json +import os +import platform +import re +import stat +import sys +import uuid +from datetime import datetime, timezone +from hashlib import sha256 +from pathlib import Path +from typing import Any, Dict, List, Optional, Set, Tuple +from urllib.error import HTTPError, URLError +from urllib.request import Request, urlopen + + +DEFAULT_BASE_URL = "https://api.mem0.ai" +POSTHOG_API_KEY = "phc_hgJkUVJFYtmaJqrvf6CYN67TIQ8yhXAkWzUn9AMU4yX" +POSTHOG_CAPTURE_URL = "https://us.i.posthog.com/i/v0/e/" +SCRIPT_VERSION = "oss-to-platform-migrate-v1" +EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$") +DEFAULT_QDRANT_COLLECTION = "mem0" +DEFAULT_EXPORT_PAGE_SIZE = 100 +DEFAULT_PLATFORM_PAGE_SIZE = 100 + + +class MigrationError(Exception): + pass + + +class AuthError(MigrationError): + pass + + +class HTTPStatusError(MigrationError): + def __init__(self, status_code: int, detail: str): + self.status_code = status_code + self.detail = detail + super().__init__(detail) + + +def utc_now() -> str: + return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + + +def print_info(message: str) -> None: + print(message) + + +def print_error(message: str) -> None: + print(f"Error: {message}", file=sys.stderr) + + +def mem0_dir() -> Path: + configured = os.environ.get("MEM0_DIR") + if configured: + return Path(configured).expanduser() + return Path.home() / ".mem0" + + +def config_path() -> Path: + return mem0_dir() / "config.json" + + +def load_config(path: Path) -> Dict[str, Any]: + if not path.exists(): + return {} + try: + data = json.loads(path.read_text(encoding="utf-8")) + return data if isinstance(data, dict) else {} + except Exception: + return {} + + +def write_config(path: Path, config: Dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + os.chmod(path.parent, stat.S_IRWXU) + path.write_text(json.dumps(config, indent=4) + "\n", encoding="utf-8") + os.chmod(path, stat.S_IRUSR | stat.S_IWUSR) + + +def ensure_oss_anon_id(path: Path, config: Dict[str, Any]) -> str: + existing = config.get("user_id") + if isinstance(existing, str) and existing: + return existing + + generated = str(uuid.uuid4()) + config["user_id"] = generated + try: + write_config(path, config) + except Exception: + # Keep the run usable even on read-only home directories. Telemetry can + # still use the transient ID for this run. + pass + return generated + + +def read_anon_ids(config: Dict[str, Any], fallback_oss_id: str) -> Dict[str, Any]: + telemetry = config.get("telemetry") + if not isinstance(telemetry, dict): + telemetry = {} + aliased_pairs = telemetry.get("aliased_pairs") + if not isinstance(aliased_pairs, list): + aliased_pairs = [] + return { + "oss": config.get("user_id") if isinstance(config.get("user_id"), str) else fallback_oss_id, + "cli": telemetry.get("anonymous_id") if isinstance(telemetry.get("anonymous_id"), str) else None, + "aliased_pairs": [item for item in aliased_pairs if isinstance(item, str)], + } + + +def primary_anon_id(anon_ids: Dict[str, Any]) -> str: + return anon_ids.get("oss") or anon_ids.get("cli") or "anonymous_user" + + +def unique_anon_ids(anon_ids: Dict[str, Any]) -> List[str]: + seen: Set[str] = set() + result: List[str] = [] + for anon_id in (anon_ids.get("oss"), anon_ids.get("cli")): + if not anon_id or anon_id in seen: + continue + seen.add(anon_id) + result.append(anon_id) + return result + + +def alias_pair_marker(anon_id: str, email: str) -> str: + return sha256(f"{anon_id}\0{email}".encode("utf-8")).hexdigest() + + +def is_aliased(config: Dict[str, Any], anon_id: str, email: str) -> bool: + telemetry = config.get("telemetry") + if not isinstance(telemetry, dict): + return False + 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(path: Path, config: Dict[str, Any], anon_id: str, email: str) -> None: + 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(path, config) + + +def normalize_email(email: str) -> str: + normalized = email.strip().lower() + if not EMAIL_RE.match(normalized): + raise MigrationError(f"Invalid email address: {email!r}") + return normalized + + +def parse_error_body(raw: bytes, fallback: str) -> str: + text = raw.decode("utf-8", errors="replace").strip() + if not text: + return fallback + try: + parsed = json.loads(text) + if isinstance(parsed, dict): + for key in ("error", "detail", "message"): + value = parsed.get(key) + if value: + return str(value) + except Exception: + pass + return text + + +def request_json( + method: str, + url: str, + *, + payload: Optional[Dict[str, Any]] = None, + headers: Optional[Dict[str, str]] = None, + timeout: float = 30.0, +) -> Dict[str, Any]: + request_headers = {"Accept": "application/json"} + if headers: + request_headers.update(headers) + + data = None + if payload is not None: + data = json.dumps(payload).encode("utf-8") + request_headers["Content-Type"] = "application/json" + + req = Request(url, data=data, headers=request_headers, method=method) + try: + with urlopen(req, timeout=timeout) as resp: + raw = resp.read() + except HTTPError as exc: + detail = parse_error_body(exc.read(), exc.reason or f"HTTP {exc.code}") + raise HTTPStatusError(exc.code, detail) from exc + except URLError as exc: + raise MigrationError(f"Network request failed for {url}: {exc.reason}. Check your connection and rerun the command.") from exc + + if not raw: + return {} + try: + parsed = json.loads(raw.decode("utf-8")) + except json.JSONDecodeError as exc: + raise MigrationError(f"Expected JSON response from {url}") from exc + return parsed if isinstance(parsed, dict) else {} + + +def post_json_best_effort(url: str, payload: Dict[str, Any], timeout: float = 5.0) -> bool: + data = json.dumps(payload).encode("utf-8") + req = Request(url, data=data, headers={"Content-Type": "application/json"}, method="POST") + try: + with urlopen(req, timeout=timeout) as resp: + resp.read() + return 200 <= resp.status < 300 + except Exception: + return False + + +def telemetry_enabled() -> bool: + raw = os.environ.get("MEM0_TELEMETRY", "true") + return raw.lower() in ("true", "1", "yes") + + +def telemetry_url() -> str: + return os.environ.get("MEM0_MIGRATE_TELEMETRY_URL", POSTHOG_CAPTURE_URL) + + +def base_event_properties(anon_ids: Dict[str, Any], email: Optional[str] = None) -> Dict[str, Any]: + properties: Dict[str, Any] = { + "client_source": "python", + "client_version": SCRIPT_VERSION, + "migration_phase": "migration", + "local_anonymous_id": primary_anon_id(anon_ids), + "oss_anonymous_id": anon_ids.get("oss"), + "cli_anonymous_id": anon_ids.get("cli"), + "python_version": sys.version, + "os": sys.platform, + "os_version": platform.version(), + "os_release": platform.release(), + "machine": platform.machine(), + "timestamp": utc_now(), + "$lib": "posthog-python", + } + if email: + properties["authenticated_email"] = email + return properties + + +def capture_migration_event( + event_name: str, + anon_ids: Dict[str, Any], + *, + email: Optional[str] = None, + additional: Optional[Dict[str, Any]] = None, +) -> bool: + if not telemetry_enabled(): + return False + distinct_id = email or primary_anon_id(anon_ids) + properties = base_event_properties(anon_ids, email) + if additional: + properties.update(additional) + payload = { + "api_key": POSTHOG_API_KEY, + "distinct_id": distinct_id, + "event": event_name, + "properties": properties, + } + return post_json_best_effort(telemetry_url(), payload) + + +def capture_identify(anon_id: str, email: str) -> bool: + if not telemetry_enabled() or not anon_id or not email or anon_id == email: + return False + payload = { + "api_key": POSTHOG_API_KEY, + "distinct_id": email, + "event": "$identify", + "properties": { + "$anon_distinct_id": anon_id, + "client_source": "python", + "client_version": SCRIPT_VERSION, + "$lib": "posthog-python", + }, + } + return post_json_best_effort(telemetry_url(), payload) + + +def stitch_identities(config_path_value: Path, config: Dict[str, Any], anon_ids: Dict[str, Any], email: str) -> None: + for anon_id in unique_anon_ids(anon_ids): + if anon_id == email or is_aliased(config, anon_id, email): + continue + if capture_identify(anon_id, email): + try: + mark_aliased(config_path_value, config, anon_id, email) + except Exception: + pass + + +def auth_headers(language: str = "python") -> Dict[str, str]: + return { + "X-Mem0-Source": "migration", + "X-Mem0-Client-Language": language, + "X-Mem0-Client-Version": SCRIPT_VERSION, + } + + +def platform_headers(api_key: str, language: str = "python") -> Dict[str, str]: + return { + "Authorization": f"Token {api_key}", + "X-Mem0-Source": "migration", + "X-Mem0-Client-Language": language, + "X-Mem0-Client-Version": SCRIPT_VERSION, + } + + +def api_base_url(args: argparse.Namespace, config: Dict[str, Any]) -> str: + platform_config = config.get("platform") if isinstance(config.get("platform"), dict) else {} + value = args.base_url or os.environ.get("MEM0_BASE_URL") or platform_config.get("base_url") or DEFAULT_BASE_URL + return str(value).rstrip("/") + + +def configured_api_key(args: argparse.Namespace, config: Dict[str, Any]) -> Tuple[Optional[str], str]: + if args.api_key: + return args.api_key, "flag" + env_key = os.environ.get("MEM0_API_KEY") + if env_key: + return env_key, "env" + platform_config = config.get("platform") if isinstance(config.get("platform"), dict) else {} + file_key = platform_config.get("api_key") + if isinstance(file_key, str) and file_key: + return file_key, "config" + return None, "" + + +def validate_api_key(api_key: str, base_url: str) -> str: + try: + result = request_json( + "GET", + f"{base_url}/v1/ping/", + headers={"Authorization": f"Token {api_key}"}, + timeout=5.0, + ) + except HTTPStatusError as exc: + if exc.status_code == 401: + raise AuthError("Invalid or expired API key.") from exc + raise + + email = result.get("user_email") + if not isinstance(email, str) or not email: + raise MigrationError("Platform authenticated the API key but did not return an account email.") + return normalize_email(email) + + +def prompt_line(prompt: str) -> str: + try: + with open("/dev/tty", "r", encoding="utf-8") as tty_in, open("/dev/tty", "w", encoding="utf-8") as tty_out: + tty_out.write(prompt) + tty_out.flush() + line = tty_in.readline() + except OSError as exc: + raise MigrationError( + "No interactive terminal is available. Re-run with --email and --code, or use --yes with a valid API key." + ) from exc + + if line == "": + raise MigrationError("No input received from terminal.") + return line.strip() + + +def prompt_line_default(prompt: str, default: str) -> str: + answer = prompt_line(f"{prompt} [{default}]: ") + return answer or default + + +def prompt_secret(prompt: str) -> str: + ensure_interactive_terminal() + try: + with open("/dev/tty", "w", encoding="utf-8") as tty_out: + return getpass.getpass(prompt, stream=tty_out).strip() + except OSError as exc: + raise MigrationError( + "No interactive terminal is available. Re-run with the required flags or environment variables." + ) from exc + + +def has_interactive_terminal() -> bool: + try: + with open("/dev/tty", "r", encoding="utf-8"), open("/dev/tty", "w", encoding="utf-8"): + return True + except OSError: + return False + + +def ensure_interactive_terminal() -> None: + try: + with open("/dev/tty", "r", encoding="utf-8"), open("/dev/tty", "w", encoding="utf-8"): + return + except OSError as exc: + raise MigrationError( + "No interactive terminal is available. Re-run with --email and --code, or use --yes with a valid API key." + ) from exc + + +def prompt_yes_no(prompt: str, *, default: bool = True) -> bool: + suffix = " [Y/n]: " if default else " [y/N]: " + answer = prompt_line(prompt + suffix).strip().lower() + if not answer: + return default + return answer in ("y", "yes") + + +def request_email_code(email: str, base_url: str) -> None: + try: + request_json( + "POST", + f"{base_url}/api/v1/auth/email_code/", + payload={"email": email}, + headers=auth_headers(), + ) + except HTTPStatusError as exc: + if exc.status_code == 429: + raise MigrationError("Too many attempts. Try again in a few minutes.") from exc + raise MigrationError(f"Failed to send verification code: {exc.detail}") from exc + + +def verify_email_code(email: str, code: str, base_url: str) -> str: + try: + result = request_json( + "POST", + f"{base_url}/api/v1/auth/email_code/verify/", + payload={"email": email, "code": code.strip()}, + headers=auth_headers(), + ) + except HTTPStatusError as exc: + if exc.status_code == 429: + raise MigrationError("Too many attempts. Try again in a few minutes.") from exc + raise MigrationError(f"Verification failed: {exc.detail}") from exc + + api_key = result.get("api_key") + if not isinstance(api_key, str) or not api_key: + raise MigrationError("Auth succeeded but no API key was returned. Contact support.") + return api_key + + +def email_code_auth(args: argparse.Namespace, base_url: str) -> Tuple[str, str, str]: + email = normalize_email(args.email) if args.email else normalize_email(prompt_line("Email: ")) + code = args.code + + if not code: + ensure_interactive_terminal() + request_email_code(email, base_url) + print_info("Verification code sent. Check your email.") + code = prompt_line("Verification code: ") + if not code: + raise MigrationError("Verification code is required.") + + api_key = verify_email_code(email, code, base_url) + resolved_email = validate_api_key(api_key, base_url) + return api_key, resolved_email, "email_code" + + +def existing_session_auth(args: argparse.Namespace, api_key: str, key_source: str, base_url: str) -> Optional[Tuple[str, str, str]]: + try: + email = validate_api_key(api_key, base_url) + except AuthError: + if key_source in ("flag", "env"): + raise + print_info("Stored Mem0 Platform API key is invalid or expired. Continuing with email login.") + return None + except MigrationError: + if key_source in ("flag", "env"): + raise + print_info("Could not validate stored Mem0 Platform session. Continuing with email login.") + return None + + if args.email: + return None + + if not args.yes: + print_info("") + print_info("Found an existing Mem0 Platform session.") + print_info("") + print_info(f"Account: {email}") + print_info("This migration will copy your local OSS memories into this Platform account.") + print_info("") + if not prompt_yes_no(f"Continue with {email}?", default=True): + return None + + return api_key, email, "existing_api_key" + + +def resolve_auth(args: argparse.Namespace, config: Dict[str, Any], base_url: str) -> Tuple[str, str, str]: + if args.api_key and args.email: + raise MigrationError("Cannot use both --api-key and --email.") + if args.code and not args.email: + raise MigrationError("--code requires --email.") + + api_key, key_source = configured_api_key(args, config) + if api_key: + result = existing_session_auth(args, api_key, key_source, base_url) + if result: + return result + + return email_code_auth(args, base_url) + + +def default_export_path() -> Path: + timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + return mem0_dir() / "migrations" / f"mem0-qdrant-export-{timestamp}.json" + + +def strip_env_assignment(value: str, name: str) -> str: + prefix = f"{name}=" + return value[len(prefix):] if value.startswith(prefix) else value + + +def qdrant_url(args: argparse.Namespace) -> str: + value = args.qdrant_url + env_value = os.environ.get("QDRANT_URL") + if not value and env_value: + if has_interactive_terminal(): + normalized_env_value = strip_env_assignment(env_value.strip(), "QDRANT_URL") + print_info("") + print_info("Found QDRANT_URL in your environment:") + print_info(normalized_env_value) + if prompt_yes_no("Use this Qdrant URL?", default=True): + value = env_value + else: + value = env_value + if not value: + if not has_interactive_terminal(): + raise MigrationError("Hosted Qdrant export requires --qdrant-url or QDRANT_URL.") + print_info("") + print_info("Python OSS hosted Qdrant export setup") + print_info("Enter the Qdrant Cloud details used by your OSS Memory configuration.") + value = prompt_line("Qdrant URL: ") + value = strip_env_assignment(value.strip(), "QDRANT_URL") + if not value: + raise MigrationError("Qdrant URL is required.") + return value.rstrip("/") + + +def qdrant_api_key(args: argparse.Namespace) -> str: + value = args.qdrant_api_key + env_value = os.environ.get("QDRANT_API_KEY") + if not value and env_value: + if has_interactive_terminal(): + print_info("") + print_info("Found QDRANT_API_KEY in your environment.") + if prompt_yes_no("Use this Qdrant API key?", default=True): + value = env_value + else: + value = env_value + if not value: + if not has_interactive_terminal(): + raise MigrationError("Hosted Qdrant export requires --qdrant-api-key or QDRANT_API_KEY.") + value = prompt_secret("Qdrant API key: ") + value = strip_env_assignment(value.strip(), "QDRANT_API_KEY") + if not value: + raise MigrationError("Qdrant API key is required.") + return value + + +def qdrant_collection(args: argparse.Namespace) -> str: + value = args.qdrant_collection or os.environ.get("QDRANT_COLLECTION") + if not value: + if has_interactive_terminal(): + value = prompt_line_default("Qdrant collection", DEFAULT_QDRANT_COLLECTION) + else: + value = DEFAULT_QDRANT_COLLECTION + value = value.strip() + if not value: + raise MigrationError("Qdrant collection name cannot be empty.") + return value + + +def qdrant_headers(api_key: str) -> Dict[str, str]: + return {"api-key": api_key} + + +def qdrant_request_json( + method: str, + url: str, + *, + api_key: str, + payload: Optional[Dict[str, Any]] = None, +) -> Dict[str, Any]: + try: + return request_json(method, url, payload=payload, headers=qdrant_headers(api_key), timeout=30.0) + except HTTPStatusError as exc: + if exc.status_code in (401, 403): + raise MigrationError("Qdrant authentication failed. Check your QDRANT_API_KEY.") from exc + if exc.status_code == 404: + raise MigrationError(f"Qdrant resource not found: {url}") from exc + raise MigrationError(f"Qdrant request failed: {exc.detail}") from exc + + +def preflight_qdrant_collection(base_url: str, api_key: str, collection_name: str) -> None: + qdrant_request_json( + "GET", + f"{base_url}/collections/{collection_name}", + api_key=api_key, + ) + + +def qdrant_filter(args: argparse.Namespace) -> Optional[Dict[str, Any]]: + filters = [] + for key, value in (("user_id", args.user_id), ("agent_id", args.agent_id), ("run_id", args.run_id)): + if value: + filters.append({"key": key, "match": {"value": value}}) + + if not filters: + if args.all: + return None + if not has_interactive_terminal(): + raise MigrationError("Export requires --user-id, --agent-id, --run-id, or --all.") + print_info("") + print_info("Choose which OSS memories to export.") + print_info("Enter a user_id to export one user, or press Enter to export the full collection.") + user_id = prompt_line("User ID to export (press Enter for full collection): ") + if user_id: + args.user_id = user_id + filters.append({"key": "user_id", "match": {"value": user_id}}) + else: + args.all = True + return None + + return {"must": filters} + + +def scroll_qdrant_points( + base_url: str, + api_key: str, + collection_name: str, + scroll_filter: Optional[Dict[str, Any]], + page_size: int, +) -> List[Dict[str, Any]]: + points: List[Dict[str, Any]] = [] + offset = None + + while True: + payload: Dict[str, Any] = { + "limit": page_size, + "with_payload": True, + "with_vector": False, + } + if scroll_filter is not None: + payload["filter"] = scroll_filter + if offset is not None: + payload["offset"] = offset + + response = qdrant_request_json( + "POST", + f"{base_url}/collections/{collection_name}/points/scroll", + api_key=api_key, + payload=payload, + ) + result = response.get("result") + if not isinstance(result, dict): + raise MigrationError("Qdrant scroll response did not include a result object.") + + page_points = result.get("points") + if not isinstance(page_points, list): + raise MigrationError("Qdrant scroll response did not include a points list.") + points.extend(point for point in page_points if isinstance(point, dict)) + + offset = result.get("next_page_offset") + if offset is None: + break + + return points + + +def normalize_qdrant_point(point: Dict[str, Any]) -> Dict[str, Any]: + payload = point.get("payload") + if not isinstance(payload, dict): + payload = {} + + promoted_payload_keys = ("user_id", "agent_id", "run_id", "actor_id", "role") + core_and_promoted_keys = { + "data", + "hash", + "created_at", + "updated_at", + "id", + "text_lemmatized", + "attributed_to", + *promoted_payload_keys, + } + + record: Dict[str, Any] = { + "id": str(point.get("id")), + "memory": payload.get("data", ""), + "hash": payload.get("hash"), + "created_at": payload.get("created_at"), + "updated_at": payload.get("updated_at"), + "source": "python_oss_qdrant", + "source_payload": payload, + } + + for key in promoted_payload_keys: + if key in payload: + record[key] = payload[key] + + metadata = {key: value for key, value in payload.items() if key not in core_and_promoted_keys} + if metadata: + record["metadata"] = metadata + + return record + + +def write_export_file(output_path: Path, artifact: Dict[str, Any]) -> None: + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text(json.dumps(artifact, indent=2) + "\n", encoding="utf-8") + os.chmod(output_path, stat.S_IRUSR | stat.S_IWUSR) + + +def export_qdrant_memories(args: argparse.Namespace, anon_ids: Dict[str, Any]) -> Tuple[Path, int]: + base_url = qdrant_url(args) + api_key = qdrant_api_key(args) + collection_name = qdrant_collection(args) + output_path = Path(args.output).expanduser() if args.output else default_export_path() + page_size = args.qdrant_page_size + if page_size < 1: + raise MigrationError("--qdrant-page-size must be greater than 0.") + + scroll_filter = qdrant_filter(args) + preflight_qdrant_collection(base_url, api_key, collection_name) + points = scroll_qdrant_points(base_url, api_key, collection_name, scroll_filter, page_size) + records = [normalize_qdrant_point(point) for point in points] + + artifact = { + "version": 1, + "kind": "mem0_oss_qdrant_export", + "exported_at": utc_now(), + "source": { + "sdk": "python", + "vector_store": "qdrant", + "storage": "hosted", + "collection": collection_name, + "url": base_url, + "filters": { + "user_id": args.user_id, + "agent_id": args.agent_id, + "run_id": args.run_id, + "all": args.all, + }, + }, + "local_anonymous_id": primary_anon_id(anon_ids), + "anonymous_ids": { + "oss": anon_ids.get("oss"), + "cli": anon_ids.get("cli"), + }, + "record_count": len(records), + "records": records, + } + write_export_file(output_path, artifact) + return output_path, len(records) + + +def platform_request_json( + method: str, + url: str, + *, + api_key: str, + payload: Optional[Dict[str, Any]] = None, +) -> Dict[str, Any]: + try: + return request_json(method, url, payload=payload, headers=platform_headers(api_key), timeout=30.0) + except HTTPStatusError as exc: + if exc.status_code in (401, 403): + raise AuthError("Platform authentication failed. Check your MEM0_API_KEY or rerun email login.") from exc + raise MigrationError(f"Platform request failed: {exc.detail}") from exc + + +def import_input_path(args: argparse.Namespace) -> Path: + if not args.input: + raise MigrationError("--input is required for --import-only.") + return Path(args.input).expanduser() + + +def read_import_records(input_path: Path) -> Tuple[Dict[str, Any], List[Dict[str, Any]]]: + try: + parsed = json.loads(input_path.read_text(encoding="utf-8")) + except FileNotFoundError as exc: + raise MigrationError(f"Import file not found: {input_path}") from exc + except json.JSONDecodeError as exc: + raise MigrationError(f"Import file is not valid JSON: {input_path}") from exc + + if isinstance(parsed, list): + records = parsed + artifact: Dict[str, Any] = {} + elif isinstance(parsed, dict): + artifact = parsed + raw_records = parsed.get("records") + if not isinstance(raw_records, list): + raise MigrationError("Import artifact must contain a records array.") + records = raw_records + else: + raise MigrationError("Import file must be a JSON object or array.") + + return artifact, [record for record in records if isinstance(record, dict)] + + +def migration_import_key(source: Dict[str, Any], record: Dict[str, Any]) -> str: + raw = ":".join( + [ + str(source.get("sdk") or "unknown"), + str(source.get("vector_store") or "unknown"), + str(source.get("collection") or "unknown"), + str(record.get("id") or ""), + ] + ) + return sha256(raw.encode("utf-8")).hexdigest() + + +def parse_timestamp(value: Any) -> Optional[int]: + if not isinstance(value, str) or not value: + return None + try: + return int(datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()) + except ValueError: + return None + + +def build_import_metadata(source: Dict[str, Any], record: Dict[str, Any]) -> Dict[str, Any]: + existing_metadata = record.get("metadata") if isinstance(record.get("metadata"), dict) else {} + metadata = dict(existing_metadata) + metadata.update( + { + "mem0_migration_source": record.get("source") or "python_oss_qdrant", + "mem0_migration_collection": source.get("collection"), + "mem0_migration_local_id": record.get("id"), + "mem0_migration_local_hash": record.get("hash"), + "mem0_migration_import_key": migration_import_key(source, record), + } + ) + return {key: value for key, value in metadata.items() if value is not None} + + +def build_import_payload(source: Dict[str, Any], record: Dict[str, Any]) -> Dict[str, Any]: + memory = record.get("memory") + if not isinstance(memory, str) or not memory.strip(): + raise MigrationError("record is missing memory text") + + payload: Dict[str, Any] = { + "messages": [{"role": "user", "content": memory}], + "metadata": build_import_metadata(source, record), + "infer": False, + "source": "migration", + } + + for key in ("user_id", "agent_id", "run_id"): + value = record.get(key) + if value: + payload[key] = value + + timestamp = parse_timestamp(record.get("created_at")) + if timestamp is not None: + payload["timestamp"] = timestamp + + if not any(payload.get(key) for key in ("user_id", "agent_id", "run_id")): + raise MigrationError("record must have at least one of user_id, agent_id, or run_id") + + return payload + + +def payload_scope(payload: Dict[str, Any]) -> Tuple[str, str]: + for key in ("user_id", "agent_id", "run_id"): + value = payload.get(key) + if value: + return key, str(value) + raise MigrationError("import payload must have at least one of user_id, agent_id, or run_id") + + +def platform_get_all_scope(base_url: str, api_key: str, scope_field: str, scope_value: str) -> List[Dict[str, Any]]: + memories: List[Dict[str, Any]] = [] + page = 1 + while True: + response = platform_request_json( + "POST", + f"{base_url}/v3/memories/?page={page}&page_size={DEFAULT_PLATFORM_PAGE_SIZE}", + api_key=api_key, + payload={"filters": {scope_field: scope_value}, "source": "migration"}, + ) + page_results = response.get("results") + if not isinstance(page_results, list): + raise MigrationError("Platform get_all response did not include a results list.") + memories.extend(memory for memory in page_results if isinstance(memory, dict)) + if not response.get("next"): + return memories + page += 1 + + +def existing_migration_memories( + base_url: str, + api_key: str, + payloads: List[Dict[str, Any]], +) -> Dict[str, Dict[str, Any]]: + scopes: List[Tuple[str, str]] = [] + seen_scopes: Set[Tuple[str, str]] = set() + for payload in payloads: + scope = payload_scope(payload) + if scope not in seen_scopes: + seen_scopes.add(scope) + scopes.append(scope) + + existing: Dict[str, Dict[str, Any]] = {} + for scope_field, scope_value in scopes: + print_info(f"Checking existing Platform memories for {scope_field}={scope_value!r}...") + for memory in platform_get_all_scope(base_url, api_key, scope_field, scope_value): + metadata = memory.get("metadata") + if not isinstance(metadata, dict): + continue + key = metadata.get("mem0_migration_import_key") + if isinstance(key, str) and key: + existing[key] = memory + return existing + + +def review_file_path(input_path: Path) -> Path: + timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + return input_path.with_name(f"{input_path.stem}-import-review-{timestamp}.json") + + +def write_import_review_file(input_path: Path, review_records: List[Dict[str, Any]], summary: Dict[str, Any]) -> Optional[Path]: + if not review_records: + return None + path = review_file_path(input_path) + payload = { + "version": 1, + "kind": "mem0_platform_import_review", + "created_at": utc_now(), + "summary": summary, + "records": review_records, + } + path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") + os.chmod(path, stat.S_IRUSR | stat.S_IWUSR) + return path + + +def import_platform_memories(input_path: Path, api_key: str, base_url: str) -> Dict[str, Any]: + artifact, records = read_import_records(input_path) + source = artifact.get("source") if isinstance(artifact.get("source"), dict) else {} + + payloads: List[Tuple[Dict[str, Any], Dict[str, Any]]] = [] + review_records: List[Dict[str, Any]] = [] + invalid = 0 + for record in records: + try: + payloads.append((record, build_import_payload(source, record))) + except MigrationError as exc: + invalid += 1 + review_records.append({"status": "invalid", "error": str(exc), "record": record}) + + existing = existing_migration_memories(base_url, api_key, [payload for _record, payload in payloads]) if payloads else {} + + imported = 0 + skipped_existing = 0 + changed_existing = 0 + failed = 0 + statuses: Counter[str] = Counter() + + for record, payload in payloads: + key = payload["metadata"]["mem0_migration_import_key"] + existing_memory = existing.get(key) + if existing_memory: + existing_metadata = existing_memory.get("metadata") if isinstance(existing_memory.get("metadata"), dict) else {} + existing_hash = existing_metadata.get("mem0_migration_local_hash") + local_hash = payload["metadata"].get("mem0_migration_local_hash") + if existing_hash == local_hash: + skipped_existing += 1 + else: + changed_existing += 1 + review_records.append( + { + "status": "changed_existing", + "platform_memory_id": existing_memory.get("id"), + "existing_hash": existing_hash, + "local_hash": local_hash, + "record": record, + } + ) + continue + + try: + response = platform_request_json("POST", f"{base_url}/v3/memories/add/", api_key=api_key, payload=payload) + imported += 1 + statuses[str(response.get("status", "unknown"))] += 1 + except MigrationError as exc: + failed += 1 + review_records.append({"status": "failed", "error": str(exc), "record": record}) + + summary = { + "input_records": len(records), + "valid_records": len(payloads), + "invalid": invalid, + "imported": imported, + "skipped_existing": skipped_existing, + "changed_existing": changed_existing, + "failed": failed, + "response_statuses": dict(statuses), + } + review_path = write_import_review_file(input_path, review_records, summary) + summary["review_path"] = str(review_path) if review_path else None + return summary + + +def print_import_summary(summary: Dict[str, Any]) -> None: + print_info(f"Imported: {summary['imported']}") + print_info(f"Skipped existing identical: {summary['skipped_existing']}") + print_info(f"Changed existing: {summary['changed_existing']}") + print_info(f"Invalid: {summary['invalid']}") + print_info(f"Failed: {summary['failed']}") + if summary.get("review_path"): + print_info(f"Review file: {summary['review_path']}") + + +def parse_args(argv: List[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser( + prog="oss-to-platform-migrate.sh", + description="Migrate Python OSS hosted-Qdrant memories to a Mem0 Platform account.", + ) + parser.add_argument("--auth-only", action="store_true", help="Run only auth/account resolution.") + parser.add_argument("--export-only", action="store_true", help="Run only Python OSS hosted-Qdrant JSON export.") + parser.add_argument("--import-only", action="store_true", help="Run only Platform import from a migration JSON file.") + parser.add_argument("--email", help="Email address for email-code authentication.") + parser.add_argument("--code", help="Verification code for non-interactive email-code authentication.") + parser.add_argument("--api-key", help="Platform API key to use for this run only.") + parser.add_argument("--base-url", help=f"Mem0 Platform API base URL. Defaults to {DEFAULT_BASE_URL}.") + parser.add_argument("--yes", action="store_true", help="Accept an existing valid Platform session without prompting.") + parser.add_argument("--qdrant-url", help="Hosted Qdrant URL for Python OSS memory export.") + parser.add_argument("--qdrant-api-key", help="Hosted Qdrant API key for this run only.") + parser.add_argument("--qdrant-collection", help=f"Qdrant collection name. Defaults to {DEFAULT_QDRANT_COLLECTION}.") + parser.add_argument("--qdrant-page-size", type=int, default=DEFAULT_EXPORT_PAGE_SIZE, help="Qdrant scroll page size.") + parser.add_argument("--user-id", help="Export memories matching this user_id.") + parser.add_argument("--agent-id", help="Export memories matching this agent_id.") + parser.add_argument("--run-id", help="Export memories matching this run_id.") + parser.add_argument("--all", action="store_true", help="Export the full Qdrant collection without a scope filter.") + parser.add_argument("--output", help="Path for the migration JSON export file.") + parser.add_argument("--input", help="Path to a migration JSON file to import into Platform.") + return parser.parse_args(argv) + + +def main(argv: List[str]) -> int: + args = parse_args(argv) + exclusive_modes = [args.auth_only, args.export_only, args.import_only] + if sum(1 for enabled in exclusive_modes if enabled) > 1: + print_error("Use only one of --auth-only, --export-only, or --import-only.") + return 1 + + path = config_path() + config = load_config(path) + oss_anon_id = ensure_oss_anon_id(path, config) + anon_ids = read_anon_ids(config, oss_anon_id) + base_url = api_base_url(args, config) + + print_info("Mem0 OSS to Platform migration") + if args.export_only: + print_info("Export phase") + elif args.import_only: + print_info("Import phase") + else: + print_info("Auth phase") + print_info("Supported environments: macOS, Linux, and Windows via WSL/Git Bash with bash and python3.") + if args.auth_only: + print_info("No memories are imported during this auth step.") + elif args.export_only: + print_info("No memories are imported to Platform during this step.") + else: + print_info("Memories will be imported to your authenticated Mem0 Platform account.") + print_info("") + + capture_migration_event("oss.migrate.started", anon_ids, additional={"base_url": base_url}) + + try: + email = None + api_key = None + output_path = None + export_count = 0 + import_summary: Optional[Dict[str, Any]] = None + + if args.import_only: + input_path = import_input_path(args) + api_key, key_source = configured_api_key(args, config) + if not api_key: + raise MigrationError("--import-only requires --api-key, MEM0_API_KEY, or a stored Platform API key.") + try: + email = validate_api_key(api_key, base_url) + except MigrationError: + if key_source in ("flag", "env"): + raise + email = None + if email: + capture_migration_event( + "oss.migrate.authenticated", + anon_ids, + email=email, + additional={"auth_method": f"{key_source}_api_key", "base_url": base_url}, + ) + stitch_identities(path, config, anon_ids, email) + print_info("Phase 1/1: Import memories into Mem0 Platform") + import_summary = import_platform_memories(input_path, api_key, base_url) + elif not args.export_only: + phase_label = "Phase 1/1" if args.auth_only else "Phase 1/3" + print_info(f"{phase_label}: Authenticate with Mem0 Platform") + api_key, email, auth_method = resolve_auth(args, config, base_url) + capture_migration_event( + "oss.migrate.authenticated", + anon_ids, + email=email, + additional={"auth_method": auth_method, "base_url": base_url}, + ) + stitch_identities(path, config, anon_ids, email) + print_info(f"Authenticated as {email}") + print_info("") + + if not args.auth_only: + if args.export_only: + print_info("Phase 1/1: Export Python OSS memories from hosted Qdrant") + output_path, export_count = export_qdrant_memories(args, anon_ids) + elif not args.import_only: + print_info("Phase 2/3: Export Python OSS memories from hosted Qdrant") + output_path, export_count = export_qdrant_memories(args, anon_ids) + print_info(f"Exported {export_count} memories to {output_path}") + print_info("") + print_info("Phase 3/3: Import memories into Mem0 Platform") + import_summary = import_platform_memories(output_path, api_key or "", base_url) + + if import_summary is not None: + capture_migration_event( + "oss.migrate.completed", + anon_ids, + email=email, + additional={ + "base_url": base_url, + "phase": "import", + "exported": export_count, + "imported": import_summary["imported"], + "skipped_existing": import_summary["skipped_existing"], + "changed_existing": import_summary["changed_existing"], + "invalid": import_summary["invalid"], + "failed": import_summary["failed"], + }, + ) + except MigrationError as exc: + capture_migration_event( + "oss.migrate.failed", + anon_ids, + email=email, + additional={"error": str(exc), "base_url": base_url}, + ) + print_error(str(exc)) + return 1 + + if args.auth_only: + print_info("Auth-only mode complete. Export and import were not run.") + elif args.export_only: + print_info(f"Exported {export_count} memories to {output_path}") + print_info("Export-only mode complete. Platform import was not run.") + elif args.import_only: + print_import_summary(import_summary or {}) + print_info("Import-only mode complete. Export was not run.") + else: + print_info("") + print_info(f"Export file: {output_path}") + print_info(f"Exported: {export_count}") + print_import_summary(import_summary or {}) + print_info("Migration complete.") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:])) +PY diff --git a/tests/test_oss_to_platform_migrate.py b/tests/test_oss_to_platform_migrate.py new file mode 100644 index 000000000..8e740bf76 --- /dev/null +++ b/tests/test_oss_to_platform_migrate.py @@ -0,0 +1,838 @@ +from __future__ import annotations + +import json +import os +import subprocess +import threading +from hashlib import sha256 +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Any +from urllib.parse import parse_qs, urlparse + + +SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "oss-to-platform-migrate.sh" + + +class MigrationHTTPServer: + def __init__( + self, + *, + ping_emails: dict[str, str] | None = None, + verify_api_key: str = "verified-key", + verify_status: int = 200, + qdrant_api_key: str = "qdrant-key", + qdrant_collection: str = "mem0", + qdrant_pages: list[dict[str, Any]] | None = None, + platform_memories: list[dict[str, Any]] | None = None, + ) -> None: + self.ping_emails = ping_emails or {} + self.verify_api_key = verify_api_key + self.verify_status = verify_status + self.qdrant_api_key = qdrant_api_key + self.qdrant_collection = qdrant_collection + self.qdrant_pages = qdrant_pages or [{"points": [], "next_page_offset": None}] + self.platform_memories = platform_memories or [] + self.requests: list[dict[str, Any]] = [] + self._server = ThreadingHTTPServer(("127.0.0.1", 0), self._handler()) + self.url = f"http://127.0.0.1:{self._server.server_port}" + self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) + + def __enter__(self) -> "MigrationHTTPServer": + self._thread.start() + return self + + def __exit__(self, *_exc: object) -> None: + self._server.shutdown() + self._server.server_close() + self._thread.join(timeout=5) + + def _handler(self) -> type[BaseHTTPRequestHandler]: + owner = self + + class Handler(BaseHTTPRequestHandler): + def log_message(self, _format: str, *_args: object) -> None: + return + + def _read_json(self) -> dict[str, Any]: + length = int(self.headers.get("Content-Length", "0")) + raw = self.rfile.read(length) if length else b"" + if not raw: + return {} + return json.loads(raw.decode("utf-8")) + + def _send_json(self, status: int, payload: dict[str, Any]) -> None: + body = json.dumps(payload).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def _record(self, body: dict[str, Any] | None = None) -> None: + owner.requests.append( + { + "method": self.command, + "path": self.path, + "headers": dict(self.headers), + "body": body or {}, + } + ) + + def do_GET(self) -> None: + self._record() + if self.path == "/v1/ping/": + auth = self.headers.get("Authorization", "") + token = auth.removeprefix("Token ") + email = owner.ping_emails.get(token) + if email: + self._send_json(200, {"user_email": email}) + else: + self._send_json(401, {"detail": "Invalid token"}) + return + if self.path == f"/collections/{owner.qdrant_collection}": + if self.headers.get("api-key") != owner.qdrant_api_key: + self._send_json(401, {"status": {"error": "unauthorized"}}) + else: + self._send_json(200, {"result": {"status": "green"}, "status": "ok"}) + return + self._send_json(404, {"detail": "Not found"}) + + def do_POST(self) -> None: + body = self._read_json() + self._record(body) + parsed = urlparse(self.path) + if self.path == "/posthog": + self._send_json(200, {"ok": True}) + return + if self.path == "/api/v1/auth/email_code/": + self._send_json(200, {"sent": True}) + return + if self.path == "/api/v1/auth/email_code/verify/": + if owner.verify_status != 200: + self._send_json(owner.verify_status, {"error": "bad verification code"}) + else: + self._send_json(200, {"api_key": owner.verify_api_key}) + return + if self.path == f"/collections/{owner.qdrant_collection}/points/scroll": + if self.headers.get("api-key") != owner.qdrant_api_key: + self._send_json(401, {"status": {"error": "unauthorized"}}) + return + offset = body.get("offset") + page_index = int(offset) if offset is not None else 0 + page = owner.qdrant_pages[page_index] + self._send_json( + 200, + { + "result": { + "points": page["points"], + "next_page_offset": page.get("next_page_offset"), + }, + "status": "ok", + }, + ) + return + if parsed.path == "/v3/memories/": + query = parse_qs(parsed.query) + page = int(query.get("page", ["1"])[0]) + page_size = int(query.get("page_size", ["100"])[0]) + filters = body.get("filters") if isinstance(body.get("filters"), dict) else {} + filtered = owner.platform_memories + for key in ("user_id", "agent_id", "run_id"): + if key in filters: + filtered = [memory for memory in filtered if memory.get(key) == filters[key]] + start = (page - 1) * page_size + end = start + page_size + page_results = filtered[start:end] + next_url = ( + f"{owner.url}/v3/memories/?page={page + 1}&page_size={page_size}" + if end < len(filtered) + else None + ) + self._send_json( + 200, + { + "count": len(filtered), + "next": next_url, + "previous": None, + "results": page_results, + }, + ) + return + if parsed.path == "/v3/memories/add/": + memory_id = f"platform-{len(owner.platform_memories) + 1}" + message = (body.get("messages") or [{}])[0] + memory = { + "id": memory_id, + "memory": message.get("content"), + "metadata": body.get("metadata"), + "user_id": body.get("user_id"), + "agent_id": body.get("agent_id"), + "run_id": body.get("run_id"), + } + owner.platform_memories.append(memory) + self._send_json( + 200, + { + "message": "Memories stored successfully", + "status": "SUCCEEDED", + "event_id": "event-1", + "results": [{"id": memory_id, "data": {"memory": message.get("content")}, "event": "ADD"}], + }, + ) + return + self._send_json(404, {"detail": "Not found"}) + + return Handler + + +def alias_marker(anon_id: str, email: str) -> str: + return sha256(f"{anon_id}\0{email}".encode("utf-8")).hexdigest() + + +def write_config(mem0_dir: Path, data: dict[str, Any]) -> None: + mem0_dir.mkdir(parents=True, exist_ok=True) + (mem0_dir / "config.json").write_text(json.dumps(data), encoding="utf-8") + + +def read_config(mem0_dir: Path) -> dict[str, Any]: + return json.loads((mem0_dir / "config.json").read_text(encoding="utf-8")) + + +def run_migration_script( + tmp_path: Path, + server: MigrationHTTPServer, + *args: str, + config: dict[str, Any] | None = None, + raw_config: str | None = None, +) -> tuple[subprocess.CompletedProcess[str], Path]: + mem0_dir = tmp_path / "mem0" + if config is not None: + write_config(mem0_dir, config) + if raw_config is not None: + mem0_dir.mkdir(parents=True, exist_ok=True) + (mem0_dir / "config.json").write_text(raw_config, encoding="utf-8") + + env = os.environ.copy() + env.update( + { + "MEM0_DIR": str(mem0_dir), + "MEM0_MIGRATE_TELEMETRY_URL": f"{server.url}/posthog", + } + ) + env.pop("MEM0_API_KEY", None) + env.pop("MEM0_BASE_URL", None) + + result = subprocess.run( + ["bash", str(SCRIPT), "--auth-only", "--base-url", server.url, *args], + capture_output=True, + text=True, + env=env, + start_new_session=True, + timeout=20, + check=False, + ) + return result, mem0_dir + + +def run_export_script( + tmp_path: Path, + server: MigrationHTTPServer, + *args: str, + config: dict[str, Any] | None = None, + qdrant_api_key: str = "qdrant-key", +) -> tuple[subprocess.CompletedProcess[str], Path, Path]: + mem0_dir = tmp_path / "mem0" + output_path = tmp_path / "export.json" + if config is not None: + write_config(mem0_dir, config) + + env = os.environ.copy() + env.update( + { + "MEM0_DIR": str(mem0_dir), + "MEM0_MIGRATE_TELEMETRY_URL": f"{server.url}/posthog", + "QDRANT_API_KEY": qdrant_api_key, + } + ) + env.pop("MEM0_API_KEY", None) + env.pop("MEM0_BASE_URL", None) + + result = subprocess.run( + [ + "bash", + str(SCRIPT), + "--export-only", + "--qdrant-url", + server.url, + "--qdrant-collection", + server.qdrant_collection, + "--output", + str(output_path), + *args, + ], + capture_output=True, + text=True, + env=env, + start_new_session=True, + timeout=20, + check=False, + ) + return result, mem0_dir, output_path + + +def run_import_script( + tmp_path: Path, + server: MigrationHTTPServer, + input_path: Path, + *args: str, + config: dict[str, Any] | None = None, + api_key: str = "import-key", +) -> tuple[subprocess.CompletedProcess[str], Path]: + mem0_dir = tmp_path / "mem0" + if config is not None: + write_config(mem0_dir, config) + + env = os.environ.copy() + env.update( + { + "MEM0_DIR": str(mem0_dir), + "MEM0_MIGRATE_TELEMETRY_URL": f"{server.url}/posthog", + "MEM0_API_KEY": api_key, + } + ) + env.pop("MEM0_BASE_URL", None) + + result = subprocess.run( + [ + "bash", + str(SCRIPT), + "--import-only", + "--base-url", + server.url, + "--input", + str(input_path), + *args, + ], + capture_output=True, + text=True, + env=env, + start_new_session=True, + timeout=20, + check=False, + ) + return result, mem0_dir + + +def run_full_script( + tmp_path: Path, + server: MigrationHTTPServer, + *args: str, + config: dict[str, Any] | None = None, + qdrant_api_key: str = "qdrant-key", +) -> tuple[subprocess.CompletedProcess[str], Path, Path]: + mem0_dir = tmp_path / "mem0" + output_path = tmp_path / "full-export.json" + if config is not None: + write_config(mem0_dir, config) + + env = os.environ.copy() + env.update( + { + "MEM0_DIR": str(mem0_dir), + "MEM0_MIGRATE_TELEMETRY_URL": f"{server.url}/posthog", + "QDRANT_API_KEY": qdrant_api_key, + } + ) + env.pop("MEM0_API_KEY", None) + env.pop("MEM0_BASE_URL", None) + + result = subprocess.run( + [ + "bash", + str(SCRIPT), + "--base-url", + server.url, + "--qdrant-url", + server.url, + "--qdrant-collection", + server.qdrant_collection, + "--output", + str(output_path), + *args, + ], + capture_output=True, + text=True, + env=env, + start_new_session=True, + timeout=20, + check=False, + ) + return result, mem0_dir, output_path + + +def posthog_events(server: MigrationHTTPServer) -> list[dict[str, Any]]: + return [request["body"] for request in server.requests if request["path"] == "/posthog"] + + +def test_existing_api_key_authenticates_and_stitches_ids(tmp_path: Path) -> None: + config = { + "user_id": "oss-123", + "platform": {"api_key": "stored-key", "base_url": "https://api.mem0.ai"}, + "telemetry": {"anonymous_id": "cli-456"}, + } + + with MigrationHTTPServer(ping_emails={"stored-key": "bob@example.com"}) as server: + result, mem0_dir = run_migration_script(tmp_path, server, "--yes", config=config) + + assert result.returncode == 0, result.stderr + assert "Authenticated as bob@example.com" in result.stdout + assert not any(request["path"] == "/api/v1/auth/email_code/verify/" for request in server.requests) + + updated = read_config(mem0_dir) + assert updated["platform"] == config["platform"] + assert alias_marker("oss-123", "bob@example.com") in updated["telemetry"]["aliased_pairs"] + assert alias_marker("cli-456", "bob@example.com") in updated["telemetry"]["aliased_pairs"] + + events = posthog_events(server) + event_names = [event["event"] for event in events] + assert "oss.migrate.started" in event_names + assert "oss.migrate.authenticated" in event_names + assert event_names.count("$identify") == 2 + + authenticated = next(event for event in events if event["event"] == "oss.migrate.authenticated") + assert authenticated["distinct_id"] == "bob@example.com" + assert authenticated["properties"]["local_anonymous_id"] == "oss-123" + assert authenticated["properties"]["authenticated_email"] == "bob@example.com" + + +def test_email_code_authenticates_without_persisting_credentials(tmp_path: Path) -> None: + with MigrationHTTPServer(ping_emails={"verified-key": "alice@example.com"}) as server: + result, mem0_dir = run_migration_script( + tmp_path, + server, + "--email", + "Alice@Example.COM", + "--code", + "123456", + ) + + assert result.returncode == 0, result.stderr + assert "Authenticated as alice@example.com" in result.stdout + + verify_request = next(request for request in server.requests if request["path"] == "/api/v1/auth/email_code/verify/") + assert verify_request["body"] == {"email": "alice@example.com", "code": "123456"} + assert not any(request["path"] == "/api/v1/auth/email_code/" for request in server.requests) + + updated = read_config(mem0_dir) + assert "api_key" not in updated.get("platform", {}) + assert "user_email" not in updated.get("platform", {}) + assert updated["user_id"] + assert alias_marker(updated["user_id"], "alice@example.com") in updated["telemetry"]["aliased_pairs"] + + events = posthog_events(server) + assert [event["event"] for event in events].count("$identify") == 1 + authenticated = next(event for event in events if event["event"] == "oss.migrate.authenticated") + assert authenticated["properties"]["auth_method"] == "email_code" + + +def test_invalid_stored_key_falls_back_to_email_code(tmp_path: Path) -> None: + config = { + "user_id": "oss-fallback", + "platform": {"api_key": "bad-key", "base_url": "https://api.mem0.ai"}, + } + + with MigrationHTTPServer(ping_emails={"verified-key": "new@example.com"}) as server: + result, mem0_dir = run_migration_script( + tmp_path, + server, + "--email", + "new@example.com", + "--code", + "123456", + config=config, + ) + + assert result.returncode == 0, result.stderr + assert "Stored Mem0 Platform API key is invalid or expired" in result.stdout + assert "Authenticated as new@example.com" in result.stdout + + ping_tokens = [ + request["headers"]["Authorization"].removeprefix("Token ") + for request in server.requests + if request["path"] == "/v1/ping/" + ] + assert ping_tokens == ["bad-key", "verified-key"] + + updated = read_config(mem0_dir) + assert updated["platform"] == config["platform"] + assert alias_marker("oss-fallback", "new@example.com") in updated["telemetry"]["aliased_pairs"] + + +def test_email_code_failure_reports_failed_telemetry(tmp_path: Path) -> None: + with MigrationHTTPServer(verify_status=400) as server: + result, mem0_dir = run_migration_script( + tmp_path, + server, + "--email", + "fail@example.com", + "--code", + "bad", + ) + + assert result.returncode == 1 + assert "Verification failed: bad verification code" in result.stderr + + updated = read_config(mem0_dir) + assert "telemetry" not in updated or "aliased_pairs" not in updated["telemetry"] + + events = posthog_events(server) + event_names = [event["event"] for event in events] + assert "oss.migrate.started" in event_names + assert "oss.migrate.failed" in event_names + assert "$identify" not in event_names + failed = next(event for event in events if event["event"] == "oss.migrate.failed") + assert "Verification failed" in failed["properties"]["error"] + + +def test_malformed_config_does_not_crash_and_authenticates(tmp_path: Path) -> None: + with MigrationHTTPServer(ping_emails={"verified-key": "malformed@example.com"}) as server: + result, mem0_dir = run_migration_script( + tmp_path, + server, + "--email", + "malformed@example.com", + "--code", + "123456", + raw_config="{not valid json", + ) + + assert result.returncode == 0, result.stderr + assert "Authenticated as malformed@example.com" in result.stdout + + updated = read_config(mem0_dir) + assert updated["user_id"] + assert alias_marker(updated["user_id"], "malformed@example.com") in updated["telemetry"]["aliased_pairs"] + + +def test_weird_telemetry_shape_does_not_crash(tmp_path: Path) -> None: + config = {"user_id": "oss-weird-telemetry", "telemetry": "not-an-object"} + + with MigrationHTTPServer(ping_emails={"verified-key": "weird@example.com"}) as server: + result, mem0_dir = run_migration_script( + tmp_path, + server, + "--email", + "weird@example.com", + "--code", + "123456", + config=config, + ) + + assert result.returncode == 0, result.stderr + updated = read_config(mem0_dir) + assert isinstance(updated["telemetry"], dict) + assert alias_marker("oss-weird-telemetry", "weird@example.com") in updated["telemetry"]["aliased_pairs"] + + +def test_missing_python3_prints_clear_shell_error(tmp_path: Path) -> None: + env = os.environ.copy() + env["PATH"] = str(tmp_path) + + result = subprocess.run( + ["/bin/bash", str(SCRIPT), "--help"], + capture_output=True, + text=True, + env=env, + timeout=20, + check=False, + ) + + assert result.returncode == 1 + assert "python3 is required to run the Mem0 migration" in result.stderr + + +def test_curl_piped_help_works() -> None: + result = subprocess.run( + ["bash", "-c", f"curl -fsSL file://{SCRIPT} | bash -s -- --help"], + capture_output=True, + text=True, + timeout=20, + check=False, + ) + + assert result.returncode == 0, result.stderr + assert "Migrate Python OSS hosted-Qdrant memories" in result.stdout + + +def test_export_qdrant_memories_to_json_without_vectors_or_api_key(tmp_path: Path) -> None: + pages = [ + { + "points": [ + { + "id": "point-1", + "vector": [0.1, 0.2], + "payload": { + "data": "User likes dark mode", + "hash": "hash-1", + "created_at": "2026-05-01T00:00:00Z", + "updated_at": "2026-05-01T00:00:00Z", + "user_id": "alice", + "agent_id": "agent-1", + "run_id": "run-1", + "actor_id": "actor-1", + "role": "user", + "topic": "preferences", + "text_lemmatized": "user like dark mode", + }, + } + ], + "next_page_offset": 1, + }, + { + "points": [ + { + "id": "point-2", + "vector": [0.3, 0.4], + "payload": { + "data": "User prefers concise answers", + "hash": "hash-2", + "user_id": "alice", + "metadata_note": "extra", + }, + } + ], + "next_page_offset": None, + }, + ] + + with MigrationHTTPServer(qdrant_pages=pages) as server: + result, _mem0_dir, output_path = run_export_script( + tmp_path, + server, + "--user-id", + "alice", + "--qdrant-page-size", + "1", + config={"user_id": "oss-export-user"}, + ) + + assert result.returncode == 0, result.stderr + assert "Exported 2 memories" in result.stdout + + artifact = json.loads(output_path.read_text(encoding="utf-8")) + assert artifact["kind"] == "mem0_oss_qdrant_export" + assert artifact["source"]["sdk"] == "python" + assert artifact["source"]["vector_store"] == "qdrant" + assert artifact["source"]["storage"] == "hosted" + assert artifact["source"]["filters"]["user_id"] == "alice" + assert artifact["record_count"] == 2 + assert artifact["local_anonymous_id"] == "oss-export-user" + + first = artifact["records"][0] + assert first["id"] == "point-1" + assert first["memory"] == "User likes dark mode" + assert first["hash"] == "hash-1" + assert first["user_id"] == "alice" + assert first["agent_id"] == "agent-1" + assert first["run_id"] == "run-1" + assert first["actor_id"] == "actor-1" + assert first["role"] == "user" + assert first["metadata"] == {"topic": "preferences"} + assert all("vector" not in record for record in artifact["records"]) + assert "qdrant-key" not in output_path.read_text(encoding="utf-8") + + scroll_requests = [request for request in server.requests if request["path"].endswith("/points/scroll")] + assert len(scroll_requests) == 2 + assert scroll_requests[0]["body"]["with_vector"] is False + assert scroll_requests[0]["body"]["filter"] == {"must": [{"key": "user_id", "match": {"value": "alice"}}]} + assert scroll_requests[1]["body"]["offset"] == 1 + + +def test_export_requires_scope_or_all(tmp_path: Path) -> None: + with MigrationHTTPServer() as server: + result, _mem0_dir, output_path = run_export_script(tmp_path, server) + + assert result.returncode == 1 + assert "Export requires --user-id, --agent-id, --run-id, or --all" in result.stderr + assert not output_path.exists() + assert not any(request["path"].endswith("/points/scroll") for request in server.requests) + + +def test_export_all_uses_no_qdrant_filter(tmp_path: Path) -> None: + with MigrationHTTPServer(qdrant_pages=[{"points": [], "next_page_offset": None}]) as server: + result, _mem0_dir, output_path = run_export_script(tmp_path, server, "--all") + + assert result.returncode == 0, result.stderr + artifact = json.loads(output_path.read_text(encoding="utf-8")) + assert artifact["record_count"] == 0 + assert artifact["records"] == [] + + scroll_request = next(request for request in server.requests if request["path"].endswith("/points/scroll")) + assert "filter" not in scroll_request["body"] + + +def test_export_invalid_qdrant_credentials_fail_clearly(tmp_path: Path) -> None: + with MigrationHTTPServer(qdrant_api_key="correct-key") as server: + result, _mem0_dir, output_path = run_export_script( + tmp_path, + server, + "--user-id", + "alice", + qdrant_api_key="wrong-key", + ) + + assert result.returncode == 1 + assert "Qdrant authentication failed" in result.stderr + assert not output_path.exists() + + +def test_import_platform_memories_from_export_json(tmp_path: Path) -> None: + input_path = tmp_path / "import.json" + input_path.write_text( + json.dumps( + { + "source": {"sdk": "python", "vector_store": "qdrant", "collection": "mem0_test"}, + "records": [ + { + "id": "local-1", + "memory": "User likes barbecue", + "hash": "hash-1", + "created_at": "2026-05-07T00:00:00Z", + "user_id": "alice", + "metadata": {"topic": "food"}, + } + ], + } + ), + encoding="utf-8", + ) + + with MigrationHTTPServer(ping_emails={"import-key": "alice@example.com"}) as server: + result, _mem0_dir = run_import_script(tmp_path, server, input_path) + + assert result.returncode == 0, result.stderr + assert "Imported: 1" in result.stdout + assert "Failed: 0" in result.stdout + + add_request = next(request for request in server.requests if request["path"] == "/v3/memories/add/") + body = add_request["body"] + assert body["messages"] == [{"role": "user", "content": "User likes barbecue"}] + assert body["user_id"] == "alice" + assert body["infer"] is False + assert body["source"] == "migration" + assert body["timestamp"] == 1778112000 + assert body["metadata"]["topic"] == "food" + assert body["metadata"]["mem0_migration_source"] == "python_oss_qdrant" + assert body["metadata"]["mem0_migration_collection"] == "mem0_test" + assert body["metadata"]["mem0_migration_local_id"] == "local-1" + assert body["metadata"]["mem0_migration_local_hash"] == "hash-1" + + events = posthog_events(server) + assert "oss.migrate.completed" in [event["event"] for event in events] + + +def test_import_skips_existing_identical_memory(tmp_path: Path) -> None: + input_path = tmp_path / "import.json" + source = {"sdk": "python", "vector_store": "qdrant", "collection": "mem0_test"} + record = {"id": "local-1", "memory": "User likes barbecue", "hash": "hash-1", "user_id": "alice"} + input_path.write_text(json.dumps({"source": source, "records": [record]}), encoding="utf-8") + import_key = sha256("python:qdrant:mem0_test:local-1".encode("utf-8")).hexdigest() + + existing = [ + { + "id": "platform-1", + "memory": "User likes barbecue", + "user_id": "alice", + "metadata": { + "mem0_migration_import_key": import_key, + "mem0_migration_local_hash": "hash-1", + }, + } + ] + + with MigrationHTTPServer(ping_emails={"import-key": "alice@example.com"}, platform_memories=existing) as server: + result, _mem0_dir = run_import_script(tmp_path, server, input_path) + + assert result.returncode == 0, result.stderr + assert "Imported: 0" in result.stdout + assert "Skipped existing identical: 1" in result.stdout + assert "Changed existing: 0" in result.stdout + assert not any(request["path"] == "/v3/memories/add/" for request in server.requests) + + +def test_import_reports_changed_existing_without_update_or_add(tmp_path: Path) -> None: + input_path = tmp_path / "import.json" + source = {"sdk": "python", "vector_store": "qdrant", "collection": "mem0_test"} + record = {"id": "local-1", "memory": "User likes brisket", "hash": "hash-new", "user_id": "alice"} + input_path.write_text(json.dumps({"source": source, "records": [record]}), encoding="utf-8") + import_key = sha256("python:qdrant:mem0_test:local-1".encode("utf-8")).hexdigest() + + existing = [ + { + "id": "platform-1", + "memory": "User likes barbecue", + "user_id": "alice", + "metadata": { + "mem0_migration_import_key": import_key, + "mem0_migration_local_hash": "hash-old", + }, + } + ] + + with MigrationHTTPServer(ping_emails={"import-key": "alice@example.com"}, platform_memories=existing) as server: + result, _mem0_dir = run_import_script(tmp_path, server, input_path) + + assert result.returncode == 0, result.stderr + assert "Imported: 0" in result.stdout + assert "Skipped existing identical: 0" in result.stdout + assert "Changed existing: 1" in result.stdout + assert not any(request["path"] == "/v3/memories/add/" for request in server.requests) + review_path_line = next(line for line in result.stdout.splitlines() if line.startswith("Review file: ")) + review_path = Path(review_path_line.removeprefix("Review file: ")) + review = json.loads(review_path.read_text(encoding="utf-8")) + assert review["records"][0]["status"] == "changed_existing" + assert review["records"][0]["platform_memory_id"] == "platform-1" + + +def test_full_flow_auth_export_and_imports_memories(tmp_path: Path) -> None: + qdrant_pages = [ + { + "points": [ + { + "id": "point-1", + "payload": { + "data": "User likes barbecue", + "hash": "hash-1", + "created_at": "2026-05-07T00:00:00Z", + "user_id": "alice", + }, + } + ], + "next_page_offset": None, + } + ] + + with MigrationHTTPServer(ping_emails={"verified-key": "alice@example.com"}, qdrant_pages=qdrant_pages) as server: + result, _mem0_dir, output_path = run_full_script( + tmp_path, + server, + "--email", + "alice@example.com", + "--code", + "123456", + "--user-id", + "alice", + ) + + assert result.returncode == 0, result.stderr + assert "Phase 1/3: Authenticate with Mem0 Platform" in result.stdout + assert "Phase 2/3: Export Python OSS memories from hosted Qdrant" in result.stdout + assert "Phase 3/3: Import memories into Mem0 Platform" in result.stdout + assert "Imported: 1" in result.stdout + assert output_path.exists() + assert any(request["path"] == "/v3/memories/add/" for request in server.requests) + events = posthog_events(server) + event_names = [event["event"] for event in events] + assert "oss.migrate.authenticated" in event_names + assert "oss.migrate.completed" in event_names