refactor: unify llm provider runtime access

tests/fixtures/architecture/dependency-baseline.json

@@ -14,7 +14,7 @@
     "workflow_to_db": []
   },
   "edge_count": 6549,
-  "edge_sha256": "280799d1a7d3a993096834b5e0f961fcc82319d437b4d5f568c78d8a7bfa07a2",
+  "edge_sha256": "fb3c6b77623cc0ad1707fe8cdfc7e8a1c6be429a34da05b4fca5ed8ef512e734",
   "edges": [
     "app -> app.runtime",
     "app -> app.runtime.compat",
@@ -1869,7 +1869,7 @@
     "app.api.endpoints.history -> app.schemas.token",
     "app.api.endpoints.llm -> app.agent",
     "app.api.endpoints.llm -> app.agent.llm",
-    "app.api.endpoints.llm -> app.agent.llm.provider",
+    "app.api.endpoints.llm -> app.agent.llm.gateway",
     "app.api.endpoints.llm -> app.api",
     "app.api.endpoints.llm -> app.api.dependencies",
     "app.api.endpoints.llm -> app.api.dependencies.auth",

tests/test_architecture_dependencies.py

@@ -1177,6 +1177,30 @@ def test_host_consumers_get_agent_manager_through_application_facade():
     assert violations == {}
 
 
+def test_host_consumers_resolve_llm_provider_runtime_through_gateway():
+    """宿主不得绕过 gateway 或 LLM 公共导出穿透 provider 实现。"""
+    violations: dict[str, set[str]] = {}
+    graph = _build_module_graph()
+    for module_name, path in _discover_modules().items():
+        if module_name.startswith(("app.agent", "app.startup")):
+            continue
+        tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+        imported = {
+            alias.name
+            for node in ast.walk(tree)
+            if isinstance(node, ast.ImportFrom)
+            and node.module in {"app.agent.llm", "app.agent.llm.provider"}
+            for alias in node.names
+            if alias.name == "LLMProviderManager"
+        }
+        forbidden = set(imported)
+        if "app.agent.llm.provider" in graph[module_name]:
+            forbidden.add("app.agent.llm.provider")
+        if forbidden:
+            violations[module_name] = forbidden
+    assert violations == {}
+
+
 def test_agent_tools_do_not_import_entrypoint_internals():
     """Agent 工具不得穿透导入 HTTP 端点、调度器与命令注册表内部实现。
 

tests/test_llm_provider_manage.py

@@ -13,6 +13,7 @@ from unittest.mock import AsyncMock, patch
 import pytest
 
 from app import schemas
+from app.agent.llm.gateway import register_llm_provider_runtime
 from app.agent.llm.helper import LLMHelper
 from app.agent.llm.provider import LLMProviderManager
 from app.runtime.config import settings
@@ -281,6 +282,87 @@ def test_llm_manage_endpoint_accepts_empty_target(monkeypatch):
     assert "callback_url" not in captured["params"]
 
 
+def test_llm_manage_endpoint_uses_registered_provider_runtime():
+    """管理端点必须使用组合根登记的 runtime，不得自行构造 provider Singleton。"""
+    captured = {}
+
+    class ProviderRuntime:
+        """记录端点调用的最小 provider runtime。"""
+
+        async def provider_manage(self, provider, action, **params):
+            """记录统一管理参数并返回可识别结果。"""
+            captured.update(provider=provider, action=action, params=params)
+            return {"success": True, "message": "", "data": {"runtime": "registered"}}
+
+    from app.api.endpoints import llm as llm_endpoint
+
+    previous = register_llm_provider_runtime(ProviderRuntime)
+    try:
+        request = SimpleNamespace(url_for=lambda *_args, **_kwargs: "unused")
+        payload = schemas.ManageRequest(target="", action="list_providers")
+
+        response = asyncio.run(llm_endpoint.manage_provider(request, payload, _="token"))
+    finally:
+        register_llm_provider_runtime(previous)
+
+    assert response.data == {"runtime": "registered"}
+    assert captured == {
+        "provider": "",
+        "action": "list_providers",
+        "params": {},
+    }
+
+
+def test_llm_oauth_callback_uses_registered_provider_runtime():
+    """OAuth 回调必须与管理端点复用组合根登记的 provider runtime。"""
+    captured = {}
+
+    class ProviderRuntime:
+        """记录 OAuth 回调参数的最小 provider runtime。"""
+
+        async def handle_chatgpt_callback(
+            self,
+            provider_id,
+            code,
+            state,
+            error,
+            error_description,
+        ):
+            """记录回调并返回可渲染的成功结果。"""
+            captured.update(
+                provider_id=provider_id,
+                code=code,
+                state=state,
+                error=error,
+                error_description=error_description,
+            )
+            return True, "registered runtime"
+
+    from app.api.endpoints import llm as llm_endpoint
+
+    previous = register_llm_provider_runtime(ProviderRuntime)
+    try:
+        response = asyncio.run(
+            llm_endpoint.llm_provider_auth_callback(
+                provider_id="chatgpt",
+                code="oauth-code",
+                state="oauth-state",
+            )
+        )
+    finally:
+        register_llm_provider_runtime(previous)
+
+    assert response.status_code == 200
+    assert b"registered runtime" in response.body
+    assert captured == {
+        "provider_id": "chatgpt",
+        "code": "oauth-code",
+        "state": "oauth-state",
+        "error": None,
+        "error_description": None,
+    }
+
+
 def test_llm_manage_endpoint_response_model_accepts_list_data():
     """目录查询动作 data 为列表，响应模型须同时覆盖列表与映射形态。