fix: acquire session lock in background and cron wake paths

astrbot/core/astr_agent_tool_exec.py

@@ -50,6 +50,7 @@
 from astrbot.core.utils.astrbot_path import get_astrbot_temp_path
 from astrbot.core.utils.history_saver import persist_agent_history
 from astrbot.core.utils.image_ref_utils import is_supported_image_ref
+from astrbot.core.utils.session_lock import session_lock_manager
 from astrbot.core.utils.string_utils import normalize_and_dedupe_strings
 
 
@@ -555,65 +556,66 @@ async def _wake_main_agent_for_background_result(
             provider_settings=provider_settings,
         )
 
-        req = ProviderRequest()
-        conv = await _get_session_conv(event=cron_event, plugin_context=ctx)
-        req.conversation = conv
-        context = json.loads(conv.history)
-        if context:
-            req.contexts = context
-            context_dump = req._print_friendly_context()
-            req.contexts = []
-            req.system_prompt += (
-                "\n\nBellow is you and user previous conversation history:\n"
-                f"{context_dump}"
-            )
+        async with session_lock_manager.acquire_lock(event.unified_msg_origin):
+            req = ProviderRequest()
+            conv = await _get_session_conv(event=cron_event, plugin_context=ctx)
+            req.conversation = conv
+            context = json.loads(conv.history)
+            if context:
+                req.contexts = context
+                context_dump = req._print_friendly_context()
+                req.contexts = []
+                req.system_prompt += (
+                    "\n\nBellow is you and user previous conversation history:\n"
+                    f"{context_dump}"
+                )
 
-        bg = json.dumps(extras["background_task_result"], ensure_ascii=False)
-        req.system_prompt += BACKGROUND_TASK_RESULT_WOKE_SYSTEM_PROMPT.format(
-            background_task_result=bg
-        )
-        req.prompt = (
-            "Proceed according to your system instructions. "
-            "Output using same language as previous conversation. "
-            "If you need to deliver the result to the user immediately, "
-            "you MUST use `send_message_to_user` tool to send the message directly to the user, "
-            "otherwise the user will not see the result. "
-            "After completing your task, summarize and output your actions and results. "
-        )
-        if not req.func_tool:
-            req.func_tool = ToolSet()
-        req.func_tool.add_tool(
-            ctx.get_llm_tool_manager().get_builtin_tool(SendMessageToUserTool)
-        )
+            bg = json.dumps(extras["background_task_result"], ensure_ascii=False)
+            req.system_prompt += BACKGROUND_TASK_RESULT_WOKE_SYSTEM_PROMPT.format(
+                background_task_result=bg
+            )
+            req.prompt = (
+                "Proceed according to your system instructions. "
+                "Output using same language as previous conversation. "
+                "If you need to deliver the result to the user immediately, "
+                "you MUST use `send_message_to_user` tool to send the message directly to the user, "
+                "otherwise the user will not see the result. "
+                "After completing your task, summarize and output your actions and results. "
+            )
+            if not req.func_tool:
+                req.func_tool = ToolSet()
+            req.func_tool.add_tool(
+                ctx.get_llm_tool_manager().get_builtin_tool(SendMessageToUserTool)
+            )
 
-        result = await build_main_agent(
-            event=cron_event, plugin_context=ctx, config=config, req=req
-        )
-        if not result:
-            logger.error(f"Failed to build main agent for background task {tool_name}.")
-            return
+            result = await build_main_agent(
+                event=cron_event, plugin_context=ctx, config=config, req=req
+            )
+            if not result:
+                logger.error(
+                    f"Failed to build main agent for background task {tool_name}."
+                )
+                return
 
-        runner = result.agent_runner
-        async for _ in runner.step_until_done(30):
-            # agent will send message to user via using tools
-            pass
-        llm_resp = runner.get_final_llm_resp()
-        task_meta = extras.get("background_task_result", {})
-        summary_note = (
-            f"[BackgroundTask] {summary_name} "
-            f"(task_id={task_meta.get('task_id', task_id)}) finished. "
-            f"Result: {task_meta.get('result') or result_text or 'no content'}"
-        )
-        if llm_resp and llm_resp.completion_text:
-            summary_note += (
-                f"I finished the task, here is the result: {llm_resp.completion_text}"
+            runner = result.agent_runner
+            async for _ in runner.step_until_done(30):
+                # agent will send message to user via using tools
+                pass
+            llm_resp = runner.get_final_llm_resp()
+            task_meta = extras.get("background_task_result", {})
+            summary_note = (
+                f"[BackgroundTask] {summary_name} "
+                f"(task_id={task_meta.get('task_id', task_id)}) finished. "
+                f"Result: {task_meta.get('result') or result_text or 'no content'}"
+            )
+            if llm_resp and llm_resp.completion_text:
+                summary_note += f"I finished the task, here is the result: {llm_resp.completion_text}"
+            await persist_agent_history(
+                ctx.conversation_manager,
+                event=cron_event,
+                req=req,
+                summary_note=summary_note,
             )
-        await persist_agent_history(
-            ctx.conversation_manager,
-            event=cron_event,
-            req=req,
-            summary_note=summary_note,
-        )
         if not llm_resp:
             logger.warning("background task agent got no response")
             return

astrbot/core/cron/manager.py

@@ -19,6 +19,7 @@
 from astrbot.core.platform.message_type import MessageType
 from astrbot.core.provider.entites import ProviderRequest
 from astrbot.core.utils.history_saver import persist_agent_history
+from astrbot.core.utils.session_lock import session_lock_manager
 
 if TYPE_CHECKING:
     from astrbot.core.star.context import Context
@@ -448,66 +449,67 @@ async def _woke_main_agent(
             streaming_response=False,
             provider_settings=provider_settings,
         )
-        req = ProviderRequest()
-        conv = await _get_session_conv(event=cron_event, plugin_context=self.ctx)
-        req.conversation = conv
-        # finetine the messages
-        context = json.loads(conv.history)
-        if context:
-            req.contexts = context
-            context_dump = req._print_friendly_context()
-            req.contexts = []
-            req.system_prompt += (
-                "\n\nBellow is you and user previous conversation history:\n"
-                f"---\n"
-                f"{context_dump}\n"
-                f"---\n"
+        async with session_lock_manager.acquire_lock(umo):
+            req = ProviderRequest()
+            conv = await _get_session_conv(event=cron_event, plugin_context=self.ctx)
+            req.conversation = conv
+            # finetine the messages
+            context = json.loads(conv.history)
+            if context:
+                req.contexts = context
+                context_dump = req._print_friendly_context()
+                req.contexts = []
+                req.system_prompt += (
+                    "\n\nBellow is you and user previous conversation history:\n"
+                    f"---\n"
+                    f"{context_dump}\n"
+                    f"---\n"
+                )
+            cron_job_str = json.dumps(extras.get("cron_job", {}), ensure_ascii=False)
+            req.system_prompt += PROACTIVE_AGENT_CRON_WOKE_SYSTEM_PROMPT.format(
+                cron_job=cron_job_str
             )
-        cron_job_str = json.dumps(extras.get("cron_job", {}), ensure_ascii=False)
-        req.system_prompt += PROACTIVE_AGENT_CRON_WOKE_SYSTEM_PROMPT.format(
-            cron_job=cron_job_str
-        )
-        req.prompt = (
-            "You are now responding to a scheduled task. "
-            "Proceed according to your system instructions. "
-            "Output using same language as previous conversation. "
-            "After completing your task, summarize and output your actions and results."
-        )
-        if delivery_session_str:
-            if not req.func_tool:
-                req.func_tool = ToolSet()
-            req.func_tool.add_tool(
-                self.ctx.get_llm_tool_manager().get_builtin_tool(SendMessageToUserTool)
+            req.prompt = (
+                "You are now responding to a scheduled task. "
+                "Proceed according to your system instructions. "
+                "Output using same language as previous conversation. "
+                "After completing your task, summarize and output your actions and results."
             )
+            if delivery_session_str:
+                if not req.func_tool:
+                    req.func_tool = ToolSet()
+                req.func_tool.add_tool(
+                    self.ctx.get_llm_tool_manager().get_builtin_tool(
+                        SendMessageToUserTool
+                    )
+                )
 
-        result = await build_main_agent(
-            event=cron_event, plugin_context=self.ctx, config=config, req=req
-        )
-        if not result:
-            logger.error("Failed to build main agent for cron job.")
-            return
-
-        runner = result.agent_runner
-        async for _ in runner.step_until_done(30):
-            # agent will send message to user via using tools
-            pass
-        llm_resp = runner.get_final_llm_resp()
-        cron_meta = extras.get("cron_job", {}) if extras else {}
-        summary_note = (
-            f"[CronJob] {cron_meta.get('name') or cron_meta.get('id', 'unknown')}: {cron_meta.get('description', '')} "
-            f" triggered at {cron_meta.get('run_started_at', 'unknown time')}, "
-        )
-        if llm_resp and llm_resp.role == "assistant":
-            summary_note += (
-                f"I finished this job, here is the result: {llm_resp.completion_text}"
+            result = await build_main_agent(
+                event=cron_event, plugin_context=self.ctx, config=config, req=req
             )
+            if not result:
+                logger.error("Failed to build main agent for cron job.")
+                return
 
-        await persist_agent_history(
-            self.ctx.conversation_manager,
-            event=cron_event,
-            req=req,
-            summary_note=summary_note,
-        )
+            runner = result.agent_runner
+            async for _ in runner.step_until_done(30):
+                # agent will send message to user via using tools
+                pass
+            llm_resp = runner.get_final_llm_resp()
+            cron_meta = extras.get("cron_job", {}) if extras else {}
+            summary_note = (
+                f"[CronJob] {cron_meta.get('name') or cron_meta.get('id', 'unknown')}: {cron_meta.get('description', '')} "
+                f" triggered at {cron_meta.get('run_started_at', 'unknown time')}, "
+            )
+            if llm_resp and llm_resp.role == "assistant":
+                summary_note += f"I finished this job, here is the result: {llm_resp.completion_text}"
+
+            await persist_agent_history(
+                self.ctx.conversation_manager,
+                event=cron_event,
+                req=req,
+                summary_note=summary_note,
+            )
         if not llm_resp:
             logger.warning("Cron job agent got no response")
             return

tests/test_background_wake_lock.py

@@ -0,0 +1,92 @@
+"""
+回归测试: 后台任务唤醒路径应使用会话锁
+
+背景: 用户消息处理路径(internal.py)通过 session_lock_manager 按会话串行化,
+但后台任务唤醒路径(_wake_main_agent_for_background_result)直接跑 agent,
+未获取会话锁 —— 与用户消息并发处理时导致上下文丢失。
+
+本测试断言: 修复后, 唤醒流程必须获取会话锁 (acquire_lock 被调用)。
+修复前该测试失败, 修复后通过。
+"""
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from astrbot.core.astr_agent_tool_exec import FunctionToolExecutor
+
+
+def _make_run_context():
+    """构造最小可用的 run_context mock"""
+    event = SimpleNamespace(
+        unified_msg_origin="Pstar:FriendMessage:TEST",
+        role="friend",
+        get_extra=lambda key: None,
+    )
+    ctx = SimpleNamespace(
+        get_config=lambda umo: {},
+        get_llm_tool_manager=MagicMock(),
+        conversation_manager=MagicMock(update_conversation=AsyncMock()),
+    )
+    agent_ctx = SimpleNamespace(event=event, context=ctx)
+    return SimpleNamespace(
+        context=agent_ctx,
+        tool_call_timeout=60,
+    )
+
+
+def _make_runner_mock():
+    """构造假 agent runner: step_until_done 异步生成器"""
+    runner = MagicMock()
+
+    async def _step_until_done(*args, **kwargs):
+        yield None
+
+    runner.step_until_done.side_effect = _step_until_done
+    runner.get_final_llm_resp.return_value = SimpleNamespace(completion_text="done")
+    return runner
+
+
+@pytest.mark.asyncio
+async def test_background_wake_acquires_session_lock():
+    """后台任务唤醒必须获取会话锁(与用户消息路径一致)"""
+    run_context = _make_run_context()
+    runner = _make_runner_mock()
+
+    # 用 AsyncMock 追踪 acquire_lock 是否被调用
+    lock_mgr = MagicMock()
+    lock_cm = AsyncMock()
+    lock_cm.__aenter__.return_value = None
+    lock_mgr.acquire_lock.return_value = lock_cm
+
+    with (
+        # create=True: 当前代码尚未引入 session_lock_manager, patch 尚不存在的属性
+        patch("astrbot.core.astr_agent_tool_exec.session_lock_manager", lock_mgr, create=True),
+        # _get_session_conv / build_main_agent 在函数内部 import, 需 patch 源模块
+        patch(
+            "astrbot.core.astr_main_agent._get_session_conv",
+            new=AsyncMock(return_value=SimpleNamespace(history="[]", cid="conv-1")),
+        ),
+        patch(
+            "astrbot.core.astr_main_agent.build_main_agent",
+            new=AsyncMock(return_value=SimpleNamespace(agent_runner=runner)),
+        ),
+        patch("astrbot.core.astr_agent_tool_exec.CronMessageEvent"),
+        patch("astrbot.core.astr_agent_tool_exec.MessageSession"),
+    ):
+        await FunctionToolExecutor._wake_main_agent_for_background_result(
+            run_context,
+            task_id="task-1",
+            tool_name="transfer_to_x",
+            result_text="some result",
+            tool_args={},
+            note="background task finished",
+            summary_name="Dedicated to subagent `x`",
+        )
+
+    # 核心断言: 修复后必须获取会话锁, 且锁粒度为该会话
+    lock_mgr.acquire_lock.assert_called_once()
+    args = lock_mgr.acquire_lock.call_args[0]
+    assert "Pstar:FriendMessage:TEST" in args, (
+        f"会话锁应按 unified_msg_origin 获取, 实际参数: {args}"
+    )