refactor(startup): centralize chain network composition

app/startup/composition/network.py

@@ -1,10 +1,12 @@
 """网络应用端口的宿主组合根。"""
 
 from collections.abc import Mapping
-from typing import Any, Callable, Optional, cast
+from pathlib import Path
+from typing import Any, Callable, Optional, Union, cast
 
 from app.adapters.network.http import AsyncRequestUtils, RequestUtils
 from app.adapters.network.ip import IpUtils
+from app.adapters.system.host import SystemUtils
 from app.application.configuration import get_runtime_settings
 from app.application.image import (
     ImageResponsePort,
@@ -25,6 +27,33 @@ from app.application.network import (
     configure_network_test_service,
     reset_network_test_service,
 )
+from app.chain.download.ports import (
+    DownloadArchivePort,
+    DownloadHttpPort,
+    DownloadResponsePort,
+    configure_download_ports,
+    reset_download_ports,
+)
+from app.chain.message import (
+    MessageHttpPort,
+    MessageResponsePort,
+    configure_message_http_port,
+    reset_message_http_port,
+)
+from app.chain.scraping import (
+    ScrapingHttpPort,
+    ScrapingResponsePort,
+    ScrapingStreamResponsePort,
+    configure_scraping_http_port,
+    reset_scraping_http_port,
+)
+from app.chain.system import (
+    SystemEnvironmentPort,
+    SystemHttpPort,
+    SystemResponsePort,
+    configure_system_ports,
+    reset_system_ports,
+)
 from app.runtime.log import logger
 
 
@@ -140,6 +169,154 @@ class _MessageIngressAdapter:
                 logger.debug(f"释放本地消息入口响应失败：{error}")
 
 
+class _DownloadHttpAdapter:
+    """把 RequestUtils 收窄为下载链同步 HTTP 端口。"""
+
+    @staticmethod
+    def _request(
+        *,
+        cookies: Optional[Union[str, dict[str, str]]],
+        ua: Optional[str],
+        headers: Optional[dict[str, str]],
+        proxies: Optional[dict[str, str]],
+        timeout: Optional[int],
+    ) -> RequestUtils:
+        """按下载链传入的代理、认证与超时参数构造一次请求。"""
+        options: dict[str, Any] = {
+            "cookies": cookies,
+            "ua": ua,
+            "headers": headers,
+            "proxies": proxies,
+            "timeout": timeout,
+        }
+        return RequestUtils(**options)
+
+    def get(
+        self,
+        url: str,
+        *,
+        cookies: Optional[Union[str, dict[str, str]]] = None,
+        ua: Optional[str] = None,
+        headers: Optional[dict[str, str]] = None,
+        proxies: Optional[dict[str, str]] = None,
+        timeout: Optional[int] = None,
+        params: Optional[dict[str, Any]] = None,
+        raise_exception: bool = False,
+    ) -> Optional[DownloadResponsePort]:
+        """发送下载链 GET 请求并原样保留响应三态。"""
+        request = self._request(
+            cookies=cookies, ua=ua, headers=headers, proxies=proxies, timeout=timeout
+        )
+        kwargs: dict[str, Any] = {"raise_exception": raise_exception}
+        if params is not None:
+            kwargs["params"] = params
+        response = request.get_res(url, **kwargs)
+        return cast(Optional[DownloadResponsePort], response)
+
+    def post(
+        self,
+        url: str,
+        *,
+        cookies: Optional[Union[str, dict[str, str]]] = None,
+        ua: Optional[str] = None,
+        headers: Optional[dict[str, str]] = None,
+        proxies: Optional[dict[str, str]] = None,
+        timeout: Optional[int] = None,
+        params: Optional[dict[str, Any]] = None,
+    ) -> Optional[DownloadResponsePort]:
+        """发送下载链 POST 请求并原样保留响应三态。"""
+        request = self._request(
+            cookies=cookies, ua=ua, headers=headers, proxies=proxies, timeout=timeout
+        )
+        kwargs: dict[str, Any] = {}
+        if params is not None:
+            kwargs["params"] = params
+        response = request.post_res(url, **kwargs)
+        return cast(Optional[DownloadResponsePort], response)
+
+
+class _DownloadArchiveAdapter:
+    """把 SystemUtils 收窄为下载链字幕归档端口。"""
+
+    def unpack(
+        self,
+        archive_file: Path,
+        extract_dir: Path,
+        *,
+        archive_format: Optional[str],
+    ) -> None:
+        """使用宿主统一归档实现解压字幕文件。"""
+        SystemUtils.unpack_archive(
+            archive_file, extract_dir, archive_format=archive_format
+        )
+
+    def list_files(self, directory: Path, extensions: tuple[str, ...]) -> list[Path]:
+        """使用宿主统一文件扫描实现列出字幕文件。"""
+        return SystemUtils.list_files(directory, list(extensions))
+
+
+class _MessageHttpAdapter:
+    """把 RequestUtils 收窄为消息附件同步 GET 端口。"""
+
+    def get(self, url: str, *, timeout: int) -> Optional[MessageResponsePort]:
+        """按消息链固定超时读取附件响应。"""
+        response = RequestUtils(timeout=timeout).get_res(url)
+        return cast(Optional[MessageResponsePort], response)
+
+
+class _ScrapingHttpAdapter:
+    """把 RequestUtils 收窄为刮削链普通与流式 GET 端口。"""
+
+    def get(
+        self,
+        url: str,
+        *,
+        proxies: Optional[dict[str, str]],
+        ua: str,
+        timeout: int,
+    ) -> Optional[ScrapingResponsePort]:
+        """读取需要完整载荷的音乐封面响应。"""
+        options: dict[str, Any] = {"proxies": proxies, "ua": ua, "timeout": timeout}
+        response = RequestUtils(**options).get_res(url)
+        return cast(Optional[ScrapingResponsePort], response)
+
+    def stream(
+        self,
+        url: str,
+        *,
+        proxies: Optional[dict[str, str]],
+        ua: str,
+    ) -> ScrapingStreamResponsePort:
+        """打开由刮削链上下文关闭的流式图片响应。"""
+        options: dict[str, Any] = {"proxies": proxies, "ua": ua}
+        response = RequestUtils(**options).get_stream(url=url)
+        return cast(ScrapingStreamResponsePort, response)
+
+
+class _SystemHttpAdapter:
+    """把 RequestUtils 收窄为系统链发布版本 GET 端口。"""
+
+    def get(
+        self,
+        url: str,
+        *,
+        proxies: Optional[dict[str, str]],
+        headers: Mapping[str, str],
+    ) -> Optional[SystemResponsePort]:
+        """按系统配置的代理与 GitHub 请求头读取发布列表。"""
+        options: dict[str, Any] = {"proxies": proxies, "headers": dict(headers)}
+        response = RequestUtils(**options).get_res(url)
+        return cast(Optional[SystemResponsePort], response)
+
+
+class _SystemEnvironmentAdapter:
+    """把 SystemUtils 收窄为系统链容器环境判断端口。"""
+
+    def is_docker(self) -> bool:
+        """返回宿主统一环境探针的 Docker 判断。"""
+        return bool(SystemUtils.is_docker())
+
+
 def configure_application_network_ports() -> None:
     """装配网络探测、图片读取、内部地址判断和消息回环传输端口。"""
     network_test_transport: NetworkTestTransport = _NetworkTestTransportAdapter()
@@ -165,3 +342,30 @@ def reset_application_network_ports() -> None:
     reset_message_ingress_port()
     reset_image_ports()
     reset_network_test_service()
+
+
+def configure_chain_network_composition() -> None:
+    """原子装配四个 Chain 的同步网络与系统技术端口。"""
+    reset_chain_network_composition()
+    try:
+        configure_download_ports(
+            http=cast(DownloadHttpPort, _DownloadHttpAdapter()),
+            archive=cast(DownloadArchivePort, _DownloadArchiveAdapter()),
+        )
+        configure_message_http_port(cast(MessageHttpPort, _MessageHttpAdapter()))
+        configure_scraping_http_port(cast(ScrapingHttpPort, _ScrapingHttpAdapter()))
+        configure_system_ports(
+            http=cast(SystemHttpPort, _SystemHttpAdapter()),
+            environment=cast(SystemEnvironmentPort, _SystemEnvironmentAdapter()),
+        )
+    except Exception:
+        reset_chain_network_composition()
+        raise
+
+
+def reset_chain_network_composition() -> None:
+    """清除四个 Chain 的技术端口，支持重复 lifespan 与失败回滚。"""
+    reset_system_ports()
+    reset_scraping_http_port()
+    reset_message_http_port()
+    reset_download_ports()

app/startup/initializers/network.py

@@ -1,227 +1,16 @@
-"""装配 Chain 使用的同步网络与相关系统窄端口。"""
+"""Chain 网络端口的生命周期入口。"""
 
-from collections.abc import Mapping
-from pathlib import Path
-from typing import Any, Optional, Union, cast
-
-from app.adapters.network.http import RequestUtils
-from app.adapters.system.host import SystemUtils
-from app.chain.download.ports import (
-    DownloadArchivePort,
-    DownloadHttpPort,
-    DownloadResponsePort,
-    configure_download_ports,
-    reset_download_ports,
+from app.startup.composition.network import (
+    configure_chain_network_composition,
+    reset_chain_network_composition,
 )
-from app.chain.message import (
-    MessageHttpPort,
-    MessageResponsePort,
-    configure_message_http_port,
-    reset_message_http_port,
-)
-from app.chain.scraping import (
-    ScrapingHttpPort,
-    ScrapingResponsePort,
-    ScrapingStreamResponsePort,
-    configure_scraping_http_port,
-    reset_scraping_http_port,
-)
-from app.chain.system import (
-    SystemEnvironmentPort,
-    SystemHttpPort,
-    SystemResponsePort,
-    configure_system_ports,
-    reset_system_ports,
-)
-
-
-class _DownloadHttpAdapter:
-    """把 RequestUtils 收窄为下载链同步 HTTP 端口。"""
-
-    @staticmethod
-    def _request(
-        *,
-        cookies: Optional[Union[str, dict[str, str]]],
-        ua: Optional[str],
-        headers: Optional[dict[str, str]],
-        proxies: Optional[dict[str, str]],
-        timeout: Optional[int],
-    ) -> RequestUtils:
-        """按下载链传入的代理、认证与超时参数构造一次请求。"""
-        options: dict[str, Any] = {
-            "cookies": cookies,
-            "ua": ua,
-            "headers": headers,
-            "proxies": proxies,
-            "timeout": timeout,
-        }
-        return RequestUtils(**options)
-
-    def get(
-        self,
-        url: str,
-        *,
-        cookies: Optional[Union[str, dict[str, str]]] = None,
-        ua: Optional[str] = None,
-        headers: Optional[dict[str, str]] = None,
-        proxies: Optional[dict[str, str]] = None,
-        timeout: Optional[int] = None,
-        params: Optional[dict[str, Any]] = None,
-        raise_exception: bool = False,
-    ) -> Optional[DownloadResponsePort]:
-        """发送下载链 GET 请求并原样保留响应三态。"""
-        request = self._request(
-            cookies=cookies,
-            ua=ua,
-            headers=headers,
-            proxies=proxies,
-            timeout=timeout,
-        )
-        kwargs: dict[str, Any] = {"raise_exception": raise_exception}
-        if params is not None:
-            kwargs["params"] = params
-        response = request.get_res(url, **kwargs)
-        return cast(Optional[DownloadResponsePort], response)
-
-    def post(
-        self,
-        url: str,
-        *,
-        cookies: Optional[Union[str, dict[str, str]]] = None,
-        ua: Optional[str] = None,
-        headers: Optional[dict[str, str]] = None,
-        proxies: Optional[dict[str, str]] = None,
-        timeout: Optional[int] = None,
-        params: Optional[dict[str, Any]] = None,
-    ) -> Optional[DownloadResponsePort]:
-        """发送下载链 POST 请求并原样保留响应三态。"""
-        request = self._request(
-            cookies=cookies,
-            ua=ua,
-            headers=headers,
-            proxies=proxies,
-            timeout=timeout,
-        )
-        kwargs: dict[str, Any] = {}
-        if params is not None:
-            kwargs["params"] = params
-        response = request.post_res(url, **kwargs)
-        return cast(Optional[DownloadResponsePort], response)
-
-
-class _DownloadArchiveAdapter:
-    """把 SystemUtils 收窄为下载链字幕归档端口。"""
-
-    def unpack(
-        self,
-        archive_file: Path,
-        extract_dir: Path,
-        *,
-        archive_format: Optional[str],
-    ) -> None:
-        """使用宿主统一归档实现解压字幕文件。"""
-        SystemUtils.unpack_archive(
-            archive_file,
-            extract_dir,
-            archive_format=archive_format,
-        )
-
-    def list_files(self, directory: Path, extensions: tuple[str, ...]) -> list[Path]:
-        """使用宿主统一文件扫描实现列出字幕文件。"""
-        return SystemUtils.list_files(directory, list(extensions))
-
-
-class _MessageHttpAdapter:
-    """把 RequestUtils 收窄为消息附件同步 GET 端口。"""
-
-    def get(self, url: str, *, timeout: int) -> Optional[MessageResponsePort]:
-        """按消息链固定超时读取附件响应。"""
-        response = RequestUtils(timeout=timeout).get_res(url)
-        return cast(Optional[MessageResponsePort], response)
-
-
-class _ScrapingHttpAdapter:
-    """把 RequestUtils 收窄为刮削链普通与流式 GET 端口。"""
-
-    def get(
-        self,
-        url: str,
-        *,
-        proxies: Optional[dict[str, str]],
-        ua: str,
-        timeout: int,
-    ) -> Optional[ScrapingResponsePort]:
-        """读取需要完整载荷的音乐封面响应。"""
-        options: dict[str, Any] = {
-            "proxies": proxies,
-            "ua": ua,
-            "timeout": timeout,
-        }
-        response = RequestUtils(**options).get_res(url)
-        return cast(Optional[ScrapingResponsePort], response)
-
-    def stream(
-        self,
-        url: str,
-        *,
-        proxies: Optional[dict[str, str]],
-        ua: str,
-    ) -> ScrapingStreamResponsePort:
-        """打开由刮削链上下文关闭的流式图片响应。"""
-        options: dict[str, Any] = {"proxies": proxies, "ua": ua}
-        response = RequestUtils(**options).get_stream(url=url)
-        return cast(ScrapingStreamResponsePort, response)
-
-
-class _SystemHttpAdapter:
-    """把 RequestUtils 收窄为系统链发布版本 GET 端口。"""
-
-    def get(
-        self,
-        url: str,
-        *,
-        proxies: Optional[dict[str, str]],
-        headers: Mapping[str, str],
-    ) -> Optional[SystemResponsePort]:
-        """按系统配置的代理与 GitHub 请求头读取发布列表。"""
-        options: dict[str, Any] = {
-            "proxies": proxies,
-            "headers": dict(headers),
-        }
-        response = RequestUtils(**options).get_res(url)
-        return cast(Optional[SystemResponsePort], response)
-
-
-class _SystemEnvironmentAdapter:
-    """把 SystemUtils 收窄为系统链容器环境判断端口。"""
-
-    def is_docker(self) -> bool:
-        """返回宿主统一环境探针的 Docker 判断。"""
-        return bool(SystemUtils.is_docker())
 
 
 def init_chain_network_ports() -> None:
-    """原子式装配四个 Chain 的六条 Adapter 静态依赖边。"""
-    reset_chain_network_ports()
-    try:
-        configure_download_ports(
-            http=cast(DownloadHttpPort, _DownloadHttpAdapter()),
-            archive=cast(DownloadArchivePort, _DownloadArchiveAdapter()),
-        )
-        configure_message_http_port(cast(MessageHttpPort, _MessageHttpAdapter()))
-        configure_scraping_http_port(cast(ScrapingHttpPort, _ScrapingHttpAdapter()))
-        configure_system_ports(
-            http=cast(SystemHttpPort, _SystemHttpAdapter()),
-            environment=cast(SystemEnvironmentPort, _SystemEnvironmentAdapter()),
-        )
-    except Exception:
-        reset_chain_network_ports()
-        raise
+    """委托组合根装配 Chain 同步网络与系统端口。"""
+    configure_chain_network_composition()
 
 
 def reset_chain_network_ports() -> None:
-    """清除四个 Chain 的技术端口，支持重复 lifespan 与失败回滚。"""
-    reset_system_ports()
-    reset_scraping_http_port()
-    reset_message_http_port()
-    reset_download_ports()
+    """委托组合根释放 Chain 同步网络与系统端口。"""
+    reset_chain_network_composition()

docs/rules/05-architecture.md

@@ -155,10 +155,10 @@ wallpaper-provider publication. `initializers/modules.py` passes the already con
 `startup/composition/security.py` owns authentication, user lookup, PassKey and Web
 access provider assembly; it returns the persistence factories needed by the runtime
 owner, which alone constructs `AuthenticationRuntime`.
-`startup/composition/network.py` owns concrete network-test, image, internal-address
-and message-ingress Adapter wiring
- while retaining lazy RuntimeSettings reads. Initializers call these owners in
- startup order and must not recreate their concrete construction.
+`startup/composition/network.py` owns concrete network-test, image, internal-address,
+message-ingress and Chain synchronous network/system Adapter wiring while retaining
+lazy RuntimeSettings reads. Initializers call these owners in startup order and must
+not recreate their concrete construction.
 `startup/composition/domain.py` owns DNS/System/Rust Adapter assembly together
 with the single `RecognitionRuleService` used to publish media-recognition rule
 providers. `initializers/domain.py` is only the lifecycle hook and must not

tests/test_architecture_dependencies.py

@@ -2463,6 +2463,44 @@ def test_domain_initializer_delegates_construction_to_composition():
     assert "app.application.recognition" in composition_imports
 
 
+def test_network_initializer_delegates_construction_to_composition():
+    """网络 initializer 只保留生命周期入口，不得重新拥有具体技术适配器。"""
+    initializer_path = APP_ROOT / "startup" / "initializers" / "network.py"
+    initializer_tree = ast.parse(
+        initializer_path.read_text(encoding="utf-8-sig"),
+        filename=str(initializer_path),
+    )
+    imported_modules = {
+        node.module
+        for node in ast.walk(initializer_tree)
+        if isinstance(node, ast.ImportFrom) and node.module
+    }
+    assert imported_modules == {"app.startup.composition.network"}
+    calls = {
+        ast.unparse(node.func)
+        for node in ast.walk(initializer_tree)
+        if isinstance(node, ast.Call)
+    }
+    assert calls == {
+        "configure_chain_network_composition",
+        "reset_chain_network_composition",
+    }
+
+    composition_path = APP_ROOT / "startup" / "composition" / "network.py"
+    composition_tree = ast.parse(
+        composition_path.read_text(encoding="utf-8-sig"),
+        filename=str(composition_path),
+    )
+    composition_imports = {
+        node.module
+        for node in ast.walk(composition_tree)
+        if isinstance(node, ast.ImportFrom) and node.module
+    }
+    assert "app.adapters.network.http" in composition_imports
+    assert "app.adapters.system.host" in composition_imports
+    assert "app.chain.download.ports" in composition_imports
+
+
 PROCESS_LEVEL_ROOTS = (
     "app.api",
     "app.chain",

tests/test_chain_network_ports.py

@@ -9,6 +9,7 @@ import app.chain.download.ports as download_ports
 from app.chain import message as message_module
 from app.chain import scraping as scraping_module
 from app.chain import system as system_module
+from app.startup.composition import network as network_composition
 from app.startup.initializers import network as network_initializer
 
 
@@ -105,7 +106,7 @@ def test_initializer_supports_repeated_init_and_reset() -> None:
 def test_initializer_rolls_back_partial_configuration(monkeypatch) -> None:
     """任一端口装配异常后不得留下可用的半套运行时。"""
     monkeypatch.setattr(
-        network_initializer,
+        network_composition,
         "configure_scraping_http_port",
         Mock(side_effect=RuntimeError("boom")),
     )