fix: enable plugin hooks for active_agent cron jobs by dispatching through PipelineScheduler

[kaixinyujue]

关联 Issue

Fixes #7419

Modifications / 改动点

此 PR 修复了 active_agent 类型的定时任务（cron job）绕过 PipelineScheduler 标准处理流程的问题，导致插件钩子（plugin hooks）无法被触发。

问题背景：

在之前的实现中，active_agent 类型的定时任务直接调用了 _woke_main_agent() 方法，使 Agent 在隔离环境中执行，未经过标准的 PipelineScheduler 管道。这导致以下问题：

• 插件事件处理器未被调用（如 on_llm_response、on_decorating_result 等钩子失效）
• 消息中的表情符号（如 &&happy&& &&sad&&）无法被 meme_manager 等插件解析和渲染为表情图片
• 结果装饰阶段（result decoration）被跳过，影响消息格式化和后处理

影响范围：

此问题影响了所有依赖插件钩子的定时任务场景，特别是需要表情包转译、消息装饰等功能的插件无法在定时触发时正常工作。

---

修改文件：

1. core/cron/events.py

   • 将 PlatformMetadata.name 从硬编码的 "cron" 改为 session.platform_name
   • 目的： 确保 unified_msg_origin 与原始平台格式一致，使插件能够正确查找配置和执行权限检查

2. core/cron/manager.py

   • 在 start() 方法中初始化事件队列：self._event_queue: Queue = ctx.get_event_queue()
   • 在 _run_active_agent_job() 中将直接调用 _woke_main_agent() 替换为 _dispatch_to_pipeline()
   • 新增 _dispatch_to_pipeline() 方法：
     • 创建包含正确 session 和元数据的 CronMessageEvent
     • 通过 unified_msg_origin 查询配置以确定用户角色
     • 将事件放入事件队列，由 PipelineScheduler 进行完整的管道处理

核心变更逻辑：

# Before: 直接调用，绕过管道
await self._woke_main_agent(session, prompt)

# After: 通过标准管道分发
await self._dispatch_to_pipeline(session, prompt, cron_config)

Screenshots or Test Results / 运行截图或测试结果

如图所示，修复后的定时任务可以正常触发表情包了：

验证项：

• 消息中的图片标记（如 &&happy&&）被 meme_manager 插件正确解析为表情图片
• 表情图片正常发送，渲染结果符合预期

---

Checklist / 检查清单

• 👀 My changes have been well-tested, and "Verification Steps" and "Screenshots" have been provided above.
  / 我的更改经过了良好的测试，并已在上方提供了“验证步骤”和“运行截图”。

• 🤓 I have ensured that no new dependencies are introduced, OR if new dependencies are introduced, they have been added to the appropriate locations in requirements.txt and pyproject.toml.
  / 我确保没有引入新依赖库，或者引入了新依赖库的同时将其添加到 requirements.txt 和 pyproject.toml 文件相应位置。

• 😮 My changes do not introduce malicious code.
  / 我的更改没有引入恶意代码。

Summary by Sourcery

Route active_agent cron jobs through the standard pipeline so they participate in normal event handling and role resolution.

Bug Fixes:

• Fix plugin hooks not firing for active_agent cron jobs by dispatching their messages through the PipelineScheduler event queue instead of invoking the agent directly.

Enhancements:

• Align cron message platform metadata with the underlying session platform so plugins can correctly recognize and handle cron-originated messages.

[sourcery-ai]

Hey - I've found 2 issues, and left some high level feedback:

• In _dispatch_to_pipeline, consider guarding against self._event_queue not being initialized (e.g., if start() hasn’t been called yet) and failing fast with a clear log or exception instead of relying on an implicit attribute error.
• The role resolution logic in _dispatch_to_pipeline (admins_id, origin == 'api') looks duplicated from other parts of the system; if there is an existing helper or central place for this, reusing it would reduce the chance of role/permission drift between cron and non-cron flows.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- In `_dispatch_to_pipeline`, consider guarding against `self._event_queue` not being initialized (e.g., if `start()` hasn’t been called yet) and failing fast with a clear log or exception instead of relying on an implicit attribute error.
- The role resolution logic in `_dispatch_to_pipeline` (`admins_id`, `origin == 'api'`) looks duplicated from other parts of the system; if there is an existing helper or central place for this, reusing it would reduce the chance of role/permission drift between cron and non-cron flows.

## Individual Comments

### Comment 1
<location path="astrbot/core/cron/manager.py" line_range="398-403" />
<code_context>
             logger.warning("Cron job agent got no response")
             return

+    async def _dispatch_to_pipeline(
+        self,
+        *,
+        message: str,
+        session_str: str,
+        extras: dict,
+    ) -> None:
+        """将定时任务消息放入事件队列，使其经过完整的 PipelineScheduler 流程。"""
</code_context>
<issue_to_address>
**suggestion:** Align parameter naming/typing with the ability to accept `MessageSession` objects.

The parameter is typed and named as a `str`, but the implementation also accepts `MessageSession` (`isinstance(session_str, MessageSession)`). Please rename it (e.g. `session` or `session_or_str`) and change the annotation to a union type so the signature matches actual usage and is clear to callers and type checkers.

Suggested implementation:

```python
        await self._dispatch_to_pipeline(
            message=note,
            session=session_str,
            extras=extras,
        )
        logger.warning("Cron job agent got no response")
        return

```

```python
    async def _dispatch_to_pipeline(
        self,
        *,
        message: str,
        session: "MessageSession | str",
        extras: dict,
    ) -> None:
        """将定时任务消息放入事件队列，使其经过完整的 PipelineScheduler 流程。"""
        from astrbot.core.cron.events import CronMessageEvent
        from astrbot.core.platform.message_session import MessageSession

        try:
            session = (
                session
                if isinstance(session, MessageSession)
                else MessageSession.from_str(session)
            )
        except Exception as e:

```

1. If your project is not yet on Python 3.10+ or not using `from __future__ import annotations`, adjust the type annotation to `session: "MessageSession" | str` or use `Union["MessageSession", str]` and ensure `Union` is imported from `typing`.
2. If there are other call sites of `_dispatch_to_pipeline` in this file or elsewhere, update them to use the renamed keyword argument `session=` instead of `session_str=`.
</issue_to_address>

### Comment 2
<location path="astrbot/core/cron/manager.py" line_range="440-441" />
<code_context>
+
+        # 将事件放入事件队列，由 EventBus 调度到 PipelineScheduler
+        await self._event_queue.put(cron_event)
+        logger.debug(
+            f"Cron job {extras.get('cron_job', {}).get('id')} dispatched to pipeline."
+        )
+
</code_context>
<issue_to_address>
**issue (bug_risk):** Avoid `AttributeError` when logging if `extras` is `None`.

Above you already guard against `extras` being `None` when building `cron_payload`. Here the log line calls `extras.get(...)` unconditionally, which will raise if `extras` is `None`. Please either use a local `safe_extras = extras or {}` (and reuse it for both `cron_payload` and this log), or add the same `if extras else {}` pattern here.
</issue_to_address>

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}

[sourcery-ai]

suggestion: Align parameter naming/typing with the ability to accept MessageSession objects.

The parameter is typed and named as a str, but the implementation also accepts MessageSession (isinstance(session_str, MessageSession)). Please rename it (e.g. session or session_or_str) and change the annotation to a union type so the signature matches actual usage and is clear to callers and type checkers.

Suggested implementation:

        await self._dispatch_to_pipeline(
            message=note,
            session=session_str,
            extras=extras,
        )
        logger.warning("Cron job agent got no response")
        return

    async def _dispatch_to_pipeline(
        self,
        *,
        message: str,
        session: "MessageSession | str",
        extras: dict,
    ) -> None:
        """将定时任务消息放入事件队列，使其经过完整的 PipelineScheduler 流程。"""
        from astrbot.core.cron.events import CronMessageEvent
        from astrbot.core.platform.message_session import MessageSession

        try:
            session = (
                session
                if isinstance(session, MessageSession)
                else MessageSession.from_str(session)
            )
        except Exception as e:

1. If your project is not yet on Python 3.10+ or not using from __future__ import annotations, adjust the type annotation to session: "MessageSession" | str or use Union["MessageSession", str] and ensure Union is imported from typing.
2. If there are other call sites of _dispatch_to_pipeline in this file or elsewhere, update them to use the renamed keyword argument session= instead of session_str=.

[sourcery-ai]

issue (bug_risk): Avoid AttributeError when logging if extras is None.

Above you already guard against extras being None when building cron_payload. Here the log line calls extras.get(...) unconditionally, which will raise if extras is None. Please either use a local safe_extras = extras or {} (and reuse it for both cron_payload and this log), or add the same if extras else {} pattern here.

[gemini-code-assist]

Code Review

This pull request refactors the cron job execution flow to use the system's event pipeline instead of a direct agent call. Key changes include updating CronMessageEvent to use dynamic platform names, initializing an event queue in CronJobManager, and introducing _dispatch_to_pipeline to handle event routing. Review feedback suggests optimizing the new method by removing redundant local imports, improving type hinting, and ensuring that the pipeline stages correctly handle cron-specific logic like system prompts and history persistence. Additionally, it was recommended to remove the now-obsolete _woke_main_agent method.

[gemini-code-assist]

此方法有以下改进建议：

1. 冗余导入：CronMessageEvent 和 MessageSession 已经在文件顶部导入，无需在此处进行局部导入。
2. 参数命名与类型提示：建议将 session_str 更名为 session，并明确类型提示为 str | MessageSession。
3. 逻辑缺失风险：原有的 _woke_main_agent 方法包含特定的系统提示词（PROACTIVE_AGENT_CRON_WOKE_SYSTEM_PROMPT） and 历史记录持久化逻辑。通过管道分发后，请确保 Pipeline 中的 Stage（如 MainAgentStage）能够识别 CronMessageEvent 并应用这些特定的逻辑，否则可能会导致 Agent 失去 cron 任务的上下文或历史记录无法正确保存。
4. 清理死代码：_woke_main_agent 方法现在已不再被内部调用，建议在确认逻辑已成功迁移到 Pipeline 后将其删除。

    async def _dispatch_to_pipeline(
        self,
        *,
        message: str,
        session: str | MessageSession,
        extras: dict[str, Any],
    ) -> None:
        """将定时任务消息放入事件队列，使其经过完整的 PipelineScheduler 流程。"""

        try:
            session_obj = (
                session
                if isinstance(session, MessageSession)
                else MessageSession.from_str(session)
            )
        except Exception as e:
            logger.error(f"Invalid session for cron job: {e}")
            return

        cron_event = CronMessageEvent(
            context=self.ctx,
            session=session_obj,
            message=message,
            extras=extras or {},
            message_type=session_obj.message_type,
        )

        # 判断用户角色
        umo = cron_event.unified_msg_origin
        cfg = self.ctx.get_config(umo=umo)
        cron_payload = extras.get("cron_payload", {})
        sender_id = cron_payload.get("sender_id")
        admin_ids = cfg.get("admins_id", [])
        if admin_ids:
            cron_event.role = "admin" if sender_id in admin_ids else "member"
        if cron_payload.get("origin", "tool") == "api":
            cron_event.role = "admin"

        # 将事件放入事件队列，由 EventBus 调度到 PipelineScheduler
        await self._event_queue.put(cron_event)
        logger.debug(
            f"Cron job {extras.get('cron_job', {}).get('id')} dispatched to pipeline."
        )

[kaixinyujue]

gemini好强哇（。＾▽＾）

[kaixinyujue]

为什么我按照gemini的建议，在_dispatch_to_pipeline中把原有_woke_main_agent的逻辑补上之后，QQ机器人又没法正常转译定时任务的表情包了┭┮﹏┭┮

[kaixinyujue]

我知道问题在哪儿了，不是没法转译，是QQ机器人消息发重了，第二条含_woke_main_agent原始逻辑的消息没有转译

[kaixinyujue]

针对gemini-code-assist所提出的改进建议进行了修改

1. 建议1和4已修改

2. 建议2没有必要改，原来的代码用的就是session_str，保留历史沿革

3. 建议3经过分析，用新的_dispatch_to_pipeline代替原来的_woke_main_agent后，并没有逻辑缺失

替代方案的完整性分析

1. 系统提示词处理

• 原逻辑：_woke_main_agent() 直接注入 PROACTIVE_AGENT_CRON_WOKE_SYSTEM_PROMPT
• 新逻辑：CronMessageEvent 在构造时已携带 additional_prompt=PROACTIVE_AGENT_CRON_WOKE_SYSTEM_PROMPT
• 结论：系统提示词通过事件对象传递，MainAgentStage 处理时会正常读取并应用，无需特殊路径

2. 历史记录持久化

• 原逻辑：显式调用 persist_agent_history() 保存记录
• 新逻辑：PipelineScheduler 在 agent_runner 完成后自动执行 _save_to_history()
  • 代码路径：astrbot/core/pipeline/process_stage/method/agent_sub_stages/internal.py:362-369
• 结论：持久化逻辑已内置于标准流程，显式调用反而可能导致重复保存

3. 事件上下文完整性

• CronMessageEvent 继承自 MessageEvent，包含完整的会话上下文（session、platform_name 等）
• unified_msg_origin 使用 session.platform_name 而非固定 "cron"，确保配置查找和权限判断正常工作（但是我不太确定会不会有其他插件真的依赖这个固定的 "cron" ）

4. 验证结果

• ✅ on_llm_response 钩子正常触发（日志确认）
• ✅ on_decorating_result 钩子正常触发（表情包成功发送）
• ✅ 历史记录自动保存（PipelineScheduler 内置逻辑）

结论

_dispatch_to_pipeline() 并非简单绕过原有逻辑，而是将定时任务纳入标准处理流程。所有必要的上下文、提示词、持久化操作均通过 PipelineScheduler 的既有机制完成，代码更简洁且维护成本更低。

[gemini-code-assist]

感谢你对该问题的深入分析和详细的逻辑验证。既然你已经确认 PipelineScheduler 的内置机制（如 _save_to_history）能够覆盖原有的显式调用，且系统提示词通过 CronMessageEvent 正确传递，那么将定时任务纳入标准管道确实是更优雅且易于维护的方案。

关于你提到的 unified_msg_origin 使用 session.platform_name 的潜在风险：只要确保在 PipelineScheduler 处理过程中，插件能够通过该平台名称正确识别上下文并执行权限校验，这种做法是符合设计预期的。如果后续发现有特定插件确实强依赖于硬编码的 "cron" 标识，我们届时再考虑通过 extras 传递额外元数据作为兼容性补充。

目前的修改逻辑清晰，且通过将任务分发至事件队列，成功解决了插件钩子无法触发的问题，这正是该 PR 的核心目标。请继续保持对定时任务在不同平台下表现的观察。