Merge branch 'main' into pr-892

nanobot/agent/loop.py

@@ -1,30 +1,36 @@
 """Agent loop: the core processing engine."""
 
-import asyncio
+from __future__ import annotations
-from contextlib import AsyncExitStack
-import json
-import json_repair
-from pathlib import Path
-import re
-from typing import Any, Awaitable, Callable
 
+import asyncio
+import json
+import re
+from contextlib import AsyncExitStack
+from pathlib import Path
+from typing import TYPE_CHECKING, Awaitable, Callable
+
+import json_repair
 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.agent.context import ContextBuilder
-from nanobot.agent.tools.registry import ToolRegistry
-from nanobot.agent.tools.filesystem import ReadFileTool, WriteFileTool, EditFileTool, ListDirTool
-from nanobot.agent.tools.shell import ExecTool
-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 Session, SessionManager
 
+if TYPE_CHECKING:
+    from nanobot.config.schema import ExecToolConfig
+    from nanobot.cron.service import CronService
+
 
 class AgentLoop:
     """
@@ -49,14 +55,13 @@ class AgentLoop:
         max_tokens: int = 4096,
         memory_window: int = 50,
         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,
     ):
         from nanobot.config.schema import ExecToolConfig
-        from nanobot.cron.service import CronService
         self.bus = bus
         self.provider = provider
         self.workspace = workspace
@@ -84,7 +89,7 @@ class AgentLoop:
             exec_config=self.exec_config,
             restrict_to_workspace=restrict_to_workspace,
         )
-        
+
         self._running = False
         self._mcp_servers = mcp_servers or {}
         self._mcp_stack: AsyncExitStack | None = None
@@ -92,7 +97,7 @@ class AgentLoop:
         self._mcp_connecting = False
         self._consolidating: set[str] = set()  # Session keys with consolidation in progress
         self._register_default_tools()
-    
+
     def _register_default_tools(self) -> None:
         """Register the default set of tools."""
         # File tools (workspace for relative paths, restrict if configured)
@@ -101,30 +106,30 @@ class AgentLoop:
         self.tools.register(WriteFileTool(workspace=self.workspace, allowed_dir=allowed_dir))
         self.tools.register(EditFileTool(workspace=self.workspace, allowed_dir=allowed_dir))
         self.tools.register(ListDirTool(workspace=self.workspace, 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,
         ))
-        
+
         # Web tools
         self.tools.register(WebSearchTool(api_key=self.brave_api_key))
         self.tools.register(WebFetchTool())
-        
+
         # 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:
         """Connect to configured MCP servers (one-time, lazy)."""
         if self._mcp_connected or self._mcp_connecting or not self._mcp_servers:
@@ -283,7 +288,7 @@ class AgentLoop:
                     ))
             except asyncio.TimeoutError:
                 continue
-    
+
     async def close_mcp(self) -> None:
         """Close MCP connections."""
         if self._mcp_stack:
@@ -297,7 +302,7 @@ class AgentLoop:
         """Stop the agent loop."""
         self._running = False
         logger.info("Agent loop stopping")
-    
+
     async def _process_message(
         self,
         msg: InboundMessage,
@@ -306,25 +311,25 @@ class AgentLoop:
     ) -> OutboundMessage | None:
         """
         Process a single inbound message.
-        
+
         Args:
             msg: The inbound message to process.
             session_key: Override session key (used by process_direct).
             on_progress: Optional callback for intermediate output (defaults to bus publish).
-        
+
         Returns:
             The response message, or None if no response needed.
         """
         # System messages route back via chat_id ("channel:chat_id")
         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("Processing message from {}:{}: {}", msg.channel, msg.sender_id, preview)
-        
+
         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":
@@ -345,7 +350,7 @@ class AgentLoop:
         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")
-        
+
         if len(session.messages) > self.memory_window and session.key not in self._consolidating:
             self._consolidating.add(session.key)
 
@@ -358,6 +363,10 @@ class AgentLoop:
             asyncio.create_task(_consolidate_and_unlock())
 
         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()
+
         initial_messages = self.context.build_messages(
             history=session.get_history(max_messages=self.memory_window),
             current_message=msg.content,
@@ -378,31 +387,35 @@ class AgentLoop:
 
         if final_content is None:
             final_content = "I've completed processing but have no response to give."
-        
+
         preview = final_content[:120] + "..." if len(final_content) > 120 else final_content
         logger.info("Response to {}:{}: {}", msg.channel, msg.sender_id, preview)
-        
+
         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)
-        
+
+        if message_tool := self.tools.get("message"):
+            if isinstance(message_tool, MessageTool) and message_tool._sent_in_turn:
+                return None
+
         return OutboundMessage(
             channel=msg.channel,
             chat_id=msg.chat_id,
             content=final_content,
             metadata=msg.metadata or {},  # Pass through for channel-specific needs (e.g. Slack thread_ts)
         )
-    
+
     async def _process_system_message(self, msg: InboundMessage) -> OutboundMessage | None:
         """
         Process a system message (e.g., subagent announce).
-        
+
         The chat_id field contains "original_channel:original_chat_id" to route
         the response back to the correct destination.
         """
         logger.info("Processing system message from {}", msg.sender_id)
-        
+
         # Parse origin from chat_id (format: "channel:chat_id")
         if ":" in msg.chat_id:
             parts = msg.chat_id.split(":", 1)
@@ -412,7 +425,7 @@ class AgentLoop:
             # Fallback
             origin_channel = "cli"
             origin_chat_id = msg.chat_id
-        
+
         session_key = f"{origin_channel}:{origin_chat_id}"
         session = self.sessions.get_or_create(session_key)
         self._set_tool_context(origin_channel, origin_chat_id, msg.metadata.get("message_id"))
@@ -426,17 +439,17 @@ class AgentLoop:
 
         if final_content is None:
             final_content = "Background task completed."
-        
+
         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.
 
@@ -546,14 +559,14 @@ Respond with ONLY valid JSON, no markdown fences."""
     ) -> str:
         """
         Process a message directly (for CLI or cron usage).
-        
+
         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).
             on_progress: Optional callback for intermediate output.
-        
+
         Returns:
             The agent's response.
         """
@@ -564,6 +577,6 @@ Respond with ONLY valid JSON, no markdown fences."""
             chat_id=chat_id,
             content=content
         )
-        
+
         response = await self._process_message(msg, session_key=session_key, on_progress=on_progress)
         return response.content if response else ""

nanobot/agent/tools/message.py

@@ -1,6 +1,6 @@
 """Message tool for sending messages to users."""
 
-from typing import Any, Callable, Awaitable
+from typing import Any, Awaitable, Callable
 
 from nanobot.agent.tools.base import Tool
 from nanobot.bus.events import OutboundMessage
@@ -8,37 +8,42 @@ from nanobot.bus.events import OutboundMessage
 
 class MessageTool(Tool):
     """Tool to send messages to users on chat channels."""
-    
+
     def __init__(
-        self, 
+        self,
         send_callback: Callable[[OutboundMessage], Awaitable[None]] | None = None,
         default_channel: str = "",
         default_chat_id: str = "",
-        default_message_id: str | None = None
+        default_message_id: str | None = None,
     ):
         self._send_callback = send_callback
         self._default_channel = default_channel
         self._default_chat_id = default_chat_id
         self._default_message_id = default_message_id
-    
+        self._sent_in_turn: bool = False
+
     def set_context(self, channel: str, chat_id: str, message_id: str | None = None) -> None:
         """Set the current message context."""
         self._default_channel = channel
         self._default_chat_id = chat_id
         self._default_message_id = message_id
-    
+
     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:
         return "message"
-    
+
     @property
     def description(self) -> str:
         return "Send a message to the user. Use this when you want to communicate something."
-    
+
     @property
     def parameters(self) -> dict[str, Any]:
         return {
@@ -64,11 +69,11 @@ class MessageTool(Tool):
             },
             "required": ["content"]
         }
-    
+
     async def execute(
-        self, 
+        self,
-        content: str, 
+        content: str,
-        channel: str | None = None, 
+        channel: str | None = None,
         chat_id: str | None = None,
         message_id: str | None = None,
         media: list[str] | None = None,
@@ -77,13 +82,13 @@ class MessageTool(Tool):
         channel = channel or self._default_channel
         chat_id = chat_id or self._default_chat_id
         message_id = message_id or self._default_message_id
-        
+
         if not channel or not chat_id:
             return "Error: No target channel/chat specified"
-        
+
         if not self._send_callback:
             return "Error: Message sending not configured"
-        
+
         msg = OutboundMessage(
             channel=channel,
             chat_id=chat_id,
@@ -93,9 +98,10 @@ class MessageTool(Tool):
                 "message_id": message_id,
             }
         )
-        
+
         try:
             await self._send_callback(msg)
+            self._sent_in_turn = True
             media_info = f" with {len(media)} attachments" if media else ""
             return f"Message sent to {channel}:{chat_id}{media_info}"
         except Exception as e:

nanobot/channels/base.py

@@ -105,8 +105,9 @@ class BaseChannel(ABC):
         """
         if not self.is_allowed(sender_id):
             logger.warning(
-                f"Access denied for sender {sender_id} on channel {self.name}. "
+                "Access denied for sender {} on channel {}. "
-                f"Add them to allowFrom list in config to grant access."
+                "Add them to allowFrom list in config to grant access.",
+                sender_id, self.name,
             )
             return
         

nanobot/channels/dingtalk.py

@@ -58,7 +58,8 @@ class NanobotDingTalkHandler(CallbackHandler):
 
             if not content:
                 logger.warning(
-                    f"Received empty or unsupported message type: {chatbot_msg.message_type}"
+                    "Received empty or unsupported message type: {}",
+                    chatbot_msg.message_type,
                 )
                 return AckMessage.STATUS_OK, "OK"
 
@@ -126,7 +127,8 @@ class DingTalkChannel(BaseChannel):
             self._http = httpx.AsyncClient()
 
             logger.info(
-                f"Initializing DingTalk Stream Client with Client ID: {self.config.client_id}..."
+                "Initializing DingTalk Stream Client with Client ID: {}...",
+                self.config.client_id,
             )
             credential = Credential(self.config.client_id, self.config.client_secret)
             self._client = DingTalkStreamClient(credential)

nanobot/channels/discord.py

@@ -17,6 +17,29 @@ from nanobot.config.schema import DiscordConfig
 
 DISCORD_API_BASE = "https://discord.com/api/v10"
 MAX_ATTACHMENT_BYTES = 20 * 1024 * 1024  # 20MB
+MAX_MESSAGE_LEN = 2000  # Discord message character limit
+
+
+def _split_message(content: str, max_len: int = MAX_MESSAGE_LEN) -> list[str]:
+    """Split content into chunks within max_len, preferring line breaks."""
+    if not content:
+        return []
+    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 <= 0:
+            pos = cut.rfind(' ')
+        if pos <= 0:
+            pos = max_len
+        chunks.append(content[:pos])
+        content = content[pos:].lstrip()
+    return chunks
 
 
 class DiscordChannel(BaseChannel):
@@ -79,34 +102,48 @@ class DiscordChannel(BaseChannel):
             return
 
         url = f"{DISCORD_API_BASE}/channels/{msg.chat_id}/messages"
-        payload: dict[str, Any] = {"content": msg.content}
-
-        if msg.reply_to:
-            payload["message_reference"] = {"message_id": msg.reply_to}
-            payload["allowed_mentions"] = {"replied_user": False}
-
         headers = {"Authorization": f"Bot {self.config.token}"}
 
         try:
-            for attempt in range(3):
+            chunks = _split_message(msg.content or "")
-                try:
+            if not chunks:
-                    response = await self._http.post(url, headers=headers, json=payload)
+                return
-                    if response.status_code == 429:
+
-                        data = response.json()
+            for i, chunk in enumerate(chunks):
-                        retry_after = float(data.get("retry_after", 1.0))
+                payload: dict[str, Any] = {"content": chunk}
-                        logger.warning("Discord rate limited, retrying in {}s", retry_after)
+
-                        await asyncio.sleep(retry_after)
+                # Only set reply reference on the first chunk
-                        continue
+                if i == 0 and msg.reply_to:
-                    response.raise_for_status()
+                    payload["message_reference"] = {"message_id": msg.reply_to}
-                    return
+                    payload["allowed_mentions"] = {"replied_user": False}
-                except Exception as e:
+
-                    if attempt == 2:
+                if not await self._send_payload(url, headers, payload):
-                        logger.error("Error sending Discord message: {}", e)
+                    break  # Abort remaining chunks on failure
-                    else:
-                        await asyncio.sleep(1)
         finally:
             await self._stop_typing(msg.chat_id)
 
+    async def _send_payload(
+        self, url: str, headers: dict[str, str], payload: dict[str, Any]
+    ) -> bool:
+        """Send a single Discord API payload with retry on rate-limit. Returns True on success."""
+        for attempt in range(3):
+            try:
+                response = await self._http.post(url, headers=headers, json=payload)
+                if response.status_code == 429:
+                    data = response.json()
+                    retry_after = float(data.get("retry_after", 1.0))
+                    logger.warning("Discord rate limited, retrying in {}s", retry_after)
+                    await asyncio.sleep(retry_after)
+                    continue
+                response.raise_for_status()
+                return True
+            except Exception as e:
+                if attempt == 2:
+                    logger.error("Error sending Discord message: {}", e)
+                else:
+                    await asyncio.sleep(1)
+        return False
+
     async def _gateway_loop(self) -> None:
         """Main gateway loop: identify, heartbeat, dispatch events."""
         if not self._ws:

nanobot/channels/slack.py

@@ -84,11 +84,24 @@ class SlackChannel(BaseChannel):
             channel_type = slack_meta.get("channel_type")
             # Only reply in thread for channel/group messages; DMs don't use threads
             use_thread = thread_ts and channel_type != "im"
-            await self._web_client.chat_postMessage(
+            thread_ts_param = thread_ts if use_thread else None
-                channel=msg.chat_id,
+
-                text=self._to_mrkdwn(msg.content),
+            if msg.content:
-                thread_ts=thread_ts if use_thread else None,
+                await self._web_client.chat_postMessage(
-            )
+                    channel=msg.chat_id,
+                    text=self._to_mrkdwn(msg.content),
+                    thread_ts=thread_ts_param,
+                )
+
+            for media_path in msg.media or []:
+                try:
+                    await self._web_client.files_upload_v2(
+                        channel=msg.chat_id,
+                        file=media_path,
+                        thread_ts=thread_ts_param,
+                    )
+                except Exception as e:
+                    logger.error("Failed to upload file {}: {}", media_path, e)
         except Exception as e:
             logger.error("Error sending Slack message: {}", e)
 

nanobot/providers/litellm_provider.py

@@ -111,7 +111,7 @@ class LiteLLMProvider(LLMProvider):
     def _supports_cache_control(self, model: str) -> bool:
         """Return True when the provider supports cache_control on content blocks."""
         if self._gateway is not None:
-            return False
+            return self._gateway.supports_prompt_caching
         spec = find_by_model(model)
         return spec is not None and spec.supports_prompt_caching
 

nanobot/providers/registry.py

@@ -100,6 +100,7 @@ PROVIDERS: tuple[ProviderSpec, ...] = (
         default_api_base="https://openrouter.ai/api/v1",
         strip_model_prefix=False,
         model_overrides=(),
+        supports_prompt_caching=True,
     ),
 
     # AiHubMix: global gateway, OpenAI-compatible interface.

nanobot/session/manager.py

@@ -154,6 +154,7 @@ class SessionManager:
         with open(path, "w", encoding="utf-8") as f:
             metadata_line = {
                 "_type": "metadata",
+                "key": session.key,
                 "created_at": session.created_at.isoformat(),
                 "updated_at": session.updated_at.isoformat(),
                 "metadata": session.metadata,
@@ -186,8 +187,9 @@ class SessionManager:
                     if first_line:
                         data = json.loads(first_line)
                         if data.get("_type") == "metadata":
+                            key = data.get("key") or path.stem.replace("_", ":", 1)
                             sessions.append({
-                                "key": path.stem.replace("_", ":"),
+                                "key": key,
                                 "created_at": data.get("created_at"),
                                 "updated_at": data.get("updated_at"),
                                 "path": str(path)