feat: refactor agent runner, enhance follow-up, skill manager and history saver

- Remove empty completion retry logic from ToolLoopAgentRunner
- Enhance astr_main_agent with additional context handling
- Extend follow_up pipeline stage with new processing logic
- Add new capabilities to skill_manager
- Improve history_saver utility

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: AstrBot Local

astrbot/core/agent/runners/tool_loop_agent_runner.py

@@ -279,7 +279,10 @@ async def _iter_llm_responses_with_fallback(
                         candidate_id,
                     )
                     try:
-                        from astrbot.core.agent.context.truncator import ContextTruncator
+                        from astrbot.core.agent.context.truncator import (
+                            ContextTruncator,
+                        )
+
                         _truncator = ContextTruncator()
                         _before_total = len(self.run_context.messages)
                         # Aggressively halve until small enough (up to 5 rounds)
@@ -292,21 +295,27 @@ async def _iter_llm_responses_with_fallback(
                             _after = len(self.run_context.messages)
                             logger.info(
                                 "Forced context truncation round %d: %d -> %d messages.",
-                                _halve_round + 1, _before, _after,
+                                _halve_round + 1,
+                                _before,
+                                _after,
                             )
                             if _after <= 4:
                                 break  # Can't shrink further
                             # Retry same candidate
                             try:
-                                async for resp in self._iter_llm_responses(include_model=idx == 0):
+                                async for resp in self._iter_llm_responses(
+                                    include_model=idx == 0
+                                ):
                                     if resp.is_chunk:
                                         has_stream_output = True
                                         yield resp
                                         continue
                                     yield resp
                                     logger.info(
                                         "Context truncation succeeded after %d round(s): %d -> %d messages.",
-                                        _halve_round + 1, _before_total, _after,
+                                        _halve_round + 1,
+                                        _before_total,
+                                        _after,
                                     )
                                     _compression_success = True
                                     return
@@ -327,13 +336,15 @@ async def _iter_llm_responses_with_fallback(
                                     last_exception = retry_exc
                                     logger.warning(
                                         "Chat Model %s retry after compression failed: %s",
-                                        candidate_id, retry_exc,
+                                        candidate_id,
+                                        retry_exc,
                                     )
                                     break
                                 last_exception = retry_exc
                                 logger.warning(
                                     "Chat Model %s still token-exceeded after round %d, halving again...",
-                                    candidate_id, _halve_round + 1,
+                                    candidate_id,
+                                    _halve_round + 1,
                                 )
                                 continue
                     except Exception as compress_exc:
@@ -375,15 +386,25 @@ def follow_up(
         *,
         message_text: str,
     ) -> FollowUpTicket | None:
-        """Queue a follow-up message for the next tool result."""
+        """Queue a follow-up message to be injected into the next tool result.
+
+        Returns None if the agent is already done (message arrived too late) or
+        if the message text is empty.
+        """
         if self.done():
+            logger.debug("follow_up: agent already done, message discarded.")
             return None
         text = (message_text or "").strip()
         if not text:
             return None
         ticket = FollowUpTicket(seq=self._follow_up_seq, text=text)
         self._follow_up_seq += 1
         self._pending_follow_ups.append(ticket)
+        logger.debug(
+            "follow_up: queued ticket seq=%d, pending=%d",
+            ticket.seq,
+            len(self._pending_follow_ups),
+        )
         return ticket
 
     def _resolve_unconsumed_follow_ups(self) -> None:
@@ -402,14 +423,16 @@ def _consume_follow_up_notice(self) -> str:
         for ticket in follow_ups:
             ticket.consumed = True
             ticket.resolved.set()
+
         follow_up_lines = "\n".join(
             f"{idx}. {ticket.text}" for idx, ticket in enumerate(follow_ups, start=1)
         )
+        count = len(follow_ups)
+        plural = "messages" if count > 1 else "message"
         return (
-            "\n\n[SYSTEM NOTICE] User sent follow-up messages while tool execution "
-            "was in progress. Prioritize these follow-up instructions in your next "
-            "actions. In your very next action, briefly acknowledge to the user "
-            "that their follow-up message(s) were received before continuing.\n"
+            f"\n\n[FOLLOW-UP x{count}] The user sent {count} {plural} while you were working. "
+            "Incorporate them as supplementary instructions seamlessly — "
+            "do NOT stop, restart, or explicitly acknowledge receipt; just act naturally.\n"
             f"{follow_up_lines}"
         )
 
@@ -714,7 +737,7 @@ async def step_until_done(
             self.run_context.messages.append(
                 Message(
                     role="user",
-                    content="工具调用次数已达到上限，请停止使用工具，并根据已经收集到的信息，对你的任务和发现进行总结，然后直接回复用户。",
+                    content="工具调用次数已达到上限，请停止使用工具，并根据已经收集到的信息，对你的任务和发现进行总结，然后直接回复用户。(Tool call limit reached. Stop using tools and summarize your findings directly for the user.)",
                 )
             )
             # 再执行最后一步

astrbot/core/astr_main_agent.py

@@ -59,7 +59,11 @@
 from astrbot.core.platform.astr_message_event import AstrMessageEvent
 from astrbot.core.provider import Provider
 from astrbot.core.provider.entities import ProviderRequest
-from astrbot.core.skills.skill_manager import SkillManager, build_skills_prompt
+from astrbot.core.skills.skill_manager import (
+    SkillManager,
+    build_skills_prompt,
+    get_skills_fingerprint,
+)
 from astrbot.core.star.context import Context
 from astrbot.core.star.star_handler import star_map
 from astrbot.core.tools.cron_tools import (
@@ -353,6 +357,37 @@ async def _ensure_persona_and_skills(
                     "You cannot use shell or Python to perform skills. "
                     "If you need to use these capabilities, ask the user to enable Computer Use in the AstrBot WebUI -> Config."
                 )
+
+    # --- Dynamic skill update detection (lightweight) ---
+    # Compute current fingerprint using only stat() calls (no file I/O).
+    # If the fingerprint changed since the last turn (stored on the event extra),
+    # inject a one-line system reminder into extra_user_content_parts so the LLM
+    # is aware that the skill list may have changed.  This avoids rebuilding the
+    # full system prompt mid-conversation (which many providers ignore anyway).
+    current_fp = get_skills_fingerprint()
+    prev_fp = event.get_extra("_skills_fp")
+    if prev_fp is not None and prev_fp != current_fp:
+        # Skills changed since the last request in this session — notify the LLM.
+        skill_names = [s.name for s in skills]
+        skills_list_str = ", ".join(skill_names) if skill_names else "(none)"
+        req.extra_user_content_parts.append(
+            TextPart(
+                text=(
+                    "<system_reminder>"
+                    "The available skill list has been updated since the last turn. "
+                    f"Current active skills: {skills_list_str}. "
+                    "Please refer to this updated list for any skill-related requests."
+                    "</system_reminder>"
+                )
+            )
+        )
+        logger.debug(
+            "Skills fingerprint changed (%s -> %s), injected update reminder.",
+            prev_fp,
+            current_fp,
+        )
+    event.set_extra("_skills_fp", current_fp)
+    # --- end dynamic skill update detection ---
     tmgr = plugin_context.get_llm_tool_manager()
 
     # inject toolset in the persona

astrbot/core/pipeline/process_stage/follow_up.py

@@ -41,6 +41,28 @@ def register_active_runner(umo: str, runner: AgentRunner) -> None:
 def unregister_active_runner(umo: str, runner: AgentRunner) -> None:
     if _ACTIVE_AGENT_RUNNERS.get(umo) is runner:
         _ACTIVE_AGENT_RUNNERS.pop(umo, None)
+        # Best-effort cleanup: if no follow-up state is pending any more, drop the
+        # UMO entry to avoid accumulating stale entries when sessions end abnormally.
+        state = _FOLLOW_UP_ORDER_STATE.get(umo)
+        if state is not None:
+            statuses = state.get("statuses")
+            if not statuses:
+                _FOLLOW_UP_ORDER_STATE.pop(umo, None)
+            else:
+                # There are still pending/active entries — notify the condition so
+                # any waiter in _activate_and_wait_follow_up_turn can re-check and
+                # potentially hit the timeout branch.
+                condition = state.get("condition")
+                if isinstance(condition, asyncio.Condition):
+                    async def _notify_condition(cond: asyncio.Condition) -> None:
+                        async with cond:
+                            cond.notify_all()
+                    try:
+                        loop = asyncio.get_event_loop()
+                        if loop.is_running():
+                            loop.create_task(_notify_condition(condition))
+                    except RuntimeError:
+                        pass
 
 
 def _get_follow_up_order_state(umo: str) -> dict[str, object]:
@@ -108,6 +130,10 @@ async def _mark_follow_up_consumed(umo: str, seq: int) -> None:
             _FOLLOW_UP_ORDER_STATE.pop(umo, None)
 
 
+_FOLLOW_UP_WAIT_TIMEOUT: float = 30.0
+"""Max seconds a follow-up turn will wait for its predecessor to finish."""
+
+
 async def _activate_and_wait_follow_up_turn(umo: str, seq: int) -> None:
     state = _FOLLOW_UP_ORDER_STATE.get(umo)
     if not state:
@@ -121,12 +147,27 @@ async def _activate_and_wait_follow_up_turn(umo: str, seq: int) -> None:
             statuses[seq] = "active"
 
         # Strict ordering: only the head (`next_turn`) can continue.
+        # Use a timeout to guard against predecessor runner crashes.
+        deadline = asyncio.get_event_loop().time() + _FOLLOW_UP_WAIT_TIMEOUT
         while True:
             next_turn = state["next_turn"]
             assert isinstance(next_turn, int)
             if next_turn == seq:
                 break
-            await condition.wait()
+            remaining = deadline - asyncio.get_event_loop().time()
+            if remaining <= 0:
+                # Predecessor never finished; forcibly advance to avoid permanent hang.
+                logger.warning(
+                    "Follow-up wait timeout for umo=%s seq=%s; advancing turn to prevent hang.",
+                    umo,
+                    seq,
+                )
+                state["next_turn"] = seq
+                break
+            try:
+                await asyncio.wait_for(condition.wait(), timeout=remaining)
+            except asyncio.TimeoutError:
+                pass  # re-check on next iteration
 
 
 async def _finish_follow_up_turn(umo: str, seq: int) -> None:
@@ -152,8 +193,19 @@ async def _monitor_follow_up_ticket(
     ticket: FollowUpTicket,
     order_seq: int,
 ) -> None:
-    """Advance consumed slots immediately on resolution to avoid wake-order drift."""
+    """Advance consumed slots immediately on resolution to avoid wake-order drift.
+
+    Only marks the order slot consumed here when *ticket* was consumed by the
+    runner (i.e. injected into a tool-result).  If the ticket was *not* consumed
+    (i.e. the runner finished without ever flushing pending follow-ups),
+    ``prepare_follow_up_capture`` handles the mark via its own branch, so we
+    must not double-call ``_mark_follow_up_consumed`` here.
+    """
     await ticket.resolved.wait()
+    # Guard: only act when consumed=True AND prepare_follow_up_capture has not
+    # already handled this seq (it sets consumed_marked and calls us via the
+    # captured branch).  The state dict check inside _mark_follow_up_consumed
+    # is idempotent, so a double-call is safe, but we skip it when not needed.
     if ticket.consumed:
         await _mark_follow_up_consumed(umo, order_seq)
 

astrbot/core/skills/skill_manager.py

@@ -539,3 +539,44 @@ def install_skill_from_zip(self, zip_path: str, *, overwrite: bool = True) -> st
 
         self.set_skill_active(skill_name, True)
         return skill_name
+
+
+# ---------------------------------------------------------------------------
+# Lightweight skill-set fingerprint for dynamic update detection
+# ---------------------------------------------------------------------------
+
+def get_skills_fingerprint(skills_root: str | None = None) -> str:
+    """Return a lightweight fingerprint of the current skill set.
+
+    Uses only os.stat() calls (no file I/O) so it is extremely cheap.
+    The fingerprint changes whenever:
+      - skills.json is modified (new skill installed / toggled)
+      - a new SKILL.md appears or disappears under skills_root
+    """
+    import hashlib
+
+    data_path = Path(get_astrbot_data_path())
+    config_path = str(data_path / SKILLS_CONFIG_FILENAME)
+    root = Path(skills_root or get_astrbot_skills_path())
+
+    parts: list[str] = []
+
+    # 1. skills.json mtime (catches installs / enable-disable)
+    try:
+        parts.append(str(os.path.getmtime(config_path)))
+    except OSError:
+        parts.append("no-config")
+
+    # 2. Count of skill dirs that have SKILL.md (catches new folders)
+    try:
+        count = sum(
+            1
+            for e in root.iterdir()
+            if e.is_dir() and (e / "SKILL.md").exists()
+        )
+        parts.append(str(count))
+    except OSError:
+        parts.append("0")
+
+    raw = "|".join(parts)
+    return hashlib.md5(raw.encode()).hexdigest()[:12]

astrbot/core/utils/history_saver.py

@@ -13,19 +13,39 @@ async def persist_agent_history(
     req: ProviderRequest,
     summary_note: str,
 ) -> None:
-    """Persist agent interaction into conversation history."""
+    """Persist agent interaction into conversation history.
+
+    Fix: always re-fetch the latest history from DB before appending, to
+    avoid the race-condition where a CronJob snapshot overwrites newer
+    user-bot turns that were saved while the cron was running.
+    """
     if not req or not req.conversation:
         return
 
-    history = []
+    cid = req.conversation.cid
+    umo = event.unified_msg_origin
+
+    # Re-fetch the freshest history from DB (avoids stale-snapshot overwrite)
+    history: list[dict] = []
     try:
-        history = json.loads(req.conversation.history or "[]")
+        fresh_conv = await conversation_manager.get_conversation(umo, cid)
+        if fresh_conv:
+            history = json.loads(fresh_conv.history or "[]")
+        else:
+            # Fallback: use the snapshot embedded in req (better than nothing)
+            logger.warning(
+                "persist_agent_history: could not re-fetch conversation %s, "
+                "falling back to snapshot history.",
+                cid,
+            )
+            history = json.loads(req.conversation.history or "[]")
     except Exception as exc:  # noqa: BLE001
-        logger.warning("Failed to parse conversation history: %s", exc)
+        logger.warning("Failed to load fresh conversation history: %s", exc)
+
     history.append({"role": "user", "content": "Output your last task result below."})
     history.append({"role": "assistant", "content": summary_note})
     await conversation_manager.update_conversation(
-        event.unified_msg_origin,
-        req.conversation.cid,
+        umo,
+        cid,
         history=history,
     )