fix: acquire session lock in cron wake path

The cron wake path (_woke_main_agent) also bypassed session_lock_manager,
running concurrently with user message processing on the same conversation.
Wrap the read-build-run-persist flow in the per-session lock, matching
the user message path in internal.py.

- test_cron_wake_lock: asserts acquire_lock is called

Author: T52T52

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_cron_wake_lock.py

@@ -0,0 +1,87 @@
+"""
+回归测试: cron 定时任务唤醒路径也应使用会话锁
+
+背景: 与后台任务唤醒(_wake_main_agent_for_background_result)相同,
+cron 定时任务触发时直接跑 agent, 未获取会话锁 —— 与用户消息并发
+处理时可能导致上下文丢失。
+
+本测试断言: 修复后, cron 唤醒流程必须获取会话锁 (acquire_lock 被调用)。
+"""
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from astrbot.core.cron.manager import CronJobManager
+
+
+def _make_manager():
+    """构造最小可用的 CronJobManager mock"""
+    ctx = SimpleNamespace(
+        get_config=lambda umo: {"admins_id": [], "provider_settings": {}},
+        get_llm_tool_manager=MagicMock(),
+        conversation_manager=MagicMock(update_conversation=AsyncMock()),
+    )
+    mgr = CronJobManager.__new__(CronJobManager)  # 跳过 __init__, 只设 ctx
+    mgr.ctx = ctx
+    return mgr
+
+
+def _make_runner_mock():
+    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", role="assistant"
+    )
+    return runner
+
+
+@pytest.mark.asyncio
+async def test_cron_wake_acquires_session_lock():
+    """cron 定时任务唤醒必须获取会话锁(与用户消息/后台唤醒路径一致)"""
+    mgr = _make_manager()
+    runner = _make_runner_mock()
+
+    lock_mgr = MagicMock()
+    lock_cm = AsyncMock()
+    lock_cm.__aenter__.return_value = None
+    lock_mgr.acquire_lock.return_value = lock_cm
+
+    fake_cron_event = SimpleNamespace(
+        unified_msg_origin="Pstar:FriendMessage:TEST", role="member"
+    )
+
+    with (
+        patch(
+            "astrbot.core.cron.manager.session_lock_manager", lock_mgr, create=True
+        ),
+        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.cron.manager.CronMessageEvent",
+            return_value=fake_cron_event,
+        ),
+        # MessageSession 需为真实类型(函数内 isinstance 判断), from_str 可解析字符串
+    ):
+        await mgr._woke_main_agent(
+            message="test cron job",
+            session_str="Pstar:FriendMessage:TEST",
+            extras={"cron_job": {"id": "job-1", "name": "t", "run_started_at": "t"}},
+        )
+
+    # 核心断言: 修复后必须获取会话锁, 且锁粒度为该会话
+    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}"
+    )