Merge remote-tracking branch 'origin/main' into pr-1253

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
 

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"):
+        for name in ("message", "spawn", "cron"):
-            if isinstance(message_tool, MessageTool):
+            if tool := self.tools.get(name):
-                message_tool.set_context(channel, chat_id, message_id)
+                if hasattr(tool, "set_context"):
-
+                    tool.set_context(channel, chat_id, *([message_id] if name == "message" else []))
-        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)
 
     @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

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()

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 = []

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

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):

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: