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

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

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

nanobot/agent/loop.py

@@ -40,8 +40,8 @@ class AgentLoop:
     5. Sends responses back
     """
 
-    # Server-side context management: Anthropic preserves all thinking blocks
-    # 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},
             },
         ]
@@ -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):
@@ -545,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}")
@@ -574,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,
@@ -794,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,
@@ -828,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.
@@ -908,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)

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 []
@@ -178,19 +188,22 @@ class Mem0MemoryStore:
                 max_tokens=2000,
                 temperature=0.3,
             )
-            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 []
@@ -201,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
 
@@ -293,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.
         """
@@ -313,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:
@@ -386,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/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/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/anthropic_oauth.py

@@ -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,

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."""
@@ -70,6 +73,7 @@ class Session:
     def clear(self) -> None:
         """Clear all messages and reset session to initial state."""
         self.messages = []
+        self.last_consolidated = 0
         self.updated_at = datetime.now()
 
 
@@ -135,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:
@@ -147,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)
 
@@ -156,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)
@@ -172,6 +178,7 @@ 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:

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"

tests/test_anthropic_oauth.py

@@ -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_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_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_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_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_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_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