nanobot/agent/loop.py

@@ -1,40 +1,33 @@
 """Agent loop: the core processing engine."""
 
-from __future__ import annotations
-
 import asyncio
 import json
-import re
+import time
-from contextlib import AsyncExitStack
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Awaitable, Callable
+from typing import Any
 
 from loguru import logger
 
-from nanobot.agent.context import ContextBuilder
-from nanobot.agent.memory import MemoryStore
-from nanobot.agent.subagent import SubagentManager
-from nanobot.agent.tools.cron import CronTool
-from nanobot.agent.tools.filesystem import EditFileTool, ListDirTool, ReadFileTool, WriteFileTool
-from nanobot.agent.tools.message import MessageTool
-from nanobot.agent.tools.registry import ToolRegistry
-from nanobot.agent.tools.shell import ExecTool
-from nanobot.agent.tools.spawn import SpawnTool
-from nanobot.agent.tools.web import WebFetchTool, WebSearchTool
 from nanobot.bus.events import InboundMessage, OutboundMessage
 from nanobot.bus.queue import MessageBus
 from nanobot.providers.base import LLMProvider
-from nanobot.session.manager import Session, SessionManager
+from nanobot.agent.context import ContextBuilder
-
+from nanobot.agent.tools.registry import ToolRegistry
-if TYPE_CHECKING:
+from nanobot.agent.tools.filesystem import ReadFileTool, WriteFileTool, EditFileTool, ListDirTool
-    from nanobot.config.schema import ChannelsConfig, ExecToolConfig
+from nanobot.agent.tools.shell import ExecTool
-    from nanobot.cron.service import CronService
+from nanobot.agent.tools.web import WebSearchTool, WebFetchTool
+from nanobot.agent.tools.message import MessageTool
+from nanobot.agent.tools.spawn import SpawnTool
+from nanobot.agent.tools.cron import CronTool
+from nanobot.agent.memory import MemoryStore
+from nanobot.agent.subagent import SubagentManager
+from nanobot.session.manager import SessionManager
 
 
 class AgentLoop:
     """
     The agent loop is the core processing engine.
-
+    
     It:
     1. Receives messages from the bus
     2. Builds context with history, memory, skills
@@ -42,42 +35,34 @@ class AgentLoop:
     4. Executes tool calls
     5. Sends responses back
     """
-
+    
-    _TOOL_RESULT_MAX_CHARS = 500
-
     def __init__(
         self,
         bus: MessageBus,
         provider: LLMProvider,
         workspace: Path,
         model: str | None = None,
-        max_iterations: int = 40,
+        max_iterations: int = 20,
-        temperature: float = 0.1,
+        memory_window: int = 50,
-        max_tokens: int = 4096,
-        memory_window: int = 100,
         brave_api_key: str | None = None,
-        exec_config: ExecToolConfig | None = None,
+        exec_config: "ExecToolConfig | None" = None,
-        cron_service: CronService | None = None,
+        cron_service: "CronService | None" = None,
         restrict_to_workspace: bool = False,
         session_manager: SessionManager | None = None,
-        mcp_servers: dict | None = None,
-        channels_config: ChannelsConfig | None = None,
     ):
         from nanobot.config.schema import ExecToolConfig
+        from nanobot.cron.service import CronService
         self.bus = bus
-        self.channels_config = channels_config
         self.provider = provider
         self.workspace = workspace
         self.model = model or provider.get_default_model()
         self.max_iterations = max_iterations
-        self.temperature = temperature
-        self.max_tokens = max_tokens
         self.memory_window = memory_window
         self.brave_api_key = brave_api_key
         self.exec_config = exec_config or ExecToolConfig()
         self.cron_service = cron_service
         self.restrict_to_workspace = restrict_to_workspace
-
+        
         self.context = ContextBuilder(workspace)
         self.sessions = session_manager or SessionManager(workspace)
         self.tools = ToolRegistry()
@@ -86,96 +71,238 @@ class AgentLoop:
             workspace=workspace,
             bus=bus,
             model=self.model,
-            temperature=self.temperature,
-            max_tokens=self.max_tokens,
             brave_api_key=brave_api_key,
             exec_config=self.exec_config,
             restrict_to_workspace=restrict_to_workspace,
         )
 
         self._running = False
-        self._mcp_servers = mcp_servers or {}
+        self._quota_cache: dict[str, Any] = {}  # {model: str, cached_at: float}
-        self._mcp_stack: AsyncExitStack | None = None
+        self._quota_cache_ttl: float = 300.0  # 5 minutes
-        self._mcp_connected = False
-        self._mcp_connecting = False
-        self._consolidating: set[str] = set()  # Session keys with consolidation in progress
-        self._consolidation_tasks: set[asyncio.Task] = set()  # Strong refs to in-flight tasks
-        self._consolidation_locks: dict[str, asyncio.Lock] = {}
-        self._active_tasks: dict[str, list[asyncio.Task]] = {}  # session_key -> tasks
-        self._processing_lock = asyncio.Lock()
         self._register_default_tools()
-
+    
     def _register_default_tools(self) -> None:
         """Register the default set of tools."""
+        # File tools (restrict to workspace if configured)
         allowed_dir = self.workspace if self.restrict_to_workspace else None
-        for cls in (ReadFileTool, WriteFileTool, EditFileTool, ListDirTool):
+        self.tools.register(ReadFileTool(allowed_dir=allowed_dir))
-            self.tools.register(cls(workspace=self.workspace, allowed_dir=allowed_dir))
+        self.tools.register(WriteFileTool(allowed_dir=allowed_dir))
+        self.tools.register(EditFileTool(allowed_dir=allowed_dir))
+        self.tools.register(ListDirTool(allowed_dir=allowed_dir))
+        
+        # Shell tool
         self.tools.register(ExecTool(
             working_dir=str(self.workspace),
             timeout=self.exec_config.timeout,
             restrict_to_workspace=self.restrict_to_workspace,
-            path_append=self.exec_config.path_append,
         ))
+        
+        # Web tools
         self.tools.register(WebSearchTool(api_key=self.brave_api_key))
         self.tools.register(WebFetchTool())
-        self.tools.register(MessageTool(send_callback=self.bus.publish_outbound))
+        
-        self.tools.register(SpawnTool(manager=self.subagents))
+        # Message tool
+        message_tool = MessageTool(send_callback=self.bus.publish_outbound)
+        self.tools.register(message_tool)
+        
+        # Spawn tool (for subagents)
+        spawn_tool = SpawnTool(manager=self.subagents)
+        self.tools.register(spawn_tool)
+        
+        # Cron tool (for scheduling)
         if self.cron_service:
             self.tools.register(CronTool(self.cron_service))
-
+    
-    async def _connect_mcp(self) -> None:
+    async def run(self) -> None:
-        """Connect to configured MCP servers (one-time, lazy)."""
+        """Run the agent loop, processing messages from the bus."""
-        if self._mcp_connected or self._mcp_connecting or not self._mcp_servers:
+        self._running = True
-            return
+        logger.info("Agent loop started")
-        self._mcp_connecting = True
+        
-        from nanobot.agent.tools.mcp import connect_mcp_servers
+        while self._running:
-        try:
+            try:
-            self._mcp_stack = AsyncExitStack()
+                # Wait for next message
-            await self._mcp_stack.__aenter__()
+                msg = await asyncio.wait_for(
-            await connect_mcp_servers(self._mcp_servers, self.tools, self._mcp_stack)
+                    self.bus.consume_inbound(),
-            self._mcp_connected = True
+                    timeout=1.0
-        except Exception as e:
+                )
-            logger.error("Failed to connect MCP servers (will retry next message): {}", e)
+                
-            if self._mcp_stack:
+                # Process it
                 try:
-                    await self._mcp_stack.aclose()
+                    response = await self._process_message(msg)
-                except Exception:
+                    if response:
-                    pass
+                        await self.bus.publish_outbound(response)
-                self._mcp_stack = None
+                except Exception as e:
-        finally:
+                    logger.error(f"Error processing message: {e}")
-            self._mcp_connecting = False
+                    # Send error response
+                    await self.bus.publish_outbound(OutboundMessage(
+                        channel=msg.channel,
+                        chat_id=msg.chat_id,
+                        content=f"Sorry, I encountered an error: {str(e)}"
+                    ))
+            except asyncio.TimeoutError:
+                continue
+    
+    def stop(self) -> None:
+        """Stop the agent loop."""
+        self._running = False
+        logger.info("Agent loop stopping")
 
-    def _set_tool_context(self, channel: str, chat_id: str, message_id: str | None = None) -> None:
+    def _select_model_based_on_quota(self) -> str:
-        """Update context for all tools that need routing info."""
+        """Select Opus or Sonnet based on rolling weekly quota burn rate."""
-        for name in ("message", "spawn", "cron"):
+        # Check cache
-            if tool := self.tools.get(name):
+        now = time.time()
-                if hasattr(tool, "set_context"):
+        if self._quota_cache and (now - self._quota_cache.get("cached_at", 0)) < self._quota_cache_ttl:
-                    tool.set_context(channel, chat_id, *([message_id] if name == "message" else []))
+            return self._quota_cache["model"]
 
-    @staticmethod
+        # Default models
-    def _strip_think(text: str | None) -> str | None:
+        OPUS = "claude-opus-4-6"
-        """Remove <think>…</think> blocks that some models embed in content."""
+        SONNET = "claude-sonnet-4-5"
-        if not text:
+        TOLERANCE = 1.17  # 17% overage triggers downgrade
-            return None
-        return re.sub(r"<think>[\s\S]*?</think>", "", text).strip() or None
 
-    @staticmethod
+        # Read rate limits
-    def _tool_hint(tool_calls: list) -> str:
+        rate_limits_path = self.workspace / "memory" / "rate_limits.json"
-        """Format tool calls as concise hint, e.g. 'web_search("query")'."""
+        if not rate_limits_path.exists():
-        def _fmt(tc):
+            logger.warning("rate_limits.json not found, defaulting to Sonnet")
-            val = next(iter(tc.arguments.values()), None) if tc.arguments else None
+            return SONNET
-            if not isinstance(val, str):
-                return tc.name
-            return f'{tc.name}("{val[:40]}…")' if len(val) > 40 else f'{tc.name}("{val}")'
-        return ", ".join(_fmt(tc) for tc in tool_calls)
 
-    async def _run_agent_loop(
+        try:
-        self,
+            with open(rate_limits_path) as f:
-        initial_messages: list[dict],
+                limits = json.load(f)
-        on_progress: Callable[..., Awaitable[None]] | None = None,
+
-    ) -> tuple[str | None, list[str], list[dict]]:
+            actual_usage = limits.get("weekly_all_models")
-        """Run the agent iteration loop. Returns (final_content, tools_used, messages)."""
+            weekly_reset = limits.get("weekly_reset")
-        messages = initial_messages
+
+            if actual_usage is None or weekly_reset is None:
+                logger.warning("Rate limit data incomplete, defaulting to Sonnet")
+                return SONNET
+
+            # Calculate expected usage
+            actual_pct = actual_usage * 100
+            week_start = weekly_reset - (168 * 3600)
+            hours_elapsed = max(0, min((now - week_start) / 3600, 168))
+            expected_pct = (hours_elapsed / 168) * 100
+            threshold = expected_pct * TOLERANCE
+
+            # Decision logic
+            if actual_pct > threshold:
+                model = SONNET
+                logger.info(
+                    f"Quota: {actual_pct:.1f}% used, expected {expected_pct:.1f}%, "
+                    f"threshold {threshold:.1f}% → Sonnet"
+                )
+            else:
+                model = OPUS
+                logger.info(
+                    f"Quota: {actual_pct:.1f}% used, expected {expected_pct:.1f}%, "
+                    f"threshold {threshold:.1f}% → Opus"
+                )
+
+            # Cache decision
+            self._quota_cache = {"model": model, "cached_at": now}
+            return model
+
+        except Exception as e:
+            logger.error(f"Error checking quota: {e}, defaulting to Sonnet")
+            return SONNET
+
+    def _get_quota_status(self) -> str:
+        """Return human-readable quota status."""
+        rate_limits_path = self.workspace / "memory" / "rate_limits.json"
+        if not rate_limits_path.exists():
+            return "⚠️ No quota data available yet."
+
+        try:
+            with open(rate_limits_path) as f:
+                limits = json.load(f)
+
+            actual_pct = limits.get("weekly_all_models", 0) * 100
+            reset_ts = limits.get("weekly_reset", 0)
+            now = time.time()
+
+            hours_until_reset = (reset_ts - now) / 3600
+            week_start = reset_ts - (168 * 3600)
+            hours_elapsed = max(0, (now - week_start) / 3600)
+            expected_pct = (hours_elapsed / 168) * 100
+
+            model = self._select_model_based_on_quota()
+
+            return f"""📊 Quota Status:
+• Used: {actual_pct:.1f}% (expected {expected_pct:.1f}%)
+• Resets in: {hours_until_reset:.1f}h
+• Current model: {model}
+• Burn rate: {actual_pct / max(expected_pct, 0.01):.2f}x target"""
+
+        except Exception as e:
+            return f"⚠️ Error reading quota: {e}"
+
+    async def _process_message(self, msg: InboundMessage, session_key: str | None = None) -> OutboundMessage | None:
+        """
+        Process a single inbound message.
+        
+        Args:
+            msg: The inbound message to process.
+            session_key: Override session key (used by process_direct).
+        
+        Returns:
+            The response message, or None if no response needed.
+        """
+        # Handle system messages (subagent announces)
+        # The chat_id contains the original "channel:chat_id" to route back to
+        if msg.channel == "system":
+            return await self._process_system_message(msg)
+        
+        preview = msg.content[:80] + "..." if len(msg.content) > 80 else msg.content
+        logger.info(f"Processing message from {msg.channel}:{msg.sender_id}: {preview}")
+        
+        # Get or create session
+        key = session_key or msg.session_key
+        session = self.sessions.get_or_create(key)
+        
+        # Handle slash commands
+        cmd = msg.content.strip().lower()
+        if cmd == "/new":
+            await self._consolidate_memory(session, archive_all=True)
+            session.clear()
+            self.sessions.save(session)
+            return OutboundMessage(channel=msg.channel, chat_id=msg.chat_id,
+                                  content="🐈 New session started. Memory consolidated.")
+        if cmd == "/help":
+            return OutboundMessage(channel=msg.channel, chat_id=msg.chat_id,
+                                  content="🐈 nanobot commands:\n/new — Start a new conversation\n/help — Show available commands\n/quota — Show quota status")
+        if cmd == "/quota":
+            status = self._get_quota_status()
+            return OutboundMessage(channel=msg.channel, chat_id=msg.chat_id, content=status)
+
+        # Consolidate memory before processing if session is too large
+        if len(session.messages) > self.memory_window:
+            await self._consolidate_memory(session)
+        
+        # Update tool contexts
+        message_tool = self.tools.get("message")
+        if isinstance(message_tool, MessageTool):
+            message_tool.set_context(msg.channel, msg.chat_id)
+        
+        spawn_tool = self.tools.get("spawn")
+        if isinstance(spawn_tool, SpawnTool):
+            spawn_tool.set_context(msg.channel, msg.chat_id)
+        
+        cron_tool = self.tools.get("cron")
+        if isinstance(cron_tool, CronTool):
+            cron_tool.set_context(msg.channel, msg.chat_id)
+        
+        # Build initial messages (use get_history for LLM-formatted messages)
+        messages = self.context.build_messages(
+            history=session.get_history(),
+            current_message=msg.content,
+            media=msg.media if msg.media else None,
+            channel=msg.channel,
+            chat_id=msg.chat_id,
+        )
+
+        # Select model based on quota
+        selected_model = self._select_model_based_on_quota()
+
+        # Agent loop
         iteration = 0
         final_content = None
         tools_used: list[str] = []
@@ -183,28 +310,23 @@ class AgentLoop:
         while iteration < self.max_iterations:
             iteration += 1
 
+            # Call LLM
             response = await self.provider.chat(
                 messages=messages,
                 tools=self.tools.get_definitions(),
-                model=self.model,
+                model=selected_model
-                temperature=self.temperature,
-                max_tokens=self.max_tokens,
             )
-
+            
+            # Handle tool calls
             if response.has_tool_calls:
-                if on_progress:
+                # Add assistant message with tool calls
-                    clean = self._strip_think(response.content)
-                    if clean:
-                        await on_progress(clean)
-                    await on_progress(self._tool_hint(response.tool_calls), tool_hint=True)
-
                 tool_call_dicts = [
                     {
                         "id": tc.id,
                         "type": "function",
                         "function": {
                             "name": tc.name,
-                            "arguments": json.dumps(tc.arguments, ensure_ascii=False)
+                            "arguments": json.dumps(tc.arguments)  # Must be JSON string
                         }
                     }
                     for tc in response.tool_calls
@@ -213,260 +335,212 @@ class AgentLoop:
                     messages, response.content, tool_call_dicts,
                     reasoning_content=response.reasoning_content,
                 )
-
+                
+                # Execute tools
                 for tool_call in response.tool_calls:
                     tools_used.append(tool_call.name)
                     args_str = json.dumps(tool_call.arguments, ensure_ascii=False)
-                    logger.info("Tool call: {}({})", tool_call.name, args_str[:200])
+                    logger.info(f"Tool call: {tool_call.name}({args_str[:200]})")
                     result = await self.tools.execute(tool_call.name, tool_call.arguments)
                     messages = self.context.add_tool_result(
                         messages, tool_call.id, tool_call.name, result
                     )
+                # Interleaved CoT: reflect before next action
+                messages.append({"role": "user", "content": "Reflect on the results and decide next steps."})
             else:
-                clean = self._strip_think(response.content)
+                # No tool calls, we're done
-                messages = self.context.add_assistant_message(
+                final_content = response.content
-                    messages, clean, reasoning_content=response.reasoning_content,
-                )
-                final_content = clean
                 break
-
+        
-        if final_content is None and iteration >= self.max_iterations:
-            logger.warning("Max iterations ({}) reached", self.max_iterations)
-            final_content = (
-                f"I reached the maximum number of tool call iterations ({self.max_iterations}) "
-                "without completing the task. You can try breaking the task into smaller steps."
-            )
-
-        return final_content, tools_used, messages
-
-    async def run(self) -> None:
-        """Run the agent loop, dispatching messages as tasks to stay responsive to /stop."""
-        self._running = True
-        await self._connect_mcp()
-        logger.info("Agent loop started")
-
-        while self._running:
-            try:
-                msg = await asyncio.wait_for(self.bus.consume_inbound(), timeout=1.0)
-            except asyncio.TimeoutError:
-                continue
-
-            if msg.content.strip().lower() == "/stop":
-                await self._handle_stop(msg)
-            else:
-                task = asyncio.create_task(self._dispatch(msg))
-                self._active_tasks.setdefault(msg.session_key, []).append(task)
-                task.add_done_callback(lambda t, k=msg.session_key: self._active_tasks.get(k, []) and self._active_tasks[k].remove(t) if t in self._active_tasks.get(k, []) else None)
-
-    async def _handle_stop(self, msg: InboundMessage) -> None:
-        """Cancel all active tasks and subagents for the session."""
-        tasks = self._active_tasks.pop(msg.session_key, [])
-        cancelled = sum(1 for t in tasks if not t.done() and t.cancel())
-        for t in tasks:
-            try:
-                await t
-            except (asyncio.CancelledError, Exception):
-                pass
-        sub_cancelled = await self.subagents.cancel_by_session(msg.session_key)
-        total = cancelled + sub_cancelled
-        content = f"⏹ Stopped {total} task(s)." if total else "No active task to stop."
-        await self.bus.publish_outbound(OutboundMessage(
-            channel=msg.channel, chat_id=msg.chat_id, content=content,
-        ))
-
-    async def _dispatch(self, msg: InboundMessage) -> None:
-        """Process a message under the global lock."""
-        async with self._processing_lock:
-            try:
-                response = await self._process_message(msg)
-                if response is not None:
-                    await self.bus.publish_outbound(response)
-                elif msg.channel == "cli":
-                    await self.bus.publish_outbound(OutboundMessage(
-                        channel=msg.channel, chat_id=msg.chat_id,
-                        content="", metadata=msg.metadata or {},
-                    ))
-            except asyncio.CancelledError:
-                logger.info("Task cancelled for session {}", msg.session_key)
-                raise
-            except Exception:
-                logger.exception("Error processing message for session {}", msg.session_key)
-                await self.bus.publish_outbound(OutboundMessage(
-                    channel=msg.channel, chat_id=msg.chat_id,
-                    content="Sorry, I encountered an error.",
-                ))
-
-    async def close_mcp(self) -> None:
-        """Close MCP connections."""
-        if self._mcp_stack:
-            try:
-                await self._mcp_stack.aclose()
-            except (RuntimeError, BaseExceptionGroup):
-                pass  # MCP SDK cancel scope cleanup is noisy but harmless
-            self._mcp_stack = None
-
-    def stop(self) -> None:
-        """Stop the agent loop."""
-        self._running = False
-        logger.info("Agent loop stopping")
-
-    async def _process_message(
-        self,
-        msg: InboundMessage,
-        session_key: str | None = None,
-        on_progress: Callable[[str], Awaitable[None]] | None = None,
-    ) -> OutboundMessage | None:
-        """Process a single inbound message and return the response."""
-        # System messages: parse origin from chat_id ("channel:chat_id")
-        if msg.channel == "system":
-            channel, chat_id = (msg.chat_id.split(":", 1) if ":" in msg.chat_id
-                                else ("cli", msg.chat_id))
-            logger.info("Processing system message from {}", msg.sender_id)
-            key = f"{channel}:{chat_id}"
-            session = self.sessions.get_or_create(key)
-            self._set_tool_context(channel, chat_id, msg.metadata.get("message_id"))
-            history = session.get_history(max_messages=self.memory_window)
-            messages = self.context.build_messages(
-                history=history,
-                current_message=msg.content, channel=channel, chat_id=chat_id,
-            )
-            final_content, _, all_msgs = await self._run_agent_loop(messages)
-            self._save_turn(session, all_msgs, 1 + len(history))
-            self.sessions.save(session)
-            return OutboundMessage(channel=channel, chat_id=chat_id,
-                                  content=final_content or "Background task completed.")
-
-        preview = msg.content[:80] + "..." if len(msg.content) > 80 else msg.content
-        logger.info("Processing message from {}:{}: {}", msg.channel, msg.sender_id, preview)
-
-        key = session_key or msg.session_key
-        session = self.sessions.get_or_create(key)
-
-        # Slash commands
-        cmd = msg.content.strip().lower()
-        if cmd == "/new":
-            lock = self._consolidation_locks.setdefault(session.key, asyncio.Lock())
-            self._consolidating.add(session.key)
-            try:
-                async with lock:
-                    snapshot = session.messages[session.last_consolidated:]
-                    if snapshot:
-                        temp = Session(key=session.key)
-                        temp.messages = list(snapshot)
-                        if not await self._consolidate_memory(temp, archive_all=True):
-                            return OutboundMessage(
-                                channel=msg.channel, chat_id=msg.chat_id,
-                                content="Memory archival failed, session not cleared. Please try again.",
-                            )
-            except Exception:
-                logger.exception("/new archival failed for {}", session.key)
-                return OutboundMessage(
-                    channel=msg.channel, chat_id=msg.chat_id,
-                    content="Memory archival failed, session not cleared. Please try again.",
-                )
-            finally:
-                self._consolidating.discard(session.key)
-                if not lock.locked():
-                    self._consolidation_locks.pop(session.key, None)
-
-            session.clear()
-            self.sessions.save(session)
-            self.sessions.invalidate(session.key)
-            return OutboundMessage(channel=msg.channel, chat_id=msg.chat_id,
-                                  content="New session started.")
-        if cmd == "/help":
-            return OutboundMessage(channel=msg.channel, chat_id=msg.chat_id,
-                                  content="🐈 nanobot commands:\n/new — Start a new conversation\n/stop — Stop the current task\n/help — Show available commands")
-
-        unconsolidated = len(session.messages) - session.last_consolidated
-        if (unconsolidated >= self.memory_window and session.key not in self._consolidating):
-            self._consolidating.add(session.key)
-            lock = self._consolidation_locks.setdefault(session.key, asyncio.Lock())
-
-            async def _consolidate_and_unlock():
-                try:
-                    async with lock:
-                        await self._consolidate_memory(session)
-                finally:
-                    self._consolidating.discard(session.key)
-                    if not lock.locked():
-                        self._consolidation_locks.pop(session.key, None)
-                    _task = asyncio.current_task()
-                    if _task is not None:
-                        self._consolidation_tasks.discard(_task)
-
-            _task = asyncio.create_task(_consolidate_and_unlock())
-            self._consolidation_tasks.add(_task)
-
-        self._set_tool_context(msg.channel, msg.chat_id, msg.metadata.get("message_id"))
-        if message_tool := self.tools.get("message"):
-            if isinstance(message_tool, MessageTool):
-                message_tool.start_turn()
-
-        history = session.get_history(max_messages=self.memory_window)
-        initial_messages = self.context.build_messages(
-            history=history,
-            current_message=msg.content,
-            media=msg.media if msg.media else None,
-            channel=msg.channel, chat_id=msg.chat_id,
-        )
-
-        async def _bus_progress(content: str, *, tool_hint: bool = False) -> None:
-            meta = dict(msg.metadata or {})
-            meta["_progress"] = True
-            meta["_tool_hint"] = tool_hint
-            await self.bus.publish_outbound(OutboundMessage(
-                channel=msg.channel, chat_id=msg.chat_id, content=content, metadata=meta,
-            ))
-
-        final_content, _, all_msgs = await self._run_agent_loop(
-            initial_messages, on_progress=on_progress or _bus_progress,
-        )
-
         if final_content is None:
-            final_content = "I've completed processing but have no response to give."
+            if iteration >= self.max_iterations:
-
+                final_content = f"Reached {self.max_iterations} iterations without completion."
-        self._save_turn(session, all_msgs, 1 + len(history))
+            else:
-        self.sessions.save(session)
+                final_content = "I've completed processing but have no response to give."
-
+        
-        if (mt := self.tools.get("message")) and isinstance(mt, MessageTool) and mt._sent_in_turn:
+        # Log response preview
-            return None
-
         preview = final_content[:120] + "..." if len(final_content) > 120 else final_content
-        logger.info("Response to {}:{}: {}", msg.channel, msg.sender_id, preview)
+        logger.info(f"Response to {msg.channel}:{msg.sender_id}: {preview}")
+        
+        # Save to session (include tool names so consolidation sees what happened)
+        session.add_message("user", msg.content)
+        session.add_message("assistant", final_content,
+                            tools_used=tools_used if tools_used else None)
+        self.sessions.save(session)
+        
         return OutboundMessage(
-            channel=msg.channel, chat_id=msg.chat_id, content=final_content,
+            channel=msg.channel,
-            metadata=msg.metadata or {},
+            chat_id=msg.chat_id,
+            content=final_content,
+            metadata=msg.metadata or {},  # Pass through for channel-specific needs (e.g. Slack thread_ts)
         )
-
+    
-    def _save_turn(self, session: Session, messages: list[dict], skip: int) -> None:
+    async def _process_system_message(self, msg: InboundMessage) -> OutboundMessage | None:
-        """Save new-turn messages into session, truncating large tool results."""
+        """
-        from datetime import datetime
+        Process a system message (e.g., subagent announce).
-        for m in messages[skip:]:
+        
-            entry = {k: v for k, v in m.items() if k != "reasoning_content"}
+        The chat_id field contains "original_channel:original_chat_id" to route
-            role, content = entry.get("role"), entry.get("content")
+        the response back to the correct destination.
-            if role == "tool" and isinstance(content, str) and len(content) > self._TOOL_RESULT_MAX_CHARS:
+        """
-                entry["content"] = content[:self._TOOL_RESULT_MAX_CHARS] + "\n... (truncated)"
+        logger.info(f"Processing system message from {msg.sender_id}")
-            elif role == "user":
+        
-                if isinstance(content, str) and content.startswith(ContextBuilder._RUNTIME_CONTEXT_TAG):
+        # Parse origin from chat_id (format: "channel:chat_id")
-                    continue
+        if ":" in msg.chat_id:
-                if isinstance(content, list):
+            parts = msg.chat_id.split(":", 1)
-                    entry["content"] = [
+            origin_channel = parts[0]
-                        {"type": "text", "text": "[image]"} if (
+            origin_chat_id = parts[1]
-                            c.get("type") == "image_url"
+        else:
-                            and c.get("image_url", {}).get("url", "").startswith("data:image/")
+            # Fallback
-                        ) else c for c in content
+            origin_channel = "cli"
-                    ]
+            origin_chat_id = msg.chat_id
-            entry.setdefault("timestamp", datetime.now().isoformat())
+        
-            session.messages.append(entry)
+        # Use the origin session for context
-        session.updated_at = datetime.now()
+        session_key = f"{origin_channel}:{origin_chat_id}"
-
+        session = self.sessions.get_or_create(session_key)
-    async def _consolidate_memory(self, session, archive_all: bool = False) -> bool:
+        
-        """Delegate to MemoryStore.consolidate(). Returns True on success."""
+        # Update tool contexts
-        return await MemoryStore(self.workspace).consolidate(
+        message_tool = self.tools.get("message")
-            session, self.provider, self.model,
+        if isinstance(message_tool, MessageTool):
-            archive_all=archive_all, memory_window=self.memory_window,
+            message_tool.set_context(origin_channel, origin_chat_id)
+        
+        spawn_tool = self.tools.get("spawn")
+        if isinstance(spawn_tool, SpawnTool):
+            spawn_tool.set_context(origin_channel, origin_chat_id)
+        
+        cron_tool = self.tools.get("cron")
+        if isinstance(cron_tool, CronTool):
+            cron_tool.set_context(origin_channel, origin_chat_id)
+        
+        # Build messages with the announce content
+        messages = self.context.build_messages(
+            history=session.get_history(),
+            current_message=msg.content,
+            channel=origin_channel,
+            chat_id=origin_chat_id,
         )
+        
+        # Agent loop (limited for announce handling)
+        iteration = 0
+        final_content = None
+        
+        while iteration < self.max_iterations:
+            iteration += 1
+            
+            response = await self.provider.chat(
+                messages=messages,
+                tools=self.tools.get_definitions(),
+                model=self.model
+            )
+            
+            if response.has_tool_calls:
+                tool_call_dicts = [
+                    {
+                        "id": tc.id,
+                        "type": "function",
+                        "function": {
+                            "name": tc.name,
+                            "arguments": json.dumps(tc.arguments)
+                        }
+                    }
+                    for tc in response.tool_calls
+                ]
+                messages = self.context.add_assistant_message(
+                    messages, response.content, tool_call_dicts,
+                    reasoning_content=response.reasoning_content,
+                )
+                
+                for tool_call in response.tool_calls:
+                    args_str = json.dumps(tool_call.arguments, ensure_ascii=False)
+                    logger.info(f"Tool call: {tool_call.name}({args_str[:200]})")
+                    result = await self.tools.execute(tool_call.name, tool_call.arguments)
+                    messages = self.context.add_tool_result(
+                        messages, tool_call.id, tool_call.name, result
+                    )
+                # Interleaved CoT: reflect before next action
+                messages.append({"role": "user", "content": "Reflect on the results and decide next steps."})
+            else:
+                final_content = response.content
+                break
+        
+        if final_content is None:
+            final_content = "Background task completed."
+        
+        # Save to session (mark as system message in history)
+        session.add_message("user", f"[System: {msg.sender_id}] {msg.content}")
+        session.add_message("assistant", final_content)
+        self.sessions.save(session)
+        
+        return OutboundMessage(
+            channel=origin_channel,
+            chat_id=origin_chat_id,
+            content=final_content
+        )
+    
+    async def _consolidate_memory(self, session, archive_all: bool = False) -> None:
+        """Consolidate old messages into MEMORY.md + HISTORY.md, then trim session."""
+        if not session.messages:
+            return
+        memory = MemoryStore(self.workspace)
+        if archive_all:
+            old_messages = session.messages
+            keep_count = 0
+        else:
+            keep_count = min(10, max(2, self.memory_window // 2))
+            old_messages = session.messages[:-keep_count]
+        if not old_messages:
+            return
+        logger.info(f"Memory consolidation started: {len(session.messages)} messages, archiving {len(old_messages)}, keeping {keep_count}")
+
+        # Format messages for LLM (include tool names when available)
+        lines = []
+        for m in old_messages:
+            if not m.get("content"):
+                continue
+            tools = f" [tools: {', '.join(m['tools_used'])}]" if m.get("tools_used") else ""
+            lines.append(f"[{m.get('timestamp', '?')[:16]}] {m['role'].upper()}{tools}: {m['content']}")
+        conversation = "\n".join(lines)
+        current_memory = memory.read_long_term()
+
+        prompt = f"""You are a memory consolidation agent. Process this conversation and return a JSON object with exactly two keys:
+
+1. "history_entry": A paragraph (2-5 sentences) summarizing the key events/decisions/topics. Start with a timestamp like [YYYY-MM-DD HH:MM]. Include enough detail to be useful when found by grep search later.
+
+2. "memory_update": The updated long-term memory content. Add any new facts: user location, preferences, personal info, habits, project context, technical decisions, tools/services used. If nothing new, return the existing content unchanged.
+
+## Current Long-term Memory
+{current_memory or "(empty)"}
+
+## Conversation to Process
+{conversation}
+
+Respond with ONLY valid JSON, no markdown fences."""
+
+        try:
+            response = await self.provider.chat(
+                messages=[
+                    {"role": "system", "content": "You are a memory consolidation agent. Respond only with valid JSON."},
+                    {"role": "user", "content": prompt},
+                ],
+                model="claude-haiku-4-5",
+                thinking_budget=0,
+                max_tokens=16384,
+            )
+            text = (response.content or "").strip()
+            if text.startswith("```"):
+                text = text.split("\n", 1)[-1].rsplit("```", 1)[0].strip()
+            result = json.loads(text)
+
+            if entry := result.get("history_entry"):
+                memory.append_history(entry)
+            if update := result.get("memory_update"):
+                if update != current_memory:
+                    memory.write_long_term(update)
+
+            session.messages = session.messages[-keep_count:] if keep_count else []
+            self.sessions.save(session)
+            logger.info(f"Memory consolidation done, session trimmed to {len(session.messages)} messages")
+        except Exception as e:
+            logger.error(f"Memory consolidation failed: {e}")
 
     async def process_direct(
         self,
@@ -474,10 +548,25 @@ class AgentLoop:
         session_key: str = "cli:direct",
         channel: str = "cli",
         chat_id: str = "direct",
-        on_progress: Callable[[str], Awaitable[None]] | None = None,
     ) -> str:
-        """Process a message directly (for CLI or cron usage)."""
+        """
-        await self._connect_mcp()
+        Process a message directly (for CLI or cron usage).
-        msg = InboundMessage(channel=channel, sender_id="user", chat_id=chat_id, content=content)
+        
-        response = await self._process_message(msg, session_key=session_key, on_progress=on_progress)
+        Args:
+            content: The message content.
+            session_key: Session identifier (overrides channel:chat_id for session lookup).
+            channel: Source channel (for tool context routing).
+            chat_id: Source chat ID (for tool context routing).
+        
+        Returns:
+            The agent's response.
+        """
+        msg = InboundMessage(
+            channel=channel,
+            sender_id="user",
+            chat_id=chat_id,
+            content=content
+        )
+        
+        response = await self._process_message(msg, session_key=session_key)
         return response.content if response else ""

nanobot/channels/telegram.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 import asyncio
 import re
 from loguru import logger
-from telegram import BotCommand, Update, ReplyParameters
+from telegram import BotCommand, Update
 from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes
 from telegram.request import HTTPXRequest
 
@@ -78,26 +78,6 @@ def _markdown_to_telegram_html(text: str) -> str:
     return text
 
 
-def _split_message(content: str, max_len: int = 4000) -> list[str]:
-    """Split content into chunks within max_len, preferring line breaks."""
-    if len(content) <= max_len:
-        return [content]
-    chunks: list[str] = []
-    while content:
-        if len(content) <= max_len:
-            chunks.append(content)
-            break
-        cut = content[:max_len]
-        pos = cut.rfind('\n')
-        if pos == -1:
-            pos = cut.rfind(' ')
-        if pos == -1:
-            pos = max_len
-        chunks.append(content[:pos])
-        content = content[pos:].lstrip()
-    return chunks
-
-
 class TelegramChannel(BaseChannel):
     """
     Telegram channel using long polling.
@@ -111,8 +91,8 @@ class TelegramChannel(BaseChannel):
     BOT_COMMANDS = [
         BotCommand("start", "Start the bot"),
         BotCommand("new", "Start a new conversation"),
-        BotCommand("stop", "Stop the current task"),
         BotCommand("help", "Show available commands"),
+        BotCommand("quota", "Show current quota status"),
     ]
     
     def __init__(
@@ -127,8 +107,6 @@ class TelegramChannel(BaseChannel):
         self._app: Application | None = None
         self._chat_ids: dict[str, int] = {}  # Map sender_id to chat_id for replies
         self._typing_tasks: dict[str, asyncio.Task] = {}  # chat_id -> typing loop task
-        self._media_group_buffers: dict[str, dict] = {}
-        self._media_group_tasks: dict[str, asyncio.Task] = {}
     
     async def start(self) -> None:
         """Start the Telegram bot with long polling."""
@@ -149,7 +127,8 @@ class TelegramChannel(BaseChannel):
         # Add command handlers
         self._app.add_handler(CommandHandler("start", self._on_start))
         self._app.add_handler(CommandHandler("new", self._forward_command))
-        self._app.add_handler(CommandHandler("help", self._on_help))
+        self._app.add_handler(CommandHandler("help", self._forward_command))
+        self._app.add_handler(CommandHandler("quota", self._forward_command))
         
         # Add message handler for text, photos, voice, documents
         self._app.add_handler(
@@ -168,13 +147,13 @@ class TelegramChannel(BaseChannel):
         
         # Get bot info and register command menu
         bot_info = await self._app.bot.get_me()
-        logger.info("Telegram bot @{} connected", bot_info.username)
+        logger.info(f"Telegram bot @{bot_info.username} connected")
         
         try:
             await self._app.bot.set_my_commands(self.BOT_COMMANDS)
             logger.debug("Telegram bot commands registered")
         except Exception as e:
-            logger.warning("Failed to register bot commands: {}", e)
+            logger.warning(f"Failed to register bot commands: {e}")
         
         # Start polling (this runs until stopped)
         await self._app.updater.start_polling(
@@ -193,11 +172,6 @@ class TelegramChannel(BaseChannel):
         # Cancel all typing indicators
         for chat_id in list(self._typing_tasks):
             self._stop_typing(chat_id)
-
-        for task in self._media_group_tasks.values():
-            task.cancel()
-        self._media_group_tasks.clear()
-        self._media_group_buffers.clear()
         
         if self._app:
             logger.info("Stopping Telegram bot...")
@@ -206,123 +180,56 @@ class TelegramChannel(BaseChannel):
             await self._app.shutdown()
             self._app = None
     
-    @staticmethod
-    def _get_media_type(path: str) -> str:
-        """Guess media type from file extension."""
-        ext = path.rsplit(".", 1)[-1].lower() if "." in path else ""
-        if ext in ("jpg", "jpeg", "png", "gif", "webp"):
-            return "photo"
-        if ext == "ogg":
-            return "voice"
-        if ext in ("mp3", "m4a", "wav", "aac"):
-            return "audio"
-        return "document"
-
     async def send(self, msg: OutboundMessage) -> None:
         """Send a message through Telegram."""
         if not self._app:
             logger.warning("Telegram bot not running")
             return
-
+        
+        # Stop typing indicator for this chat
         self._stop_typing(msg.chat_id)
-
+        
         try:
+            # chat_id should be the Telegram chat ID (integer)
             chat_id = int(msg.chat_id)
+            # Convert markdown to Telegram HTML
+            html_content = _markdown_to_telegram_html(msg.content)
+            await self._app.bot.send_message(
+                chat_id=chat_id,
+                text=html_content,
+                parse_mode="HTML"
+            )
         except ValueError:
-            logger.error("Invalid chat_id: {}", msg.chat_id)
+            logger.error(f"Invalid chat_id: {msg.chat_id}")
-            return
+        except Exception as e:
-
+            # Fallback to plain text if HTML parsing fails
-        reply_params = None
+            logger.warning(f"HTML parse failed, falling back to plain text: {e}")
-        if self.config.reply_to_message:
-            reply_to_message_id = msg.metadata.get("message_id")
-            if reply_to_message_id:
-                reply_params = ReplyParameters(
-                    message_id=reply_to_message_id,
-                    allow_sending_without_reply=True
-                )
-
-        # Send media files
-        for media_path in (msg.media or []):
             try:
-                media_type = self._get_media_type(media_path)
-                sender = {
-                    "photo": self._app.bot.send_photo,
-                    "voice": self._app.bot.send_voice,
-                    "audio": self._app.bot.send_audio,
-                }.get(media_type, self._app.bot.send_document)
-                param = "photo" if media_type == "photo" else media_type if media_type in ("voice", "audio") else "document"
-                with open(media_path, 'rb') as f:
-                    await sender(
-                        chat_id=chat_id, 
-                        **{param: f},
-                        reply_parameters=reply_params
-                    )
-            except Exception as e:
-                filename = media_path.rsplit("/", 1)[-1]
-                logger.error("Failed to send media {}: {}", media_path, e)
                 await self._app.bot.send_message(
-                    chat_id=chat_id,
+                    chat_id=int(msg.chat_id),
-                    text=f"[Failed to send: {filename}]",
+                    text=msg.content
-                    reply_parameters=reply_params
                 )
-
+            except Exception as e2:
-        # Send text content
+                logger.error(f"Error sending Telegram message: {e2}")
-        if msg.content and msg.content != "[empty message]":
-            for chunk in _split_message(msg.content):
-                try:
-                    html = _markdown_to_telegram_html(chunk)
-                    await self._app.bot.send_message(
-                        chat_id=chat_id, 
-                        text=html, 
-                        parse_mode="HTML",
-                        reply_parameters=reply_params
-                    )
-                except Exception as e:
-                    logger.warning("HTML parse failed, falling back to plain text: {}", e)
-                    try:
-                        await self._app.bot.send_message(
-                            chat_id=chat_id, 
-                            text=chunk,
-                            reply_parameters=reply_params
-                        )
-                    except Exception as e2:
-                        logger.error("Error sending Telegram message: {}", e2)
     
     async def _on_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
         """Handle /start command."""
         if not update.message or not update.effective_user:
             return
-
+        
         user = update.effective_user
         await update.message.reply_text(
             f"👋 Hi {user.first_name}! I'm nanobot.\n\n"
             "Send me a message and I'll respond!\n"
             "Type /help to see available commands."
         )
-
+    
-    async def _on_help(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
-        """Handle /help command, bypassing ACL so all users can access it."""
-        if not update.message:
-            return
-        await update.message.reply_text(
-            "🐈 nanobot commands:\n"
-            "/new — Start a new conversation\n"
-            "/stop — Stop the current task\n"
-            "/help — Show available commands"
-        )
-
-    @staticmethod
-    def _sender_id(user) -> str:
-        """Build sender_id with username for allowlist matching."""
-        sid = str(user.id)
-        return f"{sid}|{user.username}" if user.username else sid
-
     async def _forward_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
         """Forward slash commands to the bus for unified handling in AgentLoop."""
         if not update.message or not update.effective_user:
             return
         await self._handle_message(
-            sender_id=self._sender_id(update.effective_user),
+            sender_id=str(update.effective_user.id),
             chat_id=str(update.message.chat_id),
             content=update.message.text,
         )
@@ -335,7 +242,11 @@ class TelegramChannel(BaseChannel):
         message = update.message
         user = update.effective_user
         chat_id = message.chat_id
-        sender_id = self._sender_id(user)
+        
+        # Use stable numeric ID, but keep username for allowlist compatibility
+        sender_id = str(user.id)
+        if user.username:
+            sender_id = f"{sender_id}|{user.username}"
         
         # Store chat_id for replies
         self._chat_ids[sender_id] = chat_id
@@ -389,45 +300,23 @@ class TelegramChannel(BaseChannel):
                     transcriber = GroqTranscriptionProvider(api_key=self.groq_api_key)
                     transcription = await transcriber.transcribe(file_path)
                     if transcription:
-                        logger.info("Transcribed {}: {}...", media_type, transcription[:50])
+                        logger.info(f"Transcribed {media_type}: {transcription[:50]}...")
                         content_parts.append(f"[transcription: {transcription}]")
                     else:
                         content_parts.append(f"[{media_type}: {file_path}]")
                 else:
                     content_parts.append(f"[{media_type}: {file_path}]")
                     
-                logger.debug("Downloaded {} to {}", media_type, file_path)
+                logger.debug(f"Downloaded {media_type} to {file_path}")
             except Exception as e:
-                logger.error("Failed to download media: {}", e)
+                logger.error(f"Failed to download media: {e}")
                 content_parts.append(f"[{media_type}: download failed]")
         
         content = "\n".join(content_parts) if content_parts else "[empty message]"
         
-        logger.debug("Telegram message from {}: {}...", sender_id, content[:50])
+        logger.debug(f"Telegram message from {sender_id}: {content[:50]}...")
         
         str_chat_id = str(chat_id)
-
-        # Telegram media groups: buffer briefly, forward as one aggregated turn.
-        if media_group_id := getattr(message, "media_group_id", None):
-            key = f"{str_chat_id}:{media_group_id}"
-            if key not in self._media_group_buffers:
-                self._media_group_buffers[key] = {
-                    "sender_id": sender_id, "chat_id": str_chat_id,
-                    "contents": [], "media": [],
-                    "metadata": {
-                        "message_id": message.message_id, "user_id": user.id,
-                        "username": user.username, "first_name": user.first_name,
-                        "is_group": message.chat.type != "private",
-                    },
-                }
-                self._start_typing(str_chat_id)
-            buf = self._media_group_buffers[key]
-            if content and content != "[empty message]":
-                buf["contents"].append(content)
-            buf["media"].extend(media_paths)
-            if key not in self._media_group_tasks:
-                self._media_group_tasks[key] = asyncio.create_task(self._flush_media_group(key))
-            return
         
         # Start typing indicator before processing
         self._start_typing(str_chat_id)
@@ -447,21 +336,6 @@ class TelegramChannel(BaseChannel):
             }
         )
     
-    async def _flush_media_group(self, key: str) -> None:
-        """Wait briefly, then forward buffered media-group as one turn."""
-        try:
-            await asyncio.sleep(0.6)
-            if not (buf := self._media_group_buffers.pop(key, None)):
-                return
-            content = "\n".join(buf["contents"]) or "[empty message]"
-            await self._handle_message(
-                sender_id=buf["sender_id"], chat_id=buf["chat_id"],
-                content=content, media=list(dict.fromkeys(buf["media"])),
-                metadata=buf["metadata"],
-            )
-        finally:
-            self._media_group_tasks.pop(key, None)
-
     def _start_typing(self, chat_id: str) -> None:
         """Start sending 'typing...' indicator for a chat."""
         # Cancel any existing typing task for this chat
@@ -483,11 +357,11 @@ class TelegramChannel(BaseChannel):
         except asyncio.CancelledError:
             pass
         except Exception as e:
-            logger.debug("Typing indicator stopped for {}: {}", chat_id, e)
+            logger.debug(f"Typing indicator stopped for {chat_id}: {e}")
     
     async def _on_error(self, update: object, context: ContextTypes.DEFAULT_TYPE) -> None:
         """Log polling / handler errors instead of silently swallowing them."""
-        logger.error("Telegram error: {}", context.error)
+        logger.error(f"Telegram error: {context.error}")
 
     def _get_extension(self, media_type: str, mime_type: str | None) -> str:
         """Get file extension based on media type."""