fix(ci): restore architecture and coverage gates

app/api/endpoints/agent.py

@@ -329,7 +329,7 @@ async def run_agent_command(
     except ValueError:
         channel = None
     try:
-        data = web_agent_application.dispatch_web_agent_command(
+        data = web_agent_application.dispatch_command(
             payload.command,
             user_id=str(current_user.id),
             channel=channel,

app/api/endpoints/plugin.py

@@ -949,8 +949,6 @@ def set_plugin_config(
     """
     result = command.update(plugin_id, conf)
     return _SchemaResponse(success=result.success, message=result.message)
-
-
 @router.delete("/{plugin_id}", summary="卸载插件", response_model=_SchemaResponse[None])
 def uninstall_plugin(plugin_id: str, _: ApiPrincipal = Depends(get_current_active_superuser)) -> Any:
     """

app/api/endpoints/recommend.py

@@ -1,4 +1,5 @@
-from typing import Any, Awaitable, List, Optional
+from collections.abc import Awaitable, Callable
+from typing import Any, List, Optional
 
 from fastapi import Depends, HTTPException, status
 
@@ -43,6 +44,207 @@ async def _require_tmdb_result(operation: Awaitable[List[Any]]) -> List[Any]:
         ) from error
 
 
+def _normalize_agent_media_type(media_type: str) -> Optional[str]:
+    """将 Agent 媒体类型转换为推荐链使用的稳定标识。"""
+    if media_type == "all":
+        return media_type
+    media_type_enum = MediaType.from_agent(media_type)
+    return media_type_enum.to_agent() if media_type_enum else None
+
+
+async def _fetch_listenbrainz_chart(
+    chain: RecommendChain,
+    *,
+    page: int,
+    count: int,
+    range_name: str,
+    sort_by: str,
+    min_listen_count: int,
+    with_cover: bool,
+    entity: Optional[str],
+) -> List[Any] | _SchemaResponse:
+    """校验榜单参数并获取 ListenBrainz 榜单结果。"""
+    if range_name not in LISTENBRAINZ_CHART_RANGES:
+        return _SchemaResponse(success=False, message="无效的榜单周期")
+    if sort_by not in {"listen_count.desc", "listen_count.asc"}:
+        return _SchemaResponse(success=False, message="无效的榜单排序")
+    return await chain.async_music_chart(
+        range_name=range_name,
+        page=page,
+        count=count,
+        sort_by=sort_by,
+        min_listen_count=max(0, min_listen_count),
+        with_cover=with_cover,
+        entity=entity or MUSIC_ENTITY_RECORDING,
+    )
+
+
+async def _fetch_listenbrainz_fresh(
+    chain: RecommendChain,
+    *,
+    page: int,
+    count: int,
+    music_type: Optional[str],
+    days: int,
+    fresh_sort: str,
+    past: bool,
+    future: bool,
+    with_cover: bool,
+) -> List[Any] | _SchemaResponse:
+    """校验新发行参数并获取 ListenBrainz 新发行结果。"""
+    if music_type not in {None, MUSIC_ENTITY_ALBUM}:
+        return _SchemaResponse(success=False, message="新发行结果只支持专辑")
+    if fresh_sort not in LISTENBRAINZ_FRESH_SORTS:
+        return _SchemaResponse(success=False, message="无效的新发行排序")
+    if not past and not future:
+        return _SchemaResponse(
+            success=False,
+            message="past 和 future 不能同时为 false",
+        )
+    return await chain.async_music_fresh_releases(
+        days=max(1, min(days, LISTENBRAINZ_FRESH_MAX_DAYS)),
+        sort=fresh_sort,
+        past=past,
+        future=future,
+        page=page,
+        count=count,
+        with_cover=with_cover,
+    )
+
+
+async def _recommend_listenbrainz(
+    chain: RecommendChain,
+    *,
+    source: str,
+    media_type: str,
+    page: int,
+    count: int,
+    music_type: Optional[str],
+    range_name: str,
+    sort_by: str,
+    days: int,
+    fresh_sort: str,
+    past: bool,
+    future: bool,
+    min_listen_count: int,
+    with_cover: bool,
+) -> _SchemaResponse:
+    """执行 ListenBrainz 推荐分支并统一转换音乐结果。"""
+    if media_type not in {"all", "music"}:
+        return _SchemaResponse(
+            success=False,
+            message="ListenBrainz 来源只支持音乐媒体类型",
+        )
+    normalized_music_type = (
+        normalize_music_type(music_type, allow_artist=False) if music_type else None
+    )
+    if music_type and normalized_music_type is None:
+        return _SchemaResponse(success=False, message="无效的音乐实体类型")
+    if source == "listenbrainz_chart":
+        results = await _fetch_listenbrainz_chart(
+            chain,
+            page=page,
+            count=count,
+            range_name=range_name,
+            sort_by=sort_by,
+            min_listen_count=min_listen_count,
+            with_cover=with_cover,
+            entity=normalized_music_type,
+        )
+    else:
+        results = await _fetch_listenbrainz_fresh(
+            chain,
+            page=page,
+            count=count,
+            music_type=normalized_music_type,
+            days=days,
+            fresh_sort=fresh_sort,
+            past=past,
+            future=future,
+            with_cover=with_cover,
+        )
+    if isinstance(results, _SchemaResponse):
+        return results
+    return _SchemaResponse(
+        success=True,
+        data=[simplify_music_info(item) for item in results or []],
+    )
+
+
+def _media_source_operations(
+    chain: RecommendChain,
+    *,
+    page: int,
+    count: int,
+) -> dict[str, Callable[[], Awaitable[List[Any]]]]:
+    """构造影视推荐来源到链方法的延迟调用映射。"""
+    return {
+        "tmdb_trending": lambda: chain.async_tmdb_trending(page=page),
+        "tmdb_movies": lambda: chain.async_tmdb_movies(page=page),
+        "tmdb_tvs": lambda: chain.async_tmdb_tvs(page=page),
+        "douban_movie_hot": lambda: chain.async_douban_movie_hot(page=page, count=count),
+        "douban_tv_hot": lambda: chain.async_douban_tv_hot(page=page, count=count),
+        "douban_movie_showing": lambda: chain.async_douban_movie_showing(page=page, count=count),
+        "douban_movies": lambda: chain.async_douban_movies(page=page, count=count),
+        "douban_tvs": lambda: chain.async_douban_tvs(page=page, count=count),
+        "douban_movie_top250": lambda: chain.async_douban_movie_top250(page=page, count=count),
+        "douban_tv_weekly_chinese": lambda: chain.async_douban_tv_weekly_chinese(page=page, count=count),
+        "douban_tv_weekly_global": lambda: chain.async_douban_tv_weekly_global(page=page, count=count),
+        "douban_tv_animation": lambda: chain.async_douban_tv_animation(page=page, count=count),
+        "bangumi_calendar": lambda: chain.async_bangumi_calendar(page=page, count=count),
+    }
+
+
+async def _fetch_media_recommendations(
+    chain: RecommendChain,
+    *,
+    source: str,
+    media_type: str,
+    page: int,
+    count: int,
+) -> tuple[bool, List[Any]]:
+    """获取影视推荐结果，并处理按媒体类型拆分的豆瓣热门来源。"""
+    if source == "douban_hot":
+        results: List[Any] = []
+        if media_type in {"all", "movie"}:
+            results.extend(await chain.async_douban_movie_hot(page=page, count=count))
+        if media_type in {"all", "tv"}:
+            results.extend(await chain.async_douban_tv_hot(page=page, count=count))
+        return True, results
+    operation = _media_source_operations(chain, page=page, count=count).get(source)
+    if operation is None:
+        return False, []
+    return True, await operation()
+
+
+def _project_agent_recommendations(results: List[Any], count: int) -> List[dict[str, Any]]:
+    """将推荐链结果裁剪为 Agent 稳定响应字段。"""
+    projected = []
+    for item in (results or [])[:count]:
+        if not isinstance(item, dict):
+            continue
+        projected.append(
+            {
+                "title": item.get("title"),
+                "en_title": item.get("en_title"),
+                "year": item.get("year"),
+                "type": media_type_to_agent(item.get("type")),
+                "season": item.get("season"),
+                "tmdb_id": item.get("tmdb_id"),
+                "imdb_id": item.get("imdb_id"),
+                "douban_id": item.get("douban_id"),
+                "bangumi_id": item.get("bangumi_id"),
+                "anilist_id": item.get("anilist_id"),
+                "media_source": item.get("media_source"),
+                "media_id": item.get("media_id"),
+                "vote_average": item.get("vote_average"),
+                "poster_path": item.get("poster_path"),
+                "detail_link": item.get("detail_link"),
+            }
+        )
+    return projected
+
+
 @router.get(
     "/source",
     summary="获取推荐数据源",
@@ -86,116 +288,45 @@ async def agent_recommendations(
     """按稳定来源标识返回有界影视、动画或音乐推荐结果。"""
     page = max(1, page)
     count = 20
-    if media_type != "all":
-        media_type_enum = MediaType.from_agent(media_type)
-        if media_type_enum is None:
-            return _SchemaResponse(success=False, message="无效的媒体类型")
-        media_type = media_type_enum.to_agent()
+    normalized_media_type = _normalize_agent_media_type(media_type)
+    if normalized_media_type is None:
+        return _SchemaResponse(success=False, message="无效的媒体类型")
     chain = RecommendChain()
     if source in {"listenbrainz_chart", "listenbrainz_fresh"}:
-        if media_type not in {"all", "music"}:
-            return _SchemaResponse(
-                success=False,
-                message="ListenBrainz 来源只支持音乐媒体类型",
-            )
-        normalized_music_type = normalize_music_type(music_type, allow_artist=False) if music_type else None
-        if music_type and normalized_music_type is None:
-            return _SchemaResponse(success=False, message="无效的音乐实体类型")
-        if source == "listenbrainz_chart":
-            if range_name not in LISTENBRAINZ_CHART_RANGES:
-                return _SchemaResponse(success=False, message="无效的榜单周期")
-            if sort_by not in {"listen_count.desc", "listen_count.asc"}:
-                return _SchemaResponse(success=False, message="无效的榜单排序")
-            music_results = await chain.async_music_chart(
-                range_name=range_name,
-                page=page,
-                count=count,
-                sort_by=sort_by,
-                min_listen_count=max(0, min_listen_count),
-                with_cover=with_cover,
-                entity=normalized_music_type or MUSIC_ENTITY_RECORDING,
-            )
-        else:
-            if normalized_music_type not in {None, MUSIC_ENTITY_ALBUM}:
-                return _SchemaResponse(
-                    success=False,
-                    message="新发行结果只支持专辑",
-                )
-            if fresh_sort not in LISTENBRAINZ_FRESH_SORTS:
-                return _SchemaResponse(success=False, message="无效的新发行排序")
-            if not past and not future:
-                return _SchemaResponse(
-                    success=False,
-                    message="past 和 future 不能同时为 false",
-                )
-            music_results = await chain.async_music_fresh_releases(
-                days=max(1, min(days, LISTENBRAINZ_FRESH_MAX_DAYS)),
-                sort=fresh_sort,
-                past=past,
-                future=future,
-                page=page,
-                count=count,
-                with_cover=with_cover,
-            )
-        return _SchemaResponse(
-            success=True,
-            data=[simplify_music_info(item) for item in music_results or []],
+        return await _recommend_listenbrainz(
+            chain,
+            source=source,
+            media_type=normalized_media_type,
+            page=page,
+            count=count,
+            music_type=music_type,
+            range_name=range_name,
+            sort_by=sort_by,
+            days=days,
+            fresh_sort=fresh_sort,
+            past=past,
+            future=future,
+            min_listen_count=min_listen_count,
+            with_cover=with_cover,
         )
-    if media_type == "music":
+    if normalized_media_type == "music":
         return _SchemaResponse(
             success=False,
             message="音乐推荐需使用 ListenBrainz 来源",
         )
-    source_calls = {
-        "tmdb_trending": lambda: chain.async_tmdb_trending(page=page),
-        "tmdb_movies": lambda: chain.async_tmdb_movies(page=page),
-        "tmdb_tvs": lambda: chain.async_tmdb_tvs(page=page),
-        "douban_movie_hot": lambda: chain.async_douban_movie_hot(page=page, count=count),
-        "douban_tv_hot": lambda: chain.async_douban_tv_hot(page=page, count=count),
-        "douban_movie_showing": lambda: chain.async_douban_movie_showing(page=page, count=count),
-        "douban_movies": lambda: chain.async_douban_movies(page=page, count=count),
-        "douban_tvs": lambda: chain.async_douban_tvs(page=page, count=count),
-        "douban_movie_top250": lambda: chain.async_douban_movie_top250(page=page, count=count),
-        "douban_tv_weekly_chinese": lambda: chain.async_douban_tv_weekly_chinese(page=page, count=count),
-        "douban_tv_weekly_global": lambda: chain.async_douban_tv_weekly_global(page=page, count=count),
-        "douban_tv_animation": lambda: chain.async_douban_tv_animation(page=page, count=count),
-        "bangumi_calendar": lambda: chain.async_bangumi_calendar(page=page, count=count),
-    }
-    if source == "douban_hot":
-        results = []
-        if media_type in {"all", "movie"}:
-            results.extend(await chain.async_douban_movie_hot(page=page, count=count))
-        if media_type in {"all", "tv"}:
-            results.extend(await chain.async_douban_tv_hot(page=page, count=count))
-    else:
-        operation = source_calls.get(source)
-        if operation is None:
-            return _SchemaResponse(success=False, message=f"不支持的推荐来源: {source}")
-        results = await operation()
-    projected = []
-    for item in (results or [])[:count]:
-        if not isinstance(item, dict):
-            continue
-        projected.append(
-            {
-                "title": item.get("title"),
-                "en_title": item.get("en_title"),
-                "year": item.get("year"),
-                "type": media_type_to_agent(item.get("type")),
-                "season": item.get("season"),
-                "tmdb_id": item.get("tmdb_id"),
-                "imdb_id": item.get("imdb_id"),
-                "douban_id": item.get("douban_id"),
-                "bangumi_id": item.get("bangumi_id"),
-                "anilist_id": item.get("anilist_id"),
-                "media_source": item.get("media_source"),
-                "media_id": item.get("media_id"),
-                "vote_average": item.get("vote_average"),
-                "poster_path": item.get("poster_path"),
-                "detail_link": item.get("detail_link"),
-            }
-        )
-    return _SchemaResponse(success=True, data=projected)
+    supported, results = await _fetch_media_recommendations(
+        chain,
+        source=source,
+        media_type=normalized_media_type,
+        page=page,
+        count=count,
+    )
+    if not supported:
+        return _SchemaResponse(success=False, message=f"不支持的推荐来源: {source}")
+    return _SchemaResponse(
+        success=True,
+        data=_project_agent_recommendations(results, count),
+    )
 
 
 @router.get(

app/application/messaging/agent.py

@@ -48,6 +48,7 @@ from app.runtime.tasks import get_task_registry
 from app.schemas.message import Message
 from app.schemas.types import NotificationChannel, ReplyMode
 
+__all__ = ["dispatch_command"]
 # Agent 选择按钮回调前缀（新旧两种格式都必须继续兼容）
 AGENT_CHOICE_PREFIX = "agent_interaction:choice:"
 LEGACY_AGENT_CHOICE_PREFIX = "agent_choice:"
@@ -1704,25 +1705,6 @@ def build_web_agent_command_items() -> list[dict[str, Any]]:
         )
     return sorted(items, key=lambda item: (item["category"], item["command"]))
 
-
-def dispatch_web_agent_command(
-    command: str,
-    *,
-    user_id: str,
-    channel: Optional[NotificationChannel],
-    source: Optional[str],
-    publish_event: Callable[[Any, dict[str, Any]], Any],
-) -> dict[str, Any]:
-    """经消息应用边界校验并触发一条 Agent 斜杠命令。"""
-    return dispatch_command(
-        command,
-        user_id=user_id,
-        channel=channel,
-        source=source,
-        publish_event=publish_event,
-    )
-
-
 def extract_web_agent_slash_command(text: str) -> Optional[str]:
     """
     从 WebAgent 输入中提取斜杠命令名。

docs/architecture/optimization-checklist.md

@@ -103,7 +103,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁
 | 长方法 | 290 个超过 80 行 | AST 统计排除 `app/plugins/**`；65 个超过 150 行，21 个超过 250 行 |
 | 全量 mypy 历史债务 | 9,983 / 591 文件 | canonical `SearchChain` Facade 已补齐显式类型转发；strict frontier 当前覆盖 41 个文件，低水位只允许继续下降 |
 | Ruff 历史诊断 | 630 | 低水位门禁通过，但规则集只覆盖 `E4/E7/E9/F/I` |
-| 覆盖率低水位 | Application 81.86%，Domain 81.03% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 |
+| 覆盖率低水位 | Application 81.88%，Domain 81.03% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 |
 
 ### 3.3 热点文件
 

tests/fixtures/architecture/coverage-baseline.json

@@ -1,8 +1,8 @@
 {
   "application": {
-    "covered_lines": 12996,
-    "percent": 81.86,
-    "statements": 15875
+    "covered_lines": 13687,
+    "percent": 81.88,
+    "statements": 16716
   },
   "domain": {
     "covered_lines": 3686,

tests/test_agent_application_services.py

@@ -0,0 +1,397 @@
+"""Agent 应用服务的纯逻辑、端口编排和安全投影测试。"""
+
+from contextlib import asynccontextmanager, nullcontext
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+import app.application.commands as command_application
+import app.application.filtering as filtering
+import app.application.plugin.management as plugin_management
+import app.application.settings as settings_module
+from app.application.download.tasks import DownloadTaskMutationService, DownloadTaskService
+from app.application.music.projection import simplify_music_album, simplify_music_artist, simplify_music_info
+from app.application.plugin.data import (
+    DeletePluginDataCommand,
+    PluginDataQueryService,
+    build_preview_payload,
+    clamp_preview_chars,
+)
+from app.application.security.secrets import is_secret_setting_key
+from app.domain.context import MusicAlbumInfo, MusicArtistInfo, MusicInfo, MusicRelease
+from app.domain.projection.douban import project as project_douban
+from app.schemas.rule import CustomRule, FilterRuleGroup
+from app.schemas.types import EventType, MediaSource, SystemConfigKey
+
+
+def run_async(awaitable):
+    """在普通 pytest 函数中执行一个短异步断言。"""
+    import asyncio
+    return asyncio.run(awaitable)
+
+
+def test_filtering_primitives_and_projection():
+    """过滤规则基础函数应统一输入、解析引用并隔离可变输出。"""
+    assert filtering.normalize_optional_text(None) is None
+    assert filtering.normalize_optional_text("  value  ") == "value"
+    assert filtering.normalize_media_type("movie") == "电影"
+    assert filtering.normalize_media_type("电视剧") == "电视剧"
+    assert filtering.normalize_media_type("") is None
+    with pytest.raises(ValueError, match="media_type"):
+        filtering.normalize_media_type("documentary")
+    assert filtering.validate_numeric_range("size_range", None) is None
+    assert filtering.validate_numeric_range("size_range", "1000 - 5000") == "1000 - 5000"
+    with pytest.raises(ValueError, match="格式无效"):
+        filtering.validate_numeric_range("size_range", "abc")
+    with pytest.raises(ValueError, match="起始值"):
+        filtering.validate_numeric_range("size_range", "5-2")
+    assert filtering.validate_seeders(" 12 ") == "12"
+    with pytest.raises(ValueError, match="非负整数"):
+        filtering.validate_seeders("1.2")
+    builtin = filtering.get_builtin_rules()
+    builtin["4K"]["_test"] = "changed"
+    assert "_test" not in filtering.get_builtin_rules()["4K"]
+    custom = CustomRule(id="CUSTOM", name="Custom", include="x")
+    group = FilterRuleGroup(name="main", rule_string="CUSTOM & 4K")
+    assert filtering.build_custom_rule_map([custom, CustomRule(id=None)]) == {"CUSTOM": custom}
+    assert filtering.build_rule_group_map([group, FilterRuleGroup(name=None)]) == {"main": group}
+    assert filtering.extract_rule_tokens("CUSTOM & 4K & CUSTOM") == ["CUSTOM", "4K"]
+    assert filtering.extract_rule_tokens(None) == []
+    assert filtering.parse_rule_string("CUSTOM & 4K > !BLU")["levels"][0]["priority"] == 1
+    with pytest.raises(ValueError, match="不能为空"):
+        filtering.parse_rule_string(" ")
+    with pytest.raises(ValueError, match="空层级"):
+        filtering.parse_rule_string("4K > ")
+    assert filtering.validate_rule_string("CUSTOM & 4K", ["CUSTOM", "4K"])["levels"]
+    with pytest.raises(ValueError, match="不存在"):
+        filtering.validate_rule_string("MISSING", ["4K"])
+    assert filtering.serialize_builtin_rule("4K", {"name": "4K"})["source"] == "builtin"
+    assert filtering.serialize_custom_rule(custom, ["main"])["referenced_by_rule_groups"] == ["main"]
+    assert filtering.serialize_rule_group(group)["syntax_valid"] is True
+    assert filtering.serialize_rule_group(FilterRuleGroup(name="bad", rule_string="4K > "))["syntax_valid"] is False
+    assert filtering.serialize_rule_group(FilterRuleGroup(name="empty"))["syntax_valid"] is False
+    assert filtering.replace_rule_id_in_rule_string("OLD & OLDER", "OLD", "NEW") == "NEW & OLDER"
+
+
+def test_filtering_normalizers_and_usage_collection(monkeypatch):
+    """规则实体校验应拒绝重复项，并能汇总全局与订阅引用。"""
+    rule = filtering.normalize_custom_rule("NEW", "New rule", "inc", None, "1-2", "3", "2024", [])
+    assert rule.id == "NEW"
+    with pytest.raises(ValueError, match="不能为空"):
+        filtering.normalize_custom_rule("", "New", None, None, None, None, None, [])
+    with pytest.raises(ValueError, match="不能为空"):
+        filtering.normalize_custom_rule("NEW", "", None, None, None, None, None, [])
+    with pytest.raises(ValueError, match="内置"):
+        filtering.normalize_custom_rule("4K", "Built-in", None, None, None, None, None, [])
+    with pytest.raises(ValueError, match="已存在"):
+        filtering.normalize_custom_rule("NEW", "Other", None, None, None, None, None, [rule])
+    with pytest.raises(ValueError, match="规则名称"):
+        filtering.normalize_custom_rule("OTHER", "New rule", None, None, None, None, None, [rule])
+    with pytest.raises(ValueError, match="rule_id"):
+        filtering.normalize_custom_rule("bad-id", "Bad", None, None, None, None, None, [])
+    with pytest.raises(ValueError, match="规则组名称"):
+        filtering.normalize_rule_group("", "4K", None, None, [], ["4K"])
+    with pytest.raises(ValueError, match="category"):
+        filtering.normalize_rule_group("group", "4K", None, "movie", [], ["4K"])
+    normalized_group, parsed = filtering.normalize_rule_group("group", "4K", "movie", "action", [], ["4K"])
+    assert normalized_group.media_type == "电影" and parsed["levels"]
+    with pytest.raises(ValueError, match="已存在"):
+        filtering.normalize_rule_group("group", "4K", None, None, [normalized_group], ["4K"])
+    config = MagicMock()
+    config.get.side_effect = lambda key: {
+        SystemConfigKey.SearchFilterRuleGroups: ["search"],
+        SystemConfigKey.SubscribeFilterRuleGroups: ["subscribe"],
+        SystemConfigKey.BestVersionFilterRuleGroups: ["best"],
+    }.get(key, [])
+    monkeypatch.setattr(filtering, "get_configured_system_config", lambda: config)
+
+    class SubscriptionPort:
+        """提供规则组使用统计所需的最小订阅端口。"""
+
+        async def async_list(self):
+            """返回带规则组引用的订阅快照。"""
+            return [SimpleNamespace(id=1, name="Sub", season=1, type="电影", username="u", best_version=True, filter_groups=["search", "custom"]), SimpleNamespace(filter_groups=None)]
+
+    usage = run_async(filtering.collect_rule_group_usages(SubscriptionPort(), ["search", "custom"]))
+    assert usage["search"]["used_in_global_search"] is True
+    assert usage["search"]["subscribes"][0]["subscribe_id"] == 1
+    refs = filtering.collect_custom_rule_group_refs([FilterRuleGroup(name="main", rule_string="CUSTOM & 4K"), FilterRuleGroup(name="none")], ["CUSTOM"])
+    assert refs["CUSTOM"] == ["main"]
+
+
+@pytest.mark.asyncio
+async def test_filter_rule_service_queries_and_mutations(monkeypatch):
+    """规则服务应覆盖查询、增删改和重命名引用的事务边界。"""
+    old_rule = CustomRule(id="OLD", name="Old", include="old")
+    old_group = FilterRuleGroup(name="group", rule_string="OLD & 4K")
+    monkeypatch.setattr(filtering, "get_custom_rules", lambda: [old_rule])
+    monkeypatch.setattr(filtering, "get_rule_groups", lambda: [old_group])
+    config = MagicMock()
+    config.async_set = AsyncMock(return_value=True)
+    monkeypatch.setattr(filtering, "get_configured_system_config", lambda: config)
+    publish = AsyncMock()
+    mutation = MagicMock()
+    mutation.apply = AsyncMock(return_value=SimpleNamespace(to_dict=lambda: {"changed": True}))
+
+    @asynccontextmanager
+    async def mutation_scope():
+        """提供可观测的规则组异步事务。"""
+        yield mutation
+
+    class SubscriptionPort:
+        """提供空订阅列表的查询端口。"""
+
+        async def async_list(self):
+            """返回空订阅集合。"""
+            return []
+
+    service = filtering.FilterRuleService(SubscriptionPort(), mutation_scope, publish)
+    assert filtering.FilterRuleService.query_builtin(["4K"])["count"] == 1
+    assert filtering.FilterRuleService.query_custom(["OLD"])["count"] == 1
+    assert (await service.query_groups(include_usage=False))["count"] == 1
+    assert (await service.query_groups(["missing"], include_usage=False))["count"] == 0
+    added = await service.add_custom(rule_id="NEW", name="New", include="new", exclude=None, size_range=None, seeders=None, publish_time=None)
+    assert added["custom_rule"]["id"] == "NEW"
+    updated = await service.update_custom(current_rule_id="OLD", new_rule_id="RENAMED")
+    assert updated["rule_groups_updated_for_rule_id_rename"] == ["group"]
+    monkeypatch.setattr(filtering, "get_custom_rules", lambda: [CustomRule(id="DELETE", name="Delete")])
+    monkeypatch.setattr(filtering, "get_rule_groups", lambda: [])
+    assert (await service.delete_custom("DELETE"))["count"] == 0
+    assert (await service.add_group(name="new-group", rule_string="4K"))["rule_group"]["name"] == "new-group"
+    monkeypatch.setattr(filtering, "get_rule_groups", lambda: [FilterRuleGroup(name="old", rule_string="4K")])
+    assert (await service.update_group(current_name="old", new_name="new"))["rule_group"]["name"] == "new"
+    assert (await service.delete_group("old"))["count"] == 0
+    monkeypatch.setattr(filtering, "get_rule_groups", lambda: [])
+    with pytest.raises(ValueError, match="不存在"):
+        await service.delete_group("old")
+    with pytest.raises(ValueError, match="不存在"):
+        await service.delete_custom("missing")
+
+
+@pytest.mark.asyncio
+async def test_save_system_config_and_settings_service(monkeypatch):
+    """系统设置服务应处理摘要、脱敏、合并、列表更新和两类配置源。"""
+    runtime = MagicMock()
+    runtime.get.side_effect = lambda key: {"LLM_MODEL": "model", "PLUGIN_MARKET": "a"}.get(key)
+    runtime.update.return_value = (True, "updated")
+    system = MagicMock()
+    system.get.side_effect = lambda key: [{"name": "qb", "token": "secret"}] if key == SystemConfigKey.Downloaders else {"a": 1}
+    system.async_set = AsyncMock(return_value=True)
+    publish = AsyncMock()
+    filter_config = MagicMock()
+    filter_config.async_set = AsyncMock(return_value=True)
+    monkeypatch.setattr(filtering, "get_configured_system_config", lambda: filter_config)
+    monkeypatch.setattr(settings_module, "plugin_system_config_mutation", lambda _key: nullcontext())
+    service = settings_module.SystemSettingsService(runtime, system, publish)
+    await filtering.save_system_config(SystemConfigKey.CustomFilterRules, [None, ""], publish)
+    assert filter_config.async_set.await_count == 1
+    secret = service.query(setting_key=SystemConfigKey.Downloaders.value, include_values=True)
+    assert secret["settings"][0]["value"][0]["token"] == "***"
+    shown = service.query(setting_key=SystemConfigKey.Downloaders.value, include_values=True, show_secrets=True)
+    assert shown["settings"][0]["value"][0]["token"] == "secret"
+    assert service.query(group="ai_agent")["include_values"] is False
+    spec = settings_module.resolve_setting_spec(SystemConfigKey.Downloaders.value)
+    assert spec
+    assert service._prepare_next_value(spec, {"name": "old", "x": 1}, {"name": "old", "y": 2}, "merge_dict", ["x"], None, None) == {"name": "old", "y": 2}
+    assert service._prepare_next_value(spec, [{"name": "old"}], {"name": "old", "x": 2}, "upsert_list_item", None, None, None) == [{"name": "old", "x": 2}]
+    assert service._prepare_next_value(spec, [{"name": "old"}], {"name": "old"}, "remove_list_item", None, None, None) == []
+    with pytest.raises(ValueError, match="不支持"):
+        service._prepare_next_value(spec, None, None, "bad", None, None, None)
+    system.get.side_effect = [[], [{"name": "new"}]]
+    result = await service.update(setting_key=SystemConfigKey.Downloaders.value, value={"name": "new"}, operation="upsert_list_item")
+    assert result["changed"] is True
+    runtime.get.side_effect = lambda key: "old"
+    assert (await service.update(setting_key="PLUGIN_MARKET", value="new"))["changed"] is True
+
+
+def test_settings_catalog_redaction_and_projection():
+    """设置目录应支持分类别名、匹配字段和递归敏感值脱敏。"""
+    assert settings_module.normalize_group("基础配置") == "settings"
+    assert settings_module.normalize_group("全部") == "all"
+    with pytest.raises(ValueError, match="group"):
+        settings_module.normalize_group("unknown")
+    assert settings_module.resolve_setting_spec("Downloaders") is not None
+    assert settings_module.list_setting_specs("downloaders")[0].group == "downloaders"
+    assert settings_module.list_setting_specs("ai_agent", keyword="llm")
+    assert settings_module.get_default_list_match_field(SystemConfigKey.Downloaders.value) == "name"
+    assert settings_module.redact_secret_value({"apiKey": "secret", "url": "x", "nested": ["plain"]})["apiKey"] == "***"
+    assert settings_module.normalize_group(None) == "all"
+    assert settings_module.resolve_setting_spec(None) is None
+    spec = settings_module.resolve_setting_spec(SystemConfigKey.UserSiteAuthParams.value)
+    assert spec and settings_module.should_redact_setting(spec, [{"token": "secret"}])
+    assert is_secret_setting_key("accessToken") and not is_secret_setting_key("token_count")
+
+
+@pytest.mark.asyncio
+async def test_plugin_management_and_data_services(monkeypatch):
+    """插件管理、来源补齐和数据预览应覆盖成功与安全失败路径。"""
+    plugin = SimpleNamespace(id="Demo", plugin_name="Demo Plugin", plugin_desc="desc", plugin_version="1", plugin_author="author", installed=True, has_update=True, state=True, repo_url=None, add_time=1)
+    class SourceCandidate:
+        """提供插件来源检查所需的公开投影。"""
+
+        id = "Demo"
+        plugin_name = "Demo Plugin"
+        repo_url = "https://github.com/demo/repo"
+        has_update = True
+        release = "r1"
+        system_version = "3"
+        system_version_compatible = True
+        system_version_message = None
+
+        def public_dict(self):
+            """返回来源候选的脱敏字典。"""
+            return {"id": self.id, "repo_url": self.repo_url}
+
+    source = SourceCandidate()
+    manager = MagicMock()
+    manager.get_local_plugins.return_value = [plugin]
+    manager.get_local_repo_plugins.return_value = [source]
+    manager.async_get_online_plugins = AsyncMock(return_value=[])
+    manager.process_plugins_list.return_value = [source]
+    monkeypatch.setattr(plugin_management, "get_plugin_manager", lambda: manager)
+    assert plugin_management.get_plugin_snapshot("Demo")["plugin_id"] == "Demo"
+    assert plugin_management.summarize_plugin(plugin)["source"] == "market"
+    assert plugin_management.is_exact_plugin_match(plugin, "demo plugin")
+    assert plugin_management.search_plugin_candidates("demo", [plugin])[0]["exact"] is True
+    assert plugin_management.summarize_candidates(plugin_management.search_plugin_candidates("demo", [plugin]), 1)[0]["id"] == "Demo"
+    assert await plugin_management.enrich_installed_plugin_sources([plugin]) == [plugin]
+    assert plugin.repo_url == source.repo_url
+    assert await plugin_management.load_market_plugins() == [source]
+    assert plugin_management.list_installed_plugins() == [plugin]
+    install_service = MagicMock()
+    install_service.install = AsyncMock(return_value=SimpleNamespace(success=True, message="ok", refreshed_only=False))
+    install_service.inspect_source = AsyncMock(return_value=SimpleNamespace(online_candidates=[source], local_candidate=None, selection=SimpleNamespace(status=SimpleNamespace(value="selected"), reason="exact"), inventory_complete=True))
+    monkeypatch.setattr(plugin_management, "get_plugin_install_service", lambda: install_service)
+    assert await plugin_management.install_plugin_runtime("Demo", source.repo_url) == (True, "ok", False)
+    assert (await plugin_management.inspect_plugin_sources("Demo"))["selection_status"] == "selected"
+    repo = MagicMock()
+    unit = MagicMock()
+    DeletePluginDataCommand(repo, unit).execute("Demo")
+    unit.commit.assert_called_once()
+    assert clamp_preview_chars(1) == 512
+    truncated, total, returned, preview = build_preview_payload({"value": "x" * 1000}, 512)
+    assert truncated and total > returned and "截断" in preview
+    query_repo = MagicMock()
+    query_repo.get = AsyncMock(return_value={"key": "value"})
+    query_repo.list = AsyncMock(return_value={"a": 1})
+    query = PluginDataQueryService(query_repo, lambda _id: {"plugin_id": "Demo"})
+    assert (await query.query("Demo", key="config"))["found"]
+    query_repo.get.return_value = None
+    assert not (await query.query("Demo", key="missing"))["found"]
+    assert (await query.query("Demo"))["count"] == 1
+    with pytest.raises(ValueError, match="不存在"):
+        await PluginDataQueryService(query_repo, lambda _id: None).query("Demo")
+
+
+def test_remaining_application_guard_paths(monkeypatch):
+    """应用服务的空结果、回滚和匹配参数保护应保持可观测。"""
+    helper = MagicMock()
+    helper.get_custom_rules.return_value = []
+    helper.get_rule_groups.return_value = []
+    monkeypatch.setattr(filtering, "RuleHelper", lambda: helper)
+    assert filtering.get_custom_rules() == []
+    assert filtering.get_rule_groups() == []
+    assert settings_module.SystemSettingsService._normalize_systemconfig_value([]) is None
+    with pytest.raises(ValueError, match="匹配字段"):
+        settings_module.SystemSettingsService._resolve_list_match(
+            settings_module.SettingSpec("CUSTOM", "settings", "x", "x"),
+            "upsert_list_item",
+            {"value": 1},
+            None,
+            None,
+        )
+    empty_service = DownloadTaskService(
+        MagicMock(return_value=[]),
+        lambda _hashes: {},
+        MagicMock(),
+        MagicMock(),
+        MagicMock(),
+    )
+    assert empty_service.downloading() == []
+
+
+def test_command_application_facade_and_dispatch(monkeypatch):
+    """命令应用门面应支持注册、查询、初始化和事件派发。"""
+
+    class CommandRegistry:
+        """提供命令门面所需的最小注册表实现。"""
+
+        calls = []
+
+        def get_commands(self):
+            """返回一个可触发的命令定义。"""
+            return {"/demo": {"description": "Demo", "pid": "plugin"}}
+
+        def get(self, name):
+            """按名称返回命令定义。"""
+            return self.get_commands().get(name)
+
+        def init_commands(self, plugin_id=None):
+            """记录命令初始化范围。"""
+            self.calls.append(plugin_id)
+
+    command_application.register_command_class(CommandRegistry)
+    try:
+        assert isinstance(command_application.get_command_object(), CommandRegistry)
+        assert command_application.get_commands()["/demo"]["description"] == "Demo"
+        assert command_application.get_command("/demo")["pid"] == "plugin"
+        command_application.init_commands("plugin")
+        published = []
+        result = command_application.dispatch_command(
+            "demo --flag",
+            user_id="1",
+            source="test",
+            publish_event=lambda event_type, payload: published.append((event_type, payload)),
+        )
+        assert result["command"] == "/demo --flag"
+        assert result["plugin_id"] == "plugin"
+        assert published == [
+            (
+                EventType.CommandExcute,
+                {"cmd": "/demo --flag", "user": "1", "channel": None, "source": "test"},
+            )
+        ]
+        with pytest.raises(ValueError, match="命令不能为空"):
+            command_application.dispatch_command("  ", user_id="1", publish_event=lambda *_args: None)
+        with pytest.raises(ValueError, match="不存在"):
+            command_application.dispatch_command("/missing", user_id="1", publish_event=lambda *_args: None)
+    finally:
+        command_application.reset_command_class()
+
+    with pytest.raises(RuntimeError, match="未初始化"):
+        command_application.get_command_object()
+
+
+def test_music_projection_and_domain_projection():
+    """音乐和豆瓣投影应保留稳定身份并裁剪大字段。"""
+    track = MusicInfo(media_source=MediaSource.MusicBrainz, media_id="track", title="Track", artists=["Artist"], year=2024)
+    assert simplify_music_info(track)["title"] == "Track"
+    album = MusicAlbumInfo(media_source=MediaSource.MusicBrainz, media_id="album", title="Album", artists=["Artist"], tracks=[track] * 3, releases=[MusicRelease(media_id="release", title="Release")])
+    assert simplify_music_album(album, track_limit=2)["tracks_truncated"] is True
+    artist = MusicArtistInfo(media_source=MediaSource.MusicBrainz, media_id="artist", name="Artist", raw_data={"secret": 1})
+    assert simplify_music_artist(artist)["subscribable"] is False
+    projected = project_douban({}, {"id": "1", "title": "Movie", "subtype": "movie", "rating": {"value": 8.5}, "pic": {"large": "poster"}})
+    assert projected["poster_path"] == "poster"
+    assert projected["douban_id"] == "1"
+    assert project_douban({}, {}) == {}
+
+
+def test_download_task_services_validate_and_delegate(monkeypatch):
+    """下载任务服务应补齐历史媒体并严格校验高级修改。"""
+    torrent = SimpleNamespace(hash="a" * 40, downloader="qb")
+    history = SimpleNamespace(media_source=MediaSource.TMDB, media_id="1", type="电影", title="Movie", seasons="1", episodes="2", poster="p", image="b", torrent_site="site", userid="u", username="name")
+    list_torrents = MagicMock(return_value=[torrent])
+    service = DownloadTaskService(list_torrents, lambda _hashes: {torrent.hash: history}, MagicMock(return_value=True), MagicMock(return_value=True), MagicMock(return_value=True))
+    assert service.downloading()[0].media.title == "Movie"
+    assert service.set_downloading(torrent.hash, "start") is True
+    assert service.set_downloading(torrent.hash, "bad") is False
+    assert service.remove_downloading(torrent.hash) is True
+    mutation = DownloadTaskMutationService(list_torrents=lambda **_kwargs: [torrent], set_tags=MagicMock(return_value=True), set_downloading=MagicMock(return_value=True), update_torrent=MagicMock(return_value={"limits": True, "trackers": False}))
+    assert mutation.update(hash_value=torrent.hash, action="start", tags=["tag"], download_limit=1)["results"]
+    with pytest.raises(ValueError, match="hash"):
+        mutation.update(hash_value="bad", action="start")
+    with pytest.raises(ValueError, match="至少"):
+        mutation.update(hash_value=torrent.hash)
+    with pytest.raises(ValueError, match="action"):
+        mutation.update(hash_value=torrent.hash, action="pause")