fix: remove dead code, fix JSON parsing, add facts validation

Uses the main agent's existing LLM provider to extract facts from
conversations, then stores them with infer=False — bypassing mem0's
default GPT-nano call.

- extract_facts(): sends conversation to provider.chat() with one-liner prompt
- store_facts(): stores each fact via mem0 with infer=False
- consolidate(): calls extract_facts + store_facts instead of add_conversation

nanobot/agent/loop.py

@@ -40,8 +40,8 @@ class AgentLoop:
     5. Sends responses back
     """
 
-    # Server-side context management: Anthropic trims old tool results and preserves all
-    # thinking blocks (keep="all" maximises cache hits). Client keeps full history.
+    # Server-side context management: Anthropic preserves all thinking blocks
+    # and clears old tool results only when approaching the 200k context limit.
     CONTEXT_MANAGEMENT = {
         "edits": [
             {
@@ -50,7 +50,11 @@ class AgentLoop:
             },
             {
                 "type": "clear_tool_uses_20250919",
-                "trigger": {"type": "input_tokens", "value": 80000},
+                # Raised from 80k to 195k to avoid premature cache invalidation.
+                # For conversations with few tool uses (e.g., 18 uses over 182k tokens),
+                # cache stability (saves 169k/turn) >> clearing benefit (13-26k one-time).
+                # Leaves 5k headroom before hitting 200k standard context limit.
+                "trigger": {"type": "input_tokens", "value": 195000},
                 "keep": {"type": "tool_uses", "value": 5},
             },
         ]
@@ -151,9 +155,28 @@ class AgentLoop:
         # Register native Anthropic tools
         self.tools.register(BashTool20250124())
         self.tools.register(EditTool20250728())
-        self.tools.register(ComputerTool20251124())
+        # self.tools.register(ComputerTool20251124())  # Disabled - VM unavailable
 
-        logger.info("Registered native Anthropic tools: bash, text_editor, computer")
+        logger.info("Registered native Anthropic tools: bash, text_editor")
+
+        # Register mem0 memory tools (if enabled)
+        from nanobot.agent.memory_mem0 import HAS_MEM0
+        if self.mem0_config and self.mem0_config.get("enabled") and HAS_MEM0:
+            from nanobot.agent.memory_mem0 import Mem0MemoryStore
+            from nanobot.agent.tools.memory_tools import (
+                Mem0ToolContext, MemorySearchTool, MemoryListTool,
+                MemoryAddTool, MemoryUpdateTool, MemoryDeleteTool,
+                MemoryConsolidateTool,
+            )
+            store = Mem0MemoryStore(self.workspace, config=self.mem0_config)
+            self._mem0_ctx = Mem0ToolContext(store, self._consolidate_memory)
+            self.tools.register(MemorySearchTool(self._mem0_ctx))
+            self.tools.register(MemoryListTool(self._mem0_ctx))
+            self.tools.register(MemoryAddTool(self._mem0_ctx))
+            self.tools.register(MemoryUpdateTool(self._mem0_ctx))
+            self.tools.register(MemoryDeleteTool(self._mem0_ctx))
+            self.tools.register(MemoryConsolidateTool(self._mem0_ctx))
+            logger.info("Registered mem0 memory tools")
     
     async def run(self) -> None:
         """Run the agent loop, processing messages from the bus."""
@@ -332,6 +355,9 @@ class AgentLoop:
         if isinstance(cron_tool, CronTool):
             cron_tool.set_context(msg.channel, msg.chat_id)
 
+        if hasattr(self, '_mem0_ctx'):
+            self._mem0_ctx.set_context(msg.channel, msg.chat_id, session)
+
         # Track media for this turn (screenshots from computer tool)
         media_paths_for_turn: list[str] = []
 
@@ -541,15 +567,38 @@ class AgentLoop:
             reasoning_content=final_reasoning,
         )
 
-        # Save to session: user message + full tool chain (tool_use, tool_results, thinking, final reply)
+        # Save to session: mem0 context (if present) + user message + full tool chain
         # Store current_message (not msg.content) so the time prefix is preserved
-        # and cache keys match on subsequent turns
         # Include sender_id to distinguish real user messages from system-generated ones
+
+        # Find and save mem0 injection (appears just before current user message)
+        # build_messages returns: [...history, mem0_user, mem0_asst, current_user]
+        # turn_start = len(messages), so mem0 is at turn_start-3 and turn_start-2
+        # This makes mem0 part of immutable history, stabilizing cache across turns
+        if turn_start >= 3:
+            potential_mem0_user = messages[turn_start - 3]
+            potential_mem0_asst = messages[turn_start - 2]
+            if (potential_mem0_user.get("role") == "user" and
+                potential_mem0_user.get("content") == "[Memory context]" and
+                potential_mem0_asst.get("role") == "assistant"):
+                session.add_raw_message(potential_mem0_user)
+                session.add_raw_message(potential_mem0_asst)
+
         session.add_message("user", current_message, sender_id=msg.sender_id)
         for chain_msg in messages[turn_start:]:
             session.add_raw_message(chain_msg)
         self.sessions.save(session)
 
+        # Deferred trim: if memory_consolidate ran mid-turn, it set a pending trim
+        # flag instead of mutating the live session. Now that the turn's tool chain
+        # is fully saved, we can safely trim at a clean boundary.
+        pending = getattr(session, '_pending_trim', 0)
+        if pending > 0:
+            session.messages = self._trim_to_clean_boundary(session.messages, pending)
+            session._pending_trim = 0
+            self.sessions.save(session)
+            logger.info(f"Deferred trim applied, session now {len(session.messages)} messages")
+
         return OutboundMessage(
             channel=msg.channel,
             chat_id=msg.chat_id,
@@ -737,7 +786,17 @@ class AgentLoop:
             reasoning_content=final_reasoning,
         )
 
-        # Save to session: user message + full tool chain
+        # Save to session: mem0 (if present) + user message + full tool chain
+        # Find and save mem0 injection for cache stability
+        if turn_start >= 3:
+            potential_mem0_user = messages[turn_start - 3]
+            potential_mem0_asst = messages[turn_start - 2]
+            if (potential_mem0_user.get("role") == "user" and
+                potential_mem0_user.get("content") == "[Memory context]" and
+                potential_mem0_asst.get("role") == "assistant"):
+                session.add_raw_message(potential_mem0_user)
+                session.add_raw_message(potential_mem0_asst)
+
         session.add_message("user", f"[System: {msg.sender_id}] {msg.content}")
         for chain_msg in messages[turn_start:]:
             session.add_raw_message(chain_msg)
@@ -751,6 +810,34 @@ class AgentLoop:
             metadata=outbound_metadata,
         )
     
+    @staticmethod
+    def _trim_to_clean_boundary(messages: list[dict], keep_count: int) -> list[dict]:
+        """Trim messages to approximately keep_count, starting at a user message boundary.
+
+        Naive slicing (messages[-keep_count:]) can cut into a tool chain, leaving
+        orphaned tool_result messages at the start. This finds the nearest user
+        message (role="user") at or before the cut point and trims there.
+        """
+        if not messages or keep_count <= 0:
+            return []
+        if keep_count >= len(messages):
+            return messages
+
+        cut = len(messages) - keep_count
+        # Walk forward from cut to find a "user" role message (start of a turn)
+        # that isn't a tool result. Tool results have role="tool", user messages
+        # have role="user" — but after conversion, tool results ARE user messages.
+        # In session storage, they're still role="tool", so we look for role="user".
+        for i in range(cut, len(messages)):
+            if messages[i].get("role") == "user":
+                return messages[i:]
+        # If no user message found after cut, try walking backward
+        for i in range(cut - 1, -1, -1):
+            if messages[i].get("role") == "user":
+                return messages[i:]
+        # Fallback: return everything (shouldn't happen in practice)
+        return messages
+
     async def _consolidate_memory(self, session, archive_all: bool = False) -> None:
         """Consolidate session into MEMORY.md + HISTORY.md.
 
@@ -774,6 +861,17 @@ class AgentLoop:
                 archive_all=archive_all,
                 memory_window=self.memory_window,
             )
+            # archive_all (/new) runs at a turn boundary — safe to trim now.
+            # Mid-turn (memory_consolidate tool) — defer trim to end of turn
+            # to avoid orphaning tool_use IDs in the active tool chain.
+            if archive_all:
+                session.messages = []
+                self.sessions.save(session)
+                logger.info("Mem0 consolidation done, session cleared (archive_all)")
+            else:
+                keep_count = min(10, max(2, self.memory_window // 2))
+                session._pending_trim = keep_count
+                logger.info(f"Mem0 consolidation done, trim deferred (keep={keep_count})")
             return
         else:
             memory = MemoryStore(self.workspace)
@@ -864,7 +962,7 @@ Respond with ONLY valid JSON, no markdown fences."""
                 if update != current_memory:
                     memory.write_long_term(update)
 
-            session.messages = session.messages[-keep_count:] if keep_count else []
+            session.messages = self._trim_to_clean_boundary(session.messages, keep_count) if keep_count else []
             self.sessions.save(session)
             logger.info(f"Memory consolidation done, session trimmed to {len(session.messages)} messages")
         except Exception as e:

nanobot/agent/memory_mem0.py

@@ -47,6 +47,7 @@ class Mem0MemoryStore:
 
         today = datetime.now().strftime("%Y-%m-%d")
         custom_prompt = f"Extract dated facts from this conversation as JSON: {{\"facts\": [...]}}. Today is {today}.\n\n"
+        self.custom_prompt = custom_prompt
 
         # Initialize mem0 with optional config + custom prompt
         # Extract only MemoryConfig-relevant fields
@@ -57,9 +58,6 @@ class Mem0MemoryStore:
             if key in raw_config:
                 mem0_cfg_dict[key] = raw_config[key]
         logger.debug(f"Extracted for MemoryConfig: {list(mem0_cfg_dict.keys())}")
-        logger.debug(f"Custom prompt length: {len(custom_prompt)} chars")
-        self.custom_prompt = custom_prompt
-        mem0_cfg_dict["custom_fact_extraction_prompt"] = custom_prompt
         mem0_config = MemoryConfig(**mem0_cfg_dict)
         logger.debug(f"MemoryConfig created: vector_store={mem0_config.vector_store.provider if mem0_config.vector_store else None}")
         self.memory = Memory(config=mem0_config)
@@ -156,21 +154,15 @@ class Mem0MemoryStore:
         provider: Any,
         model: str,
     ) -> list[str]:
-        """
-        Extract facts from conversation using the main agent's LLM provider.
-
-        Uses the same provider/model already running (e.g. Haiku via Claude Max),
-        avoiding a separate LLM call to mem0's default GPT-nano.
-        """
+        """Extract facts from conversation using the main agent's LLM provider."""
         import json as _json
 
-        # Build conversation text for extraction
         conv_text = ""
         for msg in messages:
             role = msg.get("role", "unknown")
-            content = msg.get("content", "")
-            if isinstance(content, str) and content.strip():
-                conv_text += f"{role}: {content}\n\n"
+            content_val = msg.get("content", "")
+            if isinstance(content_val, str) and content_val.strip():
+                conv_text += f"{role}: {content_val}\n\n"
 
         if not conv_text.strip():
             return []
@@ -186,22 +178,19 @@ class Mem0MemoryStore:
                 max_tokens=2000,
                 temperature=0.3,
             )
-
-            # Parse the JSON response — LLMResponse.content is a string
-            text = response.content or ""
-            # Strip markdown code fences if present
-            text = text.strip()
+            text = (response.content or "").strip()
             if text.startswith("```"):
-                text = text.split("\n", 1)[1] if "\n" in text else text[3:]
-            if text.endswith("```"):
-                text = text[:-3]
-            text = text.strip()
-
+                text = text.split("```")[1]
+                if text.startswith("json"):
+                    text = text[4:]
+                text = text.strip()
             data = _json.loads(text)
             facts = data.get("facts", [])
+            if not isinstance(facts, list):
+                logger.warning(f"LLM returned non-list facts: {type(facts)}")
+                return []
             logger.debug(f"Extracted {len(facts)} facts using {model}")
             return facts
-
         except Exception as e:
             logger.error(f"Fact extraction failed: {e}")
             return []
@@ -212,12 +201,7 @@ class Mem0MemoryStore:
         user_id: str,
         session_id: str | None = None,
     ) -> None:
-        """
-        Store pre-extracted facts in mem0 with infer=False.
-
-        Bypasses mem0's built-in LLM extraction — facts are already
-        in final form from extract_facts().
-        """
+        """Store pre-extracted facts in mem0 with infer=False."""
         if not facts:
             return
 
@@ -309,7 +293,8 @@ class Mem0MemoryStore:
         """
         Consolidate session messages into mem0 memory.
 
-        Facts are extracted using the main agent's LLM provider, then stored with infer=False.
+        Unlike the original MemoryStore, mem0 handles extraction automatically,
+        so this just needs to feed recent messages to mem0.
 
         Returns True on success.
         """
@@ -351,21 +336,9 @@ class Mem0MemoryStore:
                 role = msg.get("role")
                 content = msg.get("content")
 
-                # Keep tool results but truncate long ones — they often contain
-                # the actual substance (file reads, search results, web pages).
-                # The extraction prompt handles ignoring code/JSON noise.
+                # Skip tool results — raw bash output, file contents, and JSON
+                # get misinterpreted by the extraction LLM as user interests
                 if role == "tool":
-                    if isinstance(content, list):
-                        text_parts = [
-                            block.get("content", "") if isinstance(block, dict) else str(block)
-                            for block in content
-                        ]
-                        content = " ".join(text_parts).strip()
-                    if isinstance(content, str) and len(content) > 2000:
-                        content = content[:2000]
-                    if not content or (isinstance(content, str) and len(content.strip()) < 10):
-                        continue
-                    mem0_messages.append({"role": "user", "content": content})
                     continue
 
                 # Skip system messages — they're boilerplate instructions, not facts

nanobot/agent/tools/anthropic/bash.py

@@ -1,114 +1,117 @@
-"""BashTool20250124 - Persistent bash session with sentinel-based output.
+"""BashTool20250124 - Persistent bash session with async buffer polling.
 
-Anthropic's native bash_20250124 tool with a long-running session.
+Based on Anthropic's reference implementation from anthropic-quickstarts.
+Uses asyncio.create_subprocess_shell + direct buffer reads instead of
+threaded readline, which avoids exhausting the default ThreadPoolExecutor.
 """
 
 import asyncio
-import subprocess
-import uuid
+import os
 from typing import Any, Literal
 
-from nanobot.agent.tools.anthropic.base import BaseAnthropicTool, ToolResult
+from nanobot.agent.tools.anthropic.base import BaseAnthropicTool, ToolResult, ToolError
 
 
 class _BashSession:
-    """Manages a persistent bash subprocess with sentinel-based output reading."""
+    """A session of a bash shell.
+
+    Uses asyncio subprocess with direct buffer polling — no threads.
+    Based on anthropics/anthropic-quickstarts computer-use-demo.
+    """
+
+    command: str = "/bin/bash"
+    _output_delay: float = 0.2  # seconds between buffer polls
+    _timeout: float = 120.0  # seconds
+    _sentinel: str = "<<exit>>"
 
     def __init__(self):
-        self.process: subprocess.Popen | None = None
-        self._start()
+        self._started = False
+        self._timed_out = False
+        self._process: asyncio.subprocess.Process | None = None
 
-    def _start(self):
-        """Start the bash process."""
-        self.process = subprocess.Popen(
-            ["bash"],
-            stdin=subprocess.PIPE,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.STDOUT,
-            text=True,
-            bufsize=1,
+    async def start(self):
+        if self._started:
+            return
+
+        self._process = await asyncio.create_subprocess_shell(
+            self.command,
+            preexec_fn=os.setsid,
+            shell=True,
+            bufsize=0,
+            stdin=asyncio.subprocess.PIPE,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
         )
+        self._started = True
 
-    def restart(self):
-        """Restart the bash session."""
-        if self.process:
-            self.process.terminate()
-            try:
-                self.process.wait(timeout=5)
-            except subprocess.TimeoutExpired:
-                self.process.kill()
-                self.process.wait()
-        self._start()
+    def stop(self):
+        """Terminate the bash shell."""
+        if not self._started:
+            return
+        if self._process and self._process.returncode is None:
+            self._process.terminate()
 
-    async def run_command(self, command: str, timeout: float = 120.0) -> str:
-        """Run a command in the persistent bash session.
+    async def run(self, command: str) -> ToolResult:
+        """Execute a command in the bash shell."""
+        if not self._started:
+            raise ToolError("Session has not started.")
+        if self._process is None or self._process.returncode is not None:
+            return ToolResult(
+                system="tool must be restarted",
+                error=f"bash has exited with returncode "
+                      f"{self._process.returncode if self._process else 'unknown'}",
+            )
+        if self._timed_out:
+            raise ToolError(
+                f"timed out: bash has not returned in {self._timeout} seconds "
+                "and must be restarted",
+            )
 
-        Uses a unique sentinel to detect command completion.
+        assert self._process.stdin
+        assert self._process.stdout
+        assert self._process.stderr
 
-        Args:
-            command: Bash command to execute
-            timeout: Maximum time to wait for command completion (seconds)
+        # Send command + sentinel on its own line so heredoc terminators
+        # aren't corrupted (EOF; echo '...' ≠ EOF)
+        self._process.stdin.write(
+            command.encode() + f"\necho '{self._sentinel}'\n".encode()
+        )
+        await self._process.stdin.drain()
 
-        Returns:
-            Command output (stdout + stderr combined)
+        # Poll stdout buffer until sentinel appears — no threads involved
+        try:
+            async with asyncio.timeout(self._timeout):
+                while True:
+                    await asyncio.sleep(self._output_delay)
+                    output = self._process.stdout._buffer.decode()
+                    if self._sentinel in output:
+                        output = output[: output.index(self._sentinel)]
+                        break
+        except asyncio.TimeoutError:
+            self._timed_out = True
+            raise ToolError(
+                f"timed out: bash has not returned in {self._timeout} seconds "
+                "and must be restarted",
+            ) from None
 
-        Raises:
-            asyncio.TimeoutError: If command doesn't complete within timeout
-            RuntimeError: If bash process has died
-        """
-        if not self.process or self.process.poll() is not None:
-            raise RuntimeError("Bash process has died")
+        if output.endswith("\n"):
+            output = output[:-1]
 
-        # Generate unique sentinel
-        sentinel = f"<<BASH_COMMAND_DONE_{uuid.uuid4().hex}>>"
+        error = self._process.stderr._buffer.decode()
+        if error.endswith("\n"):
+            error = error[:-1]
 
-        # Send command + sentinel
-        full_command = f"{command}\necho '{sentinel}'\n"
-        self.process.stdin.write(full_command)
-        self.process.stdin.flush()
+        # Clear buffers for next command
+        self._process.stdout._buffer.clear()
+        self._process.stderr._buffer.clear()
 
-        # Read output until sentinel appears
-        output_lines = []
-        start_time = asyncio.get_event_loop().time()
-
-        while True:
-            # Check timeout
-            elapsed = asyncio.get_event_loop().time() - start_time
-            if elapsed > timeout:
-                raise asyncio.TimeoutError(
-                    f"Command timed out after {timeout}s: {command[:50]}..."
-                )
-
-            # Read line (non-blocking via asyncio)
-            try:
-                line = await asyncio.wait_for(
-                    asyncio.to_thread(self.process.stdout.readline),
-                    timeout=1.0,
-                )
-            except asyncio.TimeoutError:
-                # No output yet, continue waiting
-                continue
-
-            if not line:
-                # EOF - process died
-                raise RuntimeError("Bash process terminated unexpectedly")
-
-            # Check for sentinel
-            if sentinel in line:
-                break
-
-            output_lines.append(line.rstrip("\n"))
-
-        return "\n".join(output_lines)
-
-    def __del__(self):
-        """Clean up bash process on deletion."""
-        if self.process:
-            self.process.terminate()
-            try:
-                self.process.wait(timeout=2)
-            except subprocess.TimeoutExpired:
-                self.process.kill()
+        # Return as ToolResult (our loop handles this type)
+        if error and output:
+            return ToolResult(output=f"{output}\n\nstderr: {error}")
+        elif error:
+            return ToolResult(output=error)
+        else:
+            return ToolResult(output=output if output else "(no output)")
 
 
 class BashTool20250124(BaseAnthropicTool):
@@ -124,10 +127,10 @@ class BashTool20250124(BaseAnthropicTool):
 
     api_type: Literal["bash_20250124"] = "bash_20250124"
     name: Literal["bash"] = "bash"
-    beta_flag: str = "computer-use-2025-11-24"
+    beta_flag: str | None = None
 
     def __init__(self):
-        self._session = _BashSession()
+        self._session: _BashSession | None = None
 
     async def __call__(
         self,
@@ -135,39 +138,26 @@ class BashTool20250124(BaseAnthropicTool):
         restart: bool = False,
         **kwargs: Any,
     ) -> ToolResult:
-        """Execute bash command or restart session.
-
-        Args:
-            command: Bash command to execute (optional)
-            restart: Restart the bash session (optional)
-            **kwargs: Additional arguments (ignored)
-
-        Returns:
-            ToolResult with command output or error
-        """
         if restart:
-            self._session.restart()
-            return ToolResult(output="Bash session restarted successfully.")
+            if self._session:
+                self._session.stop()
+            self._session = _BashSession()
+            await self._session.start()
+            return ToolResult(system="tool has been restarted.")
 
-        if not command:
-            return ToolResult(
-                error="Either 'command' or 'restart=True' must be provided."
-            )
+        if self._session is None:
+            self._session = _BashSession()
+            await self._session.start()
 
-        try:
-            output = await self._session.run_command(command)
-            return ToolResult(output=output if output else "(no output)")
-        except asyncio.TimeoutError as e:
-            return ToolResult(error=f"Command timed out: {e}")
-        except Exception as e:
-            return ToolResult(error=f"{e}")
+        if command is not None:
+            try:
+                return await self._session.run(command)
+            except ToolError as e:
+                return ToolResult(error=str(e))
+
+        return ToolResult(error="Either 'command' or 'restart=True' must be provided.")
 
     def to_params(self) -> dict[str, Any]:
-        """Convert to Anthropic API tool parameter format.
-
-        Returns:
-            Tool definition for Anthropic API with bash_20250124 type
-        """
         return {
             "type": self.api_type,
             "name": self.name,

nanobot/agent/tools/anthropic/computer.py

@@ -67,13 +67,14 @@ class ComputerTool20251124(BaseAnthropicTool):
         self.display_height_px = display_height_px
 
     def to_params(self):
-        """Return tool definition for API."""
+        """Return tool definition for API.
+
+        NOTE: display_width_px, display_height_px, and enable_zoom are NOT
+        valid parameters for computer_20251124 and cause API hangs if sent.
+        """
         return {
             "type": self.api_type,
             "name": self.name,
-            "display_width_px": self.display_width_px,
-            "display_height_px": self.display_height_px,
-            "enable_zoom": True,
         }
 
     async def __call__(

nanobot/agent/tools/anthropic/edit.py

@@ -20,7 +20,7 @@ class EditTool20250728(BaseAnthropicTool):
 
     api_type: Literal["text_editor_20250728"] = "text_editor_20250728"
     name: Literal["str_replace_based_edit_tool"] = "str_replace_based_edit_tool"
-    beta_flag: str = "computer-use-2025-11-24"
+    beta_flag: str | None = None
 
     async def __call__(
         self,

nanobot/agent/tools/memory_tools.py

@@ -0,0 +1,230 @@
+"""Mem0 memory tools — expose semantic memory to the agent."""
+
+from __future__ import annotations
+
+import json
+from typing import Any, TYPE_CHECKING
+
+from loguru import logger
+
+from nanobot.agent.tools.base import Tool
+
+if TYPE_CHECKING:
+    from nanobot.agent.memory_mem0 import Mem0MemoryStore
+
+
+class Mem0ToolContext:
+    """Shared mutable state injected into every mem0 tool."""
+
+    def __init__(self, store: Mem0MemoryStore, consolidate_fn):
+        self.store = store
+        self.consolidate_fn = consolidate_fn  # async (session, archive_all) -> None
+        self.user_id: str = "unknown"
+        self.session = None
+
+    def set_context(self, channel: str, chat_id: str, session=None):
+        self.user_id = f"{channel}_{chat_id}"
+        self.session = session
+
+
+class MemorySearchTool(Tool):
+    """Search memories semantically."""
+
+    name = "memory_search"
+    description = (
+        "Search your long-term memory for facts relevant to a query. "
+        "Returns the most relevant memories ranked by similarity."
+    )
+    parameters = {
+        "type": "object",
+        "properties": {
+            "query": {
+                "type": "string",
+                "description": "Natural-language search query",
+            },
+            "limit": {
+                "type": "integer",
+                "description": "Max results to return (default 5)",
+                "minimum": 1,
+                "maximum": 20,
+            },
+        },
+        "required": ["query"],
+    }
+
+    def __init__(self, ctx: Mem0ToolContext):
+        self._ctx = ctx
+
+    async def execute(self, query: str, limit: int = 5, **kw: Any) -> str:
+        results = self._ctx.store.search_memories(
+            query=query,
+            user_id=self._ctx.user_id,
+            limit=limit,
+        )
+        if not results:
+            return "No memories found."
+        lines = []
+        for i, mem in enumerate(results, 1):
+            text = mem.get("memory", "")
+            score = mem.get("score")
+            mid = mem.get("id", "")
+            score_str = f"  (score: {score:.2f})" if score else ""
+            lines.append(f"{i}. [{mid}] {text}{score_str}")
+        return "\n".join(lines)
+
+
+class MemoryListTool(Tool):
+    """List all memories for the current user."""
+
+    name = "memory_list"
+    description = (
+        "List ALL stored memories for the current user. "
+        "Use memory_search for targeted lookup; use this to browse everything."
+    )
+    parameters = {
+        "type": "object",
+        "properties": {},
+    }
+
+    def __init__(self, ctx: Mem0ToolContext):
+        self._ctx = ctx
+
+    async def execute(self, **kw: Any) -> str:
+        memories = self._ctx.store.get_all_memories(self._ctx.user_id)
+        if not memories:
+            return "No memories stored."
+        lines = []
+        for i, mem in enumerate(memories, 1):
+            text = mem.get("memory", "")
+            mid = mem.get("id", "")
+            lines.append(f"{i}. [{mid}] {text}")
+        return f"{len(memories)} memories:\n" + "\n".join(lines)
+
+
+class MemoryAddTool(Tool):
+    """Add a fact to long-term memory."""
+
+    name = "memory_add"
+    description = (
+        "Store a new fact or piece of information in long-term memory. "
+        "The content will be processed by the extraction LLM and stored as one or more facts."
+    )
+    parameters = {
+        "type": "object",
+        "properties": {
+            "content": {
+                "type": "string",
+                "description": "The fact or information to remember",
+            },
+        },
+        "required": ["content"],
+    }
+
+    def __init__(self, ctx: Mem0ToolContext):
+        self._ctx = ctx
+
+    async def execute(self, content: str, **kw: Any) -> str:
+        try:
+            result = self._ctx.store.memory.add(
+                [{"role": "user", "content": content}],
+                user_id=self._ctx.user_id,
+            )
+            facts_count = len(result.get("results", [])) if result else 0
+            return f"Added to memory. {facts_count} fact(s) extracted."
+        except Exception as e:
+            logger.error(f"memory_add failed: {e}")
+            return f"Error adding memory: {e}"
+
+
+class MemoryUpdateTool(Tool):
+    """Update an existing memory by ID."""
+
+    name = "memory_update"
+    description = (
+        "Update the content of an existing memory. "
+        "Use memory_list or memory_search first to find the memory ID."
+    )
+    parameters = {
+        "type": "object",
+        "properties": {
+            "memory_id": {
+                "type": "string",
+                "description": "The memory ID to update",
+            },
+            "content": {
+                "type": "string",
+                "description": "The new content for this memory",
+            },
+        },
+        "required": ["memory_id", "content"],
+    }
+
+    def __init__(self, ctx: Mem0ToolContext):
+        self._ctx = ctx
+
+    async def execute(self, memory_id: str, content: str, **kw: Any) -> str:
+        try:
+            self._ctx.store.update_memory(memory_id, content)
+            return f"Memory {memory_id} updated."
+        except Exception as e:
+            logger.error(f"memory_update failed: {e}")
+            return f"Error updating memory: {e}"
+
+
+class MemoryDeleteTool(Tool):
+    """Delete a memory by ID."""
+
+    name = "memory_delete"
+    description = (
+        "Delete a specific memory by its ID. "
+        "Use memory_list or memory_search first to find the memory ID."
+    )
+    parameters = {
+        "type": "object",
+        "properties": {
+            "memory_id": {
+                "type": "string",
+                "description": "The memory ID to delete",
+            },
+        },
+        "required": ["memory_id"],
+    }
+
+    def __init__(self, ctx: Mem0ToolContext):
+        self._ctx = ctx
+
+    async def execute(self, memory_id: str, **kw: Any) -> str:
+        try:
+            self._ctx.store.delete_memory(memory_id)
+            return f"Memory {memory_id} deleted."
+        except Exception as e:
+            logger.error(f"memory_delete failed: {e}")
+            return f"Error deleting memory: {e}"
+
+
+class MemoryConsolidateTool(Tool):
+    """Trigger memory consolidation for the current session."""
+
+    name = "memory_consolidate"
+    description = (
+        "Extract and store facts from the current conversation into long-term memory. "
+        "Normally this happens automatically on /new, but you can trigger it manually."
+    )
+    parameters = {
+        "type": "object",
+        "properties": {},
+    }
+
+    def __init__(self, ctx: Mem0ToolContext):
+        self._ctx = ctx
+
+    async def execute(self, **kw: Any) -> str:
+        session = self._ctx.session
+        if not session:
+            return "Error: no active session."
+        try:
+            await self._ctx.consolidate_fn(session, archive_all=False)
+            return "Memory consolidation complete."
+        except Exception as e:
+            logger.error(f"memory_consolidate failed: {e}")
+            return f"Error during consolidation: {e}"

nanobot/providers/anthropic_oauth.py

@@ -59,9 +59,83 @@ class AnthropicOAuthProvider(LLMProvider):
     async def _get_client(self) -> httpx.AsyncClient:
         """Get or create async HTTP client."""
         if self._client is None:
-            self._client = httpx.AsyncClient(timeout=300.0)
+            self._client = httpx.AsyncClient(
+                timeout=httpx.Timeout(300.0, pool=30.0),
+            )
         return self._client
 
+    async def _reset_client(self) -> None:
+        """Destroy and recreate the HTTP client after connection errors."""
+        old = self._client
+        self._client = None
+        if old:
+            try:
+                await old.aclose()
+            except Exception:
+                pass
+        logger.warning("Reset httpx client (pool recycled)")
+
+    async def _diagnose_connectivity(self) -> None:
+        """Run diagnostics when ConnectTimeout occurs to understand why."""
+        import socket
+        import asyncio
+
+        # 1. Raw socket test (bypasses httpx entirely)
+        try:
+            t0 = __import__('time').monotonic()
+            s = socket.create_connection(('api.anthropic.com', 443), timeout=10)
+            elapsed = __import__('time').monotonic() - t0
+            s.close()
+            logger.warning(f"DIAG: raw socket connect OK in {elapsed:.3f}s")
+        except Exception as e:
+            logger.error(f"DIAG: raw socket connect FAILED: {e}")
+
+        # 2. asyncio connect test (same event loop)
+        try:
+            t0 = __import__('time').monotonic()
+            reader, writer = await asyncio.wait_for(
+                asyncio.open_connection('api.anthropic.com', 443),
+                timeout=10.0,
+            )
+            elapsed = __import__('time').monotonic() - t0
+            writer.close()
+            await writer.wait_closed()
+            logger.warning(f"DIAG: asyncio connect OK in {elapsed:.3f}s")
+        except Exception as e:
+            logger.error(f"DIAG: asyncio connect FAILED: {e}")
+
+        # 3. Fresh httpx client test (new pool)
+        try:
+            t0 = __import__('time').monotonic()
+            async with httpx.AsyncClient(timeout=10.0) as fresh:
+                r = await fresh.get('https://api.anthropic.com/')
+                elapsed = __import__('time').monotonic() - t0
+                logger.warning(f"DIAG: fresh httpx OK in {elapsed:.3f}s (status={r.status_code})")
+        except Exception as e:
+            logger.error(f"DIAG: fresh httpx FAILED: {e}")
+
+        # 4. DNS resolution
+        try:
+            ips = socket.getaddrinfo('api.anthropic.com', 443)
+            logger.warning(f"DIAG: DNS resolved to {len(ips)} entries, first={ips[0][4][0]}")
+        except Exception as e:
+            logger.error(f"DIAG: DNS FAILED: {e}")
+
+        # 5. Connection pool state of the broken client
+        if self._client:
+            transport = self._client._transport
+            if hasattr(transport, '_pool'):
+                pool = transport._pool
+                conns = getattr(pool, '_connections', [])
+                reqs = getattr(pool, '_requests', [])
+                logger.warning(
+                    f"DIAG: pool state: {len(conns)} connections, "
+                    f"{len(reqs)} pending requests"
+                )
+                for i, conn in enumerate(conns[:5]):
+                    state = getattr(conn, '_state', 'unknown')
+                    logger.warning(f"DIAG: conn[{i}] state={state}")
+
     def _prepare_messages(
         self,
         messages: list[dict[str, Any]]
@@ -252,18 +326,35 @@ class AnthropicOAuthProvider(LLMProvider):
         """Make request to Anthropic API."""
         client = await self._get_client()
 
-        # Cache the last user message so conversation history is cached across turns
-        if messages:
-            last = messages[-1]
-            if last.get("role") == "user":
-                content = last["content"]
-                if isinstance(content, str):
-                    last = {**last, "content": [{"type": "text", "text": content, "cache_control": {"type": "ephemeral"}}]}
-                elif isinstance(content, list) and content:
-                    new_content = list(content)
-                    new_content[-1] = {**new_content[-1], "cache_control": {"type": "ephemeral"}}
-                    last = {**last, "content": new_content}
-                messages = messages[:-1] + [last]
+        # Add cache breakpoints on the last TWO user messages (4-breakpoint strategy):
+        # BP3: Second-to-last user message (stable history from previous turn)
+        # BP4: Last user message (current turn, will become BP3 next turn)
+        # This allows BP3 to reuse what BP4 cached last turn.
+        user_indices = [i for i, m in enumerate(messages) if m.get("role") == "user"]
+
+        if len(user_indices) >= 2:
+            # BP3: Second-to-last user message
+            idx = user_indices[-2]
+            msg = messages[idx]
+            content = msg["content"]
+            if isinstance(content, str):
+                messages[idx] = {**msg, "content": [{"type": "text", "text": content, "cache_control": {"type": "ephemeral"}}]}
+            elif isinstance(content, list) and content:
+                new_content = list(content)
+                new_content[-1] = {**new_content[-1], "cache_control": {"type": "ephemeral"}}
+                messages[idx] = {**msg, "content": new_content}
+
+        if len(user_indices) >= 1:
+            # BP4: Last user message
+            idx = user_indices[-1]
+            msg = messages[idx]
+            content = msg["content"]
+            if isinstance(content, str):
+                messages[idx] = {**msg, "content": [{"type": "text", "text": content, "cache_control": {"type": "ephemeral"}}]}
+            elif isinstance(content, list) and content:
+                new_content = list(content)
+                new_content[-1] = {**new_content[-1], "cache_control": {"type": "ephemeral"}}
+                messages[idx] = {**msg, "content": new_content}
 
         payload: dict[str, Any] = {
             "model": model,
@@ -321,11 +412,43 @@ class AnthropicOAuthProvider(LLMProvider):
             tool_names = [t.get("name", "unnamed") for t in payload["tools"]]
             logger.debug(f"Tool names in request: {tool_names}")
 
-        response = await client.post(
-            self._get_api_url(),
-            headers=headers,
-            json=payload,
-        )
+        # Debug: Log message structure to diagnose orphaned tool_result errors
+        for idx, m in enumerate(payload.get("messages", [])):
+            role = m.get("role", "?")
+            content = m.get("content", "")
+            if isinstance(content, list):
+                block_types = [b.get("type", "?") for b in content]
+                logger.debug(f"  msg[{idx}] role={role} blocks={block_types}")
+            else:
+                logger.debug(f"  msg[{idx}] role={role} text={str(content)[:80]}")
+
+        import asyncio
+        import time as _time
+        _t0 = _time.monotonic()
+        try:
+            response = await client.post(
+                self._get_api_url(),
+                headers=headers,
+                json=payload,
+            )
+        except httpx.ConnectTimeout:
+            elapsed = _time.monotonic() - _t0
+            logger.error(f"ConnectTimeout after {elapsed:.1f}s — running diagnostics")
+            await self._diagnose_connectivity()
+            await self._reset_client()
+            raise
+        except httpx.PoolTimeout:
+            elapsed = _time.monotonic() - _t0
+            logger.error(f"PoolTimeout after {elapsed:.1f}s — resetting client")
+            await self._reset_client()
+            raise
+        except (httpx.ConnectError, httpx.TimeoutException) as e:
+            elapsed = _time.monotonic() - _t0
+            logger.error(f"{type(e).__name__} after {elapsed:.1f}s")
+            raise
+        elapsed = _time.monotonic() - _t0
+        if elapsed > 30:
+            logger.warning(f"Anthropic API slow response: {elapsed:.1f}s")
 
         # Dump rate limit headers for analysis
         try: