diff --git a/README.md b/README.md index be360dc..71922fb 100644 --- a/README.md +++ b/README.md @@ -16,7 +16,7 @@ ⚡️ Delivers core agent functionality in just **~4,000** lines of code — **99% smaller** than Clawdbot's 430k+ lines. -📏 Real-time line count: **3,966 lines** (run `bash core_agent_lines.sh` to verify anytime) +📏 Real-time line count: **3,932 lines** (run `bash core_agent_lines.sh` to verify anytime) ## 📢 News diff --git a/nanobot/agent/loop.py b/nanobot/agent/loop.py index 6155f99..e3a9d67 100644 --- a/nanobot/agent/loop.py +++ b/nanobot/agent/loop.py @@ -43,6 +43,8 @@ class AgentLoop: 5. Sends responses back """ + _TOOL_RESULT_MAX_CHARS = 500 + def __init__( self, bus: MessageBus, @@ -145,17 +147,10 @@ class AgentLoop: def _set_tool_context(self, channel: str, chat_id: str, message_id: str | None = None) -> None: """Update context for all tools that need routing info.""" - if message_tool := self.tools.get("message"): - if isinstance(message_tool, MessageTool): - message_tool.set_context(channel, chat_id, message_id) - - if spawn_tool := self.tools.get("spawn"): - if isinstance(spawn_tool, SpawnTool): - spawn_tool.set_context(channel, chat_id) - - if cron_tool := self.tools.get("cron"): - if isinstance(cron_tool, CronTool): - cron_tool.set_context(channel, chat_id) + for name in ("message", "spawn", "cron"): + if tool := self.tools.get(name): + if hasattr(tool, "set_context"): + tool.set_context(channel, chat_id, *([message_id] if name == "message" else [])) @staticmethod def _strip_think(text: str | None) -> str | None: @@ -315,18 +310,6 @@ class AgentLoop: self._running = False logger.info("Agent loop stopping") - def _get_consolidation_lock(self, session_key: str) -> asyncio.Lock: - lock = self._consolidation_locks.get(session_key) - if lock is None: - lock = asyncio.Lock() - self._consolidation_locks[session_key] = lock - return lock - - def _prune_consolidation_lock(self, session_key: str, lock: asyncio.Lock) -> None: - """Drop lock entry if no longer in use.""" - if not lock.locked(): - self._consolidation_locks.pop(session_key, None) - async def _process_message( self, msg: InboundMessage, @@ -362,7 +345,7 @@ class AgentLoop: # Slash commands cmd = msg.content.strip().lower() if cmd == "/new": - lock = self._get_consolidation_lock(session.key) + lock = self._consolidation_locks.setdefault(session.key, asyncio.Lock()) self._consolidating.add(session.key) try: async with lock: @@ -383,7 +366,8 @@ class AgentLoop: ) finally: self._consolidating.discard(session.key) - self._prune_consolidation_lock(session.key, lock) + if not lock.locked(): + self._consolidation_locks.pop(session.key, None) session.clear() self.sessions.save(session) @@ -397,7 +381,7 @@ class AgentLoop: 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._get_consolidation_lock(session.key) + lock = self._consolidation_locks.setdefault(session.key, asyncio.Lock()) async def _consolidate_and_unlock(): try: @@ -405,7 +389,8 @@ class AgentLoop: await self._consolidate_memory(session) finally: self._consolidating.discard(session.key) - self._prune_consolidation_lock(session.key, lock) + 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) @@ -454,8 +439,6 @@ class AgentLoop: metadata=msg.metadata or {}, ) - _TOOL_RESULT_MAX_CHARS = 500 - def _save_turn(self, session: Session, messages: list[dict], skip: int) -> None: """Save new-turn messages into session, truncating large tool results.""" from datetime import datetime diff --git a/nanobot/agent/tools/web.py b/nanobot/agent/tools/web.py index 56956c3..7860f12 100644 --- a/nanobot/agent/tools/web.py +++ b/nanobot/agent/tools/web.py @@ -80,7 +80,7 @@ class WebSearchTool(Tool): r = await client.get( "https://api.search.brave.com/res/v1/web/search", params={"q": query, "count": n}, - headers={"Accept": "application/json", "X-Subscription-Token": api_key}, + headers={"Accept": "application/json", "X-Subscription-Token": self.api_key}, timeout=10.0 ) r.raise_for_status() diff --git a/nanobot/channels/feishu.py b/nanobot/channels/feishu.py index 480bf7b..4a6312e 100644 --- a/nanobot/channels/feishu.py +++ b/nanobot/channels/feishu.py @@ -692,7 +692,7 @@ class FeishuChannel(BaseChannel): msg_type = message.message_type # Add reaction - await self._add_reaction(message_id, "THUMBSUP") + await self._add_reaction(message_id, self.config.react_emoji) # Parse content content_parts = [] diff --git a/nanobot/channels/telegram.py b/nanobot/channels/telegram.py index 808f50c..969d853 100644 --- a/nanobot/channels/telegram.py +++ b/nanobot/channels/telegram.py @@ -127,6 +127,8 @@ 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.""" @@ -191,6 +193,11 @@ 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...") @@ -399,6 +406,28 @@ class TelegramChannel(BaseChannel): logger.debug("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) @@ -418,6 +447,21 @@ 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 diff --git a/nanobot/config/schema.py b/nanobot/config/schema.py index 61aee96..d83967c 100644 --- a/nanobot/config/schema.py +++ b/nanobot/config/schema.py @@ -42,6 +42,7 @@ class FeishuConfig(Base): encrypt_key: str = "" # Encrypt Key for event subscription (optional) verification_token: str = "" # Verification Token for event subscription (optional) allow_from: list[str] = Field(default_factory=list) # Allowed user open_ids + react_emoji: str = "THUMBSUP" # Emoji type for message reactions (e.g. THUMBSUP, OK, DONE, SMILE) class DingTalkConfig(Base): diff --git a/tests/test_consolidate_offset.py b/tests/test_consolidate_offset.py index 323519e..6755124 100644 --- a/tests/test_consolidate_offset.py +++ b/tests/test_consolidate_offset.py @@ -812,7 +812,7 @@ class TestConsolidationDeduplicationGuard: loop.sessions.save(session) # Ensure lock exists before /new. - _ = loop._get_consolidation_lock(session.key) + loop._consolidation_locks.setdefault(session.key, asyncio.Lock()) assert session.key in loop._consolidation_locks async def _ok_consolidate(sess, archive_all: bool = False) -> bool: