fix(skill-tools): follow symlinks in skill discovery via rglob_follow

agent/skill_utils.py

@@ -26,6 +26,30 @@
 
 EXCLUDED_SKILL_DIRS = frozenset((".git", ".github", ".hub", ".archive"))
 
+
+def rglob_follow(root: Path, pattern: str):
+    """Like ``Path.rglob(pattern)`` but follows symlinks into subdirectories.
+
+    Python's ``Path.rglob()`` intentionally does **not** descend into
+    symlinked directories (see https://bugs.python.org/issue40358).
+    This helper uses ``os.walk(followlinks=True)`` so that symlinked
+    skill directories (e.g. ``~/.hermes/skills/redpiggy -> workspace/skills``)
+    are discovered correctly.
+    """
+    import fnmatch
+
+    for dirpath, dirnames, filenames in os.walk(str(root), followlinks=True):
+        # Prune excluded dirs (same as rglob's natural traversal)
+        dirnames[:] = [d for d in dirnames if d not in EXCLUDED_SKILL_DIRS]
+        dirpath_p = Path(dirpath)
+        for fname in filenames:
+            if fnmatch.fnmatch(fname, pattern):
+                yield dirpath_p / fname
+        # Also match the pattern against directory names if needed
+        for dname in dirnames:
+            if fnmatch.fnmatch(dname, pattern):
+                yield dirpath_p / dname
+
 # ── Lazy YAML loader ─────────────────────────────────────────────────────
 
 _yaml_load_fn = None

gateway/platforms/base.py

@@ -23,6 +23,26 @@
 
 logger = logging.getLogger(__name__)
 
+
+def _apply_text_hooks(text: str, event: "MessageEvent", session_id: str | None = None) -> tuple[str, str]:
+    """Apply pre_gateway_text_send hooks. Returns (action, text).
+    action is 'allow', 'block', or 'rewrite'.
+    """
+    from hermes_cli.plugins import invoke_hook
+    result = invoke_hook("pre_gateway_text_send", text=text, event=event, session_id=session_id)
+    if not result:
+        return "allow", text
+    # Hooks may return list of results; take first non-None
+    for r in result:
+        if r is not None:
+            if isinstance(r, dict):
+                action = r.get("action", "allow")
+                new_text = r.get("text", text)
+                return action, new_text
+            elif isinstance(r, str):
+                return "allow", r
+    return "allow", text
+
 # Audio file extensions Hermes recognizes for native audio delivery.
 # Kept in sync with tools/send_message_tool.py and cron/scheduler.py via
 # should_send_media_as_audio() below.
@@ -2946,6 +2966,11 @@ async def _stop_typing_task() -> None:
                         except OSError:
                             pass
 
+                # Apply pre_gateway_text_send hooks
+                _hook_action, text_content = _apply_text_hooks(text_content, event)
+                if _hook_action == "block":
+                    text_content = ""
+
                 # Send the text portion
                 if text_content:
                     logger.info("[%s] Sending response (%d chars) to %s", self.name, len(text_content), event.source.chat_id)

gateway/platforms/weixin.py

@@ -1226,6 +1226,12 @@ def __init__(self, config: PlatformConfig):
             default=False,
         )
 
+        # Text debounce batching
+        self._text_batch_delay_seconds = float(os.getenv("HERMES_WEIXIN_TEXT_BATCH_DELAY_SECONDS", "3.0"))
+        self._text_batch_split_delay_seconds = float(os.getenv("HERMES_WEIXIN_TEXT_BATCH_SPLIT_DELAY_SECONDS", "5.0"))
+        self._pending_text_batches: Dict[str, MessageEvent] = {}
+        self._pending_text_batch_tasks: Dict[str, asyncio.Task] = {}
+
         if self._account_id and not self._token:
             persisted = load_weixin_account(hermes_home, self._account_id)
             if persisted:
@@ -1293,6 +1299,11 @@ async def connect(self) -> bool:
     async def disconnect(self) -> None:
         _LIVE_ADAPTERS.pop(self._token, None)
         self._running = False
+        for task in self._pending_text_batch_tasks.values():
+            if not task.done():
+                task.cancel()
+        self._pending_text_batches.clear()
+        self._pending_text_batch_tasks.clear()
         if self._poll_task and not self._poll_task.done():
             self._poll_task.cancel()
             try:
@@ -1441,7 +1452,10 @@ async def _process_message(self, message: Dict[str, Any]) -> None:
             timestamp=datetime.now(),
         )
         logger.info("[%s] inbound from=%s type=%s media=%d", self.name, _safe_id(sender_id), source.chat_type, len(media_paths))
-        await self.handle_message(event)
+        if event.message_type == MessageType.TEXT:
+            self._enqueue_text_event(event)
+        else:
+            await self.handle_message(event)
 
     def _is_dm_allowed(self, sender_id: str) -> bool:
         if self._dm_policy == "disabled":
@@ -1450,6 +1464,59 @@ def _is_dm_allowed(self, sender_id: str) -> bool:
             return sender_id in self._allow_from
         return True
 
+    # ------------------------------------------------------------------
+    # Text debounce batching
+    # ------------------------------------------------------------------
+
+    _SPLIT_THRESHOLD = 1800
+
+    def _text_batch_key(self, event: MessageEvent) -> str:
+        from gateway.session import build_session_key
+        return build_session_key(
+            event.source,
+            group_sessions_per_user=self.config.extra.get("group_sessions_per_user", True),
+            thread_sessions_per_user=self.config.extra.get("thread_sessions_per_user", False),
+        )
+
+    def _enqueue_text_event(self, event: MessageEvent) -> None:
+        key = self._text_batch_key(event)
+        existing = self._pending_text_batches.get(key)
+        chunk_len = len(event.text or "")
+        if existing is None:
+            event._last_chunk_len = chunk_len
+            self._pending_text_batches[key] = event
+        else:
+            if event.text:
+                existing.text = f"{existing.text}\n{event.text}" if existing.text else event.text
+            existing._last_chunk_len = chunk_len
+            if event.media_urls:
+                existing.media_urls.extend(event.media_urls)
+                existing.media_types.extend(event.media_types)
+        prior_task = self._pending_text_batch_tasks.get(key)
+        if prior_task and not prior_task.done():
+            prior_task.cancel()
+        self._pending_text_batch_tasks[key] = asyncio.create_task(
+            self._flush_text_batch(key)
+        )
+
+    async def _flush_text_batch(self, key: str) -> None:
+        current_task = asyncio.current_task()
+        try:
+            pending = self._pending_text_batches.get(key)
+            last_len = getattr(pending, "_last_chunk_len", 0) if pending else 0
+            if last_len >= self._SPLIT_THRESHOLD:
+                delay = self._text_batch_split_delay_seconds
+            else:
+                delay = self._text_batch_delay_seconds
+            await asyncio.sleep(delay)
+            event = self._pending_text_batches.pop(key, None)
+            if not event:
+                return
+            await self.handle_message(event)
+        finally:
+            if self._pending_text_batch_tasks.get(key) is current_task:
+                self._pending_text_batch_tasks.pop(key, None)
+
     async def _collect_media(self, item: Dict[str, Any], media_paths: List[str], media_types: List[str]) -> None:
         item_type = item.get("type")
         if item_type == ITEM_IMAGE:

gateway/platforms/whatsapp.py

@@ -278,6 +278,12 @@ def __init__(self, config: PlatformConfig):
         # notification before the normal "✓ whatsapp disconnected" fires.
         self._shutting_down: bool = False
 
+        # Text debounce batching (mirrors Telegram adapter pattern)
+        self._text_batch_delay_seconds = float(os.getenv("HERMES_WHATSAPP_TEXT_BATCH_DELAY_SECONDS", "5.0"))
+        self._text_batch_split_delay_seconds = float(os.getenv("HERMES_WHATSAPP_TEXT_BATCH_SPLIT_DELAY_SECONDS", "10.0"))
+        self._pending_text_batches: Dict[str, MessageEvent] = {}
+        self._pending_text_batch_tasks: Dict[str, asyncio.Task] = {}
+
     def _effective_reply_prefix(self) -> str:
         """Return the prefix the Node bridge will add in self-chat mode."""
         whatsapp_mode = os.getenv("WHATSAPP_MODE", "self-chat")
@@ -734,6 +740,13 @@ async def disconnect(self) -> None:
                 pass
         self._poll_task = None
 
+        # Cancel pending text batches
+        for task in self._pending_text_batch_tasks.values():
+            if not task.done():
+                task.cancel()
+        self._pending_text_batches.clear()
+        self._pending_text_batch_tasks.clear()
+
         # Close the persistent HTTP session
         if self._http_session and not self._http_session.closed:
             await self._http_session.close()
@@ -1077,7 +1090,10 @@ async def _poll_messages(self) -> None:
                         for msg_data in messages:
                             event = await self._build_message_event(msg_data)
                             if event:
-                                await self.handle_message(event)
+                                if event.message_type == MessageType.TEXT:
+                                    self._enqueue_text_event(event)
+                                else:
+                                    await self.handle_message(event)
             except asyncio.CancelledError:
                 break
             except Exception as e:
@@ -1090,6 +1106,66 @@ async def _poll_messages(self) -> None:
 
             await asyncio.sleep(1)  # Poll interval
 
+    # ── Text debounce batching ──────────────────────────────────────
+
+    _SPLIT_THRESHOLD = 6000  # WhatsApp supports ~65K; use generous threshold
+
+    def _text_batch_key(self, event: MessageEvent) -> str:
+        """Session-scoped key for text message batching."""
+        from gateway.session import build_session_key
+        return build_session_key(
+            event.source,
+            group_sessions_per_user=self.config.extra.get("group_sessions_per_user", True),
+            thread_sessions_per_user=self.config.extra.get("thread_sessions_per_user", False),
+        )
+
+    def _enqueue_text_event(self, event: MessageEvent) -> None:
+        """Buffer a text event and reset the flush timer.
+
+        When WhatsApp delivers rapid-fire messages (e.g. forwarded
+        batches), this concatenates them and waits for a short quiet
+        period before dispatching the combined message.
+        """
+        key = self._text_batch_key(event)
+        existing = self._pending_text_batches.get(key)
+        chunk_len = len(event.text or "")
+        if existing is None:
+            event._last_chunk_len = chunk_len  # type: ignore[attr-defined]
+            self._pending_text_batches[key] = event
+        else:
+            if event.text:
+                existing.text = f"{existing.text}\n{event.text}" if existing.text else event.text
+            existing._last_chunk_len = chunk_len  # type: ignore[attr-defined]
+            if event.media_urls:
+                existing.media_urls.extend(event.media_urls)
+                existing.media_types.extend(event.media_types)
+
+        prior_task = self._pending_text_batch_tasks.get(key)
+        if prior_task and not prior_task.done():
+            prior_task.cancel()
+        self._pending_text_batch_tasks[key] = asyncio.create_task(
+            self._flush_text_batch(key)
+        )
+
+    async def _flush_text_batch(self, key: str) -> None:
+        """Wait for quiet period then dispatch aggregated text."""
+        current_task = asyncio.current_task()
+        try:
+            pending = self._pending_text_batches.get(key)
+            last_len = getattr(pending, "_last_chunk_len", 0) if pending else 0
+            if last_len >= self._SPLIT_THRESHOLD:
+                delay = self._text_batch_split_delay_seconds
+            else:
+                delay = self._text_batch_delay_seconds
+            await asyncio.sleep(delay)
+            event = self._pending_text_batches.pop(key, None)
+            if not event:
+                return
+            await self.handle_message(event)
+        finally:
+            if self._pending_text_batch_tasks.get(key) is current_task:
+                self._pending_text_batch_tasks.pop(key, None)
+
     async def _build_message_event(self, data: Dict[str, Any]) -> Optional[MessageEvent]:
         """Build a MessageEvent from bridge message data, downloading images to cache."""
         try:

gateway/run.py

@@ -52,6 +52,7 @@
 from agent.account_usage import fetch_account_usage, render_account_usage_lines
 from agent.i18n import t
 from hermes_cli.config import cfg_get
+from hermes_cli.plugins import get_plugin_manager
 
 # --- Agent cache tuning ---------------------------------------------------
 # Bounds the per-session AIAgent cache to prevent unbounded growth in
@@ -3353,6 +3354,7 @@ async def start(self) -> bool:
 
         # Discover and load event hooks
         self.hooks.discover_and_load()
+        get_plugin_manager().discover_and_load()
 
 
         # Recover background processes from checkpoint (crash recovery)

gateway/stream_consumer.py

@@ -25,6 +25,12 @@
 
 logger = logging.getLogger("gateway.stream_consumer")
 
+
+def _apply_streaming_text_hooks(text: str, event: "MessageEvent", session_id: Optional[str] = None) -> tuple[str, str]:
+    """Same as _apply_text_hooks but for streaming output."""
+    from gateway.platforms.base import _apply_text_hooks
+    return _apply_text_hooks(text, event, session_id)
+
 # Sentinel to signal the stream is complete
 _DONE = object()
 
@@ -421,6 +427,11 @@ async def run(self) -> None:
                     # here instead of letting the base gateway path send the
                     # full response again.
                     if self._accumulated:
+                        # Apply pre_gateway_text_send hooks to streamed text
+                        _stream_action, self._accumulated = _apply_streaming_text_hooks(self._accumulated, None)
+                        if _stream_action == "block":
+                            self._accumulated = ""
+                            return
                         if self._fallback_final_send:
                             await self._send_fallback_final(self._accumulated)
                         elif (

hermes_cli/plugins.py

@@ -165,6 +165,7 @@ def _install_plugin_debug_handler(force: bool = False) -> None:
     #   choice: "once" | "session" | "always" | "deny" | "timeout"
     "pre_approval_request",
     "post_approval_response",
+    "pre_gateway_text_send",  # rewrite/block outbound text before platform send
 }
 
 ENTRY_POINTS_GROUP = "hermes_agent.plugins"