fix(agent): percent-encode API context headers

app/agent/api/executor.py

@@ -12,7 +12,6 @@ from app.agent.policy.api import ApiOperationRoute, resolve_api_route
 from app.application.security.token import create_access_token
 from app.runtime.log import logger
 from app.runtime.settings import get_runtime_setting
-from app.schemas.types import NotificationChannel
 
 
 class ApiExecutionError(RuntimeError):
@@ -112,12 +111,8 @@ class MoviePilotApiExecutor:
         if self._context.session_id:
             headers["X-MoviePilot-Agent-Session"] = self._context.session_id
         if self._context.channel:
-            try:
-                channel = NotificationChannel(self._context.channel)
-                headers["X-MoviePilot-Agent-Channel"] = channel.name
-            except ValueError:
-                # 未知渠道仍需经过 URL 编码，避免自定义渠道值破坏 HTTP 头的 ASCII 约束。
-                headers["X-MoviePilot-Agent-Channel"] = quote(self._context.channel, safe="")
+            # 渠道值来自通知渠道枚举，也可能来自插件扩展；统一 URL 编码可兼容两者。
+            headers["X-MoviePilot-Agent-Channel"] = quote(self._context.channel, safe="")
         if self._context.source:
             headers["X-MoviePilot-Agent-Source"] = quote(self._context.source, safe="")
         return headers

tests/test_agent_api_executor.py

@@ -138,7 +138,7 @@ def test_executor_serializes_channel_and_source_headers_as_ascii() -> None:
     with patch("app.agent.api.executor.create_access_token", return_value="token"):
         headers = executor._build_headers()
 
-    assert headers["X-MoviePilot-Agent-Channel"] == NotificationChannel.Wechat.name
+    assert headers["X-MoviePilot-Agent-Channel"] == "%E5%BE%AE%E4%BF%A1"
     assert headers["X-MoviePilot-Agent-Channel"].isascii()
     assert headers["X-MoviePilot-Agent-Source"] == "%E4%BC%81%E4%B8%9A%E5%BE%AE%E4%BF%A1"
     assert headers["X-MoviePilot-Agent-Source"].isascii()

tests/test_agent_command_endpoint.py

@@ -1,5 +1,6 @@
 from types import SimpleNamespace
 from unittest.mock import patch
+from urllib.parse import quote
 
 import pytest
 
@@ -10,7 +11,7 @@ from app.schemas.types import NotificationChannel
 
 @pytest.mark.asyncio
 async def test_agent_command_endpoint_decodes_internal_channel_headers() -> None:
-    """命令端点应解析 ASCII 渠道名，并还原编码后的来源名称。"""
+    """命令端点应解析 ASCII 编码的渠道，并还原编码后的来源名称。"""
     user = SimpleNamespace(id=1, is_superuser=True)
     command_result = {
         "message": "命令已触发",
@@ -25,7 +26,7 @@ async def test_agent_command_endpoint_decodes_internal_channel_headers() -> None
         response = await run_agent_command(
             AgentCommandRunRequest(command="/demo"),
             current_user=user,
-            agent_channel=NotificationChannel.Wechat.name,
+            agent_channel=quote(NotificationChannel.Wechat.value, safe=""),
             agent_source="%E4%BC%81%E4%B8%9A%E5%BE%AE%E4%BF%A1",
         )