From a623cfaf76ae7379a58be1e837f8a88a9b15a184 Mon Sep 17 00:00:00 2001 From: youneshima Date: Thu, 7 May 2026 19:34:09 -0700 Subject: [PATCH] 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