fix: address PR feedback - improve type checking, extract complex logic, update docs

- Add numbers.Number-based type checking with _is_numeric() helper
- Extract Redis aggregation logic into separate _search_with_redis_aggregation() method
- Add safe _get_vectorstore_index() method to avoid direct _index access
- Document hard_age_multiplier parameter in select_ids_for_forgetting docstring
- Remove stale TDD comment from test file

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <[email protected]>

Author: abrookins

agent_memory_server/long_term_memory.py

@@ -1,6 +1,7 @@
 import hashlib
 import json
 import logging
+import numbers
 import time
 from collections.abc import Iterable
 from datetime import UTC, datetime, timedelta
@@ -1428,6 +1429,11 @@ def combined_score(mem: MemoryRecordResult) -> float:
     return sorted(results, key=combined_score, reverse=True)
 
 
+def _is_numeric(value: Any) -> bool:
+    """Check if a value is numeric (int, float, or other number type)."""
+    return isinstance(value, numbers.Number)
+
+
 def select_ids_for_forgetting(
     results: Iterable[MemoryRecordResult],
     *,
@@ -1442,6 +1448,7 @@ def select_ids_for_forgetting(
       - max_inactive_days: float | None
       - budget: int | None (keep top N by recency score)
       - memory_type_allowlist: set[str] | list[str] | None (only consider these types for deletion)
+      - hard_age_multiplier: float (default 12.0) - multiplier for max_age_days to determine extremely old items
     """
     pinned_ids = pinned_ids or set()
     max_age_days = policy.get("max_age_days")
@@ -1476,9 +1483,7 @@ def select_ids_for_forgetting(
         # - If both thresholds are set, prefer not to delete recently accessed
         #   items unless they are extremely old.
         # - Extremely old: age > max_age_days * hard_age_multiplier (default 12x)
-        if isinstance(max_age_days, int | float) and isinstance(
-            max_inactive_days, int | float
-        ):
+        if _is_numeric(max_age_days) and _is_numeric(max_inactive_days):
             if age_days > float(max_age_days) * hard_age_multiplier:
                 to_delete.add(mem.id)
                 continue
@@ -1488,10 +1493,8 @@ def select_ids_for_forgetting(
                 to_delete.add(mem.id)
                 continue
         else:
-            ttl_hit = isinstance(max_age_days, int | float) and age_days > float(
-                max_age_days
-            )
-            inactivity_hit = isinstance(max_inactive_days, int | float) and (
+            ttl_hit = _is_numeric(max_age_days) and age_days > float(max_age_days)
+            inactivity_hit = _is_numeric(max_inactive_days) and (
                 inactive_days > float(max_inactive_days)
             )
             if ttl_hit or inactivity_hit:

agent_memory_server/vectorstore_adapter.py

@@ -13,7 +13,6 @@
 from langchain_core.embeddings import Embeddings
 from langchain_core.vectorstores import VectorStore
 from langchain_redis.vectorstores import RedisVectorStore
-from redisvl.query import RangeQuery, VectorQuery
 
 from agent_memory_server.filters import (
     CreatedAt,
@@ -837,6 +836,127 @@ async def update_memories(self, memories: list[MemoryRecord]) -> int:
         added = await self.add_memories(memories)
         return len(added)
 
+    def _get_vectorstore_index(self):
+        """Safely access the underlying RedisVL index from the vectorstore.
+
+        Returns:
+            RedisVL SearchIndex or None if not available
+        """
+        return getattr(self.vectorstore, "_index", None)
+
+    async def _search_with_redis_aggregation(
+        self,
+        query: str,
+        redis_filter,
+        limit: int,
+        offset: int,
+        distance_threshold: float | None,
+        recency_params: dict | None,
+    ) -> MemoryRecordResults:
+        """Perform server-side Redis aggregation search with recency scoring.
+
+        Args:
+            query: Search query text
+            redis_filter: Redis filter expression
+            limit: Maximum number of results
+            offset: Offset for pagination
+            distance_threshold: Distance threshold for range queries
+            recency_params: Parameters for recency scoring
+
+        Returns:
+            MemoryRecordResults with server-side scored results
+
+        Raises:
+            Exception: If Redis aggregation fails (caller should handle fallback)
+        """
+        from datetime import UTC as _UTC, datetime as _dt
+
+        from langchain_core.documents import Document
+        from redisvl.query import RangeQuery, VectorQuery
+
+        from agent_memory_server.utils.redis_query import RecencyAggregationQuery
+
+        index = self._get_vectorstore_index()
+        if index is None:
+            raise Exception("RedisVL index not available")
+
+        # Embed the query text to vector
+        embedding_vector = self.embeddings.embed_query(query)
+
+        # Build base KNN query (hybrid)
+        if distance_threshold is not None:
+            knn = RangeQuery(
+                vector=embedding_vector,
+                vector_field_name="vector",
+                filter_expression=redis_filter,
+                distance_threshold=float(distance_threshold),
+                num_results=limit,
+            )
+        else:
+            knn = VectorQuery(
+                vector=embedding_vector,
+                vector_field_name="vector",
+                filter_expression=redis_filter,
+                num_results=limit,
+            )
+
+        # Aggregate with APPLY/SORTBY boosted score via helper
+        now_ts = int(_dt.now(_UTC).timestamp())
+        agg = (
+            RecencyAggregationQuery.from_vector_query(
+                knn, filter_expression=redis_filter
+            )
+            .load_default_fields()
+            .apply_recency(now_ts=now_ts, params=recency_params or {})
+            .sort_by_boosted_desc()
+            .paginate(offset, limit)
+        )
+
+        raw = (
+            await index.aaggregate(agg)
+            if hasattr(index, "aaggregate")
+            else index.aggregate(agg)  # type: ignore
+        )
+
+        rows = getattr(raw, "rows", raw) or []
+        memory_results: list[MemoryRecordResult] = []
+        for row in rows:
+            fields = getattr(row, "__dict__", None) or row
+            metadata = {
+                k: fields.get(k)
+                for k in [
+                    "id_",
+                    "session_id",
+                    "user_id",
+                    "namespace",
+                    "created_at",
+                    "last_accessed",
+                    "updated_at",
+                    "pinned",
+                    "access_count",
+                    "topics",
+                    "entities",
+                    "memory_hash",
+                    "discrete_memory_extracted",
+                    "memory_type",
+                    "persisted_at",
+                    "extracted_from",
+                    "event_date",
+                ]
+                if k in fields
+            }
+            text_val = fields.get("text", "")
+            score = fields.get("__vector_score", 1.0) or 1.0
+            doc_obj = Document(page_content=text_val, metadata=metadata)
+            memory_results.append(self.document_to_memory(doc_obj, float(score)))
+
+        next_offset = offset + limit if len(memory_results) == limit else None
+        return MemoryRecordResults(
+            memories=memory_results[:limit],
+            total=offset + len(memory_results),
+            next_offset=next_offset,
+        )
+
     async def search_memories(
         self,
         query: str,
@@ -900,94 +1020,14 @@ async def search_memories(
         # If server-side recency is requested, attempt RedisVL query first (DB-level path)
         if server_side_recency:
             try:
-                index = getattr(self.vectorstore, "_index", None)
-                if index is not None:
-                    # Embed the query text to vector
-                    embedding_vector = self.embeddings.embed_query(query)
-
-                    # Build base KNN query (hybrid)
-                    if distance_threshold is not None:
-                        knn = RangeQuery(
-                            vector=embedding_vector,
-                            vector_field_name="vector",
-                            filter_expression=redis_filter,
-                            distance_threshold=float(distance_threshold),
-                            num_results=limit,
-                        )
-                    else:
-                        knn = VectorQuery(
-                            vector=embedding_vector,
-                            vector_field_name="vector",
-                            filter_expression=redis_filter,
-                            num_results=limit,
-                        )
-
-                    # Aggregate with APPLY/SORTBY boosted score via helper
-                    from datetime import UTC as _UTC, datetime as _dt
-
-                    from agent_memory_server.utils.redis_query import (
-                        RecencyAggregationQuery,
-                    )
-
-                    now_ts = int(_dt.now(_UTC).timestamp())
-                    agg = (
-                        RecencyAggregationQuery.from_vector_query(
-                            knn, filter_expression=redis_filter
-                        )
-                        .load_default_fields()
-                        .apply_recency(now_ts=now_ts, params=recency_params or {})
-                        .sort_by_boosted_desc()
-                        .paginate(offset, limit)
-                    )
-
-                    raw = (
-                        await index.aaggregate(agg)
-                        if hasattr(index, "aaggregate")
-                        else index.aggregate(agg)  # type: ignore
-                    )
-
-                    rows = getattr(raw, "rows", raw) or []
-                    memory_results: list[MemoryRecordResult] = []
-                    for row in rows:
-                        fields = getattr(row, "__dict__", None) or row
-                        metadata = {
-                            k: fields.get(k)
-                            for k in [
-                                "id_",
-                                "session_id",
-                                "user_id",
-                                "namespace",
-                                "created_at",
-                                "last_accessed",
-                                "updated_at",
-                                "pinned",
-                                "access_count",
-                                "topics",
-                                "entities",
-                                "memory_hash",
-                                "discrete_memory_extracted",
-                                "memory_type",
-                                "persisted_at",
-                                "extracted_from",
-                                "event_date",
-                            ]
-                            if k in fields
-                        }
-                        text_val = fields.get("text", "")
-                        score = fields.get("__vector_score", 1.0) or 1.0
-                        doc_obj = Document(page_content=text_val, metadata=metadata)
-                        memory_results.append(
-                            self.document_to_memory(doc_obj, float(score))
-                        )
-
-                    next_offset = (
-                        offset + limit if len(memory_results) == limit else None
-                    )
-                    return MemoryRecordResults(
-                        memories=memory_results[:limit],
-                        total=offset + len(memory_results),
-                        next_offset=next_offset,
-                    )
+                return await self._search_with_redis_aggregation(
+                    query=query,
+                    redis_filter=redis_filter,
+                    limit=limit,
+                    offset=offset,
+                    distance_threshold=distance_threshold,
+                    recency_params=recency_params,
+                )
             except Exception as e:
                 logger.warning(
                     f"RedisVL DB-level recency search failed; falling back to client-side path: {e}"

tests/test_forgetting.py

@@ -1,6 +1,5 @@
 from datetime import UTC, datetime, timedelta
 
-# TDD: These helpers/functions will be implemented in agent_memory_server.long_term_memory
 from agent_memory_server.long_term_memory import (
     rerank_with_recency,  # new: pure function
     score_recency,  # new: pure function