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

README.md

@@ -143,7 +143,7 @@ Add or merge these **two parts** into your config (other options have defaults).
 {
   "agents": {
     "defaults": {
-      "model": "anthropic/claude-opus-4-7",
+      "model": "anthropic/claude-opus-4-5",
       "provider": "openrouter"
     }
   }

docs/testing-workflow.md

@@ -1,214 +0,0 @@
-# PR Testing Workflow
-
-Guide for testing Pull Requests using the local staging environment.
-
-## Quick Start
-
-```bash
-./test-pr.sh <pr-number> "test message"
-```
-
-## Staging Environment
-
-**Location:** `/config/workspace/.nanobot-staging/`
-
-**Components:**
-- `config.json` — Staging configuration (channels disabled, shared OAuth)
-- `workspace/` — Isolated workspace for tool operations
-- `workspace/sessions/` — Session storage (separate from production)
-
-**Key differences from production:**
-- No external channels (Telegram disabled)
-- Uses `NANOBOT_CONFIG` environment variable
-- Gateway runs on localhost:18791 (vs production's 18790)
-- `restrictToWorkspace: true` for safety
-
-## Testing a PR
-
-### Method 1: Helper Script (Recommended)
-
-```bash
-# Test PR with default message
-./test-pr.sh 31
-
-# Test with custom message
-./test-pr.sh 31 "test the hidden message feature"
-```
-
-**What it does:**
-1. Fetches PR from `wylab` remote (force updates if branch exists)
-2. Checks out PR branch locally
-3. Installs in editable mode with `uv pip install -e .`
-4. Runs test with staging config via `NANOBOT_CONFIG` env var
-5. Leaves branch checked out for further testing
-
-**After testing:**
-```bash
-git checkout main  # Return to main branch
-```
-
-### Method 2: Manual Testing
-
-```bash
-# 1. Fetch and checkout PR
-cd /config/workspace/nanobot-oauth-port/nanobot-fork
-git fetch wylab pull/<N>/head:pr-<N>
-git checkout pr-<N>
-
-# 2. Install in editable mode
-uv pip install -e .
-
-# 3. Test with staging config
-NANOBOT_CONFIG=/config/workspace/.nanobot-staging/config.json \
-  .venv/bin/nanobot agent -m "test message"
-
-# 4. For multi-turn testing
-NANOBOT_CONFIG=/config/workspace/.nanobot-staging/config.json \
-  .venv/bin/nanobot agent  # Interactive mode
-
-# 5. Return to main
-git checkout main
-```
-
-### Method 3: Gateway Validation
-
-Test that gateway starts without errors:
-
-```bash
-NANOBOT_CONFIG=/config/workspace/.nanobot-staging/config.json \
-  .venv/bin/nanobot gateway
-
-# Kill with Ctrl+C when validated
-```
-
-## Verifying Cache Behavior
-
-To verify prompt caching works correctly (important for performance):
-
-```bash
-# Enable logs to see cache metrics
-NANOBOT_CONFIG=/config/workspace/.nanobot-staging/config.json \
-  .venv/bin/nanobot agent --logs -m "Turn 1: list files"
-
-# Look for cache metrics in output:
-# - cache_write: New cache entries created
-# - cache_read: Tokens read from cache
-```
-
-**What to look for:**
-- Turn 1: High `cache_write`, moderate `cache_read`
-- Turn 2+: Low `cache_write`, high `cache_read` (reusing cache)
-- `cache_read` should increase across turns as context grows
-
-**Example healthy pattern:**
-```
-Turn 1: cache_write=354 cache_read=3563
-Turn 2: cache_write=255 cache_read=3917  ← Same as Turn 1 end
-Turn 3: cache_write=113 cache_read=4172  ← Growing with context
-```
-
-## Session Management
-
-### Clear session for fresh test
-
-```bash
-rm -f /config/workspace/.nanobot-staging/workspace/sessions/cli_direct.jsonl
-```
-
-### View session contents
-
-```bash
-cat /config/workspace/.nanobot-staging/workspace/sessions/cli_direct.jsonl | jq
-```
-
-### Check for specific features (e.g., hidden signatures)
-
-```bash
-cat /config/workspace/.nanobot-staging/workspace/sessions/cli_direct.jsonl | grep "_hidden_sig"
-```
-
-## Common Testing Scenarios
-
-### Test tool execution
-
-```bash
-./test-pr.sh 31 "List all Python files in the current directory"
-```
-
-### Test multi-turn conversation
-
-```bash
-NANOBOT_CONFIG=/config/workspace/.nanobot-staging/config.json \
-  .venv/bin/nanobot agent
-
-# Then interact naturally:
-> list files in current directory
-> how many python files are there?
-> what's the total size?
-```
-
-### Test error handling
-
-```bash
-./test-pr.sh 31 "Try to read a file that doesn't exist: /nonexistent.txt"
-```
-
-### Test with thinking mode
-
-The staging config has `thinking_budget: 10000` enabled by default, so all tests use extended thinking.
-
-## Troubleshooting
-
-### "No API key configured" error
-
-- **Cause:** `NANOBOT_CONFIG` env var not set
-- **Fix:** Ensure you're using `NANOBOT_CONFIG=/config/workspace/.nanobot-staging/config.json`
-
-### "Module not found" after checkout
-
-- **Cause:** Need to reinstall after switching branches
-- **Fix:** Run `uv pip install -e .` after checkout
-
-### Changes not applying
-
-- **Cause:** Using cached `.pyc` files
-- **Fix:** Clear pycache: `find . -type d -name __pycache__ -exec rm -rf {} + 2>/dev/null || true`
-
-### Session has stale data
-
-- **Cause:** Previous test left session data
-- **Fix:** `rm /config/workspace/.nanobot-staging/workspace/sessions/cli_direct.jsonl`
-
-## Best Practices
-
-1. **Clear session between PR tests** to avoid cross-contamination
-2. **Test with tool use** to trigger agentic behavior (not just simple Q&A)
-3. **Check cache metrics** for performance-sensitive PRs
-4. **Run with `--logs`** to see detailed behavior during development
-5. **Return to main** after testing to avoid accidental commits on PR branches
-
-## Integration with CI/CD
-
-The staging environment is currently manual-only. Future enhancements:
-
-- [ ] Automated PR testing via Gitea Actions
-- [ ] Cache validation in CI pipeline
-- [ ] Multi-PR parallel testing using git worktrees
-- [ ] Regression test suite against production behavior
-
-## File Locations Reference
-
-| Path | Purpose |
-|------|---------|
-| `/config/workspace/nanobot-oauth-port/nanobot-fork/` | Local nanobot repository |
-| `/config/workspace/.nanobot-staging/` | Staging environment root |
-| `/config/workspace/.nanobot-staging/config.json` | Staging configuration |
-| `/config/workspace/.nanobot-staging/workspace/` | Staging workspace |
-| `/config/workspace/.nanobot-staging/workspace/sessions/` | Session storage |
-| `/config/workspace/nanobot-oauth-port/nanobot-fork/test-pr.sh` | Helper script |
-
-## Related Documentation
-
-- [nanobot README](../README.md) - Main project documentation
-- [CLAUDE.md](../CLAUDE.md) - Development guide for Claude Code
-- [config/schema.py](../nanobot/config/schema.py) - Configuration schema

nanobot/agent/context.py

@@ -11,7 +11,6 @@ from loguru import logger
 from nanobot.agent.memory import MemoryStore
 from nanobot.agent.memory_mem0 import Mem0MemoryStore, HAS_MEM0
 from nanobot.agent.skills import SkillsLoader
-from nanobot.agent.visibility import compute_signature
 
 
 class ContextBuilder:
@@ -227,14 +226,12 @@ visibility markers will be rejected."""
         Returns:
             Updated message list.
         """
-        msg: dict[str, Any] = {
+        messages.append({
             "role": "tool",
             "tool_call_id": tool_call_id,
             "name": tool_name,
-            "content": result,
-            "_hidden_sig": compute_signature(result if isinstance(result, str) else ""),
-        }
-        messages.append(msg)
+            "content": result
+        })
         return messages
     
     def add_assistant_message(
@@ -257,14 +254,13 @@ visibility markers will be rejected."""
             Updated message list.
         """
         msg: dict[str, Any] = {"role": "assistant", "content": content or ""}
-
+        
         if tool_calls:
             msg["tool_calls"] = tool_calls
-            msg["_hidden_sig"] = compute_signature(content or "")
-
+        
         # Thinking models reject history without this
         if reasoning_content:
             msg["reasoning_content"] = reasoning_content
-
+        
         messages.append(msg)
         return messages

nanobot/agent/loop.py

@@ -11,7 +11,7 @@ from loguru import logger
 
 from nanobot.bus.events import InboundMessage, OutboundMessage
 from nanobot.bus.queue import MessageBus
-from nanobot.providers.base import LLMProvider, LongContextError
+from nanobot.providers.base import LLMProvider
 from nanobot.agent.context import ContextBuilder
 from nanobot.agent.tools.registry import ToolRegistry
 from nanobot.agent.tools.filesystem import ReadFileTool, WriteFileTool, EditFileTool, ListDirTool
@@ -40,8 +40,8 @@ class AgentLoop:
     5. Sends responses back
     """
 
-    # Server-side context management: Anthropic preserves all thinking blocks
-    # and clears old tool results only when approaching the 200k context limit.
+    # Server-side context management: Anthropic trims old tool results and preserves all
+    # 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.
-                # 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},
+                "trigger": {"type": "input_tokens", "value": 80000},
                 "keep": {"type": "tool_uses", "value": 5},
             },
         ]
@@ -221,7 +217,7 @@ class AgentLoop:
             return self._quota_cache["model"]
 
         # Default models
-        OPUS = "claude-opus-4-7"
+        OPUS = "claude-opus-4-6"
         SONNET = "claude-sonnet-4-6"
         TOLERANCE = 1.17  # 17% overage triggers downgrade
 
@@ -346,7 +342,6 @@ class AgentLoop:
         message_tool = self.tools.get("message")
         if isinstance(message_tool, MessageTool):
             message_tool.set_context(msg.channel, msg.chat_id)
-            message_tool.start_turn()
 
         spawn_tool = self.tools.get("spawn")
         if isinstance(spawn_tool, SpawnTool):
@@ -417,35 +412,12 @@ class AgentLoop:
 
             # Call LLM
             logger.debug(f"Calling LLM with model={selected_model}, provider.thinking_budget={self.provider.thinking_budget}")
-            try:
-                response = await self.provider.chat(
-                    messages=messages,
-                    tools=self.tools.get_tools(),  # Pass tool objects for beta flag extraction
-                    model=selected_model,
-                    context_management=self.CONTEXT_MANAGEMENT,
-                )
-            except LongContextError:
-                logger.warning("Long context 429 — auto-consolidating session")
-                await self._consolidate_memory(session, archive_all=False)
-                # Apply trim immediately (normally deferred to end of turn)
-                checkpoint = getattr(session, '_trim_checkpoint', None)
-                if checkpoint is not None:
-                    old_size = len(session.messages)
-                    session.messages = session.messages[checkpoint:]
-                    session._trim_checkpoint = None
-                    self.sessions.save(session)
-                    logger.info(f"Emergency trim: {old_size} -> {len(session.messages)} messages")
-                    # Rebuild messages from trimmed session
-                    messages = self.context.build_messages(
-                        history=session.get_history(),
-                        current_message=current_message,
-                        media=msg.media if msg.media else None,
-                        channel=msg.channel,
-                        chat_id=msg.chat_id,
-                    )
-                    turn_start = len(messages)
-                    continue  # Retry LLM call with shorter context
-                raise  # No trim happened — can't recover
+            response = await self.provider.chat(
+                messages=messages,
+                tools=self.tools.get_tools(),  # Pass tool objects for beta flag extraction
+                model=selected_model,
+                context_management=self.CONTEXT_MANAGEMENT,
+            )
 
             # Handle tool calls
             if response.has_tool_calls:
@@ -568,12 +540,6 @@ class AgentLoop:
             else:
                 final_content = "I've completed processing but have no response to give."
 
-        # Check if message tool already sent to same target (suppress final reply)
-        message_tool = self.tools.get("message")
-        if isinstance(message_tool, MessageTool) and message_tool._sent_in_turn:
-            logger.info(f"Suppressing final reply to {msg.channel}:{msg.chat_id} (message tool already sent)")
-            return None
-
         # Log response preview
         preview = final_content[:120] + "..." if len(final_content) > 120 else final_content
         logger.info(f"Response to {msg.channel}:{msg.sender_id}: {preview}")
@@ -597,38 +563,24 @@ 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 checkpoint
-        # marking where to trim. Now that the turn's tool chain is fully saved,
-        # we can safely trim to that checkpoint.
-        checkpoint = getattr(session, '_trim_checkpoint', None)
-        if checkpoint is not None:
-            old_size = len(session.messages)
-            session.messages = session.messages[checkpoint:]
-            session._trim_checkpoint = None
+        # 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: {old_size} -> {len(session.messages)} messages (checkpoint={checkpoint})")
+            logger.info(f"Deferred trim applied, session now {len(session.messages)} messages")
 
         return OutboundMessage(
             channel=msg.channel,
@@ -695,32 +647,12 @@ class AgentLoop:
         while iteration < self.max_iterations:
             iteration += 1
 
-            try:
-                response = await self.provider.chat(
-                    messages=messages,
-                    tools=self.tools.get_tools(),  # Pass tool objects for beta flag extraction
-                    model=selected_model,
-                    context_management=self.CONTEXT_MANAGEMENT,
-                )
-            except LongContextError:
-                logger.warning("Long context 429 in system handler — auto-consolidating")
-                await self._consolidate_memory(session, archive_all=False)
-                checkpoint = getattr(session, '_trim_checkpoint', None)
-                if checkpoint is not None:
-                    old_size = len(session.messages)
-                    session.messages = session.messages[checkpoint:]
-                    session._trim_checkpoint = None
-                    self.sessions.save(session)
-                    logger.info(f"Emergency trim: {old_size} -> {len(session.messages)} messages")
-                    messages = self.context.build_messages(
-                        history=session.get_history(),
-                        current_message=msg.content,
-                        channel=origin_channel,
-                        chat_id=origin_chat_id,
-                    )
-                    turn_start = len(messages)
-                    continue
-                raise
+            response = await self.provider.chat(
+                messages=messages,
+                tools=self.tools.get_tools(),  # Pass tool objects for beta flag extraction
+                model=selected_model,
+                context_management=self.CONTEXT_MANAGEMENT,
+            )
 
             if response.has_tool_calls:
                 tool_call_dicts = [
@@ -837,32 +769,12 @@ class AgentLoop:
             reasoning_content=final_reasoning,
         )
 
-        # 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)
-
+        # Save to session: user message + full tool chain
         session.add_message("user", f"[System: {msg.sender_id}] {msg.content}")
         for chain_msg in messages[turn_start:]:
             session.add_raw_message(chain_msg)
         self.sessions.save(session)
 
-        # Deferred trim: same logic as _process_message
-        # System messages (including subagents) can trigger consolidation
-        checkpoint = getattr(session, '_trim_checkpoint', None)
-        if checkpoint is not None:
-            old_size = len(session.messages)
-            session.messages = session.messages[checkpoint:]
-            session._trim_checkpoint = None
-            self.sessions.save(session)
-            logger.info(f"Deferred trim applied: {old_size} -> {len(session.messages)} messages (checkpoint={checkpoint})")
-
         # Return original content (not signed) for outbound, but with suppressed metadata
         return OutboundMessage(
             channel=origin_channel,
@@ -871,26 +783,6 @@ class AgentLoop:
             metadata=outbound_metadata,
         )
     
-    @staticmethod
-    def _find_clean_boundary_before(messages: list[dict], target_pos: int) -> int:
-        """Find a clean user message boundary at or before target position.
-
-        Returns the index of a user message at or before target_pos,
-        or target_pos if no user message is found.
-        """
-        if not messages or target_pos <= 0:
-            return 0
-        if target_pos >= len(messages):
-            return len(messages)
-
-        # Walk backward from target to find a user message
-        for i in range(target_pos, -1, -1):
-            if messages[i].get("role") == "user":
-                return i
-
-        # No user message found, return target position
-        return target_pos
-
     @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.
@@ -951,13 +843,8 @@ class AgentLoop:
                 logger.info("Mem0 consolidation done, session cleared (archive_all)")
             else:
                 keep_count = min(10, max(2, self.memory_window // 2))
-                # Set checkpoint at current session size minus keep_count
-                # This preserves the intended trim point regardless of messages added later
-                checkpoint = max(0, len(session.messages) - keep_count)
-                # Find clean boundary at or before checkpoint
-                checkpoint = self._find_clean_boundary_before(session.messages, checkpoint)
-                session._trim_checkpoint = checkpoint
-                logger.info(f"Mem0 consolidation done, trim deferred (checkpoint={checkpoint}, current_size={len(session.messages)})")
+                session._pending_trim = keep_count
+                logger.info(f"Mem0 consolidation done, trim deferred (keep={keep_count})")
             return
         else:
             memory = MemoryStore(self.workspace)
@@ -1051,33 +938,9 @@ Respond with ONLY valid JSON, no markdown fences."""
             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")
-            self._spawn_ara_research_manager()
         except Exception as e:
             logger.error(f"Memory consolidation failed: {e}")
 
-    def _spawn_ara_research_manager(self) -> None:
-        """Fire-and-forget: update the nanobot ARA trace after memory consolidation.
-
-        Uses the SpawnTool which already has channel/chat_id context from the
-        last message. Silently skips if ARA path doesn't exist or spawn fails.
-        """
-        import os
-        ara_path = os.path.join(self.workspace, "ara-projects", "nanobot")
-        if not os.path.isdir(ara_path):
-            return
-        spawn_tool = self.tools.get("spawn")
-        if not isinstance(spawn_tool, SpawnTool):
-            return
-        task = (
-            f"Read /root/.nanobot/workspace/skills/ara-research-manager/SKILL.md "
-            f"and run the per-turn procedure on the nanobot ARA at {ara_path}/. "
-            f"Capture any research events from this session."
-        )
-        asyncio.create_task(
-            spawn_tool.execute(task=task, label="ara-pm", model="claude-haiku-4-5")
-        )
-        logger.debug("ARA research-manager task spawned")
-
     async def process_direct(
         self,
         content: str,

nanobot/agent/memory.py

@@ -87,7 +87,11 @@ class MemoryStore:
             keep_count = memory_window // 2
             if len(session.messages) <= keep_count:
                 return True
-            old_messages = session.messages[:-keep_count]
+            if len(session.messages) - session.last_consolidated <= 0:
+                return True
+            old_messages = session.messages[session.last_consolidated:-keep_count]
+            if not old_messages:
+                return True
             logger.info("Memory consolidation: {} to consolidate, {} keep", len(old_messages), keep_count)
 
         lines = []
@@ -138,7 +142,8 @@ class MemoryStore:
                 if update != current_memory:
                     self.write_long_term(update)
 
-            logger.info("Memory consolidation done: {} messages total", len(session.messages))
+            session.last_consolidated = 0 if archive_all else len(session.messages) - keep_count
+            logger.info("Memory consolidation done: {} messages, last_consolidated={}", len(session.messages), session.last_consolidated)
             return True
         except Exception:
             logger.exception("Memory consolidation failed")

nanobot/agent/memory_mem0.py

@@ -47,7 +47,6 @@ class Mem0MemoryStore:
 
         today = datetime.now().strftime("%Y-%m-%d")
         custom_prompt = f"Extract dated facts from this conversation as JSON: {{\"facts\": [...]}}. Today is {today}.\n\n"
-        self.custom_prompt = custom_prompt
 
         # Initialize mem0 with optional config + custom prompt
         # Extract only MemoryConfig-relevant fields
@@ -58,9 +57,14 @@ class Mem0MemoryStore:
             if key in raw_config:
                 mem0_cfg_dict[key] = raw_config[key]
         logger.debug(f"Extracted for MemoryConfig: {list(mem0_cfg_dict.keys())}")
+        logger.debug(f"Custom prompt length: {len(custom_prompt)} chars")
+        self.custom_prompt = custom_prompt
+        mem0_cfg_dict["custom_fact_extraction_prompt"] = custom_prompt
         mem0_config = MemoryConfig(**mem0_cfg_dict)
         logger.debug(f"MemoryConfig created: vector_store={mem0_config.vector_store.provider if mem0_config.vector_store else None}")
+        logger.debug(f"MemoryConfig.custom_fact_extraction_prompt is None: {mem0_config.custom_fact_extraction_prompt is None}")
         self.memory = Memory(config=mem0_config)
+        logger.debug(f"Memory.config.custom_fact_extraction_prompt is None: {self.memory.config.custom_fact_extraction_prompt is None}")
 
         logger.info("Mem0 memory system initialized with custom nanobot prompt")
 
@@ -154,15 +158,21 @@ class Mem0MemoryStore:
         provider: Any,
         model: str,
     ) -> list[str]:
-        """Extract facts from conversation using the main agent's LLM provider."""
+        """
+        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_val = msg.get("content", "")
-            if isinstance(content_val, str) and content_val.strip():
-                conv_text += f"{role}: {content_val}\n\n"
+            content = msg.get("content", "")
+            if isinstance(content, str) and content.strip():
+                conv_text += f"{role}: {content}\n\n"
 
         if not conv_text.strip():
             return []
@@ -175,23 +185,25 @@ class Mem0MemoryStore:
             response = await provider.chat(
                 messages=extraction_messages,
                 model=model,
-                max_tokens=16384,
+                max_tokens=2000,
                 temperature=0.3,
-                thinking_budget=0,
             )
-            text = (response.content or "").strip()
+
+            # 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("```")[1]
-                if text.startswith("json"):
-                    text = text[4:]
-                text = text.strip()
+                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", [])
-            if not isinstance(facts, list):
-                logger.warning(f"LLM returned non-list facts: {type(facts)}")
-                return []
             logger.debug(f"Extracted {len(facts)} facts using {model}")
             return facts
+
         except Exception as e:
             logger.error(f"Fact extraction failed: {e}")
             return []
@@ -202,7 +214,12 @@ class Mem0MemoryStore:
         user_id: str,
         session_id: str | None = None,
     ) -> None:
-        """Store pre-extracted facts in mem0 with infer=False."""
+        """
+        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
 
@@ -212,23 +229,16 @@ class Mem0MemoryStore:
 
         stored = 0
         for fact in facts:
-            # Normalize: LLM may return dicts like {"fact": "...", "date": "..."} or plain strings
-            if isinstance(fact, dict):
-                fact_text = fact.get("fact", fact.get("text", str(fact)))
-            else:
-                fact_text = str(fact)
-            if not fact_text.strip():
-                continue
             try:
                 self.memory.add(
-                    fact_text,
+                    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 '{str(fact_text)[:50]}...': {e}")
+                logger.error(f"Failed to store fact '{fact[:50]}...': {e}")
 
         logger.info(f"Stored {stored}/{len(facts)} facts for user {user_id}")
 
@@ -301,8 +311,7 @@ class Mem0MemoryStore:
         """
         Consolidate session messages into mem0 memory.
 
-        Unlike the original MemoryStore, mem0 handles extraction automatically,
-        so this just needs to feed recent messages to mem0.
+        Facts are extracted using the main agent's LLM provider, then stored with infer=False.
 
         Returns True on success.
         """
@@ -321,8 +330,8 @@ class Mem0MemoryStore:
                 if len(session.messages) <= keep_count:
                     return True
 
-                # Consolidate messages except the most recent (kept for context)
-                start_idx = 0
+                # Get unconsolidated messages
+                start_idx = session.last_consolidated
                 end_idx = len(session.messages) - keep_count
 
                 if end_idx <= start_idx:
@@ -394,8 +403,15 @@ class Mem0MemoryStore:
                 facts = await self.extract_facts(mem0_messages, provider, model)
                 self.store_facts(facts, user_id=user_id, session_id=session.key)
 
+            # Update consolidation marker
+            if archive_all:
+                session.last_consolidated = len(session.messages)
+            else:
+                session.last_consolidated = end_idx
+
             logger.info(
-                f"Mem0 consolidation done: {len(session.messages)} messages total"
+                f"Mem0 consolidation done: {len(session.messages)} messages, "
+                f"last_consolidated={session.last_consolidated}"
             )
             return True
 

nanobot/agent/subagent.py

@@ -73,7 +73,7 @@ class SubagentManager:
             origin_metadata: Optional metadata to propagate to announcement (e.g. suppress_output).
 
         Returns:
-            Task ID of the spawned subagent.
+            Status message indicating the subagent was started.
         """
         task_id = str(uuid.uuid4())[:8]
         display_label = label or task[:30] + ("..." if len(task) > 30 else "")
@@ -83,18 +83,18 @@ class SubagentManager:
             "chat_id": origin_chat_id,
             "metadata": origin_metadata or {},
         }
-
+        
         # Create background task
         bg_task = asyncio.create_task(
             self._run_subagent(task_id, task, display_label, origin, model=model)
         )
         self._running_tasks[task_id] = bg_task
-
+        
         # Cleanup when done
         bg_task.add_done_callback(lambda _: self._running_tasks.pop(task_id, None))
-
+        
         logger.info(f"Spawned subagent [{task_id}]: {display_label}")
-        return task_id
+        return f"Subagent [{display_label}] started. Task ID: {task_id}"
     
     async def _run_subagent(
         self,

nanobot/agent/tools/message.py

@@ -21,7 +21,6 @@ class MessageTool(Tool):
         self._sessions = sessions
         self._default_channel = default_channel
         self._default_chat_id = default_chat_id
-        self._sent_in_turn: bool = False
     
     def set_context(self, channel: str, chat_id: str) -> None:
         """Set the current message context."""
@@ -31,10 +30,6 @@ class MessageTool(Tool):
     def set_send_callback(self, callback: Callable[[OutboundMessage], Awaitable[None]]) -> None:
         """Set the callback for sending messages."""
         self._send_callback = callback
-
-    def start_turn(self) -> None:
-        """Reset per-turn send tracking."""
-        self._sent_in_turn = False
     
     @property
     def name(self) -> str:
@@ -97,10 +92,6 @@ class MessageTool(Tool):
         try:
             await self._send_callback(msg)
 
-            # Track if sent to same target as current context
-            if channel == self._default_channel and chat_id == self._default_chat_id:
-                self._sent_in_turn = True
-
             if self._sessions:
                 session_key = f"{channel}:{chat_id}"
                 session = self._sessions.get_or_create(session_key)

nanobot/agent/visibility.py

@@ -4,19 +4,11 @@
 import hmac
 import hashlib
 import re
+from typing import Tuple
 
 SECRET_KEY = "nanobot_visibility_secret_key_v1"
 
 
-def compute_signature(content: str) -> str:
-    """Compute HMAC signature for content (hex string, no prefix)."""
-    return hmac.new(
-        SECRET_KEY.encode(),
-        content.encode(),
-        hashlib.sha256
-    ).hexdigest()[:8]
-
-
 def sign_content(content: str) -> str:
     """
     Sign content with HMAC and prepend marker.
@@ -27,11 +19,15 @@ def sign_content(content: str) -> str:
     Returns:
         Content with signed visibility marker: "[HIDDEN:{sig}] {content}"
     """
-    sig = compute_signature(content)
+    sig = hmac.new(
+        SECRET_KEY.encode(),
+        content.encode(),
+        hashlib.sha256
+    ).hexdigest()[:8]
     return f"[HIDDEN:{sig}] {content}"
 
 
-def verify_signature(marked_content: str) -> tuple[bool, str]:
+def verify_signature(marked_content: str) -> Tuple[bool, str]:
     """
     Verify HMAC signature and extract clean content.
 
@@ -48,7 +44,12 @@ def verify_signature(marked_content: str) -> tuple[bool, str]:
         return False, marked_content
 
     claimed_sig, content = match.groups()
-    expected_sig = compute_signature(content)
+    expected_sig = hmac.new(
+        SECRET_KEY.encode(),
+        content.encode(),
+        hashlib.sha256
+    ).hexdigest()[:8]
+
     is_valid = hmac.compare_digest(claimed_sig, expected_sig)
     return is_valid, content
 

nanobot/cli/commands.py

@@ -832,7 +832,6 @@ def cron_add(
     message: str = typer.Option(..., "--message", "-m", help="Message for agent"),
     every: int = typer.Option(None, "--every", "-e", help="Run every N seconds"),
     cron_expr: str = typer.Option(None, "--cron", "-c", help="Cron expression (e.g. '0 9 * * *')"),
-    tz: str | None = typer.Option(None, "--tz", help="IANA timezone for cron (e.g. 'America/Vancouver')"),
     at: str = typer.Option(None, "--at", help="Run once at time (ISO format)"),
     deliver: bool = typer.Option(False, "--deliver", "-d", help="Deliver response to channel"),
     to: str = typer.Option(None, "--to", help="Recipient for delivery"),
@@ -843,15 +842,11 @@ def cron_add(
     from nanobot.cron.service import CronService
     from nanobot.cron.types import CronSchedule
 
-    if tz and not cron_expr:
-        console.print("[red]Error: --tz can only be used with --cron[/red]")
-        raise typer.Exit(1)
-
     # Determine schedule type
     if every:
         schedule = CronSchedule(kind="every", every_ms=every * 1000)
     elif cron_expr:
-        schedule = CronSchedule(kind="cron", expr=cron_expr, tz=tz)
+        schedule = CronSchedule(kind="cron", expr=cron_expr)
     elif at:
         import datetime
         dt = datetime.datetime.fromisoformat(at)
@@ -863,18 +858,14 @@ def cron_add(
     store_path = get_data_dir() / "cron" / "jobs.json"
     service = CronService(store_path)
 
-    try:
-        job = service.add_job(
-            name=name,
-            schedule=schedule,
-            message=message,
-            deliver=deliver,
-            to=to,
-            channel=channel,
-        )
-    except ValueError as e:
-        console.print(f"[red]Error: {e}[/red]")
-        raise typer.Exit(1) from e
+    job = service.add_job(
+        name=name,
+        schedule=schedule,
+        message=message,
+        deliver=deliver,
+        to=to,
+        channel=channel,
+    )
 
     console.print(f"[green]✓[/green] Added job '{job.name}' ({job.id})")
 

nanobot/config/loader.py

@@ -1,21 +1,13 @@
 """Configuration loading utilities."""
 
 import json
-import os
 from pathlib import Path
 
 from nanobot.config.schema import Config
 
 
 def get_config_path() -> Path:
-    """Get the configuration file path.
-
-    Checks NANOBOT_CONFIG environment variable first, otherwise defaults
-    to ~/.nanobot/config.json
-    """
-    env_path = os.getenv("NANOBOT_CONFIG")
-    if env_path:
-        return Path(env_path)
+    """Get the default configuration file path."""
     return Path.home() / ".nanobot" / "config.json"
 
 
@@ -92,18 +84,4 @@ def _migrate_config(data: dict) -> dict:
     exec_cfg = tools.get("exec", {})
     if "restrictToWorkspace" in exec_cfg and "restrictToWorkspace" not in tools:
         tools["restrictToWorkspace"] = exec_cfg.pop("restrictToWorkspace")
-
-    # Extract api_key from oauthCredentials if present
-    providers = data.get("providers", {})
-    for _, provider_config in providers.items():
-        if isinstance(provider_config, dict):
-            oauth_creds = provider_config.get("oauthCredentials")
-            if oauth_creds and isinstance(oauth_creds, dict):
-                access_token = oauth_creds.get("access_token", "")
-                # Only set api_key if not already set and access_token exists
-                if access_token and not provider_config.get("api_key"):
-                    provider_config["api_key"] = access_token
-                    # Clean up migrated data to avoid duplication
-                    del provider_config["oauthCredentials"]
-
     return data

nanobot/config/schema.py

@@ -220,7 +220,7 @@ class AgentDefaults(Base):
     """Default agent configuration."""
 
     workspace: str = "~/.nanobot/workspace"
-    model: str = "anthropic/claude-opus-4-7"
+    model: str = "anthropic/claude-opus-4-5"
     provider: str = "auto"  # Provider name (e.g. "anthropic", "openrouter") or "auto" for auto-detection
     max_tokens: int = 8192
     temperature: float = 0.1

nanobot/heartbeat/service.py

@@ -87,10 +87,6 @@ class HeartbeatService:
             logger.info("Heartbeat disabled")
             return
 
-        # Idempotent: don't create a new task if already running
-        if self._task is not None and not self._task.done():
-            return
-
         self._running = True
         self._task = asyncio.create_task(self._run_loop())
         logger.info(f"Heartbeat started (every {self.interval_s}s)")

nanobot/providers/__init__.py

@@ -1,6 +1,6 @@
 """Provider module exports."""
 
-from nanobot.providers.base import LLMProvider, LLMResponse, LongContextError, ToolCallRequest
+from nanobot.providers.base import LLMProvider, LLMResponse, ToolCallRequest
 from nanobot.providers.litellm_provider import LiteLLMProvider
 from nanobot.providers.openai_codex_provider import OpenAICodexProvider
 from nanobot.providers.anthropic_oauth import AnthropicOAuthProvider

nanobot/providers/anthropic_oauth.py

@@ -10,8 +10,8 @@ from typing import Any
 import httpx
 from loguru import logger
 
-from nanobot.providers.base import LLMProvider, LLMResponse, LongContextError, ToolCallRequest
-from nanobot.providers.oauth_utils import get_auth_headers, get_claude_code_system_prefix
+from nanobot.providers.base import LLMProvider, LLMResponse, ToolCallRequest
+from nanobot.providers.oauth_utils import get_auth_headers
 
 
 class AnthropicOAuthProvider(LLMProvider):
@@ -27,7 +27,7 @@ class AnthropicOAuthProvider(LLMProvider):
     def __init__(
         self,
         oauth_token: str,
-        default_model: str = "claude-opus-4-7",
+        default_model: str = "claude-opus-4-5",
         api_base: str | None = None,
         thinking_budget: int = 0,
     ):
@@ -51,8 +51,8 @@ class AnthropicOAuthProvider(LLMProvider):
     def _normalize_model(model: str) -> str:
         """Normalize model name for the Anthropic API.
 
-        Anthropic model IDs use hyphens (claude-sonnet-4-6), but users often
-        write dots (claude-sonnet-4.6). Normalize so both work.
+        Anthropic model IDs use hyphens (claude-sonnet-4-5), but users often
+        write dots (claude-sonnet-4.5). Normalize so both work.
         """
         return model.replace(".", "-")
 
@@ -326,35 +326,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):
-        # BP3: Second-to-last user message (stable history from previous turn)
-        # BP4: Last user message (current turn, will become BP3 next turn)
-        # This allows BP3 to reuse what BP4 cached last turn.
-        user_indices = [i for i, m in enumerate(messages) if m.get("role") == "user"]
-
-        if len(user_indices) >= 2:
-            # BP3: Second-to-last user message
-            idx = user_indices[-2]
-            msg = messages[idx]
-            content = msg["content"]
-            if isinstance(content, str):
-                messages[idx] = {**msg, "content": [{"type": "text", "text": content, "cache_control": {"type": "ephemeral"}}]}
-            elif isinstance(content, list) and content:
-                new_content = list(content)
-                new_content[-1] = {**new_content[-1], "cache_control": {"type": "ephemeral"}}
-                messages[idx] = {**msg, "content": new_content}
-
-        if len(user_indices) >= 1:
-            # BP4: Last user message
-            idx = user_indices[-1]
-            msg = messages[idx]
-            content = msg["content"]
-            if isinstance(content, str):
-                messages[idx] = {**msg, "content": [{"type": "text", "text": content, "cache_control": {"type": "ephemeral"}}]}
-            elif isinstance(content, list) and content:
-                new_content = list(content)
-                new_content[-1] = {**new_content[-1], "cache_control": {"type": "ephemeral"}}
-                messages[idx] = {**msg, "content": new_content}
+        # Cache the last user message so conversation history is cached across turns
+        if messages:
+            last = messages[-1]
+            if last.get("role") == "user":
+                content = last["content"]
+                if isinstance(content, str):
+                    last = {**last, "content": [{"type": "text", "text": content, "cache_control": {"type": "ephemeral"}}]}
+                elif isinstance(content, list) and content:
+                    new_content = list(content)
+                    new_content[-1] = {**new_content[-1], "cache_control": {"type": "ephemeral"}}
+                    last = {**last, "content": new_content}
+                messages = messages[:-1] + [last]
 
         payload: dict[str, Any] = {
             "model": model,
@@ -377,14 +360,7 @@ class AnthropicOAuthProvider(LLMProvider):
             payload["temperature"] = temperature
 
         if system:
-            payload["system"] = [
-                {"type": "text", "text": get_claude_code_system_prefix()},
-                {"type": "text", "text": system, "cache_control": {"type": "ephemeral", "ttl": "1h"}},
-            ]
-        else:
-            payload["system"] = [
-                {"type": "text", "text": get_claude_code_system_prefix()},
-            ]
+            payload["system"] = [{"type": "text", "text": system, "cache_control": {"type": "ephemeral", "ttl": "1h"}}]
 
         if tools:
             cached_tools = list(tools)
@@ -431,114 +407,70 @@ class AnthropicOAuthProvider(LLMProvider):
 
         import asyncio
         import time as _time
-
-        max_retries = 3
-        base_delay = 2.0  # seconds
-
-        for attempt in range(max_retries + 1):
-            _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 (attempt {attempt+1}/{max_retries+1})")
-                if attempt == 0:
-                    await self._diagnose_connectivity()
-                await self._reset_client()
-                if attempt < max_retries:
-                    delay = base_delay * (2 ** attempt)
-                    logger.info(f"Retrying in {delay:.1f}s...")
-                    await asyncio.sleep(delay)
-                    continue
-                raise
-            except httpx.PoolTimeout:
-                elapsed = _time.monotonic() - _t0
-                logger.error(f"PoolTimeout after {elapsed:.1f}s (attempt {attempt+1}/{max_retries+1})")
-                await self._reset_client()
-                if attempt < max_retries:
-                    delay = base_delay * (2 ** attempt)
-                    logger.info(f"Retrying in {delay:.1f}s...")
-                    await asyncio.sleep(delay)
-                    continue
-                raise
-            except (httpx.ConnectError, httpx.TimeoutException) as e:
-                elapsed = _time.monotonic() - _t0
-                logger.error(f"{type(e).__name__} after {elapsed:.1f}s (attempt {attempt+1}/{max_retries+1})")
-                if attempt < max_retries:
-                    delay = base_delay * (2 ** attempt)
-                    logger.info(f"Retrying in {delay:.1f}s...")
-                    await asyncio.sleep(delay)
-                    continue
-                raise
+        _t0 = _time.monotonic()
+        try:
+            response = await client.post(
+                self._get_api_url(),
+                headers=headers,
+                json=payload,
+            )
+        except httpx.ConnectTimeout:
             elapsed = _time.monotonic() - _t0
-            if elapsed > 30:
-                logger.warning(f"Anthropic API slow response: {elapsed:.1f}s")
+            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:
-                import datetime
-                import os
-                header_dump = {
-                    "timestamp": datetime.datetime.now(datetime.UTC).isoformat(),
-                    "status_code": response.status_code,
-                    "model": payload.get("model"),
-                    "headers": dict(response.headers),
-                }
-                dump_path = "/root/.nanobot/workspace/api_headers.jsonl"
-                with open(dump_path, "a") as f:
-                    f.write(json.dumps(header_dump) + "\n")
-                # Capture rate limit state for quota-based model switching
-                hdrs = response.headers
-                rate_limit_state = {
-                    "updated_at": datetime.datetime.utcnow().isoformat(),
-                    "model": payload.get("model"),
-                    "weekly_all_models": float(hdrs["anthropic-ratelimit-unified-7d-utilization"]) if hdrs.get("anthropic-ratelimit-unified-7d-utilization") else None,
-                    "weekly_sonnet": float(hdrs["anthropic-ratelimit-unified-7d_sonnet-utilization"]) if hdrs.get("anthropic-ratelimit-unified-7d_sonnet-utilization") else None,
-                    "session_5h": float(hdrs["anthropic-ratelimit-unified-5h-utilization"]) if hdrs.get("anthropic-ratelimit-unified-5h-utilization") else None,
-                    "weekly_reset": int(hdrs["anthropic-ratelimit-unified-7d-reset"]) if hdrs.get("anthropic-ratelimit-unified-7d-reset") else None,
-                    "session_reset": int(hdrs["anthropic-ratelimit-unified-5h-reset"]) if hdrs.get("anthropic-ratelimit-unified-5h-reset") else None,
-                    "binding_limit": hdrs.get("anthropic-ratelimit-unified-representative-claim"),
-                    "sonnet_fallback": hdrs.get("anthropic-ratelimit-unified-fallback"),
-                }
-                state_path = "/root/.nanobot/workspace/memory/rate_limits.json"
-                os.makedirs(os.path.dirname(state_path), exist_ok=True)
-                with open(state_path, "w") as f:
-                    json.dump(rate_limit_state, f, indent=2)
-            except Exception as e:
-                logger.warning("Rate limit header capture failed: {}", e)
+        # Dump rate limit headers for analysis
+        try:
+            import datetime
+            import os
+            header_dump = {
+                "timestamp": datetime.datetime.utcnow().isoformat(),
+                "status_code": response.status_code,
+                "model": payload.get("model"),
+                "headers": dict(response.headers),
+            }
+            dump_path = "/root/.nanobot/workspace/api_headers.jsonl"
+            with open(dump_path, "a") as f:
+                f.write(json.dumps(header_dump) + "\n")
+            # Capture rate limit state for quota-based model switching
+            hdrs = response.headers
+            rate_limit_state = {
+                "updated_at": datetime.datetime.utcnow().isoformat(),
+                "model": payload.get("model"),
+                "weekly_all_models": float(hdrs["anthropic-ratelimit-unified-7d-utilization"]) if hdrs.get("anthropic-ratelimit-unified-7d-utilization") else None,
+                "weekly_sonnet": float(hdrs["anthropic-ratelimit-unified-7d_sonnet-utilization"]) if hdrs.get("anthropic-ratelimit-unified-7d_sonnet-utilization") else None,
+                "session_5h": float(hdrs["anthropic-ratelimit-unified-5h-utilization"]) if hdrs.get("anthropic-ratelimit-unified-5h-utilization") else None,
+                "weekly_reset": int(hdrs["anthropic-ratelimit-unified-7d-reset"]) if hdrs.get("anthropic-ratelimit-unified-7d-reset") else None,
+                "session_reset": int(hdrs["anthropic-ratelimit-unified-5h-reset"]) if hdrs.get("anthropic-ratelimit-unified-5h-reset") else None,
+                "binding_limit": hdrs.get("anthropic-ratelimit-unified-representative-claim"),
+                "sonnet_fallback": hdrs.get("anthropic-ratelimit-unified-fallback"),
+            }
+            state_path = "/root/.nanobot/workspace/memory/rate_limits.json"
+            os.makedirs(os.path.dirname(state_path), exist_ok=True)
+            with open(state_path, "w") as f:
+                json.dump(rate_limit_state, f, indent=2)
+        except Exception as e:
+            logger.warning("Rate limit header capture failed: {}", e)
 
-            # Retry on 5xx server errors and 429 rate limits
-            if response.status_code >= 500 or response.status_code == 429:
-                error_text = response.text
-                logger.warning(f"Anthropic API {response.status_code} (attempt {attempt+1}/{max_retries+1}): {error_text[:200]}")
+        if response.status_code != 200:
+            error_text = response.text
+            raise Exception(f"Anthropic API error {response.status_code}: {error_text}")
 
-                # Long context 429 — retrying won't help, need to trim context
-                if response.status_code == 429 and "long context" in error_text.lower():
-                    raise LongContextError(f"Context too long for current plan: {error_text[:200]}")
-
-                if attempt < max_retries:
-                    if response.status_code == 429:
-                        retry_after = response.headers.get("retry-after")
-                        delay = float(retry_after) if retry_after else base_delay * (2 ** attempt)
-                    else:
-                        delay = base_delay * (2 ** attempt)
-                    logger.info(f"Retrying in {delay:.1f}s...")
-                    await asyncio.sleep(delay)
-                    continue
-                raise Exception(f"Anthropic API error {response.status_code}: {error_text}")
-
-            if response.status_code != 200:
-                error_text = response.text
-                raise Exception(f"Anthropic API error {response.status_code}: {error_text}")
-
-            return response.json()
-
-        # Should not reach here, but just in case
-        raise Exception("Exhausted all retry attempts")
+        return response.json()
 
     async def chat(
         self,
@@ -557,7 +489,7 @@ class AnthropicOAuthProvider(LLMProvider):
         if "/" in model:
             model = model.split("/")[-1]
 
-        # Normalize dots to hyphens (claude-sonnet-4.6 -> claude-sonnet-4-6)
+        # Normalize dots to hyphens (claude-sonnet-4.5 -> claude-sonnet-4-5)
         model = self._normalize_model(model)
 
         system, prepared_messages = self._prepare_messages(messages)
@@ -590,8 +522,6 @@ class AnthropicOAuthProvider(LLMProvider):
                 beta_flags=beta_flags,
             )
             return self._parse_response(response)
-        except LongContextError:
-            raise  # Let caller handle context trimming
         except Exception as e:
             logger.exception("Exception in chat():")
             error_msg = f"{type(e).__name__}: {str(e)}" if str(e) else f"{type(e).__name__} (no message)"

nanobot/providers/base.py

@@ -28,11 +28,6 @@ class LLMResponse:
         return len(self.tool_calls) > 0
 
 
-class LongContextError(Exception):
-    """Raised when the API rejects a request due to long context limits."""
-    pass
-
-
 class LLMProvider(ABC):
     """
     Abstract base class for LLM providers.

nanobot/providers/litellm_provider.py

@@ -37,7 +37,7 @@ class LiteLLMProvider(LLMProvider):
         self, 
         api_key: str | None = None, 
         api_base: str | None = None,
-        default_model: str = "anthropic/claude-opus-4-7",
+        default_model: str = "anthropic/claude-opus-4-5",
         extra_headers: dict[str, str] | None = None,
         provider_name: str | None = None,
     ):
@@ -187,7 +187,7 @@ class LiteLLMProvider(LLMProvider):
         Args:
             messages: List of message dicts with 'role' and 'content'.
             tools: Optional list of tool definitions in OpenAI format.
-            model: Model identifier (e.g., 'anthropic/claude-sonnet-4-6').
+            model: Model identifier (e.g., 'anthropic/claude-sonnet-4-5').
             max_tokens: Maximum tokens in response.
             temperature: Sampling temperature.
         

nanobot/providers/oauth_utils.py

@@ -36,11 +36,3 @@ def get_auth_headers(token: str, is_oauth: bool = False) -> dict[str, str]:
         headers["x-api-key"] = token
 
     return headers
-
-
-def get_claude_code_system_prefix() -> str:
-    """Get the required system prompt prefix for OAuth tokens.
-
-    Anthropic requires this identity declaration for OAuth auth.
-    """
-    return "You are a Claude agent, built on Anthropic's Claude Agent SDK."

nanobot/session/manager.py

@@ -19,7 +19,9 @@ class Session:
 
     Stores messages in JSONL format for easy reading and persistence.
 
-    Messages are trimmed after consolidation to keep session size manageable.
+    Important: Messages are append-only for LLM cache efficiency.
+    The consolidation process writes summaries to MEMORY.md/HISTORY.md
+    but does NOT modify the messages list or get_history() output.
     """
 
     key: str  # channel:chat_id
@@ -27,6 +29,7 @@ class Session:
     created_at: datetime = field(default_factory=datetime.now)
     updated_at: datetime = field(default_factory=datetime.now)
     metadata: dict[str, Any] = field(default_factory=dict)
+    last_consolidated: int = 0  # Number of messages already consolidated to files
     
     def add_message(self, role: str, content: str, **kwargs: Any) -> None:
         """Add a message to the session."""
@@ -59,26 +62,18 @@ class Session:
         trimming old tool chains safely at token thresholds, so we send the full
         history and let the server decide what to drop.
 
-        Messages with ``_hidden_sig`` get a ``[HIDDEN:{sig}]`` prefix applied to
-        their content so the model knows the user never saw them.  The prefix is
-        applied at read time (not stored in content) to preserve prompt-cache
-        stability: the same prefixed string is produced every turn.
-
         Returns:
             List of messages in LLM format (API-relevant fields only).
         """
-        out: list[dict[str, Any]] = []
-        for m in self.messages:
-            msg = {k: v for k, v in m.items() if k in self._API_FIELDS and v is not None}
-            sig = m.get("_hidden_sig")
-            if sig and isinstance(msg.get("content"), str):
-                msg["content"] = f"[HIDDEN:{sig}] {msg['content']}"
-            out.append(msg)
-        return out
+        return [
+            {k: v for k, v in m.items() if k in self._API_FIELDS and v is not None}
+            for m in self.messages
+        ]
     
     def clear(self) -> None:
         """Clear all messages and reset session to initial state."""
         self.messages = []
+        self.last_consolidated = 0
         self.updated_at = datetime.now()
 
 
@@ -144,6 +139,7 @@ class SessionManager:
             messages = []
             metadata = {}
             created_at = None
+            last_consolidated = 0
 
             with open(path, encoding="utf-8") as f:
                 for line in f:
@@ -156,7 +152,7 @@ class SessionManager:
                     if data.get("_type") == "metadata":
                         metadata = data.get("metadata", {})
                         created_at = datetime.fromisoformat(data["created_at"]) if data.get("created_at") else None
-                        # Ignore legacy last_consolidated field
+                        last_consolidated = data.get("last_consolidated", 0)
                     else:
                         messages.append(data)
 
@@ -165,6 +161,7 @@ class SessionManager:
                 messages=messages,
                 created_at=created_at or datetime.now(),
                 metadata=metadata,
+                last_consolidated=last_consolidated
             )
         except Exception as e:
             logger.warning("Failed to load session {}: {}", key, e)
@@ -181,31 +178,13 @@ class SessionManager:
                 "created_at": session.created_at.isoformat(),
                 "updated_at": session.updated_at.isoformat(),
                 "metadata": session.metadata,
+                "last_consolidated": session.last_consolidated
             }
             f.write(json.dumps(metadata_line, ensure_ascii=False) + "\n")
             for msg in session.messages:
                 f.write(json.dumps(msg, ensure_ascii=False) + "\n")
 
         self._cache[session.key] = session
-        self._append_audit(session)
-
-    def _append_audit(self, session: Session) -> None:
-        """Append session state to an audit log (append-only, rotated monthly)."""
-        now = datetime.now()
-        safe_key = safe_filename(session.key.replace(":", "_"))
-        audit_path = self.sessions_dir / f"{safe_key}.audit.{now:%Y-%m}.jsonl"
-        try:
-            with open(audit_path, "a", encoding="utf-8") as f:
-                marker = {
-                    "_type": "save_marker",
-                    "timestamp": now.isoformat(),
-                    "message_count": len(session.messages),
-                }
-                f.write(json.dumps(marker, ensure_ascii=False) + "\n")
-                for msg in session.messages:
-                    f.write(json.dumps(msg, ensure_ascii=False) + "\n")
-        except Exception as e:
-            logger.warning("Audit log write failed for {}: {}", session.key, e)
     
     def invalidate(self, key: str) -> None:
         """Remove a session from the in-memory cache."""

pyproject.toml

@@ -50,11 +50,6 @@ dev = [
 mem0 = [
     "mem0ai>=0.1.0",
 ]
-matrix = [
-    "matrix-nio>=0.20.0",
-    "mistune>=3.0.0",
-    "nh3>=0.2.0",
-]
 
 [project.scripts]
 nanobot = "nanobot.cli.commands:app"

test-pr.sh

@@ -1,34 +0,0 @@
-#!/bin/bash
-# test-pr.sh - Quick PR testing script for nanobot staging
-#
-# Usage: ./test-pr.sh <pr-number> [test-message]
-# Example: ./test-pr.sh 31 "test tool use feature"
-
-set -e
-
-PR_NUM="$1"
-TEST_MSG="${2:-Hello, testing PR #$PR_NUM}"
-REPO_DIR="/config/workspace/nanobot-oauth-port/nanobot-fork"
-STAGING_CONFIG="/config/workspace/.nanobot-staging/config.json"
-
-if [ -z "$PR_NUM" ]; then
-  echo "Usage: $0 <pr-number> [test-message]"
-  exit 1
-fi
-
-echo "==> Fetching PR #$PR_NUM..."
-cd "$REPO_DIR"
-git fetch wylab "+pull/$PR_NUM/head:pr-$PR_NUM"
-
-echo "==> Checking out pr-$PR_NUM..."
-git checkout "pr-$PR_NUM"
-
-echo "==> Installing in editable mode..."
-uv pip install -e . -q
-
-echo "==> Testing with message: $TEST_MSG"
-NANOBOT_CONFIG="$STAGING_CONFIG" "$REPO_DIR/.venv/bin/nanobot" agent -m "$TEST_MSG"
-
-echo ""
-echo "==> Test complete. Branch pr-$PR_NUM is still checked out."
-echo "    Run 'git checkout main' to return to main branch."

tests/test_agent_loop_tool_result.py

@@ -28,7 +28,7 @@ def mock_session_manager():
         "messages": [],
         "metadata": {},
     })
-    session_mgr.save = MagicMock()  # Synchronous in production, not async
+    session_mgr.save = AsyncMock()
     return session_mgr
 
 

tests/test_anthropic_oauth.py

@@ -10,14 +10,14 @@ def provider():
     """Create provider with test OAuth token."""
     return AnthropicOAuthProvider(
         oauth_token="sk-ant-oat01-test-token",
-        default_model="claude-opus-4-7"
+        default_model="claude-opus-4-5"
     )
 
 
 def test_provider_init(provider):
     """Provider should initialize with OAuth token."""
     assert provider.oauth_token == "sk-ant-oat01-test-token"
-    assert provider.default_model == "claude-opus-4-7"
+    assert provider.default_model == "claude-opus-4-5"
 
 
 def test_provider_uses_bearer_auth(provider):
@@ -28,8 +28,18 @@ def test_provider_uses_bearer_auth(provider):
     assert "x-api-key" not in headers
 
 
-# test_chat_prepends_system_prompt removed - feature no longer exists
-# System prompt handling is done by the agent loop, not the provider
+@pytest.mark.asyncio
+async def test_chat_prepends_system_prompt(provider):
+    """Chat should prepend Claude Code identity to system prompt."""
+    messages = [{"role": "user", "content": "Hello"}]
+
+    with patch.object(provider, "_make_request", new_callable=AsyncMock) as mock:
+        mock.return_value = {"content": [{"type": "text", "text": "Hi"}], "stop_reason": "end_turn"}
+        await provider.chat(messages)
+
+        call_args = mock.call_args
+        system = call_args[1]["system"]
+        assert "Claude Code" in system
 
 
 def test_parse_response_text(provider):

tests/test_beta_flags.py

@@ -50,12 +50,11 @@ async def test_beta_flags_collected_from_tools():
             tools=tools_with_flags
         )
 
-        # Check that beta flag was added to headers (merged with hardcoded flags)
+        # Check that beta flag was added to headers
         call_args = mock_client.post.call_args
         headers = call_args[1]["headers"]
         assert "anthropic-beta" in headers
-        # Should include hardcoded flags + tool flag, sorted alphabetically
-        assert headers["anthropic-beta"] == "claude-code-20250219,computer-use-2025-11-24,context-management-2025-06-27,oauth-2025-04-20"
+        assert headers["anthropic-beta"] == "computer-use-2025-11-24"
 
 
 @pytest.mark.asyncio
@@ -100,5 +99,5 @@ async def test_multiple_beta_flags_joined():
         call_args = mock_client.post.call_args
         headers = call_args[1]["headers"]
         assert "anthropic-beta" in headers
-        # Should include hardcoded flags + tool flags, sorted alphabetically and joined with comma
-        assert headers["anthropic-beta"] == "claude-code-20250219,context-management-2025-06-27,flag-a,flag-b,oauth-2025-04-20"
+        # Should be sorted alphabetically and joined with comma
+        assert headers["anthropic-beta"] == "flag-a,flag-b"

tests/test_commands.py

@@ -29,7 +29,6 @@ def mock_paths():
 
         config_file = base_dir / "config.json"
         workspace_dir = base_dir / "workspace"
-        workspace_dir.mkdir()  # Create workspace directory
 
         mock_cp.return_value = config_file
         mock_ws.return_value = workspace_dir
@@ -57,20 +56,21 @@ def test_onboard_fresh_install(mock_paths):
 
 
 def test_onboard_existing_config_refresh(mock_paths):
-    """Config exists, user declines overwrite — should exit without changes."""
+    """Config exists, user declines overwrite — should refresh (load-merge-save)."""
     config_file, workspace_dir = mock_paths
     config_file.write_text('{"existing": true}')
 
     result = runner.invoke(app, ["onboard"], input="n\n")
 
-    # User declined, so command exits (typer.Exit() returns 0)
     assert result.exit_code == 0
     assert "Config already exists" in result.stdout
-    assert "Overwrite?" in result.stdout
+    assert "existing values preserved" in result.stdout
+    assert workspace_dir.exists()
+    assert (workspace_dir / "AGENTS.md").exists()
 
 
 def test_onboard_existing_config_overwrite(mock_paths):
-    """Config exists, user confirms overwrite — should create new config."""
+    """Config exists, user confirms overwrite — should reset to defaults."""
     config_file, workspace_dir = mock_paths
     config_file.write_text('{"existing": true}')
 
@@ -78,20 +78,20 @@ def test_onboard_existing_config_overwrite(mock_paths):
 
     assert result.exit_code == 0
     assert "Config already exists" in result.stdout
-    assert "Created config" in result.stdout
+    assert "Config reset to defaults" in result.stdout
     assert workspace_dir.exists()
 
 
 def test_onboard_existing_workspace_safe_create(mock_paths):
-    """Workspace exists (from fixture) — should add missing templates."""
+    """Workspace exists — should not recreate, but still add missing templates."""
     config_file, workspace_dir = mock_paths
-    # workspace_dir already exists from fixture
-    # No existing config, so onboard should proceed
+    workspace_dir.mkdir(parents=True)
+    config_file.write_text("{}")
 
-    result = runner.invoke(app, ["onboard"])
+    result = runner.invoke(app, ["onboard"], input="n\n")
 
     assert result.exit_code == 0
-    assert "Created workspace" in result.stdout
+    assert "Created workspace" not in result.stdout
     assert "Created AGENTS.md" in result.stdout
     assert (workspace_dir / "AGENTS.md").exists()
 

tests/test_computer_tool.py

@@ -12,17 +12,15 @@ async def test_computer_tool_screenshot():
     tool = ComputerTool20251124(vnc_host="localhost", vnc_port=5900)
 
     # Mock VNC client
-    with patch('nanobot.agent.tools.anthropic.computer.vnc_api.connect') as mock_connect:
-        mock_client = MagicMock()
-        # Mock captureScreen to write fake PNG data to file path
-        def fake_capture(path):
-            from pathlib import Path
-            Path(path).write_bytes(b"fake_png_data")
-        mock_client.captureScreen = MagicMock(side_effect=fake_capture)
-        mock_client.mouseMove = MagicMock()
-        mock_client.keyPress = MagicMock()
-        mock_client.refreshScreen = MagicMock()
-        mock_connect.return_value = mock_client
+    with patch('nanobot.agent.tools.anthropic.computer.VNCDoToolClient') as mock_vnc:
+        mock_client = AsyncMock()
+        mock_client.captureScreen = AsyncMock(return_value=b"fake_png_data")
+
+        # Set up async context manager
+        mock_context = MagicMock()
+        mock_context.__aenter__ = AsyncMock(return_value=mock_client)
+        mock_context.__aexit__ = AsyncMock(return_value=None)
+        mock_vnc.create = MagicMock(return_value=mock_context)
 
         result = await tool(action="screenshot")
 
@@ -36,10 +34,15 @@ async def test_computer_tool_mouse_move():
     """Test computer tool can move mouse."""
     tool = ComputerTool20251124(vnc_host="localhost", vnc_port=5900)
 
-    with patch('nanobot.agent.tools.anthropic.computer.vnc_api.connect') as mock_connect:
-        mock_client = MagicMock()
-        mock_client.mouseMove = MagicMock()
-        mock_connect.return_value = mock_client
+    with patch('nanobot.agent.tools.anthropic.computer.VNCDoToolClient') as mock_vnc:
+        mock_client = AsyncMock()
+        mock_client.mouseMove = AsyncMock()
+
+        # Set up async context manager
+        mock_context = MagicMock()
+        mock_context.__aenter__ = AsyncMock(return_value=mock_client)
+        mock_context.__aexit__ = AsyncMock(return_value=None)
+        mock_vnc.create = MagicMock(return_value=mock_context)
 
         result = await tool(action="mouse_move", coordinate=[100, 200])
 
@@ -53,17 +56,21 @@ async def test_computer_tool_key():
     """Test computer tool can press keys."""
     tool = ComputerTool20251124(vnc_host="localhost", vnc_port=5900)
 
-    with patch('nanobot.agent.tools.anthropic.computer.vnc_api.connect') as mock_connect:
-        mock_client = MagicMock()
-        mock_client.keyPress = MagicMock()
-        mock_connect.return_value = mock_client
+    with patch('nanobot.agent.tools.anthropic.computer.VNCDoToolClient') as mock_vnc:
+        mock_client = AsyncMock()
+        mock_client.keyPress = AsyncMock()
+
+        # Set up async context manager
+        mock_context = MagicMock()
+        mock_context.__aenter__ = AsyncMock(return_value=mock_client)
+        mock_context.__aexit__ = AsyncMock(return_value=None)
+        mock_vnc.create = MagicMock(return_value=mock_context)
 
         result = await tool(action="key", text="Return")
 
         assert isinstance(result, ToolResult)
         assert result.error is None
-        # Implementation converts keys to lowercase
-        mock_client.keyPress.assert_called_once_with("return")
+        mock_client.keyPress.assert_called_once_with("Return")
 
 
 def test_computer_tool_to_params():

tests/test_config_loader.py

@@ -1,142 +0,0 @@
-"""Tests for config loader (get_config_path and _migrate_config)"""
-
-import os
-from pathlib import Path
-
-from nanobot.config.loader import get_config_path, _migrate_config
-
-
-def test_get_config_path_default():
-    """get_config_path returns ~/.nanobot/config.json by default"""
-    # Ensure NANOBOT_CONFIG is not set
-    env_backup = os.environ.pop("NANOBOT_CONFIG", None)
-    try:
-        path = get_config_path()
-        assert path == Path.home() / ".nanobot" / "config.json"
-    finally:
-        if env_backup:
-            os.environ["NANOBOT_CONFIG"] = env_backup
-
-
-def test_get_config_path_with_env_var():
-    """get_config_path uses NANOBOT_CONFIG env var when set"""
-    custom_path = "/tmp/test-nanobot-config.json"
-    env_backup = os.environ.get("NANOBOT_CONFIG")
-    try:
-        os.environ["NANOBOT_CONFIG"] = custom_path
-        path = get_config_path()
-        assert path == Path(custom_path)
-    finally:
-        if env_backup:
-            os.environ["NANOBOT_CONFIG"] = env_backup
-        else:
-            os.environ.pop("NANOBOT_CONFIG", None)
-
-
-def test_migrate_config_with_oauth_credentials():
-    """_migrate_config extracts api_key from oauthCredentials"""
-    data = {
-        "providers": {
-            "anthropic": {
-                "oauthCredentials": {
-                    "access_token": "sk-ant-test-token",
-                    "refresh_token": "",
-                    "expires_at": 0,
-                }
-            }
-        }
-    }
-
-    result = _migrate_config(data)
-
-    # api_key should be extracted
-    assert result["providers"]["anthropic"]["api_key"] == "sk-ant-test-token"
-    # oauthCredentials should be removed after migration
-    assert "oauthCredentials" not in result["providers"]["anthropic"]
-
-
-def test_migrate_config_without_oauth_credentials():
-    """_migrate_config leaves config unchanged when no oauthCredentials"""
-    data = {
-        "providers": {
-            "anthropic": {
-                "api_key": "sk-ant-existing-key"
-            }
-        }
-    }
-
-    result = _migrate_config(data)
-
-    # Should remain unchanged
-    assert result["providers"]["anthropic"]["api_key"] == "sk-ant-existing-key"
-    assert "oauthCredentials" not in result["providers"]["anthropic"]
-
-
-def test_migrate_config_already_migrated():
-    """_migrate_config doesn't overwrite existing api_key"""
-    data = {
-        "providers": {
-            "anthropic": {
-                "api_key": "sk-ant-existing-key",
-                "oauthCredentials": {
-                    "access_token": "sk-ant-oauth-token",
-                    "refresh_token": "",
-                    "expires_at": 0,
-                }
-            }
-        }
-    }
-
-    result = _migrate_config(data)
-
-    # Existing api_key should be preserved
-    assert result["providers"]["anthropic"]["api_key"] == "sk-ant-existing-key"
-    # oauthCredentials should NOT be removed (api_key already existed)
-    assert "oauthCredentials" in result["providers"]["anthropic"]
-
-
-def test_migrate_config_empty_access_token():
-    """_migrate_config skips empty access_token"""
-    data = {
-        "providers": {
-            "anthropic": {
-                "oauthCredentials": {
-                    "access_token": "",
-                    "refresh_token": "",
-                    "expires_at": 0,
-                }
-            }
-        }
-    }
-
-    result = _migrate_config(data)
-
-    # api_key should not be set
-    assert "api_key" not in result["providers"]["anthropic"]
-    # oauthCredentials should remain (no migration happened)
-    assert "oauthCredentials" in result["providers"]["anthropic"]
-
-
-def test_migrate_config_preserves_other_fields():
-    """_migrate_config preserves other provider config fields"""
-    data = {
-        "providers": {
-            "anthropic": {
-                "oauthCredentials": {
-                    "access_token": "sk-ant-test-token",
-                    "refresh_token": "refresh-token",
-                },
-                "customField": "customValue",
-                "anotherField": 123,
-            }
-        }
-    }
-
-    result = _migrate_config(data)
-
-    # api_key added, oauthCredentials removed
-    assert result["providers"]["anthropic"]["api_key"] == "sk-ant-test-token"
-    assert "oauthCredentials" not in result["providers"]["anthropic"]
-    # Other fields preserved
-    assert result["providers"]["anthropic"]["customField"] == "customValue"
-    assert result["providers"]["anthropic"]["anotherField"] == 123

tests/test_config_oauth_integration.py

@@ -13,7 +13,7 @@ def test_oauth_token_injected_into_config(tmp_path, monkeypatch):
     # Create a minimal config file (no api key set)
     config_path = tmp_path / "config.json"
     config_path.write_text(json.dumps({
-        "agents": {"defaults": {"model": "anthropic/claude-opus-4-7"}},
+        "agents": {"defaults": {"model": "anthropic/claude-opus-4-5"}},
         "providers": {"anthropic": {"apiKey": ""}}
     }))
 

tests/test_consolidate_offset.py

@@ -0,0 +1,828 @@
+"""Test session management with cache-friendly message handling."""
+
+import asyncio
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from pathlib import Path
+from nanobot.session.manager import Session, SessionManager
+
+# Test constants
+MEMORY_WINDOW = 50
+KEEP_COUNT = MEMORY_WINDOW // 2  # 25
+
+
+def create_session_with_messages(key: str, count: int, role: str = "user") -> Session:
+    """Create a session and add the specified number of messages.
+
+    Args:
+        key: Session identifier
+        count: Number of messages to add
+        role: Message role (default: "user")
+
+    Returns:
+        Session with the specified messages
+    """
+    session = Session(key=key)
+    for i in range(count):
+        session.add_message(role, f"msg{i}")
+    return session
+
+
+def assert_messages_content(messages: list, start_index: int, end_index: int) -> None:
+    """Assert that messages contain expected content from start to end index.
+
+    Args:
+        messages: List of message dictionaries
+        start_index: Expected first message index
+        end_index: Expected last message index
+    """
+    assert len(messages) > 0
+    assert messages[0]["content"] == f"msg{start_index}"
+    assert messages[-1]["content"] == f"msg{end_index}"
+
+
+def get_old_messages(session: Session, last_consolidated: int, keep_count: int) -> list:
+    """Extract messages that would be consolidated using the standard slice logic.
+
+    Args:
+        session: The session containing messages
+        last_consolidated: Index of last consolidated message
+        keep_count: Number of recent messages to keep
+
+    Returns:
+        List of messages that would be consolidated
+    """
+    return session.messages[last_consolidated:-keep_count]
+
+
+class TestSessionLastConsolidated:
+    """Test last_consolidated tracking to avoid duplicate processing."""
+
+    def test_initial_last_consolidated_zero(self) -> None:
+        """Test that new session starts with last_consolidated=0."""
+        session = Session(key="test:initial")
+        assert session.last_consolidated == 0
+
+    def test_last_consolidated_persistence(self, tmp_path) -> None:
+        """Test that last_consolidated persists across save/load."""
+        manager = SessionManager(Path(tmp_path))
+        session1 = create_session_with_messages("test:persist", 20)
+        session1.last_consolidated = 15
+        manager.save(session1)
+
+        session2 = manager.get_or_create("test:persist")
+        assert session2.last_consolidated == 15
+        assert len(session2.messages) == 20
+
+    def test_clear_resets_last_consolidated(self) -> None:
+        """Test that clear() resets last_consolidated to 0."""
+        session = create_session_with_messages("test:clear", 10)
+        session.last_consolidated = 5
+
+        session.clear()
+        assert len(session.messages) == 0
+        assert session.last_consolidated == 0
+
+
+class TestSessionImmutableHistory:
+    """Test Session message immutability for cache efficiency."""
+
+    def test_initial_state(self) -> None:
+        """Test that new session has empty messages list."""
+        session = Session(key="test:initial")
+        assert len(session.messages) == 0
+
+    def test_add_messages_appends_only(self) -> None:
+        """Test that adding messages only appends, never modifies."""
+        session = Session(key="test:preserve")
+        session.add_message("user", "msg1")
+        session.add_message("assistant", "resp1")
+        session.add_message("user", "msg2")
+        assert len(session.messages) == 3
+        assert session.messages[0]["content"] == "msg1"
+
+    def test_get_history_returns_most_recent(self) -> None:
+        """Test get_history returns the most recent messages."""
+        session = Session(key="test:history")
+        for i in range(10):
+            session.add_message("user", f"msg{i}")
+            session.add_message("assistant", f"resp{i}")
+
+        history = session.get_history(max_messages=6)
+        assert len(history) == 6
+        assert history[0]["content"] == "msg7"
+        assert history[-1]["content"] == "resp9"
+
+    def test_get_history_with_all_messages(self) -> None:
+        """Test get_history with max_messages larger than actual."""
+        session = create_session_with_messages("test:all", 5)
+        history = session.get_history(max_messages=100)
+        assert len(history) == 5
+        assert history[0]["content"] == "msg0"
+
+    def test_get_history_stable_for_same_session(self) -> None:
+        """Test that get_history returns same content for same max_messages."""
+        session = create_session_with_messages("test:stable", 20)
+        history1 = session.get_history(max_messages=10)
+        history2 = session.get_history(max_messages=10)
+        assert history1 == history2
+
+    def test_messages_list_never_modified(self) -> None:
+        """Test that messages list is never modified after creation."""
+        session = create_session_with_messages("test:immutable", 5)
+        original_len = len(session.messages)
+
+        session.get_history(max_messages=2)
+        assert len(session.messages) == original_len
+
+        for _ in range(10):
+            session.get_history(max_messages=3)
+        assert len(session.messages) == original_len
+
+
+class TestSessionPersistence:
+    """Test Session persistence and reload."""
+
+    @pytest.fixture
+    def temp_manager(self, tmp_path):
+        return SessionManager(Path(tmp_path))
+
+    def test_persistence_roundtrip(self, temp_manager):
+        """Test that messages persist across save/load."""
+        session1 = create_session_with_messages("test:persistence", 20)
+        temp_manager.save(session1)
+
+        session2 = temp_manager.get_or_create("test:persistence")
+        assert len(session2.messages) == 20
+        assert session2.messages[0]["content"] == "msg0"
+        assert session2.messages[-1]["content"] == "msg19"
+
+    def test_get_history_after_reload(self, temp_manager):
+        """Test that get_history works correctly after reload."""
+        session1 = create_session_with_messages("test:reload", 30)
+        temp_manager.save(session1)
+
+        session2 = temp_manager.get_or_create("test:reload")
+        history = session2.get_history(max_messages=10)
+        assert len(history) == 10
+        assert history[0]["content"] == "msg20"
+        assert history[-1]["content"] == "msg29"
+
+    def test_clear_resets_session(self, temp_manager):
+        """Test that clear() properly resets session."""
+        session = create_session_with_messages("test:clear", 10)
+        assert len(session.messages) == 10
+
+        session.clear()
+        assert len(session.messages) == 0
+
+
+class TestConsolidationTriggerConditions:
+    """Test consolidation trigger conditions and logic."""
+
+    def test_consolidation_needed_when_messages_exceed_window(self):
+        """Test consolidation logic: should trigger when messages > memory_window."""
+        session = create_session_with_messages("test:trigger", 60)
+
+        total_messages = len(session.messages)
+        messages_to_process = total_messages - session.last_consolidated
+
+        assert total_messages > MEMORY_WINDOW
+        assert messages_to_process > 0
+
+        expected_consolidate_count = total_messages - KEEP_COUNT
+        assert expected_consolidate_count == 35
+
+    def test_consolidation_skipped_when_within_keep_count(self):
+        """Test consolidation skipped when total messages <= keep_count."""
+        session = create_session_with_messages("test:skip", 20)
+
+        total_messages = len(session.messages)
+        assert total_messages <= KEEP_COUNT
+
+        old_messages = get_old_messages(session, session.last_consolidated, KEEP_COUNT)
+        assert len(old_messages) == 0
+
+    def test_consolidation_skipped_when_no_new_messages(self):
+        """Test consolidation skipped when messages_to_process <= 0."""
+        session = create_session_with_messages("test:already_consolidated", 40)
+        session.last_consolidated = len(session.messages) - KEEP_COUNT  # 15
+
+        # Add a few more messages
+        for i in range(40, 42):
+            session.add_message("user", f"msg{i}")
+
+        total_messages = len(session.messages)
+        messages_to_process = total_messages - session.last_consolidated
+        assert messages_to_process > 0
+
+        # Simulate last_consolidated catching up
+        session.last_consolidated = total_messages - KEEP_COUNT
+        old_messages = get_old_messages(session, session.last_consolidated, KEEP_COUNT)
+        assert len(old_messages) == 0
+
+
+class TestLastConsolidatedEdgeCases:
+    """Test last_consolidated edge cases and data corruption scenarios."""
+
+    def test_last_consolidated_exceeds_message_count(self):
+        """Test behavior when last_consolidated > len(messages) (data corruption)."""
+        session = create_session_with_messages("test:corruption", 10)
+        session.last_consolidated = 20
+
+        total_messages = len(session.messages)
+        messages_to_process = total_messages - session.last_consolidated
+        assert messages_to_process <= 0
+
+        old_messages = get_old_messages(session, session.last_consolidated, 5)
+        assert len(old_messages) == 0
+
+    def test_last_consolidated_negative_value(self):
+        """Test behavior with negative last_consolidated (invalid state)."""
+        session = create_session_with_messages("test:negative", 10)
+        session.last_consolidated = -5
+
+        keep_count = 3
+        old_messages = get_old_messages(session, session.last_consolidated, keep_count)
+
+        # messages[-5:-3] with 10 messages gives indices 5,6
+        assert len(old_messages) == 2
+        assert old_messages[0]["content"] == "msg5"
+        assert old_messages[-1]["content"] == "msg6"
+
+    def test_messages_added_after_consolidation(self):
+        """Test correct behavior when new messages arrive after consolidation."""
+        session = create_session_with_messages("test:new_messages", 40)
+        session.last_consolidated = len(session.messages) - KEEP_COUNT  # 15
+
+        # Add new messages after consolidation
+        for i in range(40, 50):
+            session.add_message("user", f"msg{i}")
+
+        total_messages = len(session.messages)
+        old_messages = get_old_messages(session, session.last_consolidated, KEEP_COUNT)
+        expected_consolidate_count = total_messages - KEEP_COUNT - session.last_consolidated
+
+        assert len(old_messages) == expected_consolidate_count
+        assert_messages_content(old_messages, 15, 24)
+
+    def test_slice_behavior_when_indices_overlap(self):
+        """Test slice behavior when last_consolidated >= total - keep_count."""
+        session = create_session_with_messages("test:overlap", 30)
+        session.last_consolidated = 12
+
+        old_messages = get_old_messages(session, session.last_consolidated, 20)
+        assert len(old_messages) == 0
+
+
+class TestArchiveAllMode:
+    """Test archive_all mode (used by /new command)."""
+
+    def test_archive_all_consolidates_everything(self):
+        """Test archive_all=True consolidates all messages."""
+        session = create_session_with_messages("test:archive_all", 50)
+
+        archive_all = True
+        if archive_all:
+            old_messages = session.messages
+            assert len(old_messages) == 50
+
+        assert session.last_consolidated == 0
+
+    def test_archive_all_resets_last_consolidated(self):
+        """Test that archive_all mode resets last_consolidated to 0."""
+        session = create_session_with_messages("test:reset", 40)
+        session.last_consolidated = 15
+
+        archive_all = True
+        if archive_all:
+            session.last_consolidated = 0
+
+        assert session.last_consolidated == 0
+        assert len(session.messages) == 40
+
+    def test_archive_all_vs_normal_consolidation(self):
+        """Test difference between archive_all and normal consolidation."""
+        # Normal consolidation
+        session1 = create_session_with_messages("test:normal", 60)
+        session1.last_consolidated = len(session1.messages) - KEEP_COUNT
+
+        # archive_all mode
+        session2 = create_session_with_messages("test:all", 60)
+        session2.last_consolidated = 0
+
+        assert session1.last_consolidated == 35
+        assert len(session1.messages) == 60
+        assert session2.last_consolidated == 0
+        assert len(session2.messages) == 60
+
+
+class TestCacheImmutability:
+    """Test that consolidation doesn't modify session.messages (cache safety)."""
+
+    def test_consolidation_does_not_modify_messages_list(self):
+        """Test that consolidation leaves messages list unchanged."""
+        session = create_session_with_messages("test:immutable", 50)
+
+        original_messages = session.messages.copy()
+        original_len = len(session.messages)
+        session.last_consolidated = original_len - KEEP_COUNT
+
+        assert len(session.messages) == original_len
+        assert session.messages == original_messages
+
+    def test_get_history_does_not_modify_messages(self):
+        """Test that get_history doesn't modify messages list."""
+        session = create_session_with_messages("test:history_immutable", 40)
+        original_messages = [m.copy() for m in session.messages]
+
+        for _ in range(5):
+            history = session.get_history(max_messages=10)
+            assert len(history) == 10
+
+        assert len(session.messages) == 40
+        for i, msg in enumerate(session.messages):
+            assert msg["content"] == original_messages[i]["content"]
+
+    def test_consolidation_only_updates_last_consolidated(self):
+        """Test that consolidation only updates last_consolidated field."""
+        session = create_session_with_messages("test:field_only", 60)
+
+        original_messages = session.messages.copy()
+        original_key = session.key
+        original_metadata = session.metadata.copy()
+
+        session.last_consolidated = len(session.messages) - KEEP_COUNT
+
+        assert session.messages == original_messages
+        assert session.key == original_key
+        assert session.metadata == original_metadata
+        assert session.last_consolidated == 35
+
+
+class TestSliceLogic:
+    """Test the slice logic: messages[last_consolidated:-keep_count]."""
+
+    def test_slice_extracts_correct_range(self):
+        """Test that slice extracts the correct message range."""
+        session = create_session_with_messages("test:slice", 60)
+
+        old_messages = get_old_messages(session, 0, KEEP_COUNT)
+
+        assert len(old_messages) == 35
+        assert_messages_content(old_messages, 0, 34)
+
+        remaining = session.messages[-KEEP_COUNT:]
+        assert len(remaining) == 25
+        assert_messages_content(remaining, 35, 59)
+
+    def test_slice_with_partial_consolidation(self):
+        """Test slice when some messages already consolidated."""
+        session = create_session_with_messages("test:partial", 70)
+
+        last_consolidated = 30
+        old_messages = get_old_messages(session, last_consolidated, KEEP_COUNT)
+
+        assert len(old_messages) == 15
+        assert_messages_content(old_messages, 30, 44)
+
+    def test_slice_with_various_keep_counts(self):
+        """Test slice behavior with different keep_count values."""
+        session = create_session_with_messages("test:keep_counts", 50)
+
+        test_cases = [(10, 40), (20, 30), (30, 20), (40, 10)]
+
+        for keep_count, expected_count in test_cases:
+            old_messages = session.messages[0:-keep_count]
+            assert len(old_messages) == expected_count
+
+    def test_slice_when_keep_count_exceeds_messages(self):
+        """Test slice when keep_count > len(messages)."""
+        session = create_session_with_messages("test:exceed", 10)
+
+        old_messages = session.messages[0:-20]
+        assert len(old_messages) == 0
+
+
+class TestEmptyAndBoundarySessions:
+    """Test empty sessions and boundary conditions."""
+
+    def test_empty_session_consolidation(self):
+        """Test consolidation behavior with empty session."""
+        session = Session(key="test:empty")
+
+        assert len(session.messages) == 0
+        assert session.last_consolidated == 0
+
+        messages_to_process = len(session.messages) - session.last_consolidated
+        assert messages_to_process == 0
+
+        old_messages = get_old_messages(session, session.last_consolidated, KEEP_COUNT)
+        assert len(old_messages) == 0
+
+    def test_single_message_session(self):
+        """Test consolidation with single message."""
+        session = Session(key="test:single")
+        session.add_message("user", "only message")
+
+        assert len(session.messages) == 1
+
+        old_messages = get_old_messages(session, session.last_consolidated, KEEP_COUNT)
+        assert len(old_messages) == 0
+
+    def test_exactly_keep_count_messages(self):
+        """Test session with exactly keep_count messages."""
+        session = create_session_with_messages("test:exact", KEEP_COUNT)
+
+        assert len(session.messages) == KEEP_COUNT
+
+        old_messages = get_old_messages(session, session.last_consolidated, KEEP_COUNT)
+        assert len(old_messages) == 0
+
+    def test_just_over_keep_count(self):
+        """Test session with one message over keep_count."""
+        session = create_session_with_messages("test:over", KEEP_COUNT + 1)
+
+        assert len(session.messages) == 26
+
+        old_messages = get_old_messages(session, session.last_consolidated, KEEP_COUNT)
+        assert len(old_messages) == 1
+        assert old_messages[0]["content"] == "msg0"
+
+    def test_very_large_session(self):
+        """Test consolidation with very large message count."""
+        session = create_session_with_messages("test:large", 1000)
+
+        assert len(session.messages) == 1000
+
+        old_messages = get_old_messages(session, session.last_consolidated, KEEP_COUNT)
+        assert len(old_messages) == 975
+        assert_messages_content(old_messages, 0, 974)
+
+        remaining = session.messages[-KEEP_COUNT:]
+        assert len(remaining) == 25
+        assert_messages_content(remaining, 975, 999)
+
+    def test_session_with_gaps_in_consolidation(self):
+        """Test session with potential gaps in consolidation history."""
+        session = create_session_with_messages("test:gaps", 50)
+        session.last_consolidated = 10
+
+        # Add more messages
+        for i in range(50, 60):
+            session.add_message("user", f"msg{i}")
+
+        old_messages = get_old_messages(session, session.last_consolidated, KEEP_COUNT)
+
+        expected_count = 60 - KEEP_COUNT - 10
+        assert len(old_messages) == expected_count
+        assert_messages_content(old_messages, 10, 34)
+
+
+class TestConsolidationDeduplicationGuard:
+    """Test that consolidation tasks are deduplicated and serialized."""
+
+    @pytest.mark.asyncio
+    async def test_consolidation_guard_prevents_duplicate_tasks(self, tmp_path: Path) -> None:
+        """Concurrent messages above memory_window spawn only one consolidation task."""
+        from nanobot.agent.loop import AgentLoop
+        from nanobot.bus.events import InboundMessage
+        from nanobot.bus.queue import MessageBus
+        from nanobot.providers.base import LLMResponse
+
+        bus = MessageBus()
+        provider = MagicMock()
+        provider.get_default_model.return_value = "test-model"
+        loop = AgentLoop(
+            bus=bus, provider=provider, workspace=tmp_path, model="test-model", memory_window=10
+        )
+
+        loop.provider.chat = AsyncMock(return_value=LLMResponse(content="ok", tool_calls=[]))
+        loop.tools.get_definitions = MagicMock(return_value=[])
+
+        session = loop.sessions.get_or_create("cli:test")
+        for i in range(15):
+            session.add_message("user", f"msg{i}")
+            session.add_message("assistant", f"resp{i}")
+        loop.sessions.save(session)
+
+        consolidation_calls = 0
+
+        async def _fake_consolidate(_session, archive_all: bool = False) -> None:
+            nonlocal consolidation_calls
+            consolidation_calls += 1
+            await asyncio.sleep(0.05)
+
+        loop._consolidate_memory = _fake_consolidate  # type: ignore[method-assign]
+
+        msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="hello")
+        await loop._process_message(msg)
+        await loop._process_message(msg)
+        await asyncio.sleep(0.1)
+
+        assert consolidation_calls == 1, (
+            f"Expected exactly 1 consolidation, got {consolidation_calls}"
+        )
+
+    @pytest.mark.asyncio
+    async def test_new_command_guard_prevents_concurrent_consolidation(
+        self, tmp_path: Path
+    ) -> None:
+        """/new command does not run consolidation concurrently with in-flight consolidation."""
+        from nanobot.agent.loop import AgentLoop
+        from nanobot.bus.events import InboundMessage
+        from nanobot.bus.queue import MessageBus
+        from nanobot.providers.base import LLMResponse
+
+        bus = MessageBus()
+        provider = MagicMock()
+        provider.get_default_model.return_value = "test-model"
+        loop = AgentLoop(
+            bus=bus, provider=provider, workspace=tmp_path, model="test-model", memory_window=10
+        )
+
+        loop.provider.chat = AsyncMock(return_value=LLMResponse(content="ok", tool_calls=[]))
+        loop.tools.get_definitions = MagicMock(return_value=[])
+
+        session = loop.sessions.get_or_create("cli:test")
+        for i in range(15):
+            session.add_message("user", f"msg{i}")
+            session.add_message("assistant", f"resp{i}")
+        loop.sessions.save(session)
+
+        consolidation_calls = 0
+        active = 0
+        max_active = 0
+
+        async def _fake_consolidate(_session, archive_all: bool = False) -> None:
+            nonlocal consolidation_calls, active, max_active
+            consolidation_calls += 1
+            active += 1
+            max_active = max(max_active, active)
+            await asyncio.sleep(0.05)
+            active -= 1
+
+        loop._consolidate_memory = _fake_consolidate  # type: ignore[method-assign]
+
+        msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="hello")
+        await loop._process_message(msg)
+
+        new_msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="/new")
+        await loop._process_message(new_msg)
+        await asyncio.sleep(0.1)
+
+        assert consolidation_calls == 2, (
+            f"Expected normal + /new consolidations, got {consolidation_calls}"
+        )
+        assert max_active == 1, (
+            f"Expected serialized consolidation, observed concurrency={max_active}"
+        )
+
+    @pytest.mark.asyncio
+    async def test_consolidation_tasks_are_referenced(self, tmp_path: Path) -> None:
+        """create_task results are tracked in _consolidation_tasks while in flight."""
+        from nanobot.agent.loop import AgentLoop
+        from nanobot.bus.events import InboundMessage
+        from nanobot.bus.queue import MessageBus
+        from nanobot.providers.base import LLMResponse
+
+        bus = MessageBus()
+        provider = MagicMock()
+        provider.get_default_model.return_value = "test-model"
+        loop = AgentLoop(
+            bus=bus, provider=provider, workspace=tmp_path, model="test-model", memory_window=10
+        )
+
+        loop.provider.chat = AsyncMock(return_value=LLMResponse(content="ok", tool_calls=[]))
+        loop.tools.get_definitions = MagicMock(return_value=[])
+
+        session = loop.sessions.get_or_create("cli:test")
+        for i in range(15):
+            session.add_message("user", f"msg{i}")
+            session.add_message("assistant", f"resp{i}")
+        loop.sessions.save(session)
+
+        started = asyncio.Event()
+
+        async def _slow_consolidate(_session, archive_all: bool = False) -> None:
+            started.set()
+            await asyncio.sleep(0.1)
+
+        loop._consolidate_memory = _slow_consolidate  # type: ignore[method-assign]
+
+        msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="hello")
+        await loop._process_message(msg)
+
+        await started.wait()
+        assert len(loop._consolidation_tasks) == 1, "Task must be referenced while in-flight"
+
+        await asyncio.sleep(0.15)
+        assert len(loop._consolidation_tasks) == 0, (
+            "Task reference must be removed after completion"
+        )
+
+    @pytest.mark.asyncio
+    async def test_new_waits_for_inflight_consolidation_and_preserves_messages(
+        self, tmp_path: Path
+    ) -> None:
+        """/new waits for in-flight consolidation and archives before clear."""
+        from nanobot.agent.loop import AgentLoop
+        from nanobot.bus.events import InboundMessage
+        from nanobot.bus.queue import MessageBus
+        from nanobot.providers.base import LLMResponse
+
+        bus = MessageBus()
+        provider = MagicMock()
+        provider.get_default_model.return_value = "test-model"
+        loop = AgentLoop(
+            bus=bus, provider=provider, workspace=tmp_path, model="test-model", memory_window=10
+        )
+
+        loop.provider.chat = AsyncMock(return_value=LLMResponse(content="ok", tool_calls=[]))
+        loop.tools.get_definitions = MagicMock(return_value=[])
+
+        session = loop.sessions.get_or_create("cli:test")
+        for i in range(15):
+            session.add_message("user", f"msg{i}")
+            session.add_message("assistant", f"resp{i}")
+        loop.sessions.save(session)
+
+        started = asyncio.Event()
+        release = asyncio.Event()
+        archived_count = 0
+
+        async def _fake_consolidate(sess, archive_all: bool = False) -> bool:
+            nonlocal archived_count
+            if archive_all:
+                archived_count = len(sess.messages)
+                return True
+            started.set()
+            await release.wait()
+            return True
+
+        loop._consolidate_memory = _fake_consolidate  # type: ignore[method-assign]
+
+        msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="hello")
+        await loop._process_message(msg)
+        await started.wait()
+
+        new_msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="/new")
+        pending_new = asyncio.create_task(loop._process_message(new_msg))
+
+        await asyncio.sleep(0.02)
+        assert not pending_new.done(), "/new should wait while consolidation is in-flight"
+
+        release.set()
+        response = await pending_new
+        assert response is not None
+        assert "new session started" in response.content.lower()
+        assert archived_count > 0, "Expected /new archival to process a non-empty snapshot"
+
+        session_after = loop.sessions.get_or_create("cli:test")
+        assert session_after.messages == [], "Session should be cleared after successful archival"
+
+    @pytest.mark.asyncio
+    async def test_new_does_not_clear_session_when_archive_fails(self, tmp_path: Path) -> None:
+        """/new must keep session data if archive step reports failure."""
+        from nanobot.agent.loop import AgentLoop
+        from nanobot.bus.events import InboundMessage
+        from nanobot.bus.queue import MessageBus
+        from nanobot.providers.base import LLMResponse
+
+        bus = MessageBus()
+        provider = MagicMock()
+        provider.get_default_model.return_value = "test-model"
+        loop = AgentLoop(
+            bus=bus, provider=provider, workspace=tmp_path, model="test-model", memory_window=10
+        )
+
+        loop.provider.chat = AsyncMock(return_value=LLMResponse(content="ok", tool_calls=[]))
+        loop.tools.get_definitions = MagicMock(return_value=[])
+
+        session = loop.sessions.get_or_create("cli:test")
+        for i in range(5):
+            session.add_message("user", f"msg{i}")
+            session.add_message("assistant", f"resp{i}")
+        loop.sessions.save(session)
+        before_count = len(session.messages)
+
+        async def _failing_consolidate(sess, archive_all: bool = False) -> bool:
+            if archive_all:
+                return False
+            return True
+
+        loop._consolidate_memory = _failing_consolidate  # type: ignore[method-assign]
+
+        new_msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="/new")
+        response = await loop._process_message(new_msg)
+
+        assert response is not None
+        assert "failed" in response.content.lower()
+        session_after = loop.sessions.get_or_create("cli:test")
+        assert len(session_after.messages) == before_count, (
+            "Session must remain intact when /new archival fails"
+        )
+
+    @pytest.mark.asyncio
+    async def test_new_archives_only_unconsolidated_messages_after_inflight_task(
+        self, tmp_path: Path
+    ) -> None:
+        """/new should archive only messages not yet consolidated by prior task."""
+        from nanobot.agent.loop import AgentLoop
+        from nanobot.bus.events import InboundMessage
+        from nanobot.bus.queue import MessageBus
+        from nanobot.providers.base import LLMResponse
+
+        bus = MessageBus()
+        provider = MagicMock()
+        provider.get_default_model.return_value = "test-model"
+        loop = AgentLoop(
+            bus=bus, provider=provider, workspace=tmp_path, model="test-model", memory_window=10
+        )
+
+        loop.provider.chat = AsyncMock(return_value=LLMResponse(content="ok", tool_calls=[]))
+        loop.tools.get_definitions = MagicMock(return_value=[])
+
+        session = loop.sessions.get_or_create("cli:test")
+        for i in range(15):
+            session.add_message("user", f"msg{i}")
+            session.add_message("assistant", f"resp{i}")
+        loop.sessions.save(session)
+
+        started = asyncio.Event()
+        release = asyncio.Event()
+        archived_count = -1
+
+        async def _fake_consolidate(sess, archive_all: bool = False) -> bool:
+            nonlocal archived_count
+            if archive_all:
+                archived_count = len(sess.messages)
+                return True
+
+            started.set()
+            await release.wait()
+            sess.last_consolidated = len(sess.messages) - 3
+            return True
+
+        loop._consolidate_memory = _fake_consolidate  # type: ignore[method-assign]
+
+        msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="hello")
+        await loop._process_message(msg)
+        await started.wait()
+
+        new_msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="/new")
+        pending_new = asyncio.create_task(loop._process_message(new_msg))
+        await asyncio.sleep(0.02)
+        assert not pending_new.done()
+
+        release.set()
+        response = await pending_new
+
+        assert response is not None
+        assert "new session started" in response.content.lower()
+        assert archived_count == 3, (
+            f"Expected only unconsolidated tail to archive, got {archived_count}"
+        )
+
+    @pytest.mark.asyncio
+    async def test_new_cleans_up_consolidation_lock_for_invalidated_session(
+        self, tmp_path: Path
+    ) -> None:
+        """/new should remove lock entry for fully invalidated session key."""
+        from nanobot.agent.loop import AgentLoop
+        from nanobot.bus.events import InboundMessage
+        from nanobot.bus.queue import MessageBus
+        from nanobot.providers.base import LLMResponse
+
+        bus = MessageBus()
+        provider = MagicMock()
+        provider.get_default_model.return_value = "test-model"
+        loop = AgentLoop(
+            bus=bus, provider=provider, workspace=tmp_path, model="test-model", memory_window=10
+        )
+
+        loop.provider.chat = AsyncMock(return_value=LLMResponse(content="ok", tool_calls=[]))
+        loop.tools.get_definitions = MagicMock(return_value=[])
+
+        session = loop.sessions.get_or_create("cli:test")
+        for i in range(3):
+            session.add_message("user", f"msg{i}")
+            session.add_message("assistant", f"resp{i}")
+        loop.sessions.save(session)
+
+        # Ensure lock exists before /new.
+        loop._consolidation_locks.setdefault(session.key, asyncio.Lock())
+        assert session.key in loop._consolidation_locks
+
+        async def _ok_consolidate(sess, archive_all: bool = False) -> bool:
+            return True
+
+        loop._consolidate_memory = _ok_consolidate  # type: ignore[method-assign]
+
+        new_msg = InboundMessage(channel="cli", sender_id="user", chat_id="test", content="/new")
+        response = await loop._process_message(new_msg)
+
+        assert response is not None
+        assert "new session started" in response.content.lower()
+        assert session.key not in loop._consolidation_locks

tests/test_context_prompt_cache.py

@@ -40,7 +40,7 @@ def test_system_prompt_stays_stable_when_clock_changes(tmp_path, monkeypatch) ->
 
 
 def test_runtime_context_is_separate_untrusted_user_message(tmp_path) -> None:
-    """Runtime metadata should be included in the system prompt."""
+    """Runtime metadata should be a separate user message before the actual user message."""
     workspace = _make_workspace(tmp_path)
     builder = ContextBuilder(workspace)
 
@@ -51,12 +51,16 @@ def test_runtime_context_is_separate_untrusted_user_message(tmp_path) -> None:
         chat_id="direct",
     )
 
-    # Runtime context should be in the system prompt
     assert messages[0]["role"] == "system"
-    assert "## Current Session" in messages[0]["content"]
-    assert "Channel: cli" in messages[0]["content"]
-    assert "Chat ID: direct" in messages[0]["content"]
+    assert "## Current Session" not in messages[0]["content"]
+
+    assert messages[-2]["role"] == "user"
+    runtime_content = messages[-2]["content"]
+    assert isinstance(runtime_content, str)
+    assert ContextBuilder._RUNTIME_CONTEXT_TAG in runtime_content
+    assert "Current Time:" in runtime_content
+    assert "Channel: cli" in runtime_content
+    assert "Chat ID: direct" in runtime_content
 
-    # The actual user message should be the last message
     assert messages[-1]["role"] == "user"
     assert messages[-1]["content"] == "Return exactly: OK"

tests/test_edit_tool.py

@@ -113,4 +113,4 @@ def test_edit_tool_to_params():
     params = tool.to_params()
 
     assert params["type"] == "text_editor_20250728"
-    assert params["name"] == "str_replace_based_edit_tool"
+    assert params["name"] == "str_replace_editor"

tests/test_heartbeat_service.py

@@ -3,12 +3,27 @@ import asyncio
 import pytest
 
 from nanobot.heartbeat.service import HeartbeatService
+from nanobot.providers.base import LLMResponse, ToolCallRequest
+
+
+class DummyProvider:
+    def __init__(self, responses: list[LLMResponse]):
+        self._responses = list(responses)
+
+    async def chat(self, *args, **kwargs) -> LLMResponse:
+        if self._responses:
+            return self._responses.pop(0)
+        return LLMResponse(content="", tool_calls=[])
 
 
 @pytest.mark.asyncio
 async def test_start_is_idempotent(tmp_path) -> None:
+    provider = DummyProvider([])
+
     service = HeartbeatService(
         workspace=tmp_path,
+        provider=provider,
+        model="openai/gpt-4o-mini",
         interval_s=9999,
         enabled=True,
     )
@@ -23,36 +38,80 @@ async def test_start_is_idempotent(tmp_path) -> None:
     await asyncio.sleep(0)
 
 
+@pytest.mark.asyncio
+async def test_decide_returns_skip_when_no_tool_call(tmp_path) -> None:
+    provider = DummyProvider([LLMResponse(content="no tool call", tool_calls=[])])
+    service = HeartbeatService(
+        workspace=tmp_path,
+        provider=provider,
+        model="openai/gpt-4o-mini",
+    )
+
+    action, tasks = await service._decide("heartbeat content")
+    assert action == "skip"
+    assert tasks == ""
+
+
 @pytest.mark.asyncio
 async def test_trigger_now_executes_when_decision_is_run(tmp_path) -> None:
     (tmp_path / "HEARTBEAT.md").write_text("- [ ] do thing", encoding="utf-8")
 
-    called_with: list[tuple[str, dict | None]] = []
+    provider = DummyProvider([
+        LLMResponse(
+            content="",
+            tool_calls=[
+                ToolCallRequest(
+                    id="hb_1",
+                    name="heartbeat",
+                    arguments={"action": "run", "tasks": "check open tasks"},
+                )
+            ],
+        )
+    ])
 
-    async def _on_heartbeat(prompt: str, metadata: dict | None = None) -> str:
-        called_with.append((prompt, metadata))
+    called_with: list[str] = []
+
+    async def _on_execute(tasks: str) -> str:
+        called_with.append(tasks)
         return "done"
 
     service = HeartbeatService(
         workspace=tmp_path,
-        on_heartbeat=_on_heartbeat,
+        provider=provider,
+        model="openai/gpt-4o-mini",
+        on_execute=_on_execute,
     )
 
     result = await service.trigger_now()
     assert result == "done"
-    assert len(called_with) == 1
-    prompt, metadata = called_with[0]
-    assert "HEARTBEAT.md" in prompt
-    assert metadata == {"suppress_output": True}
+    assert called_with == ["check open tasks"]
 
 
 @pytest.mark.asyncio
-async def test_trigger_now_returns_none_when_no_callback(tmp_path) -> None:
+async def test_trigger_now_returns_none_when_decision_is_skip(tmp_path) -> None:
     (tmp_path / "HEARTBEAT.md").write_text("- [ ] do thing", encoding="utf-8")
 
+    provider = DummyProvider([
+        LLMResponse(
+            content="",
+            tool_calls=[
+                ToolCallRequest(
+                    id="hb_1",
+                    name="heartbeat",
+                    arguments={"action": "skip"},
+                )
+            ],
+        )
+    ])
+
+    async def _on_execute(tasks: str) -> str:
+        return tasks
+
     service = HeartbeatService(
         workspace=tmp_path,
-        on_heartbeat=None,  # No callback
+        provider=provider,
+        model="openai/gpt-4o-mini",
+        on_execute=_on_execute,
     )
 
     assert await service.trigger_now() is None

tests/test_long_context_recovery.py

@@ -1,78 +0,0 @@
-"""Test auto-consolidation on long context 429 errors."""
-
-import pytest
-from unittest.mock import AsyncMock, MagicMock
-
-from nanobot.providers.base import LongContextError, LLMResponse
-
-
-def test_long_context_error_is_exception():
-    """LongContextError should be a distinct exception class."""
-    err = LongContextError("too long")
-    assert isinstance(err, Exception)
-    assert str(err) == "too long"
-
-
-@pytest.mark.asyncio
-async def test_provider_raises_long_context_error_on_long_context_429():
-    """Provider should raise LongContextError immediately for long-context 429s."""
-    from nanobot.providers.anthropic_oauth import AnthropicOAuthProvider
-
-    provider = AnthropicOAuthProvider(
-        oauth_token="sk-ant-oat01-test-token",
-        default_model="claude-sonnet-4-6",
-    )
-
-    mock_response = MagicMock()
-    mock_response.status_code = 429
-    mock_response.text = '{"type":"error","error":{"type":"rate_limit_error","message":"Extra usage is required for long context requests."}}'
-    mock_response.headers = {}
-
-    mock_client = AsyncMock()
-    mock_client.post.return_value = mock_response
-    provider._client = mock_client
-
-    with pytest.raises(LongContextError, match="Context too long"):
-        await provider._make_request(
-            messages=[{"role": "user", "content": "hello"}],
-        )
-
-    # Should NOT retry — only one call
-    assert mock_client.post.call_count == 1
-
-
-@pytest.mark.asyncio
-async def test_provider_retries_normal_429():
-    """Provider should still retry normal 429s (not long-context)."""
-    from nanobot.providers.anthropic_oauth import AnthropicOAuthProvider
-
-    provider = AnthropicOAuthProvider(
-        oauth_token="sk-ant-oat01-test-token",
-        default_model="claude-sonnet-4-6",
-    )
-
-    rate_limit_response = MagicMock()
-    rate_limit_response.status_code = 429
-    rate_limit_response.text = '{"type":"error","error":{"type":"rate_limit_error","message":"Rate limit exceeded"}}'
-    rate_limit_response.headers = {}
-
-    success_response = MagicMock()
-    success_response.status_code = 200
-    success_response.headers = {}
-    success_response.json.return_value = {
-        "content": [{"type": "text", "text": "ok"}],
-        "stop_reason": "end_turn",
-        "usage": {"input_tokens": 1, "output_tokens": 1},
-    }
-
-    mock_client = AsyncMock()
-    mock_client.post.side_effect = [rate_limit_response, success_response]
-    provider._client = mock_client
-
-    result = await provider._make_request(
-        messages=[{"role": "user", "content": "hello"}],
-    )
-
-    # Should have retried and succeeded
-    assert mock_client.post.call_count == 2
-    assert result["stop_reason"] == "end_turn"

tests/test_matrix_channel.py

@@ -676,7 +676,7 @@ async def test_on_media_message_respects_declared_size_limit(
     assert client.download_calls == []
     assert len(handled) == 1
     assert handled[0]["media"] == []
-    assert handled[0]["metadata"].get("attachments", []) == []
+    assert handled[0]["metadata"]["attachments"] == []
     assert "[attachment: large.bin - too large]" in handled[0]["content"]
 
 
@@ -712,7 +712,7 @@ async def test_on_media_message_uses_server_limit_when_smaller_than_local_limit(
     assert client.download_calls == []
     assert len(handled) == 1
     assert handled[0]["media"] == []
-    assert handled[0]["metadata"].get("attachments", []) == []
+    assert handled[0]["metadata"]["attachments"] == []
     assert "[attachment: large.bin - too large]" in handled[0]["content"]
 
 
@@ -746,7 +746,7 @@ async def test_on_media_message_handles_download_error(monkeypatch, tmp_path) ->
     assert len(client.download_calls) == 1
     assert len(handled) == 1
     assert handled[0]["media"] == []
-    assert handled[0]["metadata"].get("attachments", []) == []
+    assert handled[0]["metadata"]["attachments"] == []
     assert "[attachment: photo.png - download failed]" in handled[0]["content"]
 
 
@@ -830,7 +830,7 @@ async def test_on_media_message_handles_decrypt_error(monkeypatch, tmp_path) ->
 
     assert len(handled) == 1
     assert handled[0]["media"] == []
-    assert handled[0]["metadata"].get("attachments", []) == []
+    assert handled[0]["metadata"]["attachments"] == []
     assert "[attachment: secret.txt - download failed]" in handled[0]["content"]
 
 
@@ -972,6 +972,7 @@ async def test_send_passes_thread_relates_to_to_attachment_upload(monkeypatch) -
     captured: dict[str, object] = {}
 
     async def _fake_upload_and_send_attachment(
+        *,
         room_id: str,
         path: Path,
         limit_bytes: int,

tests/test_mem0_extract_facts.py

@@ -1,129 +0,0 @@
-"""Test mem0 fact extraction calls provider with thinking disabled."""
-
-import pytest
-from unittest.mock import AsyncMock, MagicMock
-from pathlib import Path
-
-from nanobot.providers.base import LLMResponse
-
-
-@pytest.fixture
-def mock_provider():
-    provider = AsyncMock()
-    provider.chat = AsyncMock(return_value=LLMResponse(
-        content='{"facts": ["user likes Python", "user works on nanobot"]}',
-        finish_reason="end_turn",
-    ))
-    return provider
-
-
-@pytest.fixture
-def mem0_store(tmp_path):
-    """Create a Mem0MemoryStore with mocked mem0 dependency."""
-    # We can't import Mem0MemoryStore at module level because it requires
-    # the mem0 package. Instead, we test extract_facts as a standalone method
-    # by constructing a minimal instance.
-    try:
-        from nanobot.agent.memory_mem0 import Mem0MemoryStore
-        store = Mem0MemoryStore(workspace=tmp_path)
-        return store
-    except ImportError:
-        pytest.skip("mem0 not installed")
-
-
-@pytest.mark.asyncio
-async def test_extract_facts_passes_thinking_budget_zero(mock_provider):
-    """extract_facts must pass thinking_budget=0 to provider.chat().
-
-    Without this, the provider inherits its instance default (e.g. 10000),
-    causing the model to spend tokens on thinking instead of outputting JSON.
-    """
-    try:
-        from nanobot.agent.memory_mem0 import Mem0MemoryStore
-    except ImportError:
-        pytest.skip("mem0 not installed")
-
-    # Create a minimal instance without full mem0 init
-    store = object.__new__(Mem0MemoryStore)
-    store.custom_prompt = "Extract facts as JSON: "
-
-    messages = [
-        {"role": "user", "content": "I like Python programming"},
-        {"role": "assistant", "content": "That's great! Python is versatile."},
-    ]
-
-    facts = await store.extract_facts(messages, mock_provider, "claude-sonnet-4-6")
-
-    # Verify provider.chat was called with thinking_budget=0
-    mock_provider.chat.assert_called_once()
-    call_kwargs = mock_provider.chat.call_args.kwargs
-    assert call_kwargs["thinking_budget"] == 0
-
-
-@pytest.mark.asyncio
-async def test_extract_facts_returns_parsed_facts(mock_provider):
-    """extract_facts should parse JSON response into a list of fact strings."""
-    try:
-        from nanobot.agent.memory_mem0 import Mem0MemoryStore
-    except ImportError:
-        pytest.skip("mem0 not installed")
-
-    store = object.__new__(Mem0MemoryStore)
-    store.custom_prompt = "Extract facts as JSON: "
-
-    messages = [
-        {"role": "user", "content": "I like Python programming"},
-    ]
-
-    facts = await store.extract_facts(messages, mock_provider, "claude-sonnet-4-6")
-
-    assert facts == ["user likes Python", "user works on nanobot"]
-
-
-@pytest.mark.asyncio
-async def test_extract_facts_handles_empty_response():
-    """extract_facts should return empty list when provider returns no content."""
-    try:
-        from nanobot.agent.memory_mem0 import Mem0MemoryStore
-    except ImportError:
-        pytest.skip("mem0 not installed")
-
-    provider = AsyncMock()
-    provider.chat = AsyncMock(return_value=LLMResponse(
-        content="",
-        finish_reason="end_turn",
-    ))
-
-    store = object.__new__(Mem0MemoryStore)
-    store.custom_prompt = "Extract facts as JSON: "
-
-    messages = [{"role": "user", "content": "Hello there"}]
-
-    facts = await store.extract_facts(messages, provider, "claude-sonnet-4-6")
-
-    assert facts == []
-
-
-@pytest.mark.asyncio
-async def test_extract_facts_skips_empty_messages():
-    """extract_facts should return empty list when all messages have empty content."""
-    try:
-        from nanobot.agent.memory_mem0 import Mem0MemoryStore
-    except ImportError:
-        pytest.skip("mem0 not installed")
-
-    provider = AsyncMock()
-
-    store = object.__new__(Mem0MemoryStore)
-    store.custom_prompt = "Extract facts as JSON: "
-
-    messages = [
-        {"role": "user", "content": ""},
-        {"role": "assistant", "content": ""},
-    ]
-
-    facts = await store.extract_facts(messages, provider, "claude-sonnet-4-6")
-
-    assert facts == []
-    # Provider should not be called when there's no content
-    provider.chat.assert_not_called()

tests/test_message_visibility.py

@@ -1,153 +0,0 @@
-"""Tests for message visibility signing (hidden intermediate messages)."""
-
-import json
-from pathlib import Path
-
-from nanobot.agent.context import ContextBuilder
-from nanobot.agent.visibility import compute_signature, sign_content
-from nanobot.session.manager import Session
-
-
-class TestComputeSignature:
-    """Tests for compute_signature()."""
-
-    def test_returns_8_char_hex(self):
-        sig = compute_signature("hello")
-        assert len(sig) == 8
-        assert all(c in "0123456789abcdef" for c in sig)
-
-    def test_deterministic(self):
-        assert compute_signature("hello") == compute_signature("hello")
-
-    def test_different_content_different_sig(self):
-        assert compute_signature("hello") != compute_signature("world")
-
-    def test_sign_content_uses_compute_signature(self):
-        """sign_content should produce [HIDDEN:{compute_signature(content)}] prefix."""
-        content = "test message"
-        sig = compute_signature(content)
-        assert sign_content(content) == f"[HIDDEN:{sig}] {content}"
-
-
-class TestAddAssistantMessage:
-    """Tests for _hidden_sig in add_assistant_message()."""
-
-    def setup_method(self):
-        self.ctx = ContextBuilder(Path("/tmp"))
-
-    def test_intermediate_message_gets_hidden_sig(self):
-        msgs: list = []
-        tool_calls = [{"id": "tc1", "type": "function", "function": {"name": "test", "arguments": "{}"}}]
-        self.ctx.add_assistant_message(msgs, "thinking...", tool_calls)
-
-        assert msgs[0].get("_hidden_sig") is not None
-        assert msgs[0]["_hidden_sig"] == compute_signature("thinking...")
-
-    def test_final_message_no_hidden_sig(self):
-        msgs: list = []
-        self.ctx.add_assistant_message(msgs, "Here is the answer", None)
-
-        assert "_hidden_sig" not in msgs[0]
-
-    def test_empty_content_signed(self):
-        msgs: list = []
-        tool_calls = [{"id": "tc1", "type": "function", "function": {"name": "test", "arguments": "{}"}}]
-        self.ctx.add_assistant_message(msgs, None, tool_calls)
-
-        assert msgs[0]["_hidden_sig"] == compute_signature("")
-
-
-class TestAddToolResult:
-    """Tests for _hidden_sig in add_tool_result()."""
-
-    def setup_method(self):
-        self.ctx = ContextBuilder(Path("/tmp"))
-
-    def test_tool_result_gets_hidden_sig(self):
-        msgs: list = []
-        self.ctx.add_tool_result(msgs, "tc1", "read_file", "file contents here")
-
-        assert msgs[0]["_hidden_sig"] == compute_signature("file contents here")
-
-    def test_tool_result_non_string_content(self):
-        msgs: list = []
-        # Multipart content (e.g. image) is a list, not a string
-        self.ctx.add_tool_result(msgs, "tc1", "screenshot", [{"type": "text", "text": "ok"}])
-
-        assert msgs[0]["_hidden_sig"] == compute_signature("")
-
-
-class TestGetHistoryPrefix:
-    """Tests for get_history() applying [HIDDEN:sig] prefix."""
-
-    def test_hidden_sig_applied_at_read_time(self):
-        session = Session(key="test")
-        sig = compute_signature("thinking...")
-        session.messages = [
-            {"role": "assistant", "content": "thinking...", "tool_calls": [{}], "_hidden_sig": sig},
-        ]
-
-        history = session.get_history()
-        assert history[0]["content"] == f"[HIDDEN:{sig}] thinking..."
-        assert "_hidden_sig" not in history[0]
-
-    def test_no_prefix_without_hidden_sig(self):
-        session = Session(key="test")
-        session.messages = [
-            {"role": "assistant", "content": "Here is the answer"},
-        ]
-
-        history = session.get_history()
-        assert history[0]["content"] == "Here is the answer"
-
-    def test_tool_result_gets_prefix(self):
-        session = Session(key="test")
-        sig = compute_signature("file contents")
-        session.messages = [
-            {"role": "tool", "tool_call_id": "tc1", "name": "read", "content": "file contents", "_hidden_sig": sig},
-        ]
-
-        history = session.get_history()
-        assert history[0]["content"] == f"[HIDDEN:{sig}] file contents"
-
-    def test_roundtrip_jsonl(self, tmp_path):
-        """Write to session JSONL, reload, verify get_history() produces correct prefix."""
-        from nanobot.session.manager import SessionManager
-
-        workspace = tmp_path / "workspace"
-        workspace.mkdir()
-        mgr = SessionManager(workspace)
-
-        session = mgr.get_or_create("test:roundtrip")
-        sig = compute_signature("intermediate")
-        session.add_raw_message({
-            "role": "assistant",
-            "content": "intermediate",
-            "tool_calls": [{"id": "tc1", "type": "function", "function": {"name": "x", "arguments": "{}"}}],
-            "_hidden_sig": sig,
-        })
-        session.add_raw_message({
-            "role": "assistant",
-            "content": "final answer",
-        })
-        mgr.save(session)
-
-        # Reload from disk
-        mgr.invalidate("test:roundtrip")
-        reloaded = mgr.get_or_create("test:roundtrip")
-        history = reloaded.get_history()
-
-        assert history[0]["content"] == f"[HIDDEN:{sig}] intermediate"
-        assert history[1]["content"] == "final answer"
-
-    def test_idempotent_across_calls(self):
-        """Same prefix produced every call (cache stability)."""
-        session = Session(key="test")
-        sig = compute_signature("msg")
-        session.messages = [
-            {"role": "assistant", "content": "msg", "_hidden_sig": sig},
-        ]
-
-        h1 = session.get_history()
-        h2 = session.get_history()
-        assert h1[0]["content"] == h2[0]["content"]

tests/test_native_tools_registration.py

@@ -43,12 +43,15 @@ def test_native_tools_registered(mock_provider, mock_bus, tmp_path):
 
     # Verify native tools are registered (using their internal names)
     assert "bash" in tool_names, "bash tool should be registered"
-    assert "str_replace_based_edit_tool" in tool_names, "str_replace_based_edit_tool tool should be registered"
-    # Note: computer tool is intentionally disabled by default (requires VNC setup)
+    assert "str_replace_editor" in tool_names, "str_replace_editor tool should be registered"
+    assert "computer" in tool_names, "computer tool should be registered"
 
     # Verify we can get the tool instances
     bash_tool = loop.tools.get("bash")
     assert isinstance(bash_tool, BashTool20250124)
 
-    editor_tool = loop.tools.get("str_replace_based_edit_tool")
+    editor_tool = loop.tools.get("str_replace_editor")
     assert isinstance(editor_tool, EditTool20250728)
+
+    computer_tool = loop.tools.get("computer")
+    assert isinstance(computer_tool, ComputerTool20251124)

tests/test_oauth_identity_block.py

@@ -1,102 +0,0 @@
-"""Test that the Anthropic OAuth identity block is always included in API requests."""
-
-import pytest
-from unittest.mock import AsyncMock, patch, MagicMock
-
-import httpx
-
-from nanobot.providers.anthropic_oauth import AnthropicOAuthProvider
-from nanobot.providers.oauth_utils import get_claude_code_system_prefix
-
-
-IDENTITY_TEXT = get_claude_code_system_prefix()
-
-
-@pytest.fixture
-def provider():
-    return AnthropicOAuthProvider(
-        oauth_token="sk-ant-oat01-test-token",
-        default_model="claude-opus-4-7",
-    )
-
-
-def _mock_response(status_code=200, json_data=None):
-    """Create a mock httpx.Response."""
-    resp = MagicMock(spec=httpx.Response)
-    resp.status_code = status_code
-    resp.headers = {}
-    resp.text = ""
-    resp.json.return_value = json_data or {
-        "content": [{"type": "text", "text": "ok"}],
-        "stop_reason": "end_turn",
-        "usage": {"input_tokens": 1, "output_tokens": 1},
-    }
-    return resp
-
-
-@pytest.mark.asyncio
-async def test_identity_block_present_with_system_prompt(provider):
-    """When a system prompt is provided, identity block is the first system block."""
-    mock_client = AsyncMock()
-    mock_client.post.return_value = _mock_response()
-    provider._client = mock_client
-
-    await provider._make_request(
-        messages=[{"role": "user", "content": "hello"}],
-        system="You are a helpful assistant.",
-    )
-
-    call_kwargs = mock_client.post.call_args
-    payload = call_kwargs.kwargs["json"] if "json" in call_kwargs.kwargs else call_kwargs[1]["json"]
-    system_blocks = payload["system"]
-
-    assert len(system_blocks) == 2
-    assert system_blocks[0]["type"] == "text"
-    assert system_blocks[0]["text"] == IDENTITY_TEXT
-    assert system_blocks[1]["text"] == "You are a helpful assistant."
-
-
-@pytest.mark.asyncio
-async def test_identity_block_present_without_system_prompt(provider):
-    """When no system prompt is provided, identity block is still included.
-
-    This is the critical fix: extract_facts and similar calls pass system=None,
-    but Anthropic requires the identity block for OAuth tokens.
-    """
-    mock_client = AsyncMock()
-    mock_client.post.return_value = _mock_response()
-    provider._client = mock_client
-
-    await provider._make_request(
-        messages=[{"role": "user", "content": "extract facts"}],
-        system=None,
-    )
-
-    call_kwargs = mock_client.post.call_args
-    payload = call_kwargs.kwargs["json"] if "json" in call_kwargs.kwargs else call_kwargs[1]["json"]
-    system_blocks = payload["system"]
-
-    assert len(system_blocks) == 1
-    assert system_blocks[0]["type"] == "text"
-    assert system_blocks[0]["text"] == IDENTITY_TEXT
-
-
-@pytest.mark.asyncio
-async def test_identity_block_present_with_empty_string_system(provider):
-    """Empty string system prompt should still include the identity block."""
-    mock_client = AsyncMock()
-    mock_client.post.return_value = _mock_response()
-    provider._client = mock_client
-
-    await provider._make_request(
-        messages=[{"role": "user", "content": "hello"}],
-        system="",
-    )
-
-    call_kwargs = mock_client.post.call_args
-    payload = call_kwargs.kwargs["json"] if "json" in call_kwargs.kwargs else call_kwargs[1]["json"]
-    system_blocks = payload["system"]
-
-    # Empty string is falsy, so should go through the else branch
-    assert len(system_blocks) == 1
-    assert system_blocks[0]["text"] == IDENTITY_TEXT

tests/test_oauth_utils.py

@@ -17,7 +17,7 @@ def test_get_auth_headers_oauth():
     assert "Authorization" in headers
     assert headers["Authorization"] == "Bearer sk-ant-oat01-xxx"
     assert "x-api-key" not in headers
-    assert headers["anthropic-beta"] == "claude-code-20250219,oauth-2025-04-20,context-management-2025-06-27"
+    assert headers["anthropic-beta"] == "claude-code-20250219,oauth-2025-04-20"
 
 
 def test_get_auth_headers_api_key():

tests/test_provider_factory.py

@@ -9,7 +9,7 @@ def test_create_provider_oauth_token():
     """OAuth tokens should create AnthropicOAuthProvider."""
     provider = create_provider(
         api_key="sk-ant-oat01-test-token",
-        model="anthropic/claude-opus-4-7"
+        model="anthropic/claude-opus-4-5"
     )
     assert isinstance(provider, AnthropicOAuthProvider)
 
@@ -18,7 +18,7 @@ def test_create_provider_regular_key():
     """Regular API keys should create LiteLLMProvider."""
     provider = create_provider(
         api_key="sk-ant-api03-regular-key",
-        model="anthropic/claude-opus-4-7"
+        model="anthropic/claude-opus-4-5"
     )
     assert isinstance(provider, LiteLLMProvider)
 
@@ -27,6 +27,6 @@ def test_create_provider_openrouter():
     """OpenRouter keys should create LiteLLMProvider."""
     provider = create_provider(
         api_key="sk-or-v1-xxx",
-        model="anthropic/claude-opus-4-7"
+        model="anthropic/claude-opus-4-5"
     )
     assert isinstance(provider, LiteLLMProvider)

tests/test_registry_native_execution.py

@@ -31,7 +31,7 @@ async def test_registry_executes_edit_tool():
     with tempfile.TemporaryDirectory() as tmpdir:
         test_file = str(Path(tmpdir) / "test.txt")
 
-        result = await registry.execute("str_replace_based_edit_tool", {
+        result = await registry.execute("str_replace_editor", {
             "command": "create",
             "path": test_file,
             "file_text": "Hello, world!"

tests/test_registry_oauth.py

@@ -5,14 +5,14 @@ from nanobot.providers.registry import should_use_oauth_provider
 
 def test_should_use_oauth_for_oat_token():
     """OAuth provider should be used for sk-ant-oat tokens."""
-    assert should_use_oauth_provider("sk-ant-oat01-xxx", "anthropic/claude-opus-4-7") is True
+    assert should_use_oauth_provider("sk-ant-oat01-xxx", "anthropic/claude-opus-4-5") is True
     assert should_use_oauth_provider("sk-ant-oat01-xxx", "claude-sonnet-4") is True
 
 
 def test_should_not_use_oauth_for_regular_key():
     """Regular API keys should not use OAuth provider."""
-    assert should_use_oauth_provider("sk-ant-api03-xxx", "claude-opus-4-7") is False
-    assert should_use_oauth_provider("sk-or-v1-xxx", "anthropic/claude-opus-4-7") is False
+    assert should_use_oauth_provider("sk-ant-api03-xxx", "claude-opus-4-5") is False
+    assert should_use_oauth_provider("sk-or-v1-xxx", "anthropic/claude-opus-4-5") is False
 
 
 def test_should_not_use_oauth_for_non_anthropic():

tests/test_session_audit_log.py

@@ -1,137 +0,0 @@
-"""Test SessionManager audit log functionality."""
-
-import json
-
-import pytest
-
-from nanobot.session.manager import Session, SessionManager
-
-
-@pytest.fixture
-def session_manager(tmp_path):
-    return SessionManager(workspace=tmp_path)
-
-
-@pytest.fixture
-def session():
-    s = Session(key="telegram:12345")
-    s.add_message("user", "Hello")
-    s.add_message("assistant", "Hi there!")
-    return s
-
-
-def test_save_creates_audit_file(session_manager, session):
-    """SessionManager.save() should create a monthly audit log file."""
-    session_manager.save(session)
-
-    audit_files = list(session_manager.sessions_dir.glob("*.audit.*.jsonl"))
-    assert len(audit_files) == 1
-    assert "telegram_12345.audit." in audit_files[0].name
-
-
-def test_audit_file_contains_save_marker(session_manager, session):
-    """Audit log should start with a save_marker line containing metadata."""
-    session_manager.save(session)
-
-    audit_files = list(session_manager.sessions_dir.glob("*.audit.*.jsonl"))
-    lines = audit_files[0].read_text().strip().split("\n")
-
-    marker = json.loads(lines[0])
-    assert marker["_type"] == "save_marker"
-    assert marker["message_count"] == 2
-    assert "timestamp" in marker
-
-
-def test_audit_file_contains_all_messages(session_manager, session):
-    """Audit log should contain all session messages after the save marker."""
-    session_manager.save(session)
-
-    audit_files = list(session_manager.sessions_dir.glob("*.audit.*.jsonl"))
-    lines = audit_files[0].read_text().strip().split("\n")
-
-    # Line 0 = save_marker, lines 1-2 = messages
-    assert len(lines) == 3
-    msg1 = json.loads(lines[1])
-    msg2 = json.loads(lines[2])
-    assert msg1["role"] == "user"
-    assert msg1["content"] == "Hello"
-    assert msg2["role"] == "assistant"
-    assert msg2["content"] == "Hi there!"
-
-
-def test_audit_file_is_append_only(session_manager, session):
-    """Multiple saves should append to the same audit file, not overwrite."""
-    session_manager.save(session)
-
-    # Add another message and save again
-    session.add_message("user", "How are you?")
-    session_manager.save(session)
-
-    audit_files = list(session_manager.sessions_dir.glob("*.audit.*.jsonl"))
-    assert len(audit_files) == 1  # Same file
-
-    lines = audit_files[0].read_text().strip().split("\n")
-
-    # First save: 1 marker + 2 messages = 3 lines
-    # Second save: 1 marker + 3 messages = 4 lines
-    # Total: 7 lines
-    assert len(lines) == 7
-
-    # Both save markers present
-    markers = [json.loads(l) for l in lines if json.loads(l).get("_type") == "save_marker"]
-    assert len(markers) == 2
-    assert markers[0]["message_count"] == 2
-    assert markers[1]["message_count"] == 3
-
-
-def test_audit_preserves_message_fields(session_manager):
-    """Audit log should preserve all message fields including reasoning_content."""
-    session = Session(key="test:preserve")
-    session.add_raw_message({
-        "role": "assistant",
-        "content": "thinking response",
-        "reasoning_content": [{"type": "thinking", "thinking": "deep thoughts"}],
-        "timestamp": "2026-03-22T12:00:00",
-    })
-
-    session_manager.save(session)
-
-    audit_files = list(session_manager.sessions_dir.glob("*.audit.*.jsonl"))
-    lines = audit_files[0].read_text().strip().split("\n")
-
-    msg = json.loads(lines[1])
-    assert msg["reasoning_content"] == [{"type": "thinking", "thinking": "deep thoughts"}]
-
-
-def test_audit_failure_does_not_break_save(session_manager, session, tmp_path):
-    """If audit logging fails, the main session save should still succeed.
-
-    _append_audit has its own try/except, so internal failures are caught.
-    We simulate a realistic failure by making the sessions dir read-only
-    for audit file creation.
-    """
-    # First save works (creates both session file and audit file)
-    session_manager.save(session)
-
-    path = session_manager._get_session_path(session.key)
-    assert path.exists()
-
-    # Remove audit files and make a blocking file at the audit path
-    # so the next audit open("a") fails
-    for af in session_manager.sessions_dir.glob("*.audit.*.jsonl"):
-        af.unlink()
-
-    # Create a directory where the audit file should be — open() will fail
-    from datetime import datetime
-    now = datetime.now()
-    bad_path = session_manager.sessions_dir / f"telegram_12345.audit.{now:%Y-%m}.jsonl"
-    bad_path.mkdir()
-
-    # Second save should succeed despite audit failure
-    session.add_message("user", "another message")
-    session_manager.save(session)
-
-    # Session file should still be written correctly
-    with open(path) as f:
-        first_line = json.loads(f.readline())
-        assert first_line["_type"] == "metadata"

tests/test_subagent_wait.py

@@ -1,139 +0,0 @@
-# tests/test_subagent_wait.py
-"""Tests for wait_for_subagents with top-level and child subagents."""
-import pytest
-from pathlib import Path
-from unittest.mock import AsyncMock, MagicMock
-from nanobot.agent.subagent import SubagentManager
-from nanobot.bus.queue import MessageBus
-from nanobot.providers.base import LLMProvider, LLMResponse
-
-
-@pytest.mark.asyncio
-async def test_wait_for_top_level_subagent():
-    """Test that wait_for works for top-level subagents spawned from telegram."""
-    bus = MessageBus()
-    provider = MagicMock(spec=LLMProvider)
-    provider.chat = AsyncMock(return_value=LLMResponse(
-        content="Task completed",
-        tool_calls=[]
-    ))
-    provider.get_default_model = MagicMock(return_value="test-model")
-    provider.thinking_budget = 0
-
-    workspace = Path("/tmp/test-subagent")
-    workspace.mkdir(exist_ok=True)
-
-    manager = SubagentManager(
-        bus=bus,
-        provider=provider,
-        workspace=workspace
-    )
-
-    # Spawn a top-level subagent (origin channel = "telegram")
-    task_id = await manager.spawn(
-        task="Test task",
-        label="test",
-        model=None,
-        origin_channel="telegram",
-        origin_chat_id="12345"
-    )
-
-    # Wait for it to complete
-    result = await manager.wait_for([task_id])
-
-    # Should find the result (not "No result found")
-    assert "No result found" not in result
-    assert task_id in result
-    assert "Task completed" in result or "completed" in result.lower()
-
-
-@pytest.mark.asyncio
-async def test_wait_for_child_subagent():
-    """Test that wait_for works for child subagents (orchestrator pattern)."""
-    bus = MessageBus()
-    provider = MagicMock(spec=LLMProvider)
-    provider.chat = AsyncMock(return_value=LLMResponse(
-        content="Child task completed",
-        tool_calls=[]
-    ))
-    provider.get_default_model = MagicMock(return_value="test-model")
-    provider.thinking_budget = 0
-
-    workspace = Path("/tmp/test-subagent")
-    workspace.mkdir(exist_ok=True)
-
-    manager = SubagentManager(
-        bus=bus,
-        provider=provider,
-        workspace=workspace
-    )
-
-    # Spawn a child subagent (origin channel = "subagent")
-    task_id = await manager.spawn(
-        task="Child test task",
-        label="test-child",
-        model=None,
-        origin_channel="subagent",
-        origin_chat_id="parent-id"
-    )
-
-    # Wait for it to complete
-    result = await manager.wait_for([task_id])
-
-    # Should find the result (not "No result found")
-    assert "No result found" not in result
-    assert task_id in result
-    assert "Child task completed" in result or "completed" in result.lower()
-
-
-@pytest.mark.asyncio
-async def test_wait_for_multiple_subagents():
-    """Test waiting for multiple subagents of different types."""
-    bus = MessageBus()
-    call_count = 0
-
-    async def chat_response(*args, **kwargs):
-        nonlocal call_count
-        call_count += 1
-        return LLMResponse(content=f"Task {call_count} completed", tool_calls=[])
-
-    provider = MagicMock(spec=LLMProvider)
-    provider.chat = AsyncMock(side_effect=chat_response)
-    provider.get_default_model = MagicMock(return_value="test-model")
-    provider.thinking_budget = 0
-
-    workspace = Path("/tmp/test-subagent")
-    workspace.mkdir(exist_ok=True)
-
-    manager = SubagentManager(
-        bus=bus,
-        provider=provider,
-        workspace=workspace
-    )
-
-    # Spawn one top-level and one child subagent
-    task_id_1 = await manager.spawn(
-        task="Top-level task",
-        label="test-top",
-        model=None,
-        origin_channel="telegram",
-        origin_chat_id="12345"
-    )
-
-    task_id_2 = await manager.spawn(
-        task="Child task",
-        label="test-child",
-        model=None,
-        origin_channel="subagent",
-        origin_chat_id="parent"
-    )
-
-    # Wait for both
-    result = await manager.wait_for([task_id_1, task_id_2])
-
-    # Should find both results
-    assert "No result found" not in result
-    assert task_id_1 in result
-    assert task_id_2 in result
-    assert "Task 1 completed" in result or "completed" in result.lower()
-    assert "Task 2 completed" in result or "completed" in result.lower()

tests/test_task_cancel.py

@@ -0,0 +1,167 @@
+"""Tests for /stop task cancellation."""
+
+from __future__ import annotations
+
+import asyncio
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+
+def _make_loop():
+    """Create a minimal AgentLoop with mocked dependencies."""
+    from nanobot.agent.loop import AgentLoop
+    from nanobot.bus.queue import MessageBus
+
+    bus = MessageBus()
+    provider = MagicMock()
+    provider.get_default_model.return_value = "test-model"
+    workspace = MagicMock()
+    workspace.__truediv__ = MagicMock(return_value=MagicMock())
+
+    with patch("nanobot.agent.loop.ContextBuilder"), \
+         patch("nanobot.agent.loop.SessionManager"), \
+         patch("nanobot.agent.loop.SubagentManager") as MockSubMgr:
+        MockSubMgr.return_value.cancel_by_session = AsyncMock(return_value=0)
+        loop = AgentLoop(bus=bus, provider=provider, workspace=workspace)
+    return loop, bus
+
+
+class TestHandleStop:
+    @pytest.mark.asyncio
+    async def test_stop_no_active_task(self):
+        from nanobot.bus.events import InboundMessage
+
+        loop, bus = _make_loop()
+        msg = InboundMessage(channel="test", sender_id="u1", chat_id="c1", content="/stop")
+        await loop._handle_stop(msg)
+        out = await asyncio.wait_for(bus.consume_outbound(), timeout=1.0)
+        assert "No active task" in out.content
+
+    @pytest.mark.asyncio
+    async def test_stop_cancels_active_task(self):
+        from nanobot.bus.events import InboundMessage
+
+        loop, bus = _make_loop()
+        cancelled = asyncio.Event()
+
+        async def slow_task():
+            try:
+                await asyncio.sleep(60)
+            except asyncio.CancelledError:
+                cancelled.set()
+                raise
+
+        task = asyncio.create_task(slow_task())
+        await asyncio.sleep(0)
+        loop._active_tasks["test:c1"] = [task]
+
+        msg = InboundMessage(channel="test", sender_id="u1", chat_id="c1", content="/stop")
+        await loop._handle_stop(msg)
+
+        assert cancelled.is_set()
+        out = await asyncio.wait_for(bus.consume_outbound(), timeout=1.0)
+        assert "stopped" in out.content.lower()
+
+    @pytest.mark.asyncio
+    async def test_stop_cancels_multiple_tasks(self):
+        from nanobot.bus.events import InboundMessage
+
+        loop, bus = _make_loop()
+        events = [asyncio.Event(), asyncio.Event()]
+
+        async def slow(idx):
+            try:
+                await asyncio.sleep(60)
+            except asyncio.CancelledError:
+                events[idx].set()
+                raise
+
+        tasks = [asyncio.create_task(slow(i)) for i in range(2)]
+        await asyncio.sleep(0)
+        loop._active_tasks["test:c1"] = tasks
+
+        msg = InboundMessage(channel="test", sender_id="u1", chat_id="c1", content="/stop")
+        await loop._handle_stop(msg)
+
+        assert all(e.is_set() for e in events)
+        out = await asyncio.wait_for(bus.consume_outbound(), timeout=1.0)
+        assert "2 task" in out.content
+
+
+class TestDispatch:
+    @pytest.mark.asyncio
+    async def test_dispatch_processes_and_publishes(self):
+        from nanobot.bus.events import InboundMessage, OutboundMessage
+
+        loop, bus = _make_loop()
+        msg = InboundMessage(channel="test", sender_id="u1", chat_id="c1", content="hello")
+        loop._process_message = AsyncMock(
+            return_value=OutboundMessage(channel="test", chat_id="c1", content="hi")
+        )
+        await loop._dispatch(msg)
+        out = await asyncio.wait_for(bus.consume_outbound(), timeout=1.0)
+        assert out.content == "hi"
+
+    @pytest.mark.asyncio
+    async def test_processing_lock_serializes(self):
+        from nanobot.bus.events import InboundMessage, OutboundMessage
+
+        loop, bus = _make_loop()
+        order = []
+
+        async def mock_process(m, **kwargs):
+            order.append(f"start-{m.content}")
+            await asyncio.sleep(0.05)
+            order.append(f"end-{m.content}")
+            return OutboundMessage(channel="test", chat_id="c1", content=m.content)
+
+        loop._process_message = mock_process
+        msg1 = InboundMessage(channel="test", sender_id="u1", chat_id="c1", content="a")
+        msg2 = InboundMessage(channel="test", sender_id="u1", chat_id="c1", content="b")
+
+        t1 = asyncio.create_task(loop._dispatch(msg1))
+        t2 = asyncio.create_task(loop._dispatch(msg2))
+        await asyncio.gather(t1, t2)
+        assert order == ["start-a", "end-a", "start-b", "end-b"]
+
+
+class TestSubagentCancellation:
+    @pytest.mark.asyncio
+    async def test_cancel_by_session(self):
+        from nanobot.agent.subagent import SubagentManager
+        from nanobot.bus.queue import MessageBus
+
+        bus = MessageBus()
+        provider = MagicMock()
+        provider.get_default_model.return_value = "test-model"
+        mgr = SubagentManager(provider=provider, workspace=MagicMock(), bus=bus)
+
+        cancelled = asyncio.Event()
+
+        async def slow():
+            try:
+                await asyncio.sleep(60)
+            except asyncio.CancelledError:
+                cancelled.set()
+                raise
+
+        task = asyncio.create_task(slow())
+        await asyncio.sleep(0)
+        mgr._running_tasks["sub-1"] = task
+        mgr._session_tasks["test:c1"] = {"sub-1"}
+
+        count = await mgr.cancel_by_session("test:c1")
+        assert count == 1
+        assert cancelled.is_set()
+
+    @pytest.mark.asyncio
+    async def test_cancel_by_session_no_tasks(self):
+        from nanobot.agent.subagent import SubagentManager
+        from nanobot.bus.queue import MessageBus
+
+        bus = MessageBus()
+        provider = MagicMock()
+        provider.get_default_model.return_value = "test-model"
+        mgr = SubagentManager(provider=provider, workspace=MagicMock(), bus=bus)
+        assert await mgr.cancel_by_session("nonexistent") == 0