feat: same-turn memory retrieval in provider mode

prefetch() now reuses the in-flight background search when it matches
the current query, or falls back to a synchronous search (capped at 3s,
split across both requests in case-insensitive mode) so memories are
injected on the same turn instead of one turn behind.

- Track _prefetch_query to scope cached results to their query
- Reset _prefetch_result when starting a new background search to
  prevent stale cross-turn injection after a timed-out join
- Guard background thread writes against superseded queries
- Add optional per-request timeout override to LocalMem0Client
- Update AGENTS.md and README to describe the same-turn flow

__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"