Merge pull request 'feat: same-turn memory retrieval in provider mode' (#6) from feature/same-turn-prefetch into main

Reviewed-on: #6

AGENTS.md

@@ -43,8 +43,8 @@ curl http://localhost:8000/health
 
 **Memory injection flow:**
 1. `queue_prefetch()` — Spawns background thread on user message
-2. `prefetch()` — Returns cached result next turn
-3. `pre_llm_call` hook — Sync prefetch for immediate injection (fallback context)
+2. `prefetch()` — Sync retrieval for the current query (same-turn injection); falls back to cached background result when no query is provided
+3. `pre_llm_call` hook — Sync prefetch for immediate injection (plugin mode)
 4. `sync_turn()` — Non-blocking server-side fact extraction
 
 **Circuit breaker:** 5 consecutive failures → 120s cooldown.

README.md

@@ -5,7 +5,7 @@ Self-hosted Mem0 memory provider for Hermes-Agent. Provides semantic memory sear
 ## Features
 
 - **Local Mem0 server** — No cloud dependency, full data privacy
-- **Async prefetch** — Memory retrieval happens in background (~40ms)
+- **Same-turn prefetch** — Memory retrieved synchronously before the LLM call (~40ms)
 - **Context injection** — Relevant memories injected directly into LLM prompt
 - **Automatic fact extraction** — Server-side LLM extracts facts from conversations
 - **Semantic search** — Find memories by meaning, not keywords
@@ -138,16 +138,15 @@ hermes gateway restart
 
 ### How It Works
 
-1. **User message received** → `queue_prefetch()` spawns background thread
-2. **Mem0 search** → Semantic search for relevant memories (~40ms)
-3. **Context injection** → Results injected via `pre_llm_call` hook
-4. **LLM receives** → User message + memory context (no tool call needed!)
+1. **User message received** → `prefetch()` searches Mem0 synchronously (~40ms)
+2. **Context injection** → Results injected before the LLM call
+3. **LLM receives** → User message + memory context (no tool call needed!)
 
 **Example**:
 ```
 User: "Hey, is a new episode out from my favorite anime?"
 
-↓ [Background: mem0.prefetch() searches for "favorite anime"]
+↓ [mem0.prefetch() searches for "favorite anime"]
 
 LLM receives:
 """

__init__.py

@@ -227,6 +227,7 @@ class Mem0LocalMemoryProvider(MemoryProvider):
             ],
         }
         self._prefetch_result = ""
+        self._prefetch_query = ""
         self._prefetch_lock = threading.Lock()
         self._prefetch_thread = None
         self._sync_thread = None
@@ -520,20 +521,39 @@ class Mem0LocalMemoryProvider(MemoryProvider):
         )
 
     def prefetch(self, query: str = "", *, session_id: str = "") -> str:
-        """Return cached prefetch result from previous turn.
+        """Return memory context for the current turn.
+
+        Reuses the in-flight background prefetch when it was started for the
+        same query, otherwise performs a synchronous search (bounded by the
+        shorter of the configured client timeout and 3s) so memories are
+        injected on the same turn. Falls back to the cached background
+        result when no query is available.
 
         Args:
-            query: Deprecated, kept for API compatibility.
+            query: Current user message, used for synchronous retrieval.
             session_id: Session identifier.
         """
-        if self._prefetch_thread and self._prefetch_thread.is_alive():
-            self._prefetch_thread.join(timeout=3.0)
-        with self._prefetch_lock:
-            result = self._prefetch_result
-            self._prefetch_result = ""
+        if query:
+            with self._prefetch_lock:
+                query_matches = self._prefetch_query == query
+            if query_matches and self._prefetch_thread and self._prefetch_thread.is_alive():
+                # Background search for this exact query is in flight — wait for it
+                self._prefetch_thread.join(timeout=3.0)
+            if query_matches:
+                with self._prefetch_lock:
+                    result = self._prefetch_result
+                    self._prefetch_result = ""
+            else:
+                # No cached result for this query — search synchronously
+                result = self.queue_prefetch_and_get(query)
+        else:
+            if self._prefetch_thread and self._prefetch_thread.is_alive():
+                self._prefetch_thread.join(timeout=3.0)
+            with self._prefetch_lock:
+                result = self._prefetch_result
+                self._prefetch_result = ""
         if not result:
             return ""
-        # Check if it's an error message
         if result.startswith("ERROR:"):
             return f"<mem0_error>\n{result[6:]}\n</mem0_error>"
         return f"<mem0_context>\n{result}\n</mem0_context>"
@@ -570,11 +590,17 @@ class Mem0LocalMemoryProvider(MemoryProvider):
             return ""
         try:
             client = self._get_client()
+            # Cap total blocking on the LLM hot path at ~3s
+            prefetch_timeout = min(client.timeout, 3.0)
+            if self._case_insensitive:
+                # case-insensitive search runs two sequential requests
+                prefetch_timeout /= 2
             results = client.search(
                 query=query,
                 user_id=self._user_id,
                 limit=self._prefetch_limit,
                 case_insensitive=self._case_insensitive,
+                timeout=prefetch_timeout,
             )
             # Filter by score threshold
             threshold = self._prefetch_score_threshold / 100.0
@@ -602,14 +628,20 @@ class Mem0LocalMemoryProvider(MemoryProvider):
         """
         if self._is_breaker_open():
             with self._prefetch_lock:
+                self._prefetch_query = query
                 self._prefetch_result = "ERROR:Memory service temporarily unavailable. Please try again later."
             return
 
         if self._is_trivial_prompt(query):
             with self._prefetch_lock:
+                self._prefetch_query = query
                 self._prefetch_result = ""
             return
 
+        with self._prefetch_lock:
+            self._prefetch_query = query
+            self._prefetch_result = ""
+
         def _run():
             try:
                 client = self._get_client()
@@ -625,16 +657,19 @@ class Mem0LocalMemoryProvider(MemoryProvider):
                 if filtered:
                     formatted = self._format_search_results(filtered, categorize=self._categorize_enabled)
                     with self._prefetch_lock:
-                        self._prefetch_result = formatted
+                        if self._prefetch_query == query:
+                            self._prefetch_result = formatted
                 else:
                     with self._prefetch_lock:
-                        self._prefetch_result = ""
+                        if self._prefetch_query == query:
+                            self._prefetch_result = ""
                 self._record_success()
             except Exception as e:
                 self._record_failure()
                 logger.debug("Mem0 prefetch failed: %s", e)
                 with self._prefetch_lock:
-                    self._prefetch_result = "ERROR:Memory service temporarily unavailable. Please try again later."
+                    if self._prefetch_query == query:
+                        self._prefetch_result = "ERROR:Memory service temporarily unavailable. Please try again later."
 
         self._prefetch_thread = threading.Thread(
             target=_run, daemon=True, name="mem0-local-prefetch"

client.py

@@ -39,17 +39,19 @@ class LocalMem0Client:
         endpoint: str,
         json: Optional[Dict] = None,
         params: Optional[Dict] = None,
+        timeout: Optional[float] = None,
     ) -> Dict:
         """Make HTTP request with error handling."""
         url = f"{self.base_url}{endpoint}"
+        effective_timeout = timeout if timeout is not None else self.timeout
         try:
             resp = self.session.request(
-                method, url, json=json, params=params, timeout=self.timeout
+                method, url, json=json, params=params, timeout=effective_timeout
             )
             resp.raise_for_status()
             return resp.json()
         except requests.exceptions.Timeout:
-            logger.error("Mem0 request timed out after %ss", self.timeout)
+            logger.error("Mem0 request timed out after %ss", effective_timeout)
             raise
         except requests.exceptions.ConnectionError as e:
             logger.error("Failed to connect to Mem0 server at %s: %s", self.base_url, e)
@@ -66,6 +68,7 @@ class LocalMem0Client:
         user_id: Optional[str] = None,
         limit: int = 5,
         case_insensitive: bool = False,
+        timeout: Optional[float] = None,
     ) -> List[Dict]:
         """Search memories by semantic similarity.
 
@@ -78,18 +81,19 @@ class LocalMem0Client:
             user_id: User identifier
             limit: Max results
             case_insensitive: If True, search with both original and lowercase query
+            timeout: Optional per-request timeout override (seconds)
         """
         if not case_insensitive:
             payload = {"query": query, "limit": limit}
             if user_id:
                 payload["user_id"] = user_id
-            result = self._request("POST", "/search", json=payload)
+            result = self._request("POST", "/search", json=payload, timeout=timeout)
             return result.get("results", [])
 
         # Case-insensitive mode: search with both original and lowercase
         # Fetch 2x limit to ensure we get top N after merging
-        results_original = self._search_with_query(query, user_id, limit * 2)
-        results_lower = self._search_with_query(query.lower(), user_id, limit * 2)
+        results_original = self._search_with_query(query, user_id, limit * 2, timeout)
+        results_lower = self._search_with_query(query.lower(), user_id, limit * 2, timeout)
 
         # Merge and deduplicate, keeping highest score
         merged = {}
@@ -109,12 +113,13 @@ class LocalMem0Client:
         query: str,
         user_id: Optional[str] = None,
         limit: int = 5,
+        timeout: Optional[float] = None,
     ) -> List[Dict]:
         """Internal search helper for case-insensitive mode."""
         payload = {"query": query, "limit": limit}
         if user_id:
             payload["user_id"] = user_id
-        result = self._request("POST", "/search", json=payload)
+        result = self._request("POST", "/search", json=payload, timeout=timeout)
         return result.get("results", [])
 
     def get_all(self, user_id: Optional[str] = None) -> List[Dict]: