chore(deps): update deps and harden security (#825)

Signed-off-by: phernandez <paul@basicmachines.co>

src/basic_memory/api/v2/routers/importer_router.py

@@ -10,6 +10,7 @@ import logging
 from fastapi import APIRouter, Form, HTTPException, UploadFile, status, Path
 
 from basic_memory.deps import (
+    AppConfigDep,
     ChatGPTImporterV2ExternalDep,
     ClaudeConversationsImporterV2ExternalDep,
     ClaudeProjectsImporterV2ExternalDep,
@@ -27,9 +28,21 @@ logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/import", tags=["import-v2"])
 
 
+async def read_import_upload(file: UploadFile, max_bytes: int) -> bytes:
+    """Read an import upload with a hard cap before JSON parsing."""
+    content = await file.read(max_bytes + 1)
+    if len(content) > max_bytes:
+        raise HTTPException(
+            status_code=status.HTTP_413_CONTENT_TOO_LARGE,
+            detail=f"Import file exceeds maximum size of {max_bytes} bytes.",
+        )
+    return content
+
+
 @router.post("/chatgpt", response_model=ChatImportResult)
 async def import_chatgpt(
     importer: ChatGPTImporterV2ExternalDep,
+    config: AppConfigDep,
     file: UploadFile,
     project_id: str = Path(..., description="Project external UUID"),
     directory: str = Form("conversations"),
@@ -49,12 +62,13 @@ async def import_chatgpt(
         HTTPException: If import fails.
     """
     logger.info(f"V2 Importing ChatGPT conversations for project {project_id}")
-    return await import_file(importer, file, directory)
+    return await import_file(importer, file, directory, config.import_upload_max_bytes)
 
 
 @router.post("/claude/conversations", response_model=ChatImportResult)
 async def import_claude_conversations(
     importer: ClaudeConversationsImporterV2ExternalDep,
+    config: AppConfigDep,
     file: UploadFile,
     project_id: str = Path(..., description="Project external UUID"),
     directory: str = Form("conversations"),
@@ -74,12 +88,13 @@ async def import_claude_conversations(
         HTTPException: If import fails.
     """
     logger.info(f"V2 Importing Claude conversations for project {project_id}")
-    return await import_file(importer, file, directory)
+    return await import_file(importer, file, directory, config.import_upload_max_bytes)
 
 
 @router.post("/claude/projects", response_model=ProjectImportResult)
 async def import_claude_projects(
     importer: ClaudeProjectsImporterV2ExternalDep,
+    config: AppConfigDep,
     file: UploadFile,
     project_id: str = Path(..., description="Project external UUID"),
     directory: str = Form("projects"),
@@ -99,12 +114,13 @@ async def import_claude_projects(
         HTTPException: If import fails.
     """
     logger.info(f"V2 Importing Claude projects for project {project_id}")
-    return await import_file(importer, file, directory)
+    return await import_file(importer, file, directory, config.import_upload_max_bytes)
 
 
 @router.post("/memory-json", response_model=EntityImportResult)
 async def import_memory_json(
     importer: MemoryJsonImporterV2ExternalDep,
+    config: AppConfigDep,
     file: UploadFile,
     project_id: str = Path(..., description="Project external UUID"),
     directory: str = Form("conversations"),
@@ -126,7 +142,7 @@ async def import_memory_json(
     logger.info(f"V2 Importing memory.json for project {project_id}")
     try:
         file_data = []
-        file_bytes = await file.read()
+        file_bytes = await read_import_upload(file, config.import_upload_max_bytes)
         file_str = file_bytes.decode("utf-8")
         for line in file_str.splitlines():
             json_data = json.loads(line)
@@ -138,6 +154,8 @@ async def import_memory_json(
                 status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                 detail=result.error_message or "Import failed",
             )
+    except HTTPException:
+        raise
     except Exception as e:
         logger.exception("V2 Import failed")
         raise HTTPException(
@@ -147,13 +165,16 @@ async def import_memory_json(
     return result
 
 
-async def import_file(importer: Importer, file: UploadFile, destination_directory: str):
+async def import_file(
+    importer: Importer, file: UploadFile, destination_directory: str, max_bytes: int
+):
     """Helper function to import a file using an importer instance.
 
     Args:
         importer: The importer instance to use
         file: The file to import
         destination_directory: Destination directory for imported content
+        max_bytes: Maximum upload size in bytes; raises HTTP 413 if exceeded
 
     Returns:
         Import result from the importer
@@ -163,7 +184,8 @@ async def import_file(importer: Importer, file: UploadFile, destination_director
     """
     try:
         # Process file
-        json_data = json.load(file.file)
+        upload_bytes = await read_import_upload(file, max_bytes)
+        json_data = json.loads(upload_bytes)
         result = await importer.import_data(json_data, destination_directory)
         if not result.success:  # pragma: no cover
             raise HTTPException(
@@ -173,6 +195,8 @@ async def import_file(importer: Importer, file: UploadFile, destination_director
 
         return result
 
+    except HTTPException:
+        raise
     except Exception as e:
         logger.exception("V2 Import failed")
         raise HTTPException(

src/basic_memory/cli/commands/cloud/rclone_installer.py

@@ -4,7 +4,7 @@ import os
 import platform
 import shutil
 import subprocess
-from typing import Optional
+from typing import Any, Optional, cast
 
 from rich.console import Console
 
@@ -53,7 +53,9 @@ def run_command(command: list[str], check: bool = True) -> subprocess.CompletedP
 
 
 def install_rclone_macos() -> None:
-    """Install rclone on macOS using Homebrew or official script."""
+    """Install rclone on macOS using package managers."""
+    install_errors: list[str] = []
+
     # Try Homebrew first
     if shutil.which("brew"):
         try:
@@ -61,35 +63,37 @@ def install_rclone_macos() -> None:
             run_command(["brew", "install", "rclone"])
             console.print("[green]rclone installed via Homebrew[/green]")
             return
-        except RcloneInstallError:
-            console.print(
-                "[yellow]Homebrew installation failed, trying official script...[/yellow]"
-            )
+        except RcloneInstallError as exc:
+            install_errors.append(f"Homebrew failed: {exc}")
+            console.print("[yellow]Homebrew installation failed, trying MacPorts...[/yellow]")
 
-    # Fallback to official script
-    console.print("[blue]Installing rclone via official script...[/blue]")
-    try:
-        run_command(["sh", "-c", "curl https://rclone.org/install.sh | sudo bash"])
-        console.print("[green]rclone installed via official script[/green]")
-    except RcloneInstallError:
-        raise RcloneInstallError(
-            "Failed to install rclone. Please install manually: brew install rclone"
-        )
+    if shutil.which("port"):
+        try:
+            console.print("[blue]Installing rclone via MacPorts...[/blue]")
+            run_command(["sudo", "port", "install", "rclone"])
+            console.print("[green]rclone installed via MacPorts[/green]")
+            return
+        except RcloneInstallError as exc:
+            install_errors.append(f"MacPorts failed: {exc}")
+            console.print("[yellow]MacPorts installation failed[/yellow]")
+
+    details = "\n".join(f"- {error}" for error in install_errors)
+    if details:
+        details = f"\n\nAttempts:\n{details}"
+    raise RcloneInstallError(
+        "Could not install rclone automatically with an available package manager.\n\n"
+        "Install rclone manually with one of:\n"
+        "  brew install rclone\n"
+        "  sudo port install rclone\n"
+        "  Download from https://rclone.org/downloads/ and add rclone to PATH"
+        f"{details}"
+    )
 
 
 def install_rclone_linux() -> None:
-    """Install rclone on Linux using package managers or official script."""
-    # Try snap first (most universal)
-    if shutil.which("snap"):
-        try:
-            console.print("[blue]Installing rclone via snap...[/blue]")
-            run_command(["sudo", "snap", "install", "rclone"])
-            console.print("[green]rclone installed via snap[/green]")
-            return
-        except RcloneInstallError:
-            console.print("[yellow]Snap installation failed, trying apt...[/yellow]")
+    """Install rclone on Linux using package managers."""
+    install_errors: list[str] = []
 
-    # Try apt (Debian/Ubuntu)
     if shutil.which("apt"):
         try:
             console.print("[blue]Installing rclone via apt...[/blue]")
@@ -97,18 +101,75 @@ def install_rclone_linux() -> None:
             run_command(["sudo", "apt", "install", "-y", "rclone"])
             console.print("[green]rclone installed via apt[/green]")
             return
-        except RcloneInstallError:
-            console.print("[yellow]apt installation failed, trying official script...[/yellow]")
+        except RcloneInstallError as exc:
+            install_errors.append(f"apt failed: {exc}")
+            console.print("[yellow]apt installation failed, trying dnf...[/yellow]")
 
-    # Fallback to official script
-    console.print("[blue]Installing rclone via official script...[/blue]")
-    try:
-        run_command(["sh", "-c", "curl https://rclone.org/install.sh | sudo bash"])
-        console.print("[green]rclone installed via official script[/green]")
-    except RcloneInstallError:
-        raise RcloneInstallError(
-            "Failed to install rclone. Please install manually: sudo snap install rclone"
-        )
+    if shutil.which("dnf"):
+        try:
+            console.print("[blue]Installing rclone via dnf...[/blue]")
+            run_command(["sudo", "dnf", "install", "-y", "rclone"])
+            console.print("[green]rclone installed via dnf[/green]")
+            return
+        except RcloneInstallError as exc:
+            install_errors.append(f"dnf failed: {exc}")
+            console.print("[yellow]dnf installation failed, trying yum...[/yellow]")
+
+    if shutil.which("yum"):
+        try:
+            console.print("[blue]Installing rclone via yum...[/blue]")
+            run_command(["sudo", "yum", "install", "-y", "rclone"])
+            console.print("[green]rclone installed via yum[/green]")
+            return
+        except RcloneInstallError as exc:
+            install_errors.append(f"yum failed: {exc}")
+            console.print("[yellow]yum installation failed, trying pacman...[/yellow]")
+
+    if shutil.which("pacman"):
+        try:
+            console.print("[blue]Installing rclone via pacman...[/blue]")
+            run_command(["sudo", "pacman", "-S", "--noconfirm", "rclone"])
+            console.print("[green]rclone installed via pacman[/green]")
+            return
+        except RcloneInstallError as exc:
+            install_errors.append(f"pacman failed: {exc}")
+            console.print("[yellow]pacman installation failed, trying zypper...[/yellow]")
+
+    if shutil.which("zypper"):
+        try:
+            console.print("[blue]Installing rclone via zypper...[/blue]")
+            run_command(["sudo", "zypper", "--non-interactive", "install", "rclone"])
+            console.print("[green]rclone installed via zypper[/green]")
+            return
+        except RcloneInstallError as exc:
+            install_errors.append(f"zypper failed: {exc}")
+            console.print("[yellow]zypper installation failed, trying snap...[/yellow]")
+
+    if shutil.which("snap"):
+        try:
+            console.print("[blue]Installing rclone via snap...[/blue]")
+            run_command(["sudo", "snap", "install", "rclone"])
+            console.print("[green]rclone installed via snap[/green]")
+            return
+        except RcloneInstallError as exc:
+            install_errors.append(f"snap failed: {exc}")
+            console.print("[yellow]snap installation failed[/yellow]")
+
+    details = "\n".join(f"- {error}" for error in install_errors)
+    if details:
+        details = f"\n\nAttempts:\n{details}"
+    raise RcloneInstallError(
+        "Could not install rclone automatically with an available package manager.\n\n"
+        "Install rclone manually with one of your OS package managers, for example:\n"
+        "  sudo apt install rclone\n"
+        "  sudo dnf install rclone\n"
+        "  sudo yum install rclone\n"
+        "  sudo pacman -S rclone\n"
+        "  sudo zypper install rclone\n"
+        "  sudo snap install rclone\n"
+        "Or download from https://rclone.org/downloads/ and add rclone to PATH"
+        f"{details}"
+    )
 
 
 def install_rclone_windows() -> None:
@@ -209,27 +270,38 @@ def refresh_windows_path() -> None:
     if platform.system().lower() != "windows":
         return
 
-    # Importing here after performing platform detection. Also note that we have to ignore pylance/pyright
-    # warnings about winreg attributes so that "errors" don't appear on non-Windows platforms.
+    # Importing here after performing platform detection. Non-Windows type checkers may still
+    # resolve a stub without registry members, so keep this platform-only module dynamic here.
     import winreg
 
+    winreg_module = cast(Any, winreg)
     user_key_path = r"Environment"
     system_key_path = r"System\CurrentControlSet\Control\Session Manager\Environment"
     new_path = ""
 
     # Read user PATH
     try:
-        reg_key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, user_key_path, 0, winreg.KEY_READ)  # type: ignore[reportAttributeAccessIssue]
-        user_path, _ = winreg.QueryValueEx(reg_key, "PATH")  # type: ignore[reportAttributeAccessIssue]
-        winreg.CloseKey(reg_key)  # type: ignore[reportAttributeAccessIssue]
+        reg_key = winreg_module.OpenKey(
+            winreg_module.HKEY_CURRENT_USER,
+            user_key_path,
+            0,
+            winreg_module.KEY_READ,
+        )
+        user_path, _ = winreg_module.QueryValueEx(reg_key, "PATH")
+        winreg_module.CloseKey(reg_key)
     except Exception:
         user_path = ""
 
     # Read system PATH
     try:
-        reg_key = winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, system_key_path, 0, winreg.KEY_READ)  # type: ignore[reportAttributeAccessIssue]
-        system_path, _ = winreg.QueryValueEx(reg_key, "PATH")  # type: ignore[reportAttributeAccessIssue]
-        winreg.CloseKey(reg_key)  # type: ignore[reportAttributeAccessIssue]
+        reg_key = winreg_module.OpenKey(
+            winreg_module.HKEY_LOCAL_MACHINE,
+            system_key_path,
+            0,
+            winreg_module.KEY_READ,
+        )
+        system_path, _ = winreg_module.QueryValueEx(reg_key, "PATH")
+        winreg_module.CloseKey(reg_key)
     except Exception:
         system_path = ""
 

src/basic_memory/cli/commands/cloud/workspace.py

@@ -100,9 +100,7 @@ def set_default_workspace(
         raise typer.Exit(1)
 
     if len(matches) > 1:
-        console.print(
-            f"[red]Error: Workspace '{identifier}' matches multiple workspaces.[/red]"
-        )
+        console.print(f"[red]Error: Workspace '{identifier}' matches multiple workspaces.[/red]")
         console.print(
             "[dim]Choose one of these matching workspaces by slug:\n"
             f"{format_workspace_selection_choices(matches)}[/dim]"

src/basic_memory/cli/commands/project.py

@@ -300,9 +300,7 @@ def _resolve_workspace_id(config, workspace: str | None) -> str | None:
                 console.print(f"[dim]Available:\n{format_workspace_choices(workspaces)}[/dim]")
             raise typer.Exit(1)
         if len(matches) > 1:
-            console.print(
-                f"[red]Error: Workspace '{workspace}' matches multiple workspaces.[/red]"
-            )
+            console.print(f"[red]Error: Workspace '{workspace}' matches multiple workspaces.[/red]")
             console.print(
                 "[dim]Choose one of these matching workspaces by slug:\n"
                 f"{format_workspace_selection_choices(matches)}[/dim]"
@@ -575,9 +573,7 @@ def list_projects(
             _, permalink = row_key
             project_name = row_names_by_key[row_key]
             is_attached_row = attached_row_by_permalink.get(permalink) == row_key
-            local_project = (
-                local_projects_by_permalink.get(permalink) if is_attached_row else None
-            )
+            local_project = local_projects_by_permalink.get(permalink) if is_attached_row else None
             cloud_project = cloud_projects_by_key.get(row_key)
             cloud_workspace = cloud_workspaces_by_key.get(row_key)
             configured_name = configured_names_by_permalink.get(permalink)
@@ -621,9 +617,7 @@ def list_projects(
             )
             is_default = bool(is_attached_row and permalink == default_permalink)
 
-            sync_supported = (
-                cloud_workspace is None or cloud_workspace.workspace_type == "personal"
-            )
+            sync_supported = cloud_workspace is None or cloud_workspace.workspace_type == "personal"
             has_sync = bool(is_attached_row and entry and entry.local_sync_path and sync_supported)
             # Determine MCP transport based on project routing mode
             if entry and entry.mode == ProjectMode.CLOUD:

src/basic_memory/config.py

@@ -24,10 +24,24 @@ APP_DATABASE_NAME = "memory.db"  # Using the same name but in the app directory
 DATA_DIR_NAME = ".basic-memory"
 CONFIG_FILE_NAME = "config.json"
 WATCH_STATUS_JSON = "watch-status.json"
+CONFIG_DIR_MODE = 0o700
+CONFIG_FILE_MODE = 0o600
 
 Environment = Literal["test", "dev", "user"]
 
 
+def _secure_config_dir(path: Path) -> None:
+    """Restrict config directory permissions on platforms with POSIX modes."""
+    if os.name != "nt":
+        path.chmod(CONFIG_DIR_MODE)
+
+
+def _secure_config_file(path: Path) -> None:
+    """Restrict config file permissions because config can contain cloud credentials."""
+    if os.name != "nt":
+        path.chmod(CONFIG_FILE_MODE)
+
+
 class ProjectMode(str, Enum):
     """Per-project routing mode."""
 
@@ -168,13 +182,15 @@ class BasicMemoryConfig(BaseSettings):
     env: Environment = Field(default="dev", description="Environment name")
 
     projects: Dict[str, ProjectEntry] = Field(
-        default_factory=lambda: {
-            "main": ProjectEntry(
-                path=str(Path(os.getenv("BASIC_MEMORY_HOME", Path.home() / "basic-memory")))
-            )
-        }
-        if os.getenv("BASIC_MEMORY_HOME")
-        else {},
+        default_factory=lambda: (
+            {
+                "main": ProjectEntry(
+                    path=str(Path(os.getenv("BASIC_MEMORY_HOME", Path.home() / "basic-memory")))
+                )
+            }
+            if os.getenv("BASIC_MEMORY_HOME")
+            else {}
+        ),
         description="Mapping of project names to their ProjectEntry configuration",
     )
     default_project: Optional[str] = Field(
@@ -278,6 +294,11 @@ class BasicMemoryConfig(BaseSettings):
         description="Optional FastEmbed embed() parallelism override.",
         gt=0,
     )
+    import_upload_max_bytes: int = Field(
+        default=100 * 1024 * 1024,
+        description="Maximum uploaded JSON export size accepted by API import endpoints.",
+        gt=0,
+    )
     semantic_vector_k: int = Field(
         default=100,
         description="Vector candidate count for vector and hybrid retrieval.",
@@ -776,6 +797,7 @@ class ConfigManager:
 
         # Ensure config directory exists
         self.config_dir.mkdir(parents=True, exist_ok=True)
+        _secure_config_dir(self.config_dir)
 
     @property
     def config(self) -> BasicMemoryConfig:
@@ -888,6 +910,7 @@ class ConfigManager:
                     # Create backup before overwriting so users can revert if needed
                     backup_path = self.config_file.with_suffix(".json.bak")
                     shutil.copy2(self.config_file, backup_path)
+                    _secure_config_file(backup_path)
                     logger.info(f"Migrating config to current format (backup: {backup_path})")
                     save_basic_memory_config(self.config_file, _CONFIG_CACHE)
 
@@ -1043,9 +1066,12 @@ def has_cloud_credentials(config: BasicMemoryConfig) -> bool:
 def save_basic_memory_config(file_path: Path, config: BasicMemoryConfig) -> None:
     """Save configuration to file."""
     try:
+        file_path.parent.mkdir(parents=True, exist_ok=True)
+        _secure_config_dir(file_path.parent)
         # Use model_dump with mode='json' to serialize datetime objects properly
         config_dict = config.model_dump(mode="json")
         file_path.write_text(json.dumps(config_dict, indent=2))
+        _secure_config_file(file_path)
     except Exception as e:  # pragma: no cover
         logger.error(f"Failed to save config: {e}")
 

src/basic_memory/mcp/project_context.py

@@ -734,7 +734,7 @@ async def _ensure_workspace_project_index(
                 )
             continue
 
-        workspace_entries = cast(tuple[WorkspaceProjectEntry, ...], result)
+        workspace_entries = result
         successful_fetches += 1
         entries_list.extend(workspace_entries)
 
@@ -989,9 +989,7 @@ async def resolve_workspace_parameter(
         selected_workspace: WorkspaceInfo | None = None
 
         if workspace:
-            matches = [
-                item for item in workspaces if workspace_matches_identifier(item, workspace)
-            ]
+            matches = [item for item in workspaces if workspace_matches_identifier(item, workspace)]
             if not matches:
                 raise ValueError(
                     f"Workspace '{workspace}' was not found.\n"

src/basic_memory/repository/metadata_filters.py

@@ -5,11 +5,17 @@ from __future__ import annotations
 from dataclasses import dataclass
 from datetime import date, datetime
 import re
-from typing import Any, Iterable, List
+from typing import Any, Iterable, List, cast
 
 
 _KEY_RE = re.compile(r"^[A-Za-z0-9_-]+(\.[A-Za-z0-9_-]+)*$")
 _NUMERIC_RE = re.compile(r"^-?\d+(\.\d+)?$")
+_COMPARISON_OPERATORS = {
+    "$gt": "gt",
+    "$gte": "gte",
+    "$lt": "lt",
+    "$lte": "lte",
+}
 
 
 @dataclass(frozen=True)
@@ -48,6 +54,11 @@ def _normalize_scalar(value: Any) -> Any:
     return value
 
 
+def _normalize_numeric(value: object) -> float:
+    """Normalize a value already proven numeric by _is_numeric_value."""
+    return float(cast(str | int | float, value))
+
+
 def parse_metadata_filters(filters: dict[str, Any]) -> List[ParsedMetadataFilter]:
     """Parse metadata filters into normalized clauses.
 
@@ -73,7 +84,12 @@ def parse_metadata_filters(filters: dict[str, Any]) -> List[ParsedMetadataFilter
         if isinstance(raw_value, dict):
             if len(raw_value) != 1:
                 raise ValueError(f"Invalid metadata filter for '{raw_key}': {raw_value}")
-            op, value = next(iter(raw_value.items()))
+            raw_op, value = next(iter(raw_value.items()))
+            if not isinstance(raw_op, str):
+                raise ValueError(
+                    f"Unsupported operator '{raw_op}' in metadata filter for '{raw_key}'"
+                )
+            op = raw_op
 
             if op == "$in":
                 if not isinstance(value, list) or not value:
@@ -83,15 +99,20 @@ def parse_metadata_filters(filters: dict[str, Any]) -> List[ParsedMetadataFilter
                 )
                 continue
 
-            if op in {"$gt", "$gte", "$lt", "$lte"}:
+            if op in _COMPARISON_OPERATORS:
                 if _is_numeric_value(value):
-                    normalized = float(value)
+                    normalized = _normalize_numeric(value)
                     comparison = "numeric"
                 else:
                     normalized = _normalize_scalar(value)
                     comparison = "text"
                 parsed.append(
-                    ParsedMetadataFilter(path_parts, op.lstrip("$"), normalized, comparison)
+                    ParsedMetadataFilter(
+                        path_parts,
+                        _COMPARISON_OPERATORS[op],
+                        normalized,
+                        comparison,
+                    )
                 )
                 continue
 
@@ -99,7 +120,7 @@ def parse_metadata_filters(filters: dict[str, Any]) -> List[ParsedMetadataFilter
                 if not isinstance(value, list) or len(value) != 2:
                     raise ValueError(f"$between requires [min, max] for '{raw_key}'")
                 if _is_numeric_collection(value):
-                    normalized = [float(v) for v in value]
+                    normalized = [_normalize_numeric(v) for v in value]
                     comparison = "numeric"
                 else:
                     normalized = [_normalize_scalar(v) for v in value]

src/basic_memory/services/context_service.py

@@ -4,7 +4,7 @@ from __future__ import annotations
 
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
-from typing import List, Optional, Tuple, TYPE_CHECKING
+from typing import Any, List, Optional, Tuple, TYPE_CHECKING
 
 
 from loguru import logger
@@ -307,7 +307,7 @@ class ContextService:
         entity_id_values = ", ".join([str(i) for i in entity_ids])
 
         # Parameters for bindings - include project_id for security filtering
-        params = {
+        params: dict[str, Any] = {
             "max_depth": max_depth,
             "max_results": max_results,
             "project_id": self.search_repository.project_id,
@@ -322,9 +322,9 @@ class ContextService:
                 since_utc = (
                     since.astimezone(timezone.utc) if since.tzinfo else since
                 )  # pragma: no cover
-                params["since_date"] = since_utc.replace(tzinfo=None)  # pyright: ignore  # pragma: no cover
+                params["since_date"] = since_utc.replace(tzinfo=None)  # pragma: no cover
             else:
-                params["since_date"] = since.isoformat()  # pyright: ignore
+                params["since_date"] = since.isoformat()
             date_filter = "AND e.created_at >= :since_date"
             relation_date_filter = "AND e_from.created_at >= :since_date"
             timeframe_condition = "AND eg.relation_date >= :since_date"

src/basic_memory/services/entity_service.py

@@ -1042,9 +1042,9 @@ class EntityService(BaseService[EntityModel]):
             for rel, resolved in zip(markdown.relations, resolved_entities):
                 # Handle exceptions from gather and None results
                 target_entity: Optional[Entity] = None
-                if not isinstance(resolved, Exception):
-                    # Type narrowing: resolved is Optional[Entity] here, not Exception
-                    target_entity = resolved  # pyright: ignore [reportAssignmentType]
+                if not isinstance(resolved, BaseException):
+                    # asyncio.gather(..., return_exceptions=True) can return any BaseException.
+                    target_entity = resolved
 
                 if target_entity is None and not resolve_targets:
                     target_entity = await self._resolve_deferred_self_relation(rel.target, entity)