feat: extract facts with main agent LLM, bypass mem0 GPT-nano

Instead of hacking mem0's provider system, use the main agent's
existing LLM (already running, already paid for) to extract facts
from conversations, then store them with infer=False.

- extract_facts(): sends conversation to provider.chat() with extraction prompt
- store_facts(): stores each fact via mem0 with infer=False
- consolidate(): calls extract_facts + store_facts instead of add_conversation
- No new files, no Dockerfile changes, no mem0 package patches

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
+    # Server-side context management: Anthropic preserves all thinking blocks
-    # thinking blocks (keep="all" maximises cache hits). Client keeps full history.
+    # 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

@@ -58,356 +58,6 @@ class Mem0MemoryStore:
                 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)
-
-        logger.info("Mem0 memory system initialized with custom nanobot prompt")
-
-    def search_memories(
-        self,
-        query: str,
-        user_id: str,
-        limit: int = 5,
-        session_id: str | None = None,
-    ) -> list[dict[str, Any]]:
-        """
-        Search for relevant memories using semantic search.
-
-        Args:
-            query: Search query (user's current message)
-            user_id: User identifier (e.g., "telegram_12345")
-            limit: Max number of memories to return
-            session_id: Optional session-specific memories
-
-        Returns:
-            List of memory dicts with 'memory' and 'score' keys
-        """
-        try:
-            # Search user-level memories
-            user_memories = self.memory.search(
-                query=query,
-                user_id=user_id,
-                limit=limit
-            )
-
-            results = []
-            if user_memories and "results" in user_memories:
-                results.extend(user_memories["results"])
-
-            # Optionally search session-level memories
-            if session_id:
-                session_memories = self.memory.search(
-                    query=query,
-                    user_id=user_id,
-                    metadata={"session_id": session_id},
-                    limit=limit // 2  # Reserve half for session context
-                )
-                if session_memories and "results" in session_memories:
-                    results.extend(session_memories["results"])
-
-            logger.debug(
-                f"Mem0 search: query='{query[:50]}...', found {len(results)} memories"
-            )
-            return results[:limit]  # Limit total results
-
-        except Exception as e:
-            logger.error(f"Mem0 search failed: {e}")
-            return []
-
-    def add_conversation(
-        self,
-        messages: list[dict[str, Any]],
-        user_id: str,
-        session_id: str | None = None,
-    ) -> None:
-        """
-        Add conversation messages to memory for automatic extraction.
-
-        Args:
-            messages: List of message dicts with 'role' and 'content'
-            user_id: User identifier
-            session_id: Optional session identifier for session-level memories
-        """
-        try:
-            metadata = {}
-            if session_id:
-                metadata["session_id"] = session_id
-
-            # mem0 automatically extracts and stores relevant facts
-            result = self.memory.add(
-                messages,
-                user_id=user_id,
-                metadata=metadata if metadata else None
-            )
-
-            facts_count = len(result.get("results", [])) if result else 0
-            logger.debug(
-                f"Mem0 add: {len(messages)} messages for user {user_id}, extracted {facts_count} facts"
-            )
-        except Exception as e:
-            logger.error(f"Mem0 add failed: {e}")
-
-    async def extract_facts(
-        self,
-        messages: list[dict[str, Any]],
-        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.
-        """
-        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"
-
-        if not conv_text.strip():
-            return []
-
-        extraction_messages = [
-            {"role": "user", "content": self.custom_prompt + conv_text}
-        ]
-
-        try:
-            response = await provider.chat(
-                messages=extraction_messages,
-                model=model,
-                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()
-            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()
-
-            data = _json.loads(text)
-            facts = data.get("facts", [])
-            logger.debug(f"Extracted {len(facts)} facts using {model}")
-            return facts
-
-        except Exception as e:
-            logger.error(f"Fact extraction failed: {e}")
-            return []
-
-    def store_facts(
-        self,
-        facts: list[str],
-        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().
-        """
-        if not facts:
-            return
-
-        metadata = {}
-        if session_id:
-            metadata["session_id"] = session_id
-
-        stored = 0
-        for fact in facts:
-            try:
-                self.memory.add(
-                    fact,
-                    user_id=user_id,
-                    infer=False,
-                    metadata=metadata if metadata else None,
-                )
-                stored += 1
-            except Exception as e:
-                logger.error(f"Failed to store fact '{fact[:50]}...': {e}")
-
-        logger.info(f"Stored {stored}/{len(facts)} facts for user {user_id}")
-
-    def get_memory_context(
-        self,
-        query: str,
-        user_id: str,
-        limit: int = 5
-    ) -> str:
-        """
-        Get formatted memory context for inclusion in system prompt.
-
-        Args:
-            query: Current user query
-            user_id: User identifier
-            limit: Max memories to include
-
-        Returns:
-            Formatted memory context string
-        """
-        memories = self.search_memories(query, user_id, limit=limit)
-
-        if not memories:
-            return ""
-
-        lines = ["## Relevant Memories"]
-        for i, mem in enumerate(memories, 1):
-            memory_text = mem.get("memory", "")
-            # Include score if available for debugging
-            score = mem.get("score", "")
-            score_str = f" (relevance: {score:.2f})" if score else ""
-            lines.append(f"{i}. {memory_text}{score_str}")
-
-        return "\n".join(lines)
-
-    def update_memory(self, memory_id: str, data: dict[str, Any]) -> None:
-        """Update a specific memory by ID."""
-        try:
-            self.memory.update(memory_id, data)
-            logger.debug(f"Mem0 update: memory_id={memory_id}")
-        except Exception as e:
-            logger.error(f"Mem0 update failed: {e}")
-
-    def delete_memory(self, memory_id: str) -> None:
-        """Delete a specific memory by ID."""
-        try:
-            self.memory.delete(memory_id)
-            logger.debug(f"Mem0 delete: memory_id={memory_id}")
-        except Exception as e:
-            logger.error(f"Mem0 delete failed: {e}")
-
-    def get_all_memories(self, user_id: str) -> list[dict[str, Any]]:
-        """Get all memories for a user."""
-        try:
-            result = self.memory.get_all(user_id=user_id)
-            return result.get("results", []) if result else []
-        except Exception as e:
-            logger.error(f"Mem0 get_all failed: {e}")
-            return []
-
-    async def consolidate(
-        self,
-        session: Session,
-        provider: LLMProvider,
-        model: str,
-        *,
-        archive_all: bool = False,
-        memory_window: int = 50,
-    ) -> bool:
-        """
-        Consolidate session messages into mem0 memory.
-
-        Facts are extracted using the main agent's LLM provider, then stored with infer=False.
-
-        Returns True on success.
-        """
-        try:
-            # Extract user_id from session key (e.g., "telegram:12345" -> "telegram_12345")
-            user_id = session.key.replace(":", "_")
-
-            # Determine which messages to consolidate
-            if archive_all:
-                messages_to_add = session.messages
-                logger.info(
-                    f"Mem0 consolidation (archive_all): {len(messages_to_add)} messages"
-                )
-            else:
-                keep_count = memory_window // 2
-                if len(session.messages) <= keep_count:
-                    return True
-
-                # Get unconsolidated messages
-                start_idx = session.last_consolidated
-                end_idx = len(session.messages) - keep_count
-
-                if end_idx <= start_idx:
-                    return True
-
-                messages_to_add = session.messages[start_idx:end_idx]
-
-                if not messages_to_add:
-                    return True
-
-                logger.info(
-                    f"Mem0 consolidation: {len(messages_to_add)} to consolidate, "
-                    f"{keep_count} keep"
-                )
-
-            # Convert to mem0 format with intelligent filtering
-            mem0_messages = []
-            for msg in messages_to_add:
-                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.
-                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
-                if role == "system":
-                    continue
-
-                # Skip messages with no content
-                if not content:
-                    continue
-
-                # Normalize assistant message content: extract text from Anthropic list format
-                if role == "assistant" and isinstance(content, list):
-                    # Anthropic format: list of {type: "text"|"tool_use", text: "..."} blocks
-                    text_parts = [
-                        block.get("text", "")
-                        for block in content
-                        if isinstance(block, dict) and block.get("type") == "text"
-                    ]
-                    content = " ".join(text_parts).strip()
-                    if not content:
-                        continue  # Skip if assistant only called tools with no text explanation
-
-                # Normalize user message content (could also be a list in some formats)
-                if isinstance(content, list):
-                    text_parts = [
-                        block.get("text", "") if isinstance(block, dict) else str(block)
-                        for block in content
-                    ]
-                    content = " ".join(text_parts).strip()
-                    if not content:
-                        continue
-
-                # Skip trivially short messages (commands like "/new")
-                if len(content.strip()) < 10:
-                    continue
-
-                mem0_messages.append({
-                    "role": role,
-                    "content": content
-                })
-
-            if mem0_messages:
                 # Extract facts using the main agent's LLM (already paid for),
                 # then store with infer=False to bypass mem0's GPT-nano
                 facts = await self.extract_facts(mem0_messages, provider, model)

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 os
-import uuid
 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._started = False
-        self._start()
+        self._timed_out = False
+        self._process: asyncio.subprocess.Process | None = None
 
-    def _start(self):
+    async def start(self):
-        """Start the bash process."""
+        if self._started:
-        self.process = subprocess.Popen(
+            return
-            ["bash"],
+
-            stdin=subprocess.PIPE,
+        self._process = await asyncio.create_subprocess_shell(
-            stdout=subprocess.PIPE,
+            self.command,
-            stderr=subprocess.STDOUT,
+            preexec_fn=os.setsid,
-            text=True,
+            shell=True,
-            bufsize=1,
+            bufsize=0,
+            stdin=asyncio.subprocess.PIPE,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
         )
+        self._started = True
 
-    def restart(self):
+    def stop(self):
-        """Restart the bash session."""
+        """Terminate the bash shell."""
-        if self.process:
+        if not self._started:
-            self.process.terminate()
+            return
-            try:
+        if self._process and self._process.returncode is None:
-                self.process.wait(timeout=5)
+            self._process.terminate()
-            except subprocess.TimeoutExpired:
-                self.process.kill()
-                self.process.wait()
-        self._start()
 
-    async def run_command(self, command: str, timeout: float = 120.0) -> str:
+    async def run(self, command: str) -> ToolResult:
-        """Run a command in the persistent bash session.
+        """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:
+        # Send command + sentinel on its own line so heredoc terminators
-            command: Bash command to execute
+        # aren't corrupted (EOF; echo '...' ≠ EOF)
-            timeout: Maximum time to wait for command completion (seconds)
+        self._process.stdin.write(
+            command.encode() + f"\necho '{self._sentinel}'\n".encode()
+        )
+        await self._process.stdin.drain()
 
-        Returns:
+        # Poll stdout buffer until sentinel appears — no threads involved
-            Command output (stdout + stderr combined)
+        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:
+        if output.endswith("\n"):
-            asyncio.TimeoutError: If command doesn't complete within timeout
+            output = output[:-1]
-            RuntimeError: If bash process has died
-        """
-        if not self.process or self.process.poll() is not None:
-            raise RuntimeError("Bash process has died")
 
-        # Generate unique sentinel
+        error = self._process.stderr._buffer.decode()
-        sentinel = f"<<BASH_COMMAND_DONE_{uuid.uuid4().hex}>>"
+        if error.endswith("\n"):
+            error = error[:-1]
 
-        # Send command + sentinel
+        # Clear buffers for next command
-        full_command = f"{command}\necho '{sentinel}'\n"
+        self._process.stdout._buffer.clear()
-        self.process.stdin.write(full_command)
+        self._process.stderr._buffer.clear()
-        self.process.stdin.flush()
 
-        # Read output until sentinel appears
+        # Return as ToolResult (our loop handles this type)
-        output_lines = []
+        if error and output:
-        start_time = asyncio.get_event_loop().time()
+            return ToolResult(output=f"{output}\n\nstderr: {error}")
-
+        elif error:
-        while True:
+            return ToolResult(output=error)
-            # Check timeout
+        else:
-            elapsed = asyncio.get_event_loop().time() - start_time
+            return ToolResult(output=output if output else "(no output)")
-            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()
 
 
 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()
+            if self._session:
-            return ToolResult(output="Bash session restarted successfully.")
+                self._session.stop()
+            self._session = _BashSession()
+            await self._session.start()
+            return ToolResult(system="tool has been restarted.")
 
-        if not command:
+        if self._session is None:
-            return ToolResult(
+            self._session = _BashSession()
-                error="Either 'command' or 'restart=True' must be provided."
+            await self._session.start()
-            )
 
-        try:
+        if command is not None:
-            output = await self._session.run_command(command)
+            try:
-            return ToolResult(output=output if output else "(no output)")
+                return await self._session.run(command)
-        except asyncio.TimeoutError as e:
+            except ToolError as e:
-            return ToolResult(error=f"Command timed out: {e}")
+                return ToolResult(error=str(e))
-        except Exception as e:
+
-            return ToolResult(error=f"{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
+        # Add cache breakpoints on the last TWO user messages (4-breakpoint strategy):
-        if messages:
+        # BP3: Second-to-last user message (stable history from previous turn)
-            last = messages[-1]
+        # BP4: Last user message (current turn, will become BP3 next turn)
-            if last.get("role") == "user":
+        # This allows BP3 to reuse what BP4 cached last turn.
-                content = last["content"]
+        user_indices = [i for i, m in enumerate(messages) if m.get("role") == "user"]
-                if isinstance(content, str):
+
-                    last = {**last, "content": [{"type": "text", "text": content, "cache_control": {"type": "ephemeral"}}]}
+        if len(user_indices) >= 2:
-                elif isinstance(content, list) and content:
+            # BP3: Second-to-last user message
-                    new_content = list(content)
+            idx = user_indices[-2]
-                    new_content[-1] = {**new_content[-1], "cache_control": {"type": "ephemeral"}}
+            msg = messages[idx]
-                    last = {**last, "content": new_content}
+            content = msg["content"]
-                messages = messages[:-1] + [last]
+            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(
+        # Debug: Log message structure to diagnose orphaned tool_result errors
-            self._get_api_url(),
+        for idx, m in enumerate(payload.get("messages", [])):
-            headers=headers,
+            role = m.get("role", "?")
-            json=payload,
+            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: