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 preserves all thinking blocks
+    # Server-side context management: Anthropic trims old tool results and preserves all
-    # and clears old tool results only when approaching the 200k context limit.
+    # thinking blocks (keep="all" maximises cache hits). Client keeps full history.
     CONTEXT_MANAGEMENT = {
         "edits": [
             {
@@ -50,11 +50,7 @@ class AgentLoop:
             },
             {
                 "type": "clear_tool_uses_20250919",
-                # Raised from 80k to 195k to avoid premature cache invalidation.
+                "trigger": {"type": "input_tokens", "value": 80000},
-                # 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},
             },
         ]
@@ -155,28 +151,9 @@ class AgentLoop:
         # Register native Anthropic tools
         self.tools.register(BashTool20250124())
         self.tools.register(EditTool20250728())
-        # self.tools.register(ComputerTool20251124())  # Disabled - VM unavailable
+        self.tools.register(ComputerTool20251124())
 
-        logger.info("Registered native Anthropic tools: bash, text_editor")
+        logger.info("Registered native Anthropic tools: bash, text_editor, computer")
-
-        # 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."""
@@ -355,9 +332,6 @@ 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] = []
 
@@ -567,38 +541,15 @@ class AgentLoop:
             reasoning_content=final_reasoning,
         )
 
-        # Save to session: mem0 context (if present) + user message + full tool chain
+        # Save to session: user message + full tool chain (tool_use, tool_results, thinking, final reply)
         # 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,
@@ -786,17 +737,7 @@ class AgentLoop:
             reasoning_content=final_reasoning,
         )
 
-        # Save to session: mem0 (if present) + user message + full tool chain
+        # Save to session: 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)
@@ -810,34 +751,6 @@ 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.
 
@@ -861,17 +774,6 @@ 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)
@@ -962,7 +864,7 @@ Respond with ONLY valid JSON, no markdown fences."""
                 if update != current_memory:
                     memory.write_long_term(update)
 
-            session.messages = self._trim_to_clean_boundary(session.messages, keep_count) if keep_count else []
+            session.messages = 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,6 +58,356 @@ 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,117 +1,114 @@
-"""BashTool20250124 - Persistent bash session with async buffer polling.
+"""BashTool20250124 - Persistent bash session with sentinel-based output.
 
-Based on Anthropic's reference implementation from anthropic-quickstarts.
+Anthropic's native bash_20250124 tool with a long-running session.
-Uses asyncio.create_subprocess_shell + direct buffer reads instead of
-threaded readline, which avoids exhausting the default ThreadPoolExecutor.
 """
 
 import asyncio
-import os
+import subprocess
+import uuid
 from typing import Any, Literal
 
-from nanobot.agent.tools.anthropic.base import BaseAnthropicTool, ToolResult, ToolError
+from nanobot.agent.tools.anthropic.base import BaseAnthropicTool, ToolResult
 
 
 class _BashSession:
-    """A session of a bash shell.
+    """Manages a persistent bash subprocess with sentinel-based output reading."""
-
-    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._started = False
+        self.process: subprocess.Popen | None = None
-        self._timed_out = False
+        self._start()
-        self._process: asyncio.subprocess.Process | None = None
 
-    async def start(self):
+    def _start(self):
-        if self._started:
+        """Start the bash process."""
-            return
+        self.process = subprocess.Popen(
-
+            ["bash"],
-        self._process = await asyncio.create_subprocess_shell(
+            stdin=subprocess.PIPE,
-            self.command,
+            stdout=subprocess.PIPE,
-            preexec_fn=os.setsid,
+            stderr=subprocess.STDOUT,
-            shell=True,
+            text=True,
-            bufsize=0,
+            bufsize=1,
-            stdin=asyncio.subprocess.PIPE,
-            stdout=asyncio.subprocess.PIPE,
-            stderr=asyncio.subprocess.PIPE,
         )
-        self._started = True
 
-    def stop(self):
+    def restart(self):
-        """Terminate the bash shell."""
+        """Restart the bash session."""
-        if not self._started:
+        if self.process:
-            return
+            self.process.terminate()
-        if self._process and self._process.returncode is None:
+            try:
-            self._process.terminate()
+                self.process.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                self.process.kill()
+                self.process.wait()
+        self._start()
 
-    async def run(self, command: str) -> ToolResult:
+    async def run_command(self, command: str, timeout: float = 120.0) -> str:
-        """Execute a command in the bash shell."""
+        """Run a command in the persistent bash session.
-        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",
-            )
 
-        assert self._process.stdin
+        Uses a unique sentinel to detect command completion.
-        assert self._process.stdout
-        assert self._process.stderr
 
-        # Send command + sentinel on its own line so heredoc terminators
+        Args:
-        # aren't corrupted (EOF; echo '...' ≠ EOF)
+            command: Bash command to execute
-        self._process.stdin.write(
+            timeout: Maximum time to wait for command completion (seconds)
-            command.encode() + f"\necho '{self._sentinel}'\n".encode()
-        )
-        await self._process.stdin.drain()
 
-        # Poll stdout buffer until sentinel appears — no threads involved
+        Returns:
-        try:
+            Command output (stdout + stderr combined)
-            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
 
-        if output.endswith("\n"):
+        Raises:
-            output = output[:-1]
+            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")
 
-        error = self._process.stderr._buffer.decode()
+        # Generate unique sentinel
-        if error.endswith("\n"):
+        sentinel = f"<<BASH_COMMAND_DONE_{uuid.uuid4().hex}>>"
-            error = error[:-1]
 
-        # Clear buffers for next command
+        # Send command + sentinel
-        self._process.stdout._buffer.clear()
+        full_command = f"{command}\necho '{sentinel}'\n"
-        self._process.stderr._buffer.clear()
+        self.process.stdin.write(full_command)
+        self.process.stdin.flush()
 
-        # Return as ToolResult (our loop handles this type)
+        # Read output until sentinel appears
-        if error and output:
+        output_lines = []
-            return ToolResult(output=f"{output}\n\nstderr: {error}")
+        start_time = asyncio.get_event_loop().time()
-        elif error:
+
-            return ToolResult(output=error)
+        while True:
-        else:
+            # Check timeout
-            return ToolResult(output=output if output else "(no output)")
+            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()
 
 
 class BashTool20250124(BaseAnthropicTool):
@@ -127,10 +124,10 @@ class BashTool20250124(BaseAnthropicTool):
 
     api_type: Literal["bash_20250124"] = "bash_20250124"
     name: Literal["bash"] = "bash"
-    beta_flag: str | None = None
+    beta_flag: str = "computer-use-2025-11-24"
 
     def __init__(self):
-        self._session: _BashSession | None = None
+        self._session = _BashSession()
 
     async def __call__(
         self,
@@ -138,26 +135,39 @@ 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:
-            if self._session:
+            self._session.restart()
-                self._session.stop()
+            return ToolResult(output="Bash session restarted successfully.")
-            self._session = _BashSession()
-            await self._session.start()
-            return ToolResult(system="tool has been restarted.")
 
-        if self._session is None:
+        if not command:
-            self._session = _BashSession()
+            return ToolResult(
-            await self._session.start()
+                error="Either 'command' or 'restart=True' must be provided."
+            )
 
-        if command is not None:
+        try:
-            try:
+            output = await self._session.run_command(command)
-                return await self._session.run(command)
+            return ToolResult(output=output if output else "(no output)")
-            except ToolError as e:
+        except asyncio.TimeoutError as e:
-                return ToolResult(error=str(e))
+            return ToolResult(error=f"Command timed out: {e}")
-
+        except Exception as e:
-        return ToolResult(error="Either 'command' or 'restart=True' must be provided.")
+            return ToolResult(error=f"{e}")
 
     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,14 +67,13 @@ 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 | None = None
+    beta_flag: str = "computer-use-2025-11-24"
 
     async def __call__(
         self,

nanobot/agent/tools/memory_tools.py

@@ -1,230 +0,0 @@
-"""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,83 +59,9 @@ 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(
+            self._client = httpx.AsyncClient(timeout=300.0)
-                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]]
@@ -326,35 +252,18 @@ class AnthropicOAuthProvider(LLMProvider):
         """Make request to Anthropic API."""
         client = await self._get_client()
 
-        # Add cache breakpoints on the last TWO user messages (4-breakpoint strategy):
+        # Cache the last user message so conversation history is cached across turns
-        # BP3: Second-to-last user message (stable history from previous turn)
+        if messages:
-        # BP4: Last user message (current turn, will become BP3 next turn)
+            last = messages[-1]
-        # This allows BP3 to reuse what BP4 cached last turn.
+            if last.get("role") == "user":
-        user_indices = [i for i, m in enumerate(messages) if m.get("role") == "user"]
+                content = last["content"]
-
+                if isinstance(content, str):
-        if len(user_indices) >= 2:
+                    last = {**last, "content": [{"type": "text", "text": content, "cache_control": {"type": "ephemeral"}}]}
-            # BP3: Second-to-last user message
+                elif isinstance(content, list) and content:
-            idx = user_indices[-2]
+                    new_content = list(content)
-            msg = messages[idx]
+                    new_content[-1] = {**new_content[-1], "cache_control": {"type": "ephemeral"}}
-            content = msg["content"]
+                    last = {**last, "content": new_content}
-            if isinstance(content, str):
+                messages = messages[:-1] + [last]
-                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,
@@ -412,43 +321,11 @@ class AnthropicOAuthProvider(LLMProvider):
             tool_names = [t.get("name", "unnamed") for t in payload["tools"]]
             logger.debug(f"Tool names in request: {tool_names}")
 
-        # Debug: Log message structure to diagnose orphaned tool_result errors
+        response = await client.post(
-        for idx, m in enumerate(payload.get("messages", [])):
+            self._get_api_url(),
-            role = m.get("role", "?")
+            headers=headers,
-            content = m.get("content", "")
+            json=payload,
-            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: