fix: PostgresGraphDB reorganizer/handler compatibility

src/memos/graph_dbs/postgres.py

@@ -26,16 +26,40 @@
 logger = get_logger(__name__)
 
 
+def _normalize_embedding_value(embedding: Any) -> list[float] | None:
+    """Coerce pgvector/psycopg2 embedding values into list[float] for GraphDBNode."""
+    if embedding is None:
+        return None
+    try:
+        if isinstance(embedding, list):
+            return [float(x) for x in embedding]
+        if isinstance(embedding, tuple):
+            return [float(x) for x in embedding]
+        if isinstance(embedding, str):
+            stripped = embedding.strip()
+            if not stripped:
+                return None
+            try:
+                parsed = json.loads(stripped)
+            except json.JSONDecodeError:
+                return None
+            if isinstance(parsed, list):
+                return [float(x) for x in parsed]
+            return None
+    except (ValueError, TypeError):
+        return None
+    return None
+
+
 def _prepare_node_metadata(metadata: dict[str, Any]) -> dict[str, Any]:
     """Ensure metadata has proper datetime fields and normalized types."""
     now = datetime.utcnow().isoformat()
     metadata.setdefault("created_at", now)
     metadata.setdefault("updated_at", now)
 
-    # Normalize embedding type
-    embedding = metadata.get("embedding")
-    if embedding and isinstance(embedding, list):
-        metadata["embedding"] = [float(x) for x in embedding]
+    normalized = _normalize_embedding_value(metadata.get("embedding"))
+    if normalized is not None:
+        metadata["embedding"] = normalized
 
     return metadata
 
@@ -180,6 +204,49 @@ def _init_schema(self):
         finally:
             self._put_conn(conn)
 
+    def get_memory_count(self, memory_type: str, user_name: str | None = None) -> int:
+        """Count memory nodes by memory_type for a user."""
+        user_name = user_name or self.user_name
+        conn = self._get_conn()
+        try:
+            with conn.cursor() as cur:
+                cur.execute(
+                    f"""
+                    SELECT COUNT(*)
+                    FROM {self.schema}.memories
+                    WHERE properties->>'memory_type' = %s
+                      AND user_name = %s
+                """,
+                    (memory_type, user_name),
+                )
+                row = cur.fetchone()
+                return int(row[0]) if row else 0
+        except Exception as e:
+            logger.error("[get_memory_count] Failed: %s", e)
+            return -1
+        finally:
+            self._put_conn(conn)
+
+    def node_not_exist(self, scope: str, user_name: str | None = None) -> bool:
+        """Return True when no activated nodes exist for the given memory_type scope."""
+        user_name = user_name or self.user_name
+        conn = self._get_conn()
+        try:
+            with conn.cursor() as cur:
+                cur.execute(
+                    f"""
+                    SELECT 1
+                    FROM {self.schema}.memories
+                    WHERE properties->>'memory_type' = %s
+                      AND user_name = %s
+                    LIMIT 1
+                """,
+                    (scope, user_name),
+                )
+                return cur.fetchone() is None
+        finally:
+            self._put_conn(conn)
+
     # =========================================================================
     # Node Management
     # =========================================================================
@@ -437,6 +504,11 @@ def _parse_row(self, row, include_embedding: bool = False) -> dict[str, Any]:
         }
         if include_embedding and len(row) > 5:
             result["metadata"]["embedding"] = row[5]
+            normalized = _normalize_embedding_value(result["metadata"].get("embedding"))
+            if normalized is not None:
+                result["metadata"]["embedding"] = normalized
+            else:
+                del result["metadata"]["embedding"]
         return result
 
     @staticmethod
@@ -674,23 +746,104 @@ def delete_edge(
         finally:
             self._put_conn(conn)
 
-    def edge_exists(self, source_id: str, target_id: str, type: str) -> bool:
-        """Check if edge exists."""
+    def edge_exists(
+        self,
+        source_id: str,
+        target_id: str,
+        type: str = "ANY",
+        direction: str = "OUTGOING",
+        user_name: str | None = None,
+    ) -> bool:
+        """Check if an edge exists between two nodes."""
+        user_name = user_name or self.user_name
+        if direction not in ("OUTGOING", "INCOMING", "ANY"):
+            raise ValueError(
+                f"Invalid direction: {direction}. Must be 'OUTGOING', 'INCOMING', or 'ANY'."
+            )
+
+        type_clause = "" if type == "ANY" else " AND e.edge_type = %s"
+        params: list[Any] = [user_name, user_name]
+
+        if direction == "OUTGOING":
+            direction_clause = "e.source_id = %s AND e.target_id = %s"
+            params.extend([source_id, target_id])
+        elif direction == "INCOMING":
+            direction_clause = "e.source_id = %s AND e.target_id = %s"
+            params.extend([target_id, source_id])
+        else:
+            direction_clause = (
+                "(e.source_id = %s AND e.target_id = %s) OR (e.source_id = %s AND e.target_id = %s)"
+            )
+            params.extend([source_id, target_id, target_id, source_id])
+
+        if type != "ANY":
+            params.append(type)
+
         conn = self._get_conn()
         try:
             with conn.cursor() as cur:
                 cur.execute(
                     f"""
-                    SELECT 1 FROM {self.schema}.edges
-                    WHERE source_id = %s AND target_id = %s AND edge_type = %s
+                    SELECT 1
+                    FROM {self.schema}.edges e
+                    JOIN {self.schema}.memories src ON src.id = e.source_id
+                    JOIN {self.schema}.memories tgt ON tgt.id = e.target_id
+                    WHERE src.user_name = %s
+                      AND tgt.user_name = %s
+                      AND ({direction_clause})
+                      {type_clause}
                     LIMIT 1
                 """,
-                    (source_id, target_id, type),
+                    params,
                 )
                 return cur.fetchone() is not None
         finally:
             self._put_conn(conn)
 
+    def get_edges(
+        self, id: str, type: str = "ANY", direction: str = "ANY", user_name: str | None = None
+    ) -> list[dict[str, str]]:
+        """Get edges connected to a node, with optional type and direction filter."""
+        user_name = user_name or self.user_name
+        if direction not in ("OUTGOING", "INCOMING", "ANY"):
+            raise ValueError("Invalid direction. Must be 'OUTGOING', 'INCOMING', or 'ANY'.")
+
+        type_clause = "" if type == "ANY" else " AND e.edge_type = %s"
+        params: list[Any] = [user_name, user_name, id]
+
+        if direction == "OUTGOING":
+            node_clause = "e.source_id = %s"
+        elif direction == "INCOMING":
+            node_clause = "e.target_id = %s"
+        else:
+            node_clause = "(e.source_id = %s OR e.target_id = %s)"
+            params.append(id)
+
+        if type != "ANY":
+            params.append(type)
+
+        conn = self._get_conn()
+        try:
+            with conn.cursor() as cur:
+                cur.execute(
+                    f"""
+                    SELECT e.source_id, e.target_id, e.edge_type
+                    FROM {self.schema}.edges e
+                    JOIN {self.schema}.memories src ON src.id = e.source_id
+                    JOIN {self.schema}.memories tgt ON tgt.id = e.target_id
+                    WHERE src.user_name = %s
+                      AND tgt.user_name = %s
+                      AND {node_clause}
+                      {type_clause}
+                """,
+                    params,
+                )
+                return [
+                    {"from": row[0], "to": row[1], "type": row[2]} for row in cur.fetchall()
+                ]
+        finally:
+            self._put_conn(conn)
+
     # =========================================================================
     # Graph Queries
     # =========================================================================
@@ -957,10 +1110,10 @@ def get_all_memory_items(
             self._put_conn(conn)
 
     def get_structure_optimization_candidates(
-        self, scope: str, include_embedding: bool = False
+        self, scope: str, include_embedding: bool = False, **kwargs
     ) -> list[dict]:
         """Find isolated nodes (no edges)."""
-        user_name = self.user_name
+        user_name = kwargs.get("user_name") or self.user_name
         conn = self._get_conn()
         try:
             with conn.cursor() as cur:
@@ -983,6 +1136,23 @@ def get_structure_optimization_candidates(
         finally:
             self._put_conn(conn)
 
+    def search_by_fulltext(
+        self,
+        query_words: list[str],
+        top_k: int = 10,
+        scope: str | None = None,
+        status: str | None = None,
+        threshold: float | None = None,
+        search_filter: dict | None = None,
+        user_name: str | None = None,
+        filter: dict | None = None,
+        knowledgebase_ids: list[str] | None = None,
+        tsquery_config: str | None = None,
+        **kwargs,
+    ) -> list[dict]:
+        """Stub for TreeTextMemory keyword recall; Postgres fulltext search is not implemented yet."""
+        return []
+
     # =========================================================================
     # Maintenance
     # =========================================================================

tests/graph_dbs/test_postgres_embedding_parse.py

@@ -0,0 +1,85 @@
+"""Regression tests for PostgresGraphDB embedding normalization."""
+
+from __future__ import annotations
+
+import json
+
+from datetime import datetime
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from memos.graph_dbs.postgres import (
+    PostgresGraphDB,
+    _normalize_embedding_value,
+    _prepare_node_metadata,
+)
+
+
+def test_normalize_embedding_value_parses_json_string() -> None:
+    assert _normalize_embedding_value("[0.5, -1.0]") == [0.5, -1.0]
+
+
+def test_normalize_embedding_value_coerces_numeric_list() -> None:
+    assert _normalize_embedding_value([2, 3]) == [2.0, 3.0]
+
+
+def test_normalize_embedding_value_rejects_invalid_string() -> None:
+    assert _normalize_embedding_value("not-json") is None
+
+
+@pytest.mark.parametrize(
+    "embedding",
+    [["a", 0.5], '["a", 0.5]'],
+)
+def test_normalize_embedding_value_rejects_non_numeric_elements(embedding) -> None:
+    assert _normalize_embedding_value(embedding) is None
+
+
+def test_prepare_node_metadata_normalizes_string_embedding() -> None:
+    metadata = _prepare_node_metadata({"embedding": "[0.25, 0.75]"})
+    assert metadata["embedding"] == [0.25, 0.75]
+
+
+def _build_db() -> PostgresGraphDB:
+    with (
+        patch("memos.graph_dbs.postgres.require_python_package", lambda *args, **kwargs: lambda fn: fn),
+        patch("psycopg2.pool.ThreadedConnectionPool", MagicMock()),
+        patch.object(PostgresGraphDB, "_init_schema", lambda self: None),
+    ):
+        config = MagicMock()
+        config.schema_name = "test_schema"
+        config.user_name = "user-1"
+        return PostgresGraphDB(config)
+
+
+def test_parse_row_normalizes_string_embedding_from_vector_column() -> None:
+    db = _build_db()
+    row: tuple[Any, ...] = (
+        "node-1",
+        "memory text",
+        json.dumps({"memory_type": "UserMemory"}),
+        datetime(2026, 8, 22, 12, 0, 0),
+        datetime(2026, 8, 22, 12, 0, 0),
+        "[0.25, 0.75]",
+    )
+
+    parsed = db._parse_row(row, include_embedding=True)
+
+    assert parsed["metadata"]["embedding"] == [0.25, 0.75]
+
+
+def test_parse_row_preserves_props_embedding_when_not_requested() -> None:
+    db = _build_db()
+    row: tuple[Any, ...] = (
+        "node-1",
+        "memory text",
+        json.dumps({"memory_type": "UserMemory", "embedding": "[0.1, 0.2]"}),
+        datetime(2026, 8, 22, 12, 0, 0),
+        datetime(2026, 8, 22, 12, 0, 0),
+    )
+
+    parsed = db._parse_row(row, include_embedding=False)
+
+    assert parsed["metadata"]["embedding"] == "[0.1, 0.2]"