diff --git a/app/adapters/cache/backends.py b/app/adapters/cache/backends.py index fa824c820..f9337000e 100644 --- a/app/adapters/cache/backends.py +++ b/app/adapters/cache/backends.py @@ -1,13 +1,12 @@ import shutil import tempfile from pathlib import Path -from typing import Any, AsyncGenerator, Generator, Optional, Tuple +from typing import Any, AsyncGenerator, Callable, Generator, Optional, Tuple import aiofiles import aioshutil from anyio import Path as AsyncPath -from app.adapters.cache.redis import AsyncRedisHelper, RedisHelper from app.runtime.cache import ( DEFAULT_CACHE_REGION, AsyncCacheBackend, @@ -18,166 +17,6 @@ from app.runtime.cache import ( from app.runtime.settings import get_runtime_setting -class RedisBackend(AtomicCacheBackend): - """通过同步 Redis 客户端实现缓存后端。""" - - def __init__(self, ttl: Optional[int] = None) -> None: - """初始化 Redis 缓存并保存默认 TTL。""" - self.ttl = ttl - self.redis_helper = RedisHelper() - - def set( - self, - key: str, - value: Any, - ttl: Optional[int] = None, - region: Optional[str] = DEFAULT_CACHE_REGION, - **kwargs, - ) -> None: - """写入缓存,非正 TTL 视为立即删除。""" - ttl = self.ttl if ttl is None else ttl - if ttl is not None and ttl <= 0: - self.redis_helper.delete(key, region=region) - return - self.redis_helper.set(key, value, ttl=ttl, region=region, **kwargs) - - def store( - self, - key: str, - value: Any, - ttl: Optional[int] = None, - region: Optional[str] = DEFAULT_CACHE_REGION, - **kwargs: Any, - ) -> None: - """严格写入 Redis,供安全敏感的一次性状态使用。""" - ttl = self.ttl if ttl is None else ttl - if ttl is not None and ttl <= 0: - self.redis_helper.consume(key, region=region) - return - self.redis_helper.store(key, value, ttl=ttl, region=region, **kwargs) - - def consume( - self, - key: str, - region: Optional[str] = DEFAULT_CACHE_REGION, - ) -> Optional[Any]: - """通过 Redis 原子命令严格领取一个缓存值。""" - return self.redis_helper.consume(key, region=region) - - def exists( - self, - key: str, - region: Optional[str] = DEFAULT_CACHE_REGION, - ) -> bool: - """判断缓存键是否存在。""" - return self.redis_helper.exists(key, region=region) - - def get( - self, - key: str, - region: Optional[str] = DEFAULT_CACHE_REGION, - ) -> Optional[Any]: - """读取缓存值,不存在时返回空值。""" - return self.redis_helper.get(key, region=region) - - def delete( - self, - key: str, - region: Optional[str] = DEFAULT_CACHE_REGION, - ) -> None: - """删除缓存键。""" - self.redis_helper.delete(key, region=region) - - def clear(self, region: Optional[str] = DEFAULT_CACHE_REGION) -> None: - """清空指定缓存区或全部缓存。""" - self.redis_helper.clear(region=region) - - def items( - self, - region: Optional[str] = DEFAULT_CACHE_REGION, - ) -> Generator[Tuple[str, Any], None, None]: - """遍历指定缓存区的键值对。""" - return self.redis_helper.items(region=region) - - def close(self) -> None: - """关闭同步 Redis 连接池。""" - self.redis_helper.close() - - @staticmethod - def is_redis() -> bool: - """标记当前后端为 Redis。""" - return True - - -class AsyncRedisBackend(AsyncCacheBackend): - """通过异步 Redis 客户端实现缓存后端。""" - - def __init__(self, ttl: Optional[int] = None) -> None: - """初始化异步 Redis 缓存并保存默认 TTL。""" - self.ttl = ttl - self.redis_helper = AsyncRedisHelper() - - async def set( - self, - key: str, - value: Any, - ttl: Optional[int] = None, - region: Optional[str] = DEFAULT_CACHE_REGION, - **kwargs, - ) -> None: - """异步写入缓存,非正 TTL 视为立即删除。""" - ttl = self.ttl if ttl is None else ttl - if ttl is not None and ttl <= 0: - await self.redis_helper.delete(key, region=region) - return - await self.redis_helper.set(key, value, ttl=ttl, region=region, **kwargs) - - async def exists( - self, - key: str, - region: Optional[str] = DEFAULT_CACHE_REGION, - ) -> bool: - """异步判断缓存键是否存在。""" - return await self.redis_helper.exists(key, region=region) - - async def get( - self, - key: str, - region: Optional[str] = DEFAULT_CACHE_REGION, - ) -> Optional[Any]: - """异步读取缓存值,不存在时返回空值。""" - return await self.redis_helper.get(key, region=region) - - async def delete( - self, - key: str, - region: Optional[str] = DEFAULT_CACHE_REGION, - ) -> None: - """异步删除缓存键。""" - await self.redis_helper.delete(key, region=region) - - async def clear(self, region: Optional[str] = DEFAULT_CACHE_REGION) -> None: - """异步清空指定缓存区或全部缓存。""" - await self.redis_helper.clear(region=region) - - async def items( - self, - region: Optional[str] = DEFAULT_CACHE_REGION, - ) -> AsyncGenerator[Tuple[str, Any], None]: - """异步遍历指定缓存区的键值对。""" - async for item in self.redis_helper.items(region=region): - yield item - - async def close(self) -> None: - """关闭异步 Redis 连接池。""" - await self.redis_helper.close() - - @staticmethod - def is_redis() -> bool: - """标记当前后端为 Redis。""" - return True - - class FileBackend(CacheBackend): """通过本地文件系统保存二进制缓存。""" @@ -351,12 +190,19 @@ class AsyncFileBackend(AsyncCacheBackend): """异步文件缓存没有需要关闭的持久连接。""" -def configure_platform_cache() -> None: - """把配置感知的 Redis 与文件适配器注册到平台缓存工厂。""" +def configure_platform_cache( + *, + backend_type_provider: Callable[[], str], + redis_factory: Optional[Callable[[Optional[int]], AtomicCacheBackend]] = None, + async_redis_factory: Optional[Callable[[Optional[int]], AsyncCacheBackend]] = None, + redis_probe: Optional[Callable[[], bool]] = None, +) -> None: + """把文件适配器和启动层选定的 Redis 适配器注册到平台缓存工厂。""" configure_cache_factories( - backend_type_provider=lambda: get_runtime_setting('CACHE_BACKEND_TYPE'), - redis_factory=lambda ttl: RedisBackend(ttl=ttl), - async_redis_factory=lambda ttl: AsyncRedisBackend(ttl=ttl), + backend_type_provider=backend_type_provider, + redis_factory=redis_factory, + async_redis_factory=async_redis_factory, + redis_probe=redis_probe, file_factory=lambda base: FileBackend( base=base or get_runtime_setting('TEMP_PATH') ), diff --git a/app/adapters/cache/redis.py b/app/adapters/cache/redis.py index a8f4971f2..9730d1812 100644 --- a/app/adapters/cache/redis.py +++ b/app/adapters/cache/redis.py @@ -11,6 +11,7 @@ from redis.asyncio import BlockingConnectionPool as AsyncBlockingConnectionPool from redis.asyncio import Redis from app.foundation.singleton import Singleton +from app.runtime.cache import DEFAULT_CACHE_REGION, AsyncCacheBackend, AtomicCacheBackend from app.runtime.log import logger from app.runtime.reload import ConfigReloadMixin from app.runtime.settings import get_runtime_setting @@ -696,3 +697,163 @@ class AsyncRedisHelper(ConfigReloadMixin, metaclass=Singleton): """ await self._close_client() logger.debug("Redis async connection closed") + + +class RedisBackend(AtomicCacheBackend): + """通过同步 Redis 客户端实现缓存后端。""" + + def __init__(self, ttl: Optional[int] = None) -> None: + """初始化 Redis 缓存并保存默认 TTL。""" + self.ttl = ttl + self.redis_helper = RedisHelper() + + def set( + self, + key: str, + value: Any, + ttl: Optional[int] = None, + region: Optional[str] = DEFAULT_CACHE_REGION, + **kwargs, + ) -> None: + """写入缓存,非正 TTL 视为立即删除。""" + ttl = self.ttl if ttl is None else ttl + if ttl is not None and ttl <= 0: + self.redis_helper.delete(key, region=region) + return + self.redis_helper.set(key, value, ttl=ttl, region=region, **kwargs) + + def store( + self, + key: str, + value: Any, + ttl: Optional[int] = None, + region: Optional[str] = DEFAULT_CACHE_REGION, + **kwargs: Any, + ) -> None: + """严格写入 Redis,供安全敏感的一次性状态使用。""" + ttl = self.ttl if ttl is None else ttl + if ttl is not None and ttl <= 0: + self.redis_helper.consume(key, region=region) + return + self.redis_helper.store(key, value, ttl=ttl, region=region, **kwargs) + + def consume( + self, + key: str, + region: Optional[str] = DEFAULT_CACHE_REGION, + ) -> Optional[Any]: + """通过 Redis 原子命令严格领取一个缓存值。""" + return self.redis_helper.consume(key, region=region) + + def exists( + self, + key: str, + region: Optional[str] = DEFAULT_CACHE_REGION, + ) -> bool: + """判断缓存键是否存在。""" + return self.redis_helper.exists(key, region=region) + + def get( + self, + key: str, + region: Optional[str] = DEFAULT_CACHE_REGION, + ) -> Optional[Any]: + """读取缓存值,不存在时返回空值。""" + return self.redis_helper.get(key, region=region) + + def delete( + self, + key: str, + region: Optional[str] = DEFAULT_CACHE_REGION, + ) -> None: + """删除缓存键。""" + self.redis_helper.delete(key, region=region) + + def clear(self, region: Optional[str] = DEFAULT_CACHE_REGION) -> None: + """清空指定缓存区或全部缓存。""" + self.redis_helper.clear(region=region) + + def items( + self, + region: Optional[str] = DEFAULT_CACHE_REGION, + ) -> Generator[Tuple[str, Any], None, None]: + """遍历指定缓存区的键值对。""" + return self.redis_helper.items(region=region) + + def close(self) -> None: + """关闭同步 Redis 连接池。""" + self.redis_helper.close() + + @staticmethod + def is_redis() -> bool: + """标记当前后端为 Redis。""" + return True + + +class AsyncRedisBackend(AsyncCacheBackend): + """通过异步 Redis 客户端实现缓存后端。""" + + def __init__(self, ttl: Optional[int] = None) -> None: + """初始化异步 Redis 缓存并保存默认 TTL。""" + self.ttl = ttl + self.redis_helper = AsyncRedisHelper() + + async def set( + self, + key: str, + value: Any, + ttl: Optional[int] = None, + region: Optional[str] = DEFAULT_CACHE_REGION, + **kwargs, + ) -> None: + """异步写入缓存,非正 TTL 视为立即删除。""" + ttl = self.ttl if ttl is None else ttl + if ttl is not None and ttl <= 0: + await self.redis_helper.delete(key, region=region) + return + await self.redis_helper.set(key, value, ttl=ttl, region=region, **kwargs) + + async def exists( + self, + key: str, + region: Optional[str] = DEFAULT_CACHE_REGION, + ) -> bool: + """异步判断缓存键是否存在。""" + return await self.redis_helper.exists(key, region=region) + + async def get( + self, + key: str, + region: Optional[str] = DEFAULT_CACHE_REGION, + ) -> Optional[Any]: + """异步读取缓存值,不存在时返回空值。""" + return await self.redis_helper.get(key, region=region) + + async def delete( + self, + key: str, + region: Optional[str] = DEFAULT_CACHE_REGION, + ) -> None: + """异步删除缓存键。""" + await self.redis_helper.delete(key, region=region) + + async def clear(self, region: Optional[str] = DEFAULT_CACHE_REGION) -> None: + """异步清空指定缓存区或全部缓存。""" + await self.redis_helper.clear(region=region) + + async def items( + self, + region: Optional[str] = DEFAULT_CACHE_REGION, + ) -> AsyncGenerator[Tuple[str, Any], None]: + """异步遍历指定缓存区的键值对。""" + async for item in self.redis_helper.items(region=region): + yield item + + async def close(self) -> None: + """关闭异步 Redis 连接池。""" + await self.redis_helper.close() + + @staticmethod + def is_redis() -> bool: + """标记当前后端为 Redis。""" + return True diff --git a/app/agent/memory/__init__.py b/app/agent/memory/__init__.py index c230fb944..28c249ae2 100644 --- a/app/agent/memory/__init__.py +++ b/app/agent/memory/__init__.py @@ -5,6 +5,7 @@ from datetime import datetime from typing import Dict, List, Optional from langchain_core.messages import BaseMessage, messages_from_dict, messages_to_dict +from pydantic import BaseModel, ConfigDict, Field, field_serializer from app.agent.tools.result import messages_for_persistence from app.application.messaging.chat import ( @@ -16,7 +17,24 @@ from app.application.messaging.chat import ( ) from app.runtime.log import logger from app.runtime.settings import get_runtime_setting -from app.schemas.agent import ConversationMemory + + +class ConversationMemory(BaseModel): + """对话记忆模型 + + 消息字段依赖 langchain_core,因此模型归属智能体记忆模块;普通 schema 导入不加载 LangChain。 + """ + + session_id: str = Field(description="会话ID") + user_id: Optional[str] = Field(default=None, description="用户ID") + messages: List[BaseMessage] = Field(default_factory=list, description="消息列表") + updated_at: datetime = Field(default_factory=datetime.now, description="更新时间") + + model_config = ConfigDict() + + @field_serializer("updated_at", when_used="json") + def serialize_datetime(self, value: datetime) -> str: + return value.isoformat() class MemoryManager: diff --git a/app/modules/redis/__init__.py b/app/modules/redis/__init__.py index 97c8fcc03..144f4a1e4 100644 --- a/app/modules/redis/__init__.py +++ b/app/modules/redis/__init__.py @@ -1,7 +1,7 @@ from typing import Tuple, Union -from app.adapters.cache.redis import RedisHelper from app.modules import _ModuleBase +from app.runtime.cache import probe_redis_cache from app.runtime.settings import get_runtime_setting from app.schemas.types import ModuleType, OtherModulesType @@ -55,6 +55,10 @@ class RedisModule(_ModuleBase): """ if get_runtime_setting('CACHE_BACKEND_TYPE') != "redis": return None - if RedisHelper().test(): + # Redis 由缓存组合根在启动时按配置接入;缓存设置需重启生效,未接入时不代为建连。 + connected = probe_redis_cache() + if connected is None: + return False, "缓存类型已改为Redis,重启后生效" + if connected: return True, "" return False, "Redis连接失败,请检查配置" diff --git a/app/runtime/cache.py b/app/runtime/cache.py index 58f0996f1..9aeb61cc1 100644 --- a/app/runtime/cache.py +++ b/app/runtime/cache.py @@ -30,26 +30,40 @@ _async_file_factory: Optional[ Callable[[Optional[Path]], "AsyncCacheBackend"] ] = None _file_ttl_provider: Callable[[], int] = lambda: DEFAULT_CACHE_TTL +_redis_probe: Optional[Callable[[], bool]] = None def configure_cache_factories( *, backend_type_provider: Callable[[], str], - redis_factory: Callable[[Optional[int]], "AtomicCacheBackend"], - async_redis_factory: Callable[[Optional[int]], "AsyncCacheBackend"], file_factory: Callable[[Optional[Path]], "CacheBackend"], async_file_factory: Callable[[Optional[Path]], "AsyncCacheBackend"], file_ttl_provider: Callable[[], int], + redis_factory: Optional[Callable[[Optional[int]], "AtomicCacheBackend"]] = None, + async_redis_factory: Optional[Callable[[Optional[int]], "AsyncCacheBackend"]] = None, + redis_probe: Optional[Callable[[], bool]] = None, ) -> None: - """注入配置读取器和具体缓存适配器工厂。""" + """注入配置读取器和具体缓存适配器工厂。 + + Redis 工厂与探测器只在启动层选用 Redis 缓存时传入;此时 ``backend_type_provider`` + 才可能返回 ``redis``,未接入 Redis 的进程不会走到 Redis 分支。 + """ global _backend_type_provider, _redis_factory, _async_redis_factory - global _file_factory, _async_file_factory, _file_ttl_provider + global _file_factory, _async_file_factory, _file_ttl_provider, _redis_probe _backend_type_provider = backend_type_provider _redis_factory = redis_factory _async_redis_factory = async_redis_factory _file_factory = file_factory _async_file_factory = async_file_factory _file_ttl_provider = file_ttl_provider + _redis_probe = redis_probe + + +def probe_redis_cache() -> Optional[bool]: + """探测启动时接入的 Redis 缓存是否可连接;本进程未接入 Redis 时返回 None。""" + if _redis_probe is None: + return None + return _redis_probe() # 上下文变量来控制缓存行为 diff --git a/app/runtime/compat/manifest.py b/app/runtime/compat/manifest.py index bff0815d8..a8bc8217b 100644 --- a/app/runtime/compat/manifest.py +++ b/app/runtime/compat/manifest.py @@ -782,7 +782,18 @@ _MESSAGE_NOTIFICATION_SYMBOL_ALIASES: Dict[str, SymbolAlias] = { }, } +# Redis 缓存后端只在启动时选用 Redis 后由缓存组合根导入;旧入口惰性解析,文件缓存实例不加载 redis。 +_REDIS_CACHE_BACKEND_SYMBOL_ALIASES: Dict[str, SymbolAlias] = { + name: SymbolAlias( + target_module="app.adapters.cache.redis", + target_name=name, + replacement=f"app.adapters.cache.redis.{name}", + ) + for name in ("AsyncRedisBackend", "RedisBackend") +} + SYMBOL_ALIASES: Dict[str, Dict[str, SymbolAlias]] = { + "app.adapters.cache.backends": _REDIS_CACHE_BACKEND_SYMBOL_ALIASES, "app.agent": { "AgentChain": SymbolAlias( target_module="app.chain.agent", @@ -978,6 +989,17 @@ SYMBOL_ALIASES: Dict[str, Dict[str, SymbolAlias]] = { for name in ("TransferTask", "TransferQueue") }, **_MESSAGE_NOTIFICATION_SYMBOL_ALIASES, + # 对话记忆模型迁到智能体记忆模块后不再进入 schema 导出清单,根包旧入口惰性解析。 + "BaseMessage": SymbolAlias( + target_module="langchain_core.messages", + target_name="BaseMessage", + replacement="langchain_core.messages.BaseMessage", + ), + "ConversationMemory": SymbolAlias( + target_module="app.agent.memory", + target_name="ConversationMemory", + replacement="app.agent.memory.ConversationMemory", + ), }, "app.schemas.transfer": { **{ @@ -1010,12 +1032,19 @@ SYMBOL_ALIASES: Dict[str, Dict[str, SymbolAlias]] = { ), }, "app.schemas.agent": { + # 对话记忆模型依赖 langchain_core,归属智能体记忆模块;旧路径惰性解析。 + "ConversationMemory": SymbolAlias( + target_module="app.agent.memory", + target_name="ConversationMemory", + replacement="app.agent.memory.ConversationMemory", + ), "ReplyMode": SymbolAlias( target_module="app.schemas.types", target_name="ReplyMode", replacement="app.schemas.types.ReplyMode", ), }, + "app.sdk.cache": _REDIS_CACHE_BACKEND_SYMBOL_ALIASES, "app.sdk.logging": { name: SymbolAlias( target_module="app.runtime.log", diff --git a/app/schemas/agent.py b/app/schemas/agent.py index 55e438fe5..b246814ce 100644 --- a/app/schemas/agent.py +++ b/app/schemas/agent.py @@ -1,29 +1,13 @@ """AI智能体相关数据模型""" from datetime import datetime -from typing import List, Literal, Optional, Union +from typing import Literal, Optional, Union -from langchain_core.messages import BaseMessage from pydantic import BaseModel, ConfigDict, Field, field_serializer from app.schemas.common import JsonData -class ConversationMemory(BaseModel): - """对话记忆模型""" - - session_id: str = Field(description="会话ID") - user_id: Optional[str] = Field(default=None, description="用户ID") - messages: List[BaseMessage] = Field(default_factory=list, description="消息列表") - updated_at: datetime = Field(default_factory=datetime.now, description="更新时间") - - model_config = ConfigDict() - - @field_serializer("updated_at", when_used="json") - def serialize_datetime(self, value: datetime) -> str: - return value.isoformat() - - class AgentState(BaseModel): """AI智能体状态模型""" diff --git a/app/schemas/exports.py b/app/schemas/exports.py index a81b124a8..7b5e312a7 100644 --- a/app/schemas/exports.py +++ b/app/schemas/exports.py @@ -55,7 +55,6 @@ SCHEMA_EXPORTS = { 'AuthProviderInfo': ('app.schemas.user', 'AuthProviderInfo'), 'AuthProviderRemote': ('app.schemas.user', 'AuthProviderRemote'), 'BaseEventData': ('app.schemas.event', 'BaseEventData'), - 'BaseMessage': ('app.schemas.agent', 'BaseMessage'), 'BaseModel': ('app.schemas.mcp', 'BaseModel'), 'BatchProgressKeyData': ('app.schemas.common', 'BatchProgressKeyData'), 'BatchTransferHistoryRedoRequest': ('app.schemas.history', 'BatchTransferHistoryRedoRequest'), @@ -125,7 +124,6 @@ SCHEMA_EXPORTS = { 'ContentType': ('app.schemas.message', 'ContentType'), 'Context': ('app.schemas.workflow', 'Context'), 'ContextSnapshot': ('app.schemas.event', 'ContextSnapshot'), - 'ConversationMemory': ('app.schemas.agent', 'ConversationMemory'), 'CookieActionResponse': ('app.schemas.servcookie', 'CookieActionResponse'), 'CookieData': ('app.schemas.servcookie', 'CookieData'), 'CookieDecryptedPayload': ('app.schemas.servcookie', 'CookieDecryptedPayload'), @@ -594,7 +592,7 @@ SCHEMA_CONFLICTS = { 'FileItem': ['app.schemas.event', 'app.schemas.file', 'app.schemas.transfer', 'app.schemas.workflow'], 'FilterRuleGroup': ['app.schemas.rule', 'app.schemas.system'], 'JsonData': ['app.schemas.agent', 'app.schemas.category', 'app.schemas.common', 'app.schemas.context', 'app.schemas.dashboard', 'app.schemas.event', 'app.schemas.history', 'app.schemas.mediaserver', 'app.schemas.message', 'app.schemas.mfa', 'app.schemas.music', 'app.schemas.plugin', 'app.schemas.search', 'app.schemas.openai', 'app.schemas.servcookie', 'app.schemas.site', 'app.schemas.transfer', 'app.schemas.user', 'app.schemas.workflow', 'app.schemas.mcp'], - 'List': ['app.schemas.agent', 'app.schemas.context', 'app.schemas.event', 'app.schemas.history', 'app.schemas.mediaserver', 'app.schemas.message', 'app.schemas.monitoring', 'app.schemas.plugin', 'app.schemas.openai', 'app.schemas.subscribe', 'app.schemas.transfer', 'app.schemas.workflow'], + 'List': ['app.schemas.context', 'app.schemas.event', 'app.schemas.history', 'app.schemas.mediaserver', 'app.schemas.message', 'app.schemas.monitoring', 'app.schemas.plugin', 'app.schemas.openai', 'app.schemas.subscribe', 'app.schemas.transfer', 'app.schemas.workflow'], 'Literal': ['app.schemas.agent', 'app.schemas.category', 'app.schemas.download', 'app.schemas.history', 'app.schemas.music', 'app.schemas.plugin', 'app.schemas.search', 'app.schemas.servcookie', 'app.schemas.subscribe', 'app.schemas.system', 'app.schemas.transfer', 'app.schemas.mcp'], 'LocaleHelper': ['app.schemas.dashboard', 'app.schemas.response'], 'MediaInfo': ['app.schemas.context', 'app.schemas.system', 'app.schemas.transfer', 'app.schemas.workflow'], diff --git a/app/sdk/cache.py b/app/sdk/cache.py index 0aac74872..73f9fbaef 100644 --- a/app/sdk/cache.py +++ b/app/sdk/cache.py @@ -2,9 +2,7 @@ from app.adapters.cache.backends import ( AsyncFileBackend, - AsyncRedisBackend, FileBackend, - RedisBackend, ) from app.runtime.cache import ( AsyncCache, @@ -24,6 +22,8 @@ from app.runtime.cache import ( is_fresh, ) +# RedisBackend / AsyncRedisBackend 由兼容层惰性解析到 app.adapters.cache.redis,显式导入仍可用; +# 不列入星号导出,避免文件缓存实例因 ``import *`` 加载 redis 客户端库。 __all__ = [ "AtomicCacheBackend", "AsyncCache", @@ -31,14 +31,12 @@ __all__ = [ "AsyncFileBackend", "AsyncFileCache", "AsyncMemoryBackend", - "AsyncRedisBackend", "Cache", "CacheBackend", "FileBackend", "FileCache", "LRUCache", "MemoryBackend", - "RedisBackend", "TTLCache", "async_fresh", "cached", diff --git a/app/startup/composition/cache.py b/app/startup/composition/cache.py index faf56d9e5..7d3febb1c 100644 --- a/app/startup/composition/cache.py +++ b/app/startup/composition/cache.py @@ -1,8 +1,80 @@ """缓存 Adapter 的宿主组合根。""" +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Optional + from app.adapters.cache.backends import configure_platform_cache +from app.runtime.settings import get_runtime_setting + +# 未选用 Redis 时的本地缓存类型;平台缓存只区分 redis 与非 redis 两条路由。 +_LOCAL_BACKEND_TYPE = "cachetools" + + +@dataclass(frozen=True, slots=True) +class CacheComposition: + """记录本进程启动时选定的缓存实现,供关闭与配置重载定位连接 owner。""" + + redis_enabled: bool # 启动时 CACHE_BACKEND_TYPE 是否为 redis + sync_redis_owner: Optional[type[Any]] = None # 同步 Redis 连接 Singleton 类型,未启用时为空 + async_redis_owner: Optional[type[Any]] = None # 异步 Redis 连接 Singleton 类型,未启用时为空 + + @property + def redis_owners(self) -> tuple[type[Any], ...]: + """返回已接入的 Redis 连接 owner 类型。""" + return tuple( + owner + for owner in (self.sync_redis_owner, self.async_redis_owner) + if owner is not None + ) + + +_composition = CacheComposition(redis_enabled=False) def configure_cache_composition() -> None: - """在业务模块导入前登记平台缓存具体实现。""" - configure_platform_cache() + """在业务模块导入前登记平台缓存具体实现。 + + 缓存设置属于重启生效的配置组,因此只在启动时按 ``CACHE_BACKEND_TYPE`` 选一次实现。 + 选用 Redis 时才导入 Redis 适配器,文件缓存实例不加载 redis 客户端库。启动时未接入 + Redis 的进程在运行中改成 redis 会继续使用本地缓存,直到重启;已接入 Redis 的进程 + 在运行中改回本地缓存则立即切换,与之前一致。 + """ + global _composition + if get_runtime_setting("CACHE_BACKEND_TYPE") != "redis": + configure_platform_cache( + backend_type_provider=_local_backend_type, + ) + _composition = CacheComposition(redis_enabled=False) + return + + from app.adapters.cache.redis import ( + AsyncRedisBackend, + AsyncRedisHelper, + RedisBackend, + RedisHelper, + ) + + configure_platform_cache( + backend_type_provider=lambda: get_runtime_setting("CACHE_BACKEND_TYPE"), + redis_factory=lambda ttl: RedisBackend(ttl=ttl), + async_redis_factory=lambda ttl: AsyncRedisBackend(ttl=ttl), + redis_probe=lambda: RedisHelper().test(), + ) + _composition = CacheComposition( + redis_enabled=True, + sync_redis_owner=RedisHelper, + async_redis_owner=AsyncRedisHelper, + ) + + +def get_cache_composition() -> CacheComposition: + """返回本进程当前的缓存组合结果。""" + return _composition + + +def _local_backend_type() -> str: + """启动时未接入 Redis,运行中的 redis 设置在重启前不生效。""" + backend_type = get_runtime_setting("CACHE_BACKEND_TYPE") + return _LOCAL_BACKEND_TYPE if backend_type == "redis" else backend_type diff --git a/app/startup/initializers/modules.py b/app/startup/initializers/modules.py index 22ac5251f..b177d1a2c 100644 --- a/app/startup/initializers/modules.py +++ b/app/startup/initializers/modules.py @@ -3,8 +3,6 @@ import inspect import sys from typing import Callable, cast -from app.adapters.cache.redis import AsyncRedisHelper, RedisHelper - # SitesHelper涉及资源包拉取,提前引入并容错提示 try: from app.application.site.sites import SitesHelper # noqa # pylint: disable=import-error,no-name-in-module @@ -67,6 +65,7 @@ from app.startup.composition.agent import ( publish_agent_services, reset_agent_services, ) +from app.startup.composition.cache import get_cache_composition from app.startup.composition.chain import ( configure_chain_runtime_context, configure_directory_classification_service, @@ -259,8 +258,10 @@ def _existing_singleton(instance_type: type) -> object | None: return getter() if callable(getter) else None -def _call_existing_singleton(instance_type: type, method_name: str) -> object: - """调用已存在 Singleton 的关闭方法;owner 未创建时视为已经收敛。""" +def _call_existing_singleton(instance_type: type | None, method_name: str) -> object: + """调用已存在 Singleton 的关闭方法;owner 未接入或未创建时视为已经收敛。""" + if instance_type is None: + return True instance = _existing_singleton(instance_type) if instance is None: return True @@ -310,8 +311,11 @@ def get_config_reload_handler_providers( system_helper = SystemHelper() providers: dict[type, Callable[[], object | None]] = { SystemHelper: lambda: system_helper, - RedisHelper: RedisHelper.get_existing_instance, - AsyncRedisHelper: AsyncRedisHelper.get_existing_instance, + # Redis 连接 owner 只在启动时选用 Redis 缓存后存在。 + **{ + owner: owner.get_existing_instance + for owner in get_cache_composition().redis_owners + }, TransferChain: TransferChain.get_existing_instance, Monitor: Monitor.get_existing_instance, } @@ -583,12 +587,12 @@ async def stop_modules() -> bool: ) await run_step( "Redis缓存连接", - lambda: _call_existing_singleton(RedisHelper, "close"), + lambda: _call_existing_singleton(get_cache_composition().sync_redis_owner, "close"), offload=True, ) await run_step( "异步Redis缓存连接", - lambda: _call_existing_singleton(AsyncRedisHelper, "close"), + lambda: _call_existing_singleton(get_cache_composition().async_redis_owner, "close"), ) # Web Agent 的取消 finally 可能还要写入最终展示快照,必须先完成任务收尾,再关闭写入准入。 web_agent_drained = await run_step( diff --git a/database/versions/279a949d81b6_2_1_1.py b/database/versions/279a949d81b6_2_1_1.py index 3391479f7..b0f6fca91 100644 --- a/database/versions/279a949d81b6_2_1_1.py +++ b/database/versions/279a949d81b6_2_1_1.py @@ -6,8 +6,8 @@ Create Date: 2025-02-14 19:02:24.989349 """ -from app.adapters.cache.backends import configure_platform_cache from app.application.torrent.download import clear_torrent_cache +from app.startup.initializers.cache import configure_cache_dependencies # revision identifiers, used by Alembic. revision = '279a949d81b6' @@ -17,8 +17,8 @@ depends_on = None def upgrade() -> None: - # 迁移执行时生命周期尚未装配,直接通过缓存端口清理一次缓存。 - configure_platform_cache() + # 迁移执行时生命周期尚未装配,按启动缓存入口装配后清理一次缓存。 + configure_cache_dependencies() clear_torrent_cache() diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index 271d1486c..63feb13f2 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -763,7 +763,7 @@ flowchart LR | 指标 | 当前值 | |---|---:| | Python 模块 | 1052 | -| 内部导入边 | 8,930 | +| 内部导入边 | 8,926 | | 非平凡 SCC | 1(精确 containment 的 TMDB 移植包环) | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 53(债务已清零,53 条精确 containment) | @@ -773,7 +773,7 @@ flowchart LR | Model/Oper 自动事务与自建 Session | 0 | | 组合根外 `SystemConfigOper()` | 0 | -整理失败反馈由 `app.application.transfer.feedback` 集中投影;已整理下载的站点字幕复用 `app.chain.transfer` 自动整理队列;网络连通性检测按 `app.application.nettest` 归档,并覆盖 HTTP 与 WebSocket 探测。未启用智能助手的消息音乐交互由 `app.chain.music_interaction` 包中的解析展示与订阅协调模块处理;宿主依赖边现为 8,930,音乐交互包未新增 SCC,插件一次性任务 owner `app.scheduler.oncejob` 带来 1 个模块和 9 条依赖边。Agent 持久回执新增 Application 端口及 DB Model/Oper/Adapter 四个冷导入模块。当前 `app.startup.lifecycle` 为 573、`app.factory` 为 585、`app.main` 为 587。性能基线只同步模块数量,原有耗时预算、历史采样和生命周期资源约束保持有效。 +整理失败反馈由 `app.application.transfer.feedback` 集中投影;已整理下载的站点字幕复用 `app.chain.transfer` 自动整理队列;网络连通性检测按 `app.application.nettest` 归档,并覆盖 HTTP 与 WebSocket 探测。未启用智能助手的消息音乐交互由 `app.chain.music_interaction` 包中的解析展示与订阅协调模块处理;宿主依赖边现为 8,926,音乐交互包未新增 SCC,插件一次性任务 owner `app.scheduler.oncejob` 带来 1 个模块和 9 条依赖边。Agent 持久回执新增 Application 端口及 DB Model/Oper/Adapter 四个冷导入模块。缓存组合根只在启动时选用 Redis 缓存才导入 Redis 适配器,依赖 langchain_core 的对话记忆模型归属 `app.agent.memory`,文件缓存且未启用智能体的实例不再加载 redis 与 langchain_core。当前 `app.startup.lifecycle` 为 572、`app.factory` 为 584、`app.main` 为 586。性能基线只同步模块数量,原有耗时预算、历史采样和生命周期资源约束保持有效。 架构专项验证分为两个 CI 投影:`Check event semantic policy` 先运行依赖、Adapter、出口和 Event 语义门禁,`Check host architecture snapshot` 再执行快照测试及一次 diff --git a/docs/refactor/optimization-checklist.md b/docs/refactor/optimization-checklist.md index 54e5f4604..fd48b4f89 100644 --- a/docs/refactor/optimization-checklist.md +++ b/docs/refactor/optimization-checklist.md @@ -94,7 +94,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | 指标 | 当前值 | 解释 | |---|---:|---| -| 宿主 Python 模块 / 内部依赖边 | 1052 / 8,930 | `dependency-baseline.json` 当前快照;系统设置合同按 `app.application.settings` 包组织,系统设置 API 端点独立按 `app.api.endpoints.settings` 归档;消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理;已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档;网络连通性检测按 `app.application.nettest` 归档 | +| 宿主 Python 模块 / 内部依赖边 | 1052 / 8,926 | `dependency-baseline.json` 当前快照;系统设置合同按 `app.application.settings` 包组织,系统设置 API 端点独立按 `app.api.endpoints.settings` 归档;消息音乐交互由 `app.chain.music_interaction` 包中的音乐解析/展示与订阅协调模块处理;已整理下载的站点字幕经 `app.chain.transfer` 自动整理队列完成命名与归档;网络连通性检测按 `app.application.nettest` 归档 | | 非平凡 SCC | 1 | 仅保留精确 containment 的 29 模块 TMDB 移植包环 | | 跨层 DB 边界债务 | 0 | Application、Chain、API、Agent、Runtime、Workflow 到 DB 的受控债务均为零 | | Model/Oper 事务债务 | 0 | 自建 Session、自动事务装饰器、直接 commit/rollback 等基线均为零 | @@ -102,7 +102,7 @@ ARCH-201 至 ARCH-204 均达到实现、验证、提交、推送和远端门禁 | Event Contract | 53 | 均已有 payload model,但当前全部是 diagnostic enforcement | | Python 源码量 | 305,884 行 | 排除 `app/plugins/**`;61 个文件超过 1,000 行,11 个超过 2,000 行 | | 长方法 | 290 个超过 80 行 | AST 统计排除 `app/plugins/**`;65 个超过 150 行,21 个超过 250 行 | -| 全量 mypy 历史债务 | 9,269 / 506 文件 | 删除独立活动日志中间件后的现状基线;消息媒体交互类型债务已清零,canonical Facade 与 endpoint 类型边界已补齐,低水位只允许继续下降 | +| 全量 mypy 历史债务 | 9,269 / 508 文件 | 删除独立活动日志中间件后的现状基线;消息媒体交互类型债务已清零,canonical Facade 与 endpoint 类型边界已补齐,低水位只允许继续下降 | | Ruff 历史诊断 | 516 | 低水位门禁通过,但规则集只覆盖 `E4/E7/E9/F/I` | | 覆盖率固定基线 | Application 80.00%,Domain 80.00% | Chain、Runtime、Agent、Adapter、Startup 未进入包级覆盖率门禁 | diff --git a/tests/conftest.py b/tests/conftest.py index 50827bdd5..28b0d0c41 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -70,6 +70,29 @@ class _TestRuntimeSettingsProxy: raise AttributeError(key) +@pytest.fixture +def compose_cache_backend(): + """按启动流程以指定缓存类型重新装配平台缓存,用例结束后恢复原设置与装配。 + + 缓存类型属于重启生效的设置,只在启动时由缓存组合根选定;测试直接改设置不会切换路由。 + 恢复不依赖 monkeypatch 的撤销顺序,避免在设置仍被替换时重新装配。 + """ + from app.runtime.config import settings + from app.startup.composition.cache import configure_cache_composition + + original = settings.CACHE_BACKEND_TYPE + + def compose(backend_type: str) -> None: + settings.CACHE_BACKEND_TYPE = backend_type + configure_cache_composition() + + try: + yield compose + finally: + settings.CACHE_BACKEND_TYPE = original + configure_cache_composition() + + @pytest.fixture(autouse=True) def install_runtime_settings_test_proxies(monkeypatch): """给历史测试 patch 点注入测试专用对象,生产代码不保留 settings 属性。""" diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 8c7d28c12..b85339cd0 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -1097,23 +1097,21 @@ "runtime_only": true } }, - "edge_count": 8930, - "edge_sha256": "54d148d934657cd03ad7c96c6e127bba632b803ac1fb82baa183ba219e2bc47f", + "edge_count": 8926, + "edge_sha256": "038f1aea5eb67a48f17223c28a6a7caa7fb35bfd5b950d2ab21b5010d1b530f2", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", "app -> app.runtime", "app -> app.runtime.compat", "app -> app.runtime.compat.imports", - "app.adapters.cache.backends -> app.adapters", - "app.adapters.cache.backends -> app.adapters.cache", - "app.adapters.cache.backends -> app.adapters.cache.redis", "app.adapters.cache.backends -> app.runtime", "app.adapters.cache.backends -> app.runtime.cache", "app.adapters.cache.backends -> app.runtime.settings", "app.adapters.cache.redis -> app.foundation", "app.adapters.cache.redis -> app.foundation.singleton", "app.adapters.cache.redis -> app.runtime", + "app.adapters.cache.redis -> app.runtime.cache", "app.adapters.cache.redis -> app.runtime.log", "app.adapters.cache.redis -> app.runtime.reload", "app.adapters.cache.redis -> app.runtime.settings", @@ -1474,8 +1472,6 @@ "app.agent.memory -> app.runtime", "app.agent.memory -> app.runtime.log", "app.agent.memory -> app.runtime.settings", - "app.agent.memory -> app.schemas", - "app.agent.memory -> app.schemas.agent", "app.agent.middleware.config -> app.agent", "app.agent.middleware.config -> app.agent.middleware", "app.agent.middleware.config -> app.agent.middleware.utils", @@ -7668,11 +7664,9 @@ "app.modules.qqbot.qqbot -> app.runtime.cache", "app.modules.qqbot.qqbot -> app.runtime.log", "app.modules.qqbot.qqbot -> app.runtime.thread", - "app.modules.redis -> app.adapters", - "app.modules.redis -> app.adapters.cache", - "app.modules.redis -> app.adapters.cache.redis", "app.modules.redis -> app.modules", "app.modules.redis -> app.runtime", + "app.modules.redis -> app.runtime.cache", "app.modules.redis -> app.runtime.settings", "app.modules.redis -> app.schemas", "app.modules.redis -> app.schemas.types", @@ -9135,6 +9129,9 @@ "app.startup.composition.cache -> app.adapters", "app.startup.composition.cache -> app.adapters.cache", "app.startup.composition.cache -> app.adapters.cache.backends", + "app.startup.composition.cache -> app.adapters.cache.redis", + "app.startup.composition.cache -> app.runtime", + "app.startup.composition.cache -> app.runtime.settings", "app.startup.composition.chain -> app.adapters", "app.startup.composition.chain -> app.adapters.external", "app.startup.composition.chain -> app.adapters.external.server", @@ -9594,8 +9591,6 @@ "app.startup.initializers.domain -> app.startup.composition", "app.startup.initializers.domain -> app.startup.composition.domain", "app.startup.initializers.modules -> app.adapters", - "app.startup.initializers.modules -> app.adapters.cache", - "app.startup.initializers.modules -> app.adapters.cache.redis", "app.startup.initializers.modules -> app.adapters.external", "app.startup.initializers.modules -> app.adapters.external.server", "app.startup.initializers.modules -> app.adapters.network", @@ -9660,6 +9655,7 @@ "app.startup.initializers.modules -> app.startup", "app.startup.initializers.modules -> app.startup.composition", "app.startup.initializers.modules -> app.startup.composition.agent", + "app.startup.initializers.modules -> app.startup.composition.cache", "app.startup.initializers.modules -> app.startup.composition.chain", "app.startup.initializers.modules -> app.startup.composition.classification", "app.startup.initializers.modules -> app.startup.composition.configuration", diff --git a/tests/fixtures/architecture/mypy-baseline.json b/tests/fixtures/architecture/mypy-baseline.json index 7b0289775..2dbf9de61 100644 --- a/tests/fixtures/architecture/mypy-baseline.json +++ b/tests/fixtures/architecture/mypy-baseline.json @@ -2,16 +2,16 @@ "app/adapters/cache/backends.py": { "import-untyped": 1, "no-any-return": 1, - "no-untyped-call": 2, - "no-untyped-def": 4, + "no-untyped-def": 2, "operator": 5, - "override": 16 + "override": 9 }, "app/adapters/cache/redis.py": { "arg-type": 9, "no-any-return": 2, - "no-untyped-call": 12, - "no-untyped-def": 15, + "no-untyped-call": 14, + "no-untyped-def": 17, + "override": 7, "return-value": 2, "str-bytes-safe": 2, "union-attr": 14 @@ -155,6 +155,9 @@ "type-arg": 7, "type-var": 1 }, + "app/agent/memory/__init__.py": { + "misc": 2 + }, "app/agent/middleware/config.py": { "misc": 1 }, @@ -1969,7 +1972,6 @@ }, "app/modules/redis/__init__.py": { "empty-body": 1, - "no-untyped-call": 1, "no-untyped-def": 1 }, "app/modules/rtorrent/__init__.py": { @@ -2587,7 +2589,7 @@ "unused-ignore": 1 }, "app/schemas/agent.py": { - "misc": 26 + "misc": 24 }, "app/schemas/cache.py": { "misc": 3 @@ -2718,6 +2720,9 @@ "no-untyped-def": 2, "type-arg": 1 }, + "app/startup/composition/cache.py": { + "no-untyped-call": 1 + }, "app/startup/composition/database.py": { "assignment": 1 }, diff --git a/tests/fixtures/architecture/runtime-contract-baseline.json b/tests/fixtures/architecture/runtime-contract-baseline.json index 0ba10dd0d..39242cf3f 100644 --- a/tests/fixtures/architecture/runtime-contract-baseline.json +++ b/tests/fixtures/architecture/runtime-contract-baseline.json @@ -885,6 +885,18 @@ } }, "symbol_aliases": { + "app.adapters.cache.backends": { + "AsyncRedisBackend": { + "replacement": "app.adapters.cache.redis.AsyncRedisBackend", + "target_module": "app.adapters.cache.redis", + "target_name": "AsyncRedisBackend" + }, + "RedisBackend": { + "replacement": "app.adapters.cache.redis.RedisBackend", + "target_module": "app.adapters.cache.redis", + "target_name": "RedisBackend" + } + }, "app.agent": { "AgentChain": { "replacement": "app.chain.agent.AgentChain", @@ -1106,6 +1118,11 @@ } }, "app.schemas": { + "BaseMessage": { + "replacement": "langchain_core.messages.BaseMessage", + "target_module": "langchain_core.messages", + "target_name": "BaseMessage" + }, "ChannelCapabilities": { "replacement": "app.schemas.notification.ChannelCapabilities", "target_module": "app.schemas.notification", @@ -1126,6 +1143,11 @@ "target_module": "app.schemas.message", "target_name": "IncomingMessage" }, + "ConversationMemory": { + "replacement": "app.agent.memory.ConversationMemory", + "target_module": "app.agent.memory", + "target_name": "ConversationMemory" + }, "MessageChannel": { "replacement": "app.schemas.types.NotificationChannel", "target_module": "app.schemas.types", @@ -1173,6 +1195,11 @@ } }, "app.schemas.agent": { + "ConversationMemory": { + "replacement": "app.agent.memory.ConversationMemory", + "target_module": "app.agent.memory", + "target_name": "ConversationMemory" + }, "ReplyMode": { "replacement": "app.schemas.types.ReplyMode", "target_module": "app.schemas.types", @@ -1280,6 +1307,18 @@ "target_name": "MessageType" } }, + "app.sdk.cache": { + "AsyncRedisBackend": { + "replacement": "app.adapters.cache.redis.AsyncRedisBackend", + "target_module": "app.adapters.cache.redis", + "target_name": "AsyncRedisBackend" + }, + "RedisBackend": { + "replacement": "app.adapters.cache.redis.RedisBackend", + "target_module": "app.adapters.cache.redis", + "target_name": "RedisBackend" + } + }, "app.sdk.logging": { "CustomFormatter": { "replacement": "app.runtime.log.CustomFormatter", @@ -9675,11 +9714,6 @@ "name": "AsyncMemoryBackend", "target": "app.runtime.cache.AsyncMemoryBackend" }, - { - "kind": "import", - "name": "AsyncRedisBackend", - "target": "app.adapters.cache.backends.AsyncRedisBackend" - }, { "kind": "import", "name": "AtomicCacheBackend", @@ -9715,11 +9749,6 @@ "name": "MemoryBackend", "target": "app.runtime.cache.MemoryBackend" }, - { - "kind": "import", - "name": "RedisBackend", - "target": "app.adapters.cache.backends.RedisBackend" - }, { "kind": "import", "name": "TTLCache", diff --git a/tests/fixtures/architecture/startup-performance-baseline.json b/tests/fixtures/architecture/startup-performance-baseline.json index 3cf9ea7f2..b12ba18ad 100644 --- a/tests/fixtures/architecture/startup-performance-baseline.json +++ b/tests/fixtures/architecture/startup-performance-baseline.json @@ -6,7 +6,7 @@ "repeat": 3, "targets": { "app.startup.lifecycle": { - "loaded_app_module_count": 573, + "loaded_app_module_count": 572, "max_ms": 1293.338, "median_ms": 1156.239, "min_ms": 1102.806, @@ -17,7 +17,7 @@ ] }, "app.factory": { - "loaded_app_module_count": 585, + "loaded_app_module_count": 584, "max_ms": 1127.911, "median_ms": 1122.382, "min_ms": 1119.221, @@ -28,7 +28,7 @@ ] }, "app.main": { - "loaded_app_module_count": 587, + "loaded_app_module_count": 586, "max_ms": 1188.652, "median_ms": 1183.509, "min_ms": 1174.522, diff --git a/tests/test_cache_binary_routing.py b/tests/test_cache_binary_routing.py index 9365d1602..e1860ee74 100644 --- a/tests/test_cache_binary_routing.py +++ b/tests/test_cache_binary_routing.py @@ -12,15 +12,11 @@ from typing import Any, Dict, Optional, Tuple import pytest -import app.adapters.cache.backends as backends_module +import app.adapters.cache.redis as redis_adapter_module import app.application.image as image_module import app.monitor.snapshot as snapshot_module -from app.adapters.cache.backends import ( - AsyncFileBackend, - AsyncRedisBackend, - FileBackend, - RedisBackend, -) +from app.adapters.cache.backends import AsyncFileBackend, FileBackend +from app.adapters.cache.redis import AsyncRedisBackend, RedisBackend from app.foundation.singleton import Singleton from app.monitor.snapshot import SnapshotStore from app.runtime.cache import ( @@ -72,12 +68,13 @@ class _FakeAsyncRedisHelper: @pytest.fixture -def redis_mode(monkeypatch): +def redis_mode(monkeypatch, compose_cache_backend): """切换到 Redis 缓存后端,并用内存替身替换 Redis 客户端。""" _FakeRedisHelper.store = {} - monkeypatch.setattr(settings, "CACHE_BACKEND_TYPE", "redis") - monkeypatch.setattr(backends_module, "RedisHelper", _FakeRedisHelper) - monkeypatch.setattr(backends_module, "AsyncRedisHelper", _FakeAsyncRedisHelper) + monkeypatch.setattr(redis_adapter_module, "RedisHelper", _FakeRedisHelper) + monkeypatch.setattr(redis_adapter_module, "AsyncRedisHelper", _FakeAsyncRedisHelper) + # 缓存类型在启动时选定,按启动流程重新装配后 Redis 路由才生效。 + compose_cache_backend("redis") return _FakeRedisHelper.store diff --git a/tests/test_cache_system.py b/tests/test_cache_system.py index 358b834b0..4e93b5c91 100644 --- a/tests/test_cache_system.py +++ b/tests/test_cache_system.py @@ -7,13 +7,14 @@ from unittest.mock import AsyncMock import pytest -from app.adapters.cache.backends import ( - AsyncFileBackend, +from app.adapters.cache.backends import AsyncFileBackend, FileBackend +from app.adapters.cache.redis import ( AsyncRedisBackend, - FileBackend, + AsyncRedisHelper, RedisBackend, + RedisHelper, + serialize, ) -from app.adapters.cache.redis import AsyncRedisHelper, RedisHelper, serialize from app.foundation.singleton import Singleton from app.runtime.cache import ( AsyncFileCache, @@ -631,29 +632,29 @@ def test_async_redis_backend_treats_zero_ttl_as_expired(): assert not helper.set_called -def test_file_cache_preserves_zero_ttl_in_redis_mode(monkeypatch): +@pytest.fixture +def redis_composed(compose_cache_backend): + """以 Redis 缓存装配平台缓存。""" + compose_cache_backend("redis") + +def test_file_cache_preserves_zero_ttl_in_redis_mode(redis_composed): """ FileCache 在 Redis 模式下不应把显式 ttl=0 替换为临时文件默认 TTL。 """ - monkeypatch.setattr(settings, "CACHE_BACKEND_TYPE", "redis") - assert FileCache(ttl=0).ttl == 0 -def test_async_file_cache_preserves_zero_ttl_in_redis_mode(monkeypatch): +def test_async_file_cache_preserves_zero_ttl_in_redis_mode(redis_composed): """ AsyncFileCache 在 Redis 模式下应与同步工厂保持相同 TTL 语义。 """ - monkeypatch.setattr(settings, "CACHE_BACKEND_TYPE", "redis") - assert AsyncFileCache(ttl=0).ttl == 0 -def test_file_cache_uses_default_ttl_when_omitted(monkeypatch): +def test_file_cache_uses_default_ttl_when_omitted(monkeypatch, redis_composed): """ 未传 TTL 时仍使用 TEMP_FILE_DAYS 配置的默认值。 """ - monkeypatch.setattr(settings, "CACHE_BACKEND_TYPE", "redis") monkeypatch.setattr(settings, "TEMP_FILE_DAYS", 7) assert FileCache().ttl == 7 * 24 * 3600 diff --git a/tests/test_config_reload_handler.py b/tests/test_config_reload_handler.py index 441a9d971..d983cebec 100644 --- a/tests/test_config_reload_handler.py +++ b/tests/test_config_reload_handler.py @@ -265,8 +265,9 @@ async def test_host_resolver_binds_current_scheduler_owner(monkeypatch): reload_config.assert_awaited_once_with() -def test_config_reload_resolver_reads_latest_singleton_owner(monkeypatch): +def test_config_reload_resolver_reads_latest_singleton_owner(monkeypatch, compose_cache_backend): """resolver 每次读取当前单例,不能保留已经换代的 Redis 实例。""" + compose_cache_backend("redis") first = object.__new__(RedisHelper) first_reload = Mock() first.on_config_changed = first_reload @@ -297,8 +298,9 @@ def test_config_reload_resolver_reads_latest_singleton_owner(monkeypatch): @pytest.mark.asyncio -async def test_config_reload_resolver_invokes_current_async_redis_owner(monkeypatch): +async def test_config_reload_resolver_invokes_current_async_redis_owner(monkeypatch, compose_cache_backend): """异步配置处理器必须绑定当前 Redis owner,并保留键筛选。""" + compose_cache_backend("redis") helper = object.__new__(AsyncRedisHelper) reload_config = AsyncMock() helper.on_config_changed = reload_config diff --git a/tests/test_configuration_initializer.py b/tests/test_configuration_initializer.py index ada89d53a..d25d22293 100644 --- a/tests/test_configuration_initializer.py +++ b/tests/test_configuration_initializer.py @@ -9,6 +9,7 @@ import pytest from app.application.configuration import configure_runtime_settings from app.startup.composition import configuration as configuration_composition from app.startup.composition import database as database_composition +from app.startup.composition.cache import CacheComposition from app.startup.initializers import modules as modules_initializer @@ -55,10 +56,15 @@ def _isolate_startup_failure_cleanup(monkeypatch) -> None: "ModuleManager", "EventManager", "ThreadHelper", - "RedisHelper", - "AsyncRedisHelper", ): monkeypatch.setattr(modules_initializer, name, absent_owner) + # Redis 连接 owner 由缓存组合根按启动配置提供,模拟已接入但尚未建连。 + composition = CacheComposition( + redis_enabled=True, + sync_redis_owner=absent_owner, + async_redis_owner=absent_owner, + ) + monkeypatch.setattr(modules_initializer, "get_cache_composition", lambda: composition) monkeypatch.setattr( modules_initializer, "stop_doh_composition", diff --git a/tests/test_event_runtime_components.py b/tests/test_event_runtime_components.py index 0ef669a38..90144a5be 100644 --- a/tests/test_event_runtime_components.py +++ b/tests/test_event_runtime_components.py @@ -5,6 +5,8 @@ import threading import types from unittest.mock import Mock +import pytest + from app.foundation.singleton import Singleton from app.runtime.event.binding import ( EventBindingResolver, @@ -322,10 +324,24 @@ def test_all_decorated_host_handler_classes_have_explicit_factories() -> None: } +@pytest.mark.parametrize( + ("backend_type", "redis_owners"), + [ + ("redis", {"AsyncRedisHelper", "RedisHelper"}), + ("cachetools", set()), + ], +) def test_all_unmanaged_config_reload_classes_have_explicit_providers( monkeypatch, + compose_cache_backend, + backend_type, + redis_owners, ) -> None: - """专属 resolver 未覆盖的配置 owner 必须全部声明生命周期 Provider。""" + """专属 resolver 未覆盖的配置 owner 必须全部声明生命周期 Provider。 + + Redis 连接 owner 只在启动时选用 Redis 缓存后才有 Provider。 + """ + compose_cache_backend(backend_type) manager = object.__new__(PluginManager) plugin_singleton_key = (PluginManager, (), frozenset()) monkeypatch.setitem(Singleton._instances, plugin_singleton_key, manager) @@ -337,14 +353,12 @@ def test_all_unmanaged_config_reload_classes_have_explicit_providers( providers = get_config_reload_handler_providers() assert {owner.__name__ for owner in providers} == { - "AsyncRedisHelper", "DohHelper", "Monitor", "PluginManager", - "RedisHelper", "SystemHelper", "TransferChain", - } + } | redis_owners def test_error_alert_dedup_preserves_distinct_events_handlers_and_errors(): diff --git a/tests/test_legacy_import_compat.py b/tests/test_legacy_import_compat.py index 10de02104..40e47e630 100644 --- a/tests/test_legacy_import_compat.py +++ b/tests/test_legacy_import_compat.py @@ -518,6 +518,8 @@ def test_symbol_alias_manifest_covers_all_moved_public_symbols(): assert set(SYMBOL_ALIASES["app.schemas"]) == { "TransferTask", "TransferQueue", + "BaseMessage", + "ConversationMemory", } | set(_MESSAGE_NOTIFICATION_SYMBOL_ALIASES) assert set(SYMBOL_ALIASES["app.schemas.transfer"]) == { "TransferTask", @@ -527,7 +529,9 @@ def test_symbol_alias_manifest_covers_all_moved_public_symbols(): "TmdbEpisode", "MediaType", } - assert set(SYMBOL_ALIASES["app.schemas.agent"]) == {"ReplyMode"} + assert set(SYMBOL_ALIASES["app.schemas.agent"]) == {"ConversationMemory", "ReplyMode"} + assert set(SYMBOL_ALIASES["app.sdk.cache"]) == {"AsyncRedisBackend", "RedisBackend"} + assert set(SYMBOL_ALIASES["app.adapters.cache.backends"]) == {"AsyncRedisBackend", "RedisBackend"} assert set(SYMBOL_ALIASES["app.sdk.logging"]) == { "CustomFormatter", "LogConfigModel", diff --git a/tests/test_lifecycle_shutdown.py b/tests/test_lifecycle_shutdown.py index 7c95729e7..84b572f37 100644 --- a/tests/test_lifecycle_shutdown.py +++ b/tests/test_lifecycle_shutdown.py @@ -16,6 +16,7 @@ from app.application.configuration import ( from app.runtime.config import settings as runtime_settings from app.runtime.tasks import configure_task_registry, get_task_registry from app.startup import lifecycle +from app.startup.composition.cache import CacheComposition from app.startup.composition.system import compose_system_service from app.startup.initializers import modules as modules_initializer @@ -1542,7 +1543,6 @@ def _patch_module_shutdown_dependencies(monkeypatch) -> dict: ("ModuleManager", "shutdown"), ("EventManager", "stop_async"), ("ThreadHelper", "shutdown"), - ("RedisHelper", "close"), ): instance = MagicMock() setattr(instance, method_name, MagicMock()) @@ -1552,6 +1552,13 @@ def _patch_module_shutdown_dependencies(monkeypatch) -> dict: key = name.removesuffix("Helper").removesuffix("Manager").lower() dependencies[key] = getattr(instance, method_name) + # Redis 连接 owner 由缓存组合根按启动配置提供,这里模拟已选用 Redis 缓存。 + redis = MagicMock() + redis.close = MagicMock() + redis_type = MagicMock(return_value=redis) + redis_type.get_existing_instance.return_value = redis + dependencies["redis"] = redis.close + stop_doh_composition = MagicMock() monkeypatch.setattr( modules_initializer, @@ -1591,7 +1598,12 @@ def _patch_module_shutdown_dependencies(monkeypatch) -> dict: async_redis.close = AsyncMock() async_redis_type = MagicMock(return_value=async_redis) async_redis_type.get_existing_instance.return_value = async_redis - monkeypatch.setattr(modules_initializer, "AsyncRedisHelper", async_redis_type) + composition = CacheComposition( + redis_enabled=True, + sync_redis_owner=redis_type, + async_redis_owner=async_redis_type, + ) + monkeypatch.setattr(modules_initializer, "get_cache_composition", lambda: composition) dependencies["async_redis"] = async_redis.close close_database = AsyncMock() monkeypatch.setattr(modules_initializer, "close_database", close_database) diff --git a/tests/test_optional_subsystem_composition.py b/tests/test_optional_subsystem_composition.py new file mode 100644 index 000000000..3eb067f63 --- /dev/null +++ b/tests/test_optional_subsystem_composition.py @@ -0,0 +1,127 @@ +"""可选子系统按配置装配的合同测试。 + +Redis 缓存只在启动时选用 Redis 后由缓存组合根接入,智能体对话记忆模型归属智能体模块; +未使用这些功能的实例不得常驻 redis 客户端库或 langchain_core。 +""" + +from __future__ import annotations + +import json +import subprocess +import sys + +from app.runtime import cache as runtime_cache +from app.runtime.config import settings +from app.startup.composition import cache as cache_composition + + +def _run_isolated(script: str) -> dict: + """在全新解释器中执行导入探针,先装测试站点垫片并隔离配置目录。""" + prelude = ( + "from app.testing.bootstrap import ensure_sites_stub, isolate_config_dir\n" + "isolate_config_dir()\n" + "ensure_sites_stub()\n" + ) + completed = subprocess.run( + [sys.executable, "-c", prelude + script], + check=True, + capture_output=True, + text=True, + ) + return json.loads(completed.stdout.strip().splitlines()[-1]) + + +def test_file_cache_startup_does_not_load_redis_or_langchain() -> None: + """文件缓存实例导入宿主应用并完成缓存装配后,不加载 redis 与 langchain_core。""" + result = _run_isolated( + """ +import json +import sys + +import app.factory +from app.startup.composition.cache import get_cache_composition + +forbidden = ("redis", "app.adapters.cache.redis", "langchain_core", "app.agent.memory") +loaded = sorted( + name + for name in sys.modules + if any(name == prefix or name.startswith(prefix + ".") for prefix in forbidden) +) +print(json.dumps({"loaded": loaded, "redis_enabled": get_cache_composition().redis_enabled})) +""" + ) + + assert result == {"loaded": [], "redis_enabled": False} + + +def test_redis_cache_composition_registers_owners_and_backends( + monkeypatch, + compose_cache_backend, +) -> None: + """启动时选用 Redis 才接入 Redis 后端与连接 owner。""" + from app.adapters.cache.redis import AsyncRedisHelper, RedisBackend, RedisHelper + + monkeypatch.setattr(RedisHelper, "_connect", lambda _self: None) + + compose_cache_backend("redis") + composition = cache_composition.get_cache_composition() + + assert composition.redis_enabled is True + assert composition.redis_owners == (RedisHelper, AsyncRedisHelper) + assert isinstance(runtime_cache.FileCache(), RedisBackend) + assert runtime_cache.probe_redis_cache() is True + + +def test_redis_setting_after_local_startup_waits_for_restart( + compose_cache_backend, +) -> None: + """启动时未接入 Redis,运行中改成 redis 继续使用本地缓存且不探测连接。""" + compose_cache_backend("cachetools") + + settings.CACHE_BACKEND_TYPE = "redis" + + assert cache_composition.get_cache_composition().redis_owners == () + assert not runtime_cache.FileCache().is_redis() + assert runtime_cache.probe_redis_cache() is None + + +def test_redis_module_reports_restart_when_redis_not_composed( + compose_cache_backend, +) -> None: + """Redis 模块健康检查不代为建连,未接入时提示重启生效。""" + from app.modules.redis import RedisModule + + compose_cache_backend("cachetools") + settings.CACHE_BACKEND_TYPE = "redis" + + assert RedisModule().test() == (False, "缓存类型已改为Redis,重启后生效") + + +def test_legacy_redis_backend_and_conversation_memory_paths_stay_compatible() -> None: + """Redis 后端与对话记忆模型的旧导入路径惰性解析到同一个对象。""" + result = _run_isolated( + """ +import json + +import app.schemas +import app.sdk.cache +from app.adapters.cache.backends import RedisBackend as backends_redis +from app.adapters.cache.redis import RedisBackend +from app.agent.memory import ConversationMemory +from app.schemas.agent import ConversationMemory as legacy_memory + +print(json.dumps({ + "sdk": app.sdk.cache.RedisBackend is RedisBackend, + "backends": backends_redis is RedisBackend, + "schema_module": legacy_memory is ConversationMemory, + "schema_package": app.schemas.ConversationMemory is ConversationMemory, +})) +""" + ) + + assert result == { + "sdk": True, + "backends": True, + "schema_module": True, + "schema_package": True, + }