fix: decode FileMemory nested results and isolate malformed prefetch config

- extend prefetch text extractor to recurse into FileMemory's nested
  {"item": {"content": ...}} shape so file-backed recall reaches the model
- coerce prefetch_limit/prefetch_token_budget via safe helper so null or
  non-numeric config never raises before the retrieval exception boundary
- add unit coverage for the nested shape and malformed numeric config

Co-authored-by: Mervin Praison <[email protected]>

src/praisonai-agents/praisonaiagents/agent/chat_mixin.py

@@ -29,6 +29,152 @@ if TYPE_CHECKING:
 class ChatMixin:
     """Mixin providing chat methods for the Agent class."""
 
+    @staticmethod
+    def _memory_prefetch_query(prompt: Any) -> str:
+        """Return the user-authored text used for turn-start memory lookup."""
+        if isinstance(prompt, str):
+            return prompt.strip()
+        if isinstance(prompt, list):
+            parts = []
+            for item in prompt:
+                if isinstance(item, dict) and item.get("type") == "text":
+                    text = item.get("text")
+                    if text:
+                        parts.append(str(text))
+            return "\n".join(parts).strip()
+        return str(prompt).strip() if prompt is not None else ""
+
+    @staticmethod
+    def _memory_prefetch_int(value: Any, default: int) -> int:
+        """Coerce an optional prefetch bound to a non-negative int.
+
+        Malformed config (None, empty, or non-numeric strings) must never abort
+        a turn, so any coercion failure falls back to the supplied default.
+        """
+        if value is None:
+            return default
+        try:
+            return max(0, int(value))
+        except (TypeError, ValueError):
+            return default
+
+    @staticmethod
+    def _memory_prefetch_text(item: Any) -> str:
+        """Extract display text from the common memory-adapter result shapes."""
+        if isinstance(item, str):
+            return item.strip()
+        if isinstance(item, dict):
+            for key in ("memory", "text", "content", "value"):
+                value = item.get(key)
+                if value is not None:
+                    return str(value).strip()
+            # FileMemory nests the payload under "item" (e.g.
+            # {"type": "long_term", "item": {"content": ...}}), so recurse
+            # once into that wrapper before giving up.
+            nested = item.get("item")
+            if isinstance(nested, dict):
+                for key in ("memory", "text", "content", "value"):
+                    value = nested.get(key)
+                    if value is not None:
+                        return str(value).strip()
+        for attr in ("memory", "text", "content", "value"):
+            value = getattr(item, attr, None)
+            if value is not None:
+                return str(value).strip()
+        return ""
+
+    def _format_memory_prefetch(self, results: Any, limit: int, token_budget: int) -> str:
+        """Format deduplicated memory hits inside a strict estimated-token budget."""
+        if not isinstance(results, (list, tuple)):
+            return ""
+
+        try:
+            from ..context.tokens import estimate_tokens_heuristic
+        except ImportError:  # pragma: no cover - core module is always present
+            estimate_tokens_heuristic = lambda text: max(1, len(text) // 4)
+
+        lines = []
+        seen = set()
+        for item in results:
+            text = self._memory_prefetch_text(item)
+            if not text or text in seen:
+                continue
+            seen.add(text)
+            lines.append(f"- {text}")
+            if len(lines) >= limit:
+                break
+
+        if not lines or token_budget <= 0:
+            return ""
+
+        rendered = "\n".join(lines)
+        if estimate_tokens_heuristic(rendered) <= token_budget:
+            return rendered
+
+        # Token estimation is monotonic for prefixes, so binary search keeps
+        # the largest safe prefix and makes truncation explicit to the model.
+        suffix = "…"
+        low, high = 0, len(rendered)
+        while low < high:
+            mid = (low + high + 1) // 2
+            candidate = rendered[:mid].rstrip() + suffix
+            if estimate_tokens_heuristic(candidate) <= token_budget:
+                low = mid
+            else:
+                high = mid - 1
+        return (rendered[:low].rstrip() + suffix) if low else ""
+
+    def _prefetch_memory(self, prompt: Any) -> str:
+        """Best-effort synchronous turn-start retrieval (default-off)."""
+        config = getattr(self, "_memory_config", None)
+        memory = getattr(self, "_memory_instance", None)
+        if not config or not getattr(config, "prefetch", False) or memory is None:
+            return ""
+
+        query = self._memory_prefetch_query(prompt)
+        if not query:
+            return ""
+        limit = self._memory_prefetch_int(getattr(config, "prefetch_limit", 5), 5)
+        budget = self._memory_prefetch_int(getattr(config, "prefetch_token_budget", 512), 512)
+        if not limit or not budget:
+            return ""
+
+        try:
+            if hasattr(memory, "search_long_term"):
+                results = memory.search_long_term(query, limit=limit)
+            elif hasattr(memory, "search"):
+                results = memory.search(query, limit=limit)
+            else:
+                return ""
+            return self._format_memory_prefetch(results, limit, budget)
+        except Exception as exc:
+            logging.debug("Memory prefetch failed; continuing without recalled context: %s", exc)
+            return ""
+
+    async def _aprefetch_memory(self, prompt: Any) -> str:
+        """Best-effort asynchronous turn-start retrieval (default-off)."""
+        config = getattr(self, "_memory_config", None)
+        memory = getattr(self, "_memory_instance", None)
+        if not config or not getattr(config, "prefetch", False) or memory is None:
+            return ""
+
+        query = self._memory_prefetch_query(prompt)
+        if not query:
+            return ""
+        limit = self._memory_prefetch_int(getattr(config, "prefetch_limit", 5), 5)
+        budget = self._memory_prefetch_int(getattr(config, "prefetch_token_budget", 512), 512)
+        if not limit or not budget:
+            return ""
+
+        try:
+            results = await self.asearch_memory(
+                query, memory_type="long_term", limit=limit
+            )
+            return self._format_memory_prefetch(results, limit, budget)
+        except Exception as exc:
+            logging.debug("Async memory prefetch failed; continuing without recalled context: %s", exc)
+            return ""
+
     def _resolve_max_steps(self, default: int = 10) -> int:
         """Resolve the unified multi-step tool budget shared by both loops.
 
@@ -87,7 +233,7 @@ class ChatMixin:
         except Exception:
             return ""
 
-    def _build_system_prompt(self, tools=None):
+    def _build_system_prompt(self, tools=None, memory_prefetch_context: str = ""):
         """Build the system prompt with tool information.
         
         Args:
@@ -178,6 +324,9 @@ Your Goal: {self.goal}"""
             learn_context = self.get_learn_context()
             if learn_context:
                 system_prompt += f"\n\n## Learned Context (Patterns and insights from past interactions)\n{learn_context}"
+
+        if memory_prefetch_context:
+            system_prompt += f"\n\n## Recalled memories\n{memory_prefetch_context}"
         
         # Add skills prompt if skills are configured
         if self._skills or self._skills_dirs:
@@ -489,7 +638,7 @@ Your Goal: {self.goal}"""
             )
             return False
 
-    def _build_messages(self, prompt, temperature=1.0, output_json=None, output_pydantic=None, tools=None, use_native_format=False):
+    def _build_messages(self, prompt, temperature=1.0, output_json=None, output_pydantic=None, tools=None, use_native_format=False, memory_prefetch_context: str = ""):
         """Build messages list for chat completion.
         
         Args:
@@ -512,6 +661,7 @@ Your Goal: {self.goal}"""
                 prompt=prompt,
                 system_prompt=self._build_system_prompt(
                     tools=tools,
+                    memory_prefetch_context=memory_prefetch_context,
                 ),
                 chat_history=self.chat_history,
                 output_json=None if use_native_format else output_json,
@@ -521,6 +671,7 @@ Your Goal: {self.goal}"""
             # Build messages manually
             system_prompt = self._build_system_prompt(
                 tools=tools,
+                memory_prefetch_context=memory_prefetch_context,
             )
             if system_prompt:
                 messages.append({"role": "system", "content": system_prompt})
@@ -2694,6 +2845,10 @@ Your Goal: {self.goal}"""
         # Use agent's stream setting if not explicitly provided
         if stream is None:
             stream = self.stream
+
+        # Query long-term memory exactly once for this turn. The default-off
+        # guard returns before touching the backend.
+        memory_prefetch_context = self._prefetch_memory(prompt)
         
         # Unified retrieval handling with policy-based decision
         # Uses token-aware context building (DRY - same path as RAG pipeline)
@@ -2799,7 +2954,9 @@ Your Goal: {self.goal}"""
                 
                 try:
                     # --- Proactive Context Budget Management (sync custom LLM path) ---
-                    system_prompt_for_llm = self._build_system_prompt(tools)
+                    system_prompt_for_llm = self._build_system_prompt(
+                        tools, memory_prefetch_context=memory_prefetch_context
+                    )
                     
                     # Apply proactive context budget analysis before any other processing
                     try:
@@ -2929,7 +3086,8 @@ Your Goal: {self.goal}"""
             # Pass llm_prompt (which includes multimodal content if attachments present)
             messages, original_prompt = self._build_messages(
                 llm_prompt, temperature, output_json, output_pydantic,
-                use_native_format=use_native_format
+                use_native_format=use_native_format,
+                memory_prefetch_context=memory_prefetch_context,
             )
             
 
@@ -3322,6 +3480,10 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
         start_time = time.time()
         reasoning_steps = reasoning_steps or self.reasoning_steps
         try:
+            # Async adapters are awaited (sync adapters are offloaded by
+            # AsyncMemoryMixin), so turn-start retrieval never blocks the loop.
+            memory_prefetch_context = await self._aprefetch_memory(prompt)
+
             # Default to self.tools if tools argument is None
             if tools is None:
                 tools = self.tools
@@ -3385,7 +3547,10 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
                     # C1 - per-call seed forwarding (async path)  
                     llm_kwargs = {
                         'prompt': prompt,
-                        'system_prompt': self._build_system_prompt(tools),
+                        'system_prompt': self._build_system_prompt(
+                            tools,
+                            memory_prefetch_context=memory_prefetch_context,
+                        ),
                         'chat_history': effective_history,
                         'temperature': temperature,
                         'tools': tools,
@@ -3463,7 +3628,13 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
 
             # For OpenAI client
             # Use the new _build_messages helper method
-            messages, original_prompt = self._build_messages(prompt, temperature, output_json, output_pydantic)
+            messages, original_prompt = self._build_messages(
+                prompt,
+                temperature,
+                output_json,
+                output_pydantic,
+                memory_prefetch_context=memory_prefetch_context,
+            )
             
             # Store chat history length for potential rollback
             chat_history_length = len(self.chat_history)
@@ -4075,6 +4246,7 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
             # Temporarily disable verbose mode to prevent console output conflicts during streaming
             original_verbose = self.verbose
             self.verbose = False
+            memory_prefetch_context = self._prefetch_memory(prompt)
             
             # For custom LLM path, use the new get_response_stream generator
             if self._using_custom_llm:
@@ -4133,7 +4305,10 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
                     response_content = ""
                     for chunk in self.llm_instance.get_response_stream(
                         prompt=actual_prompt,
-                        system_prompt=self._build_system_prompt(tool_param),
+                        system_prompt=self._build_system_prompt(
+                            tool_param,
+                            memory_prefetch_context=memory_prefetch_context,
+                        ),
                         chat_history=self.chat_history,
                         temperature=kwargs.get('temperature', 1.0),
                         tools=tool_param,
@@ -4189,7 +4364,8 @@ Output MUST be JSON with 'reflection' and 'satisfactory'.
                 
                 # Build messages using the helper method
                 messages, original_prompt = self._build_messages(actual_prompt, kwargs.get('temperature', 1.0), 
-                                                               kwargs.get('output_json'), kwargs.get('output_pydantic'))
+                                                               kwargs.get('output_json'), kwargs.get('output_pydantic'),
+                                                               memory_prefetch_context=memory_prefetch_context)
                 
                 # Store chat history length for potential rollback
                 chat_history_length = len(self.chat_history)

src/praisonai-agents/praisonaiagents/config/feature_configs.py

@@ -238,6 +238,12 @@ class MemoryConfig:
     # History injection (auto-inject session history into context)
     history: bool = False
     history_limit: int = 10
+
+    # Turn-start long-term memory retrieval. Disabled by default so existing
+    # agents incur no backend query or prompt changes.
+    prefetch: bool = False
+    prefetch_limit: int = 5
+    prefetch_token_budget: int = 512
     
     # Auto-save session name (consolidated from standalone auto_save param)
     # When set, automatically saves session to memory with this name
@@ -264,6 +270,9 @@ class MemoryConfig:
             "learn": learn_dict,
             "history": self.history,
             "history_limit": self.history_limit,
+            "prefetch": self.prefetch,
+            "prefetch_limit": self.prefetch_limit,
+            "prefetch_token_budget": self.prefetch_token_budget,
             "auto_save": self.auto_save,
         }
 

src/praisonai-agents/tests/integration/test_memory_prefetch_real.py

@@ -0,0 +1,39 @@
+"""Opt-in real-provider smoke for memory prefetch.
+
+Run with RUN_REAL_KEY_TESTS=1 and a configured provider key.
+"""
+
+import os
+
+import pytest
+
+
+@pytest.mark.skipif(
+    os.environ.get("RUN_REAL_KEY_TESTS") != "1",
+    reason="requires an explicitly enabled real provider",
+)
+def test_real_agent_uses_prefetched_memory(tmp_path, monkeypatch):
+    from praisonaiagents import Agent, MemoryConfig
+
+    monkeypatch.setenv("PRAISONAI_MEMORY_DIR", str(tmp_path))
+    config = MemoryConfig(
+        backend="file",
+        user_id="prefetch-real-smoke",
+        prefetch=True,
+        prefetch_limit=3,
+    )
+    writer = Agent(name="writer", instructions="Store user facts.", memory=config)
+    writer.store_memory(
+        "The user's project codename is NEBULA-7.",
+        memory_type="long_term",
+    )
+
+    reader = Agent(
+        name="reader",
+        instructions="Answer only from recalled memories.",
+        memory=config,
+    )
+    result = reader.start("What is my project codename?")
+
+    print(result)
+    assert "NEBULA-7" in str(result)

src/praisonai-agents/tests/unit/memory/test_prefetch.py

@@ -0,0 +1,158 @@
+"""Tests for opt-in, turn-start long-term memory prefetch."""
+
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from praisonaiagents import Agent, MemoryConfig
+from praisonaiagents.context.tokens import estimate_tokens_heuristic
+
+
+def _agent(config: MemoryConfig) -> Agent:
+    if config.user_id is None:
+        config.user_id = "prefetch-test-user"
+    agent = Agent(
+        name="prefetch-test",
+        instructions="Answer from available context.",
+        llm="gpt-4o-mini",
+        memory=config,
+    )
+    # Message construction does not need a provider client; keep the unit test
+    # offline by selecting the existing custom-LLM fallback branch.
+    agent._using_custom_llm = True
+    return agent
+
+
+def test_memory_config_prefetch_defaults_and_round_trip():
+    config = MemoryConfig()
+    assert config.prefetch is False
+    assert config.prefetch_limit == 5
+    assert config.prefetch_token_budget == 512
+
+    loaded = MemoryConfig(**MemoryConfig(
+        prefetch=True,
+        prefetch_limit=3,
+        prefetch_token_budget=40,
+    ).to_dict())
+    assert loaded.prefetch is True
+    assert loaded.prefetch_limit == 3
+    assert loaded.prefetch_token_budget == 40
+
+
+def test_prefetch_disabled_does_not_touch_backend():
+    agent = _agent(MemoryConfig(prefetch=False))
+    backend = MagicMock()
+    agent._memory_instance = backend
+
+    assert agent._prefetch_memory("remember this") == ""
+    backend.search_long_term.assert_not_called()
+
+
+def test_prefetch_injects_deduplicated_system_context_before_user_message():
+    agent = _agent(MemoryConfig(prefetch=True, prefetch_limit=2))
+    backend = MagicMock()
+    backend.search_long_term.return_value = [
+        {"memory": "Timezone: Asia/Kolkata"},
+        {"text": "Preferred format: concise"},
+        {"memory": "Timezone: Asia/Kolkata"},
+    ]
+    agent._memory_instance = backend
+
+    recalled = agent._prefetch_memory("What timezone should I use?")
+    messages, _ = agent._build_messages(
+        "What timezone should I use?",
+        memory_prefetch_context=recalled,
+    )
+
+    backend.search_long_term.assert_called_once_with(
+        "What timezone should I use?", limit=2
+    )
+    assert messages[0]["role"] == "system"
+    assert "## Recalled memories" in messages[0]["content"]
+    assert messages[0]["content"].count("Timezone: Asia/Kolkata") == 1
+    assert "Preferred format: concise" in messages[0]["content"]
+    assert messages[-1] == {
+        "role": "user",
+        "content": "What timezone should I use?",
+    }
+
+
+def test_prefetch_limit_and_token_budget_are_enforced_with_ellipsis():
+    agent = _agent(MemoryConfig(
+        prefetch=True,
+        prefetch_limit=2,
+        prefetch_token_budget=8,
+    ))
+    backend = MagicMock()
+    backend.search_long_term.return_value = [
+        {"text": "A long remembered preference that exceeds the budget"},
+        {"text": "second memory"},
+        {"text": "must not be included"},
+    ]
+    agent._memory_instance = backend
+
+    recalled = agent._prefetch_memory("preference")
+
+    assert recalled.endswith("…")
+    assert "must not be included" not in recalled
+    assert estimate_tokens_heuristic(recalled) <= 8
+
+
+def test_prefetch_decodes_file_memory_nested_item_shape():
+    agent = _agent(MemoryConfig(prefetch=True, prefetch_limit=2))
+    backend = MagicMock()
+    # FileMemory.search_long_term nests the payload under "item".
+    backend.search_long_term.return_value = [
+        {"type": "long_term", "item": {"content": "User's name is John"}, "score": 1.0},
+        {"type": "long_term", "item": {"content": "Prefers dark mode"}, "score": 0.8},
+    ]
+    agent._memory_instance = backend
+
+    recalled = agent._prefetch_memory("what do you know about me?")
+
+    assert "User's name is John" in recalled
+    assert "Prefers dark mode" in recalled
+
+
+def test_prefetch_tolerates_malformed_numeric_config():
+    agent = _agent(MemoryConfig(prefetch=True))
+    # Simulate malformed config values that would otherwise raise on int().
+    agent._memory_config.prefetch_limit = None
+    agent._memory_config.prefetch_token_budget = "not-a-number"
+    backend = MagicMock()
+    backend.search_long_term.return_value = [{"text": "recalled fact"}]
+    agent._memory_instance = backend
+
+    recalled = agent._prefetch_memory("question")
+
+    assert "recalled fact" in recalled
+
+
+def test_prefetch_backend_failure_is_non_fatal():
+    agent = _agent(MemoryConfig(prefetch=True))
+    backend = MagicMock()
+    backend.search_long_term.side_effect = RuntimeError("backend unavailable")
+    agent._memory_instance = backend
+
+    assert agent._prefetch_memory("question") == ""
+
+
+@pytest.mark.asyncio
+async def test_async_prefetch_uses_async_memory_search():
+    agent = _agent(MemoryConfig(prefetch=True, prefetch_limit=1))
+    agent.asearch_memory = AsyncMock(return_value=[{"text": "async memory"}])
+
+    recalled = await agent._aprefetch_memory("async question")
+
+    assert recalled == "- async memory"
+    agent.asearch_memory.assert_awaited_once_with(
+        "async question", memory_type="long_term", limit=1
+    )
+
+
+@pytest.mark.asyncio
+async def test_async_prefetch_failure_is_non_fatal():
+    agent = _agent(MemoryConfig(prefetch=True))
+    agent.asearch_memory = AsyncMock(side_effect=RuntimeError("offline"))
+
+    assert await agent._aprefetch_memory("async question") == ""