diff --git a/nanobot/agent/loop.py b/nanobot/agent/loop.py index d14b6ea..e5ed6e4 100644 --- a/nanobot/agent/loop.py +++ b/nanobot/agent/loop.py @@ -1,30 +1,36 @@ """Agent loop: the core processing engine.""" -import asyncio -from contextlib import AsyncExitStack -import json -import json_repair -from pathlib import Path -import re -from typing import Any, Awaitable, Callable +from __future__ import annotations +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, - cron_service: "CronService | None" = None, + exec_config: ExecToolConfig | 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 "" diff --git a/nanobot/agent/tools/message.py b/nanobot/agent/tools/message.py index 10947c4..40e76e3 100644 --- a/nanobot/agent/tools/message.py +++ b/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, - content: str, - channel: str | None = None, + self, + content: str, + 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: diff --git a/nanobot/channels/base.py b/nanobot/channels/base.py index 30fcd1a..3a5a785 100644 --- a/nanobot/channels/base.py +++ b/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}. " - f"Add them to allowFrom list in config to grant access." + "Access denied for sender {} on channel {}. " + "Add them to allowFrom list in config to grant access.", + sender_id, self.name, ) return diff --git a/nanobot/channels/dingtalk.py b/nanobot/channels/dingtalk.py index b7263b3..09c7714 100644 --- a/nanobot/channels/dingtalk.py +++ b/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) diff --git a/nanobot/channels/discord.py b/nanobot/channels/discord.py index 8baecbf..1d2d7a6 100644 --- a/nanobot/channels/discord.py +++ b/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): - 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 - except Exception as e: - if attempt == 2: - logger.error("Error sending Discord message: {}", e) - else: - await asyncio.sleep(1) + chunks = _split_message(msg.content or "") + if not chunks: + return + + for i, chunk in enumerate(chunks): + payload: dict[str, Any] = {"content": chunk} + + # Only set reply reference on the first chunk + if i == 0 and msg.reply_to: + payload["message_reference"] = {"message_id": msg.reply_to} + payload["allowed_mentions"] = {"replied_user": False} + + if not await self._send_payload(url, headers, payload): + break # Abort remaining chunks on failure 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: diff --git a/nanobot/channels/slack.py b/nanobot/channels/slack.py index 79cbe76..4fc1f41 100644 --- a/nanobot/channels/slack.py +++ b/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( - channel=msg.chat_id, - text=self._to_mrkdwn(msg.content), - thread_ts=thread_ts if use_thread else None, - ) + thread_ts_param = thread_ts if use_thread else None + + if msg.content: + 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) diff --git a/nanobot/providers/litellm_provider.py b/nanobot/providers/litellm_provider.py index edeb5c6..58c9ac2 100644 --- a/nanobot/providers/litellm_provider.py +++ b/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 diff --git a/nanobot/providers/registry.py b/nanobot/providers/registry.py index 445d977..ecf092f 100644 --- a/nanobot/providers/registry.py +++ b/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. diff --git a/nanobot/session/manager.py b/nanobot/session/manager.py index 9c1e427..18e23b2 100644 --- a/nanobot/session/manager.py +++ b/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)