fix(agent): 工具权限拦截时补齐流式换行，渠道主ID默认管理员权限

app/agent/tools/base.py

@@ -255,6 +255,10 @@ class MoviePilotTool(BaseTool, metaclass=ABCMeta):
 
         permission_result = await self._check_permission()
         if permission_result:
+            # 工具被权限门禁拦截时，模型在调用前可能已输出一段引导文本，且这里
+            # 不会产生工具消息或统计摘要；补一个换行分隔符，避免随后的失败说明
+            # 与引导文本直接连在一起。
+            self._ensure_tool_boundary_separator()
             return permission_result
 
         # 获取工具执行提示消息
@@ -397,6 +401,21 @@ class MoviePilotTool(BaseTool, metaclass=ABCMeta):
         # 独立的新 dict，跨工具状态（例如质量门槛拒绝标记）无法传播。
         self._agent_context = {} if agent_context is None else agent_context
 
+    def _ensure_tool_boundary_separator(self) -> None:
+        """
+        在流式缓冲中为工具边界补一个换行分隔符。
+
+        工具被权限门禁等前置检查拦截时不产生工具消息或统计摘要，若模型在调用前
+        已输出文本，后续内容会直接粘在前文后面；这里保证缓冲以换行结尾，让工具
+        前后的内容分行展示。缓冲为空或已以换行结尾时无需处理。
+        """
+        if (
+            self._stream_handler
+            and self._stream_handler.is_streaming
+            and self._stream_handler.last_buffer_char not in ("", "\n")
+        ):
+            self._stream_handler.emit("\n")
+
     async def is_admin_user(self) -> bool:
         """
         判断当前工具调用者是否拥有管理员级权限。
@@ -536,7 +555,7 @@ class MoviePilotTool(BaseTool, metaclass=ABCMeta):
         """
         检查当前消息渠道身份是否具备管理员权限。
 
-        :return: 当前渠道稳定用户 ID 位于显式管理员名单时返回 True
+        :return: 当前渠道稳定用户 ID 位于显式管理员名单或等于渠道主ID时返回 True
         """
         if not self._channel or not self._source:
             return False
@@ -578,15 +597,29 @@ class MoviePilotTool(BaseTool, metaclass=ABCMeta):
 
         admin_key = admin_key_map.get(channel_type)
 
+        # 各渠道主ID对应的配置键：渠道默认接收人通常是部署者本人，
+        # 即使未配置到管理员名单也应默认拥有管理员权限。
+        primary_id_keys = {
+            "telegram": ("TELEGRAM_CHAT_ID",),
+            "feishu": ("FEISHU_OPEN_ID",),
+            "wechat": ("WECHAT_BOT_CHAT_ID",),
+            "wechatclawbot": ("WECHATCLAWBOT_DEFAULT_TARGET",),
+            "qqbot": ("QQ_OPENID",),
+        }
+        primary_keys = primary_id_keys.get(channel_type, ())
+
         try:
             configs = ServiceConfigHelper.get_notification_configs()
             for config in configs:
                 if config.name == self._source and config.config:
-                    return matches_channel_admin(
-                        config.config,
-                        admin_key,
-                        user_id_str,
-                    )
+                    if matches_channel_admin(config.config, admin_key, user_id_str):
+                        return True
+                    # 管理员名单遗漏主ID时按管理员处理
+                    if any(
+                        matches_channel_admin(config.config, key, user_id_str)
+                        for key in primary_keys
+                    ):
+                        return True
         except Exception as e:
             logger.error(f"检查权限失败: {summarize_error(e)}")
 

tests/test_agent_resource_flow_permissions.py

@@ -424,6 +424,101 @@ def test_tool_explicit_non_admin_context_does_not_fallback_to_channel_lookup():
     has_channel_admin_permission.assert_not_awaited()
 
 
+def test_channel_primary_id_defaults_to_admin_without_admin_list():
+    """渠道主ID未配置到管理员名单时仍默认为管理员。"""
+    tool = QuerySitesTool(session_id="session-1", user_id="10001")
+    tool.set_message_attr(
+        channel=MessageChannel.Telegram.value,
+        source="telegram-main",
+        username="owner",
+    )
+
+    with patch(
+        "app.agent.tools.base.ServiceConfigHelper.get_notification_configs",
+        return_value=[
+            SimpleNamespace(
+                name="telegram-main",
+                config={"TELEGRAM_CHAT_ID": "10001"},
+            )
+        ],
+    ):
+        result = asyncio.run(tool._has_channel_admin_permission())
+
+    assert result is True
+
+
+def test_channel_primary_id_mismatch_remains_non_admin():
+    """非主ID用户且不在管理员名单时不能获得管理员权限。"""
+    tool = QuerySitesTool(session_id="session-1", user_id="10002")
+    tool.set_message_attr(
+        channel=MessageChannel.Telegram.value,
+        source="telegram-main",
+        username="other",
+    )
+
+    with patch(
+        "app.agent.tools.base.ServiceConfigHelper.get_notification_configs",
+        return_value=[
+            SimpleNamespace(
+                name="telegram-main",
+                config={"TELEGRAM_CHAT_ID": "10001"},
+            )
+        ],
+    ):
+        result = asyncio.run(tool._has_channel_admin_permission())
+
+    assert result is False
+
+
+def test_feishu_primary_open_id_defaults_to_admin():
+    """飞书渠道主ID使用默认接收人 OPEN_ID 判断管理员身份。"""
+    tool = QuerySitesTool(session_id="session-1", user_id="ou_owner")
+    tool.set_message_attr(
+        channel=MessageChannel.Feishu.value,
+        source="feishu-main",
+        username="owner",
+    )
+
+    with patch(
+        "app.agent.tools.base.ServiceConfigHelper.get_notification_configs",
+        return_value=[
+            SimpleNamespace(
+                name="feishu-main",
+                config={"FEISHU_OPEN_ID": "ou_owner"},
+            )
+        ],
+    ):
+        result = asyncio.run(tool._has_channel_admin_permission())
+
+    assert result is True
+
+
+def test_channel_primary_id_still_prefers_admin_list():
+    """管理员名单命中优先于主ID兜底。"""
+    tool = QuerySitesTool(session_id="session-1", user_id="10001")
+    tool.set_message_attr(
+        channel=MessageChannel.Telegram.value,
+        source="telegram-main",
+        username="owner",
+    )
+
+    with patch(
+        "app.agent.tools.base.ServiceConfigHelper.get_notification_configs",
+        return_value=[
+            SimpleNamespace(
+                name="telegram-main",
+                config={
+                    "TELEGRAM_ADMINS": "10001",
+                    "TELEGRAM_CHAT_ID": "99999",
+                },
+            )
+        ],
+    ):
+        result = asyncio.run(tool._has_channel_admin_permission())
+
+    assert result is True
+
+
 def test_admin_tool_rejects_explicit_non_admin_without_channel_context():
     """显式非管理员事实必须拒绝管理员工具，不能走无渠道兼容放行。"""
     tool = QuerySystemSettingsTool(session_id="session-1", user_id="10002")

tests/test_agent_tool_streaming.py

@@ -58,6 +58,22 @@ class DummyTool(MoviePilotTool):
         return "ok"
 
 
+class AdminOnlyDummyTool(MoviePilotTool):
+    """仅管理员可用的工具，用于权限拦截场景的流式输出测试。"""
+
+    name: str = "admin_only_tool"
+    description: str = "Admin-only tool for streaming tests."
+    require_admin: bool = True
+
+    def get_tool_message(self, **kwargs) -> str:
+        """返回固定工具执行提示。"""
+        return "run admin only tool"
+
+    async def run(self, **kwargs) -> str:
+        """返回固定工具执行结果。"""
+        return "ok"
+
+
 class TestAgentToolStreaming:
     """Agent 工具流式输出测试。"""
 
@@ -83,6 +99,61 @@ class TestAgentToolStreaming:
         assert result == "ok"
         assert buffered_message == "prefix\n\n（调用了 1 次工具）\n\n"
 
+    def test_permission_failure_adds_boundary_newline(self):
+        """校验权限拦截时在流式缓冲中补齐工具边界换行。"""
+        async def _run():
+            tool = AdminOnlyDummyTool(session_id="session-1", user_id="10001")
+            handler = StreamingHandler()
+            await handler.start_streaming()
+            handler.emit("好的，我来帮您执行")
+            tool.set_stream_handler(handler)
+            tool.set_agent_context({"is_admin": False})
+
+            result = await tool._arun()
+            # 模拟模型在工具被拦截后继续输出失败说明
+            handler.emit("抱歉，您没有执行此工具的权限")
+            buffered_message = await handler.take()
+            return result, buffered_message
+
+        result, buffered_message = asyncio.run(_run())
+
+        assert "没有执行此工具" in result
+        assert buffered_message == "好的，我来帮您执行\n抱歉，您没有执行此工具的权限"
+
+    def test_permission_failure_skips_newline_when_buffer_empty(self):
+        """校验缓冲为空时权限拦截不会补多余的换行。"""
+        async def _run():
+            tool = AdminOnlyDummyTool(session_id="session-1", user_id="10001")
+            handler = StreamingHandler()
+            await handler.start_streaming()
+            tool.set_stream_handler(handler)
+            tool.set_agent_context({"is_admin": False})
+
+            await tool._arun()
+            return await handler.take()
+
+        buffered_message = asyncio.run(_run())
+
+        assert buffered_message == ""
+
+    def test_permission_failure_reuses_existing_newline(self):
+        """校验缓冲已以换行结尾时权限拦截不再补换行。"""
+        async def _run():
+            tool = AdminOnlyDummyTool(session_id="session-1", user_id="10001")
+            handler = StreamingHandler()
+            await handler.start_streaming()
+            handler.emit("好的，我来帮您执行\n")
+            tool.set_stream_handler(handler)
+            tool.set_agent_context({"is_admin": False})
+
+            await tool._arun()
+            handler.emit("抱歉，您没有执行此工具的权限")
+            return await handler.take()
+
+        buffered_message = asyncio.run(_run())
+
+        assert buffered_message == "好的，我来帮您执行\n抱歉，您没有执行此工具的权限"
+
     def test_non_verbose_tool_call_reuses_existing_newline_before_summary(self):
         """校验非详细模式复用已有换行追加工具摘要。"""
         result, buffered_message = asyncio.run(self._run_tool("prefix\n"))