OSS to Platform Migration

Copypromoted_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

Copyartifact = {
    "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)

Copyconstructor(config: NeptuneAnalyticsConfig) {
  this.graphIdentifier = this.resolveGraphIdentifier(config);
  this.collectionName = config.collectionName || "memories";
  this.collectionLabel = `MEM0_VECTOR_${this.collectionName}`;
  this.collectionLabelExpr = this.escapeLabel(this.collectionLabel);
  this.userLabel = "MEM0_VECTOR_memory_migrations";
  this.userLabelExpr = this.escapeLabel(this.userLabel);
  this.userNodeId = "mem0-user";
  this.dimension = config.dimension || 1536;
  this.clientConfig = this.buildClientConfig(config);
  this.clientOverride = config.client;
 

Copydef 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)]
 

Copydef 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()
 

Copydef 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}
 

Copydef 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
 

Copyexport function createProvider(
  cfg: Mem0Config,
  api: OpenClawPluginApi,
): Mem0Provider {
  if (cfg.mode === "open-source") {
    // v3.0.0: use customInstructions (was customPrompt)
    return new OSSProvider(cfg.oss, cfg.customInstructions, (p) =>
      api.resolvePath(p),
    );
  }
 
  return new PlatformProvider(cfg.apiKey!, cfg.baseUrl);
}

Copyconst provider = createProvider(cfg, api);
const lifecycle = createMemoryLifecycle();
lifecycle.beginSession();
 
// Create Backend instance — PlatformBackend for platform mode, providerToBackend adapter for OSS
let backend: Backend;
if (cfg.mode === "platform") {
  backend = new PlatformBackend({
    apiKey: cfg.apiKey!,
    baseUrl: cfg.baseUrl ?? "https://api.mem0.ai",
  });
} else {
  backend = providerToBackend(provider, cfg.userId);
}

Copyexport function providerToBackend(
  provider: Mem0Provider,
  userId: string,
): Backend {
  return {
    async add(content, messages, opts = {}) {
      const msgs = messages ?? (content ? [{ role: "user", content }] : []);
      // v3.0.0: removed immutable, expiration_date
      const result = await provider.add(
        msgs as Array<{ role: string; content: string }>,
        {
          user_id: opts.userId ?? userId,
          source: "OPENCLAW",
          ...(opts.runId && { run_id: opts.runId }),
          ...(opts.metadata && { metadata: opts.metadata }),
          ...(opts.infer === false && { infer: false }),
        },
      );
      return result as unknown as Record<string, unknown>;
    },
 
    async search(query, opts = {}) {
      // v3.0.0: removed keyword_search, reranking
      const results = await provider.search(query, {
        user_id: opts.userId ?? userId,
        top_k: opts.topK,
        threshold: opts.threshold,
        filters: opts.filters,
        source: "OPENCLAW",
      });
      return results as unknown as Record<string, unknown>[];
    },
 
    async get(memoryId) {
      const item = await provider.get(memoryId);
      return item as unknown as Record<string, unknown>;
    },
 
    async listMemories(opts = {}) {
      const items = await provider.getAll({
        user_id: opts.userId ?? userId,
        page_size: opts.pageSize,
        source: "OPENCLAW",
      });
      return items as unknown as Record<string, unknown>[];
    },
 
    async update(memoryId, content, metadata) {
      if (content) await provider.update(memoryId, content);
      if (metadata) {
        // OSS provider doesn't support metadata-only updates — log warning
        console.warn(
          "providerToBackend: metadata updates are not supported in OSS mode, only text updates are applied",
        );
      }
      return { id: memoryId, updated: true };
    },
 
    async delete(memoryId, opts = {}) {
      if (opts.all) {
        await provider.deleteAll(opts.userId ?? userId);
        return { deleted: "all" };
      }
      if (memoryId) {
        await provider.delete(memoryId);
        return { deleted: memoryId };
      }
      throw new Error("Either memoryId or all is required");
    },
 
    async deleteEntities() {
      throw new Error("Entity management is only available in platform mode.");
    },
    async status() {
      return { connected: true, backend: "oss" };
    },
    async entities() {
      throw new Error("Entity management is only available in platform mode.");
    },
    async listEvents() {

Copydef 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
 

Copydef 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}