Fix stats endpoint returning zero by using count() instead of query()

The /api/v1/stats/memories endpoint always returns zeros because
_query_all_memories() calls query() with a filter but no query_vector.
In the local backend, this falls through to search_by_random which does
an HNSW graph traversal with a random vector, completely missing
filtered records in small collections.

Replace the query()-based approach with count() (which uses
aggregate_data — a proper scalar scan) per category using PathScope
URI filters. This reliably returns correct counts on all backends.

Tested on a live instance: preferences=2, entities=2, events=2 (was
all zeros).

Fixes #1255

Author: SeeYangZhi

openviking/storage/stats_aggregator.py

@@ -6,12 +6,10 @@
 aggregate memory health metrics without introducing new storage.
 """
 
-from datetime import datetime, timezone
 from typing import Any, Dict, List, Optional
 
-from openviking.retrieve.memory_lifecycle import hotness_score
 from openviking.server.identity import RequestContext
-from openviking.storage.expr import Eq
+from openviking.storage.expr import And, Eq, PathScope
 from openviking_cli.utils import get_logger
 
 logger = get_logger(__name__)
@@ -28,11 +26,6 @@
     "skills",
 ]
 
-# Hotness buckets
-COLD_THRESHOLD = 0.2
-HOT_THRESHOLD = 0.6
-
-
 class StatsAggregator:
     """Aggregates memory health statistics from VikingDB.
 
@@ -58,62 +51,36 @@ async def get_memory_stats(
             Dictionary with total counts, category breakdown,
             hotness distribution, and staleness metrics.
         """
-        now = datetime.now(timezone.utc)
-
         # Build category list to query
         categories = [category] if category else MEMORY_CATEGORIES
 
-        by_category: Dict[str, int] = {}
+        by_category: Dict[str, int] = {cat: 0 for cat in categories}
         hotness_dist = {"cold": 0, "warm": 0, "hot": 0}
         staleness = {
             "not_accessed_7d": 0,
             "not_accessed_30d": 0,
             "oldest_memory_age_days": 0,
         }
 
-        # Fetch all memories once and group by category in Python
-        all_records = await self._query_all_memories(ctx)
-        grouped: Dict[str, List[Dict[str, Any]]] = {cat: [] for cat in categories}
-        for record in all_records:
-            uri = record.get("uri", "")
-            for cat in categories:
-                if f"/{cat}/" in uri:
-                    grouped[cat].append(record)
-                    break
+        # Use count() (aggregate) instead of query() (vector search).
+        # query() with no query_vector falls through to search_by_random
+        # in the local HNSW backend, which can miss filtered results.
+        # count() uses a scalar aggregate that reliably returns correct
+        # results.
+        user_id = ctx.user.user_id
+        memory_base = f"viking://user/{user_id}/memories"
 
         for cat in categories:
-            records = grouped[cat]
-            by_category[cat] = len(records)
-
-            for record in records:
-                active_count = record.get("active_count", 0)
-                updated_at_raw = record.get("updated_at")
-                updated_at = _parse_datetime(updated_at_raw)
-                created_at_raw = record.get("created_at")
-                created_at = _parse_datetime(created_at_raw)
-
-                # Hotness distribution
-                score = hotness_score(active_count, updated_at, now=now)
-                if score < COLD_THRESHOLD:
-                    hotness_dist["cold"] += 1
-                elif score > HOT_THRESHOLD:
-                    hotness_dist["hot"] += 1
-                else:
-                    hotness_dist["warm"] += 1
-
-                # Staleness: use updated_at for access tracking
-                if updated_at:
-                    age_days = (now - updated_at).total_seconds() / 86400.0
-                    if age_days > 7:
-                        staleness["not_accessed_7d"] += 1
-                    if age_days > 30:
-                        staleness["not_accessed_30d"] += 1
-
-                # Track oldest memory by created_at
-                if created_at:
-                    age = (now - created_at).total_seconds() / 86400.0
-                    if age > staleness["oldest_memory_age_days"]:
-                        staleness["oldest_memory_age_days"] = round(age, 1)
+            try:
+                by_category[cat] = await self._vikingdb.count(
+                    filter=And([
+                        Eq("context_type", "memory"),
+                        PathScope("uri", f"{memory_base}/{cat}", depth=2),
+                    ]),
+                    ctx=ctx,
+                )
+            except Exception as e:
+                logger.error("Error counting memories for %s: %s", cat, e)
 
         total_memories = sum(by_category.values())
 
@@ -152,47 +119,3 @@ async def get_session_extraction_stats(
             "skills_used": stats.skills_used,
         }
 
-    async def _query_all_memories(
-        self,
-        ctx: RequestContext,
-    ) -> List[Dict[str, Any]]:
-        """Query all memory records in a single DB round-trip.
-
-        Uses the context_type="memory" filter. Callers group by category
-        in Python to avoid N+1 queries.
-        """
-        try:
-            return await self._vikingdb.query(
-                filter=Eq("context_type", "memory"),
-                limit=10000,
-                output_fields=[
-                    "uri",
-                    "active_count",
-                    "updated_at",
-                    "created_at",
-                    "context_type",
-                ],
-                ctx=ctx,
-            )
-        except Exception as e:
-            logger.error("Error querying memories: %s", e)
-            return []
-
-
-def _parse_datetime(value) -> Optional[datetime]:
-    """Parse a datetime value from a VikingDB record."""
-    if value is None:
-        return None
-    if isinstance(value, datetime):
-        if value.tzinfo is None:
-            return value.replace(tzinfo=timezone.utc)
-        return value
-    if isinstance(value, str):
-        try:
-            dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
-            if dt.tzinfo is None:
-                dt = dt.replace(tzinfo=timezone.utc)
-            return dt
-        except (ValueError, TypeError):
-            return None
-    return None