diff --git a/zhenxun/builtin_plugins/admin/plugin_switch/_data_source.py b/zhenxun/builtin_plugins/admin/plugin_switch/_data_source.py
index 2ea48c68..9e21947f 100644
--- a/zhenxun/builtin_plugins/admin/plugin_switch/_data_source.py
+++ b/zhenxun/builtin_plugins/admin/plugin_switch/_data_source.py
@@ -4,6 +4,7 @@ from zhenxun.models.group_console import GroupConsole
 from zhenxun.models.plugin_info import PluginInfo
 from zhenxun.models.task_info import TaskInfo
 from zhenxun.services.cache import CacheRoot
+from zhenxun.services.cache.runtime_cache import TaskInfoMemoryCache
 from zhenxun.utils.common_utils import CommonUtils
 from zhenxun.utils.enum import BlockType, CacheType, PluginType
 from zhenxun.utils.exception import GroupInfoNotFound
@@ -338,9 +339,11 @@ class PluginManager:
         """
         if is_default:
             await TaskInfo.all().update(default_status=False)
+            await TaskInfoMemoryCache.refresh()
             return "已禁用所有被动进群默认状态"
         else:
             await TaskInfo.all().update(status=False)
+            await TaskInfoMemoryCache.refresh()
             return "已全局禁用所有被动状态"
 
     @classmethod
@@ -355,9 +358,11 @@ class PluginManager:
         """
         if is_default:
             await TaskInfo.filter(name=name).update(default_status=False)
+            await TaskInfoMemoryCache.refresh()
             return f"已禁用被动进群默认状态 {name}"
         else:
             await TaskInfo.filter(name=name).update(status=False)
+            await TaskInfoMemoryCache.refresh()
             return f"已全局禁用被动状态 {name}"
 
     @classmethod
@@ -372,9 +377,11 @@ class PluginManager:
         """
         if is_default:
             await TaskInfo.all().update(default_status=True)
+            await TaskInfoMemoryCache.refresh()
             return "已开启所有被动进群默认状态"
         else:
             await TaskInfo.all().update(status=True)
+            await TaskInfoMemoryCache.refresh()
             return "已全局开启所有被动状态"
 
     @classmethod
@@ -390,9 +397,11 @@ class PluginManager:
         """
         if is_default:
             await TaskInfo.filter(name=name).update(default_status=True)
+            await TaskInfoMemoryCache.refresh()
             return f"已开启被动进群默认状态 {name}"
         else:
             await TaskInfo.filter(name=name).update(status=True)
+            await TaskInfoMemoryCache.refresh()
             return f"已全局开启被动状态 {name}"
 
     @classmethod
diff --git a/zhenxun/builtin_plugins/init/init_plugin.py b/zhenxun/builtin_plugins/init/init_plugin.py
index 95433909..f5e1b09f 100644
--- a/zhenxun/builtin_plugins/init/init_plugin.py
+++ b/zhenxun/builtin_plugins/init/init_plugin.py
@@ -14,6 +14,7 @@ from zhenxun.models.group_console import GroupConsole
 from zhenxun.models.plugin_info import PluginInfo
 from zhenxun.models.plugin_limit import PluginLimit
 from zhenxun.models.task_info import TaskInfo
+from zhenxun.services.cache.runtime_cache import TaskInfoMemoryCache
 from zhenxun.services.log import logger
 from zhenxun.utils.enum import (
     BlockType,
@@ -379,6 +380,7 @@ async def group_migration():
                 if close_task := data["close_task"]:
                     """全局被动关闭"""
                     await TaskInfo.filter(module__in=close_task).update(status=False)
+                    await TaskInfoMemoryCache.refresh()
                 group_list = await GroupConsole.filter(
                     group_id__in=old_group_list.keys()
                 )
diff --git a/zhenxun/builtin_plugins/init/init_task.py b/zhenxun/builtin_plugins/init/init_task.py
index b9bab56d..1606e41c 100644
--- a/zhenxun/builtin_plugins/init/init_task.py
+++ b/zhenxun/builtin_plugins/init/init_task.py
@@ -8,6 +8,7 @@ from nonebot_plugin_apscheduler import scheduler
 from zhenxun.configs.utils import PluginExtraData, Task
 from zhenxun.models.group_console import GroupConsole
 from zhenxun.models.task_info import TaskInfo
+from zhenxun.services.cache.runtime_cache import TaskInfoMemoryCache
 from zhenxun.services.log import logger
 from zhenxun.utils.common_utils import CommonUtils
 from zhenxun.utils.manager.priority_manager import PriorityLifecycle
@@ -89,6 +90,8 @@ async def to_db(
     if load_task:
         await TaskInfo.filter(module__in=load_task).update(load_status=True)
         await TaskInfo.filter(module__not_in=load_task).update(load_status=False)
+    if create_list or update_list or load_task:
+        await TaskInfoMemoryCache.refresh()
 
 
 async def get_run_task(task: Task, *args, **kwargs):
@@ -144,6 +147,7 @@ async def _():
         await _handle_setting(plugin, task_info_list, task_list)
     if not task_info_list:
         await TaskInfo.all().update(load_status=False)
+        await TaskInfoMemoryCache.refresh()
         return
     module_dict = {t[1]: t[0] for t in await TaskInfo.all().values_list("id", "module")}
     load_task = []
diff --git a/zhenxun/builtin_plugins/statistics/statistics_hook.py b/zhenxun/builtin_plugins/statistics/statistics_hook.py
index b71102b2..13c67ce9 100644
--- a/zhenxun/builtin_plugins/statistics/statistics_hook.py
+++ b/zhenxun/builtin_plugins/statistics/statistics_hook.py
@@ -11,6 +11,7 @@ from nonebot_plugin_session import EventSession
 from zhenxun.configs.utils import PluginExtraData
 from zhenxun.models.plugin_info import PluginInfo
 from zhenxun.models.statistics import Statistics
+from zhenxun.services.cache.runtime_cache import PluginInfoMemoryCache
 from zhenxun.services.log import logger
 from zhenxun.services.message_load import should_pause_tasks
 from zhenxun.utils.enum import PluginType
@@ -39,7 +40,11 @@ async def _(
         """过滤除poke外的notice"""
         return
     if session.id1 and matcher.plugin:
-        plugin = await PluginInfo.get_plugin(module_path=matcher.plugin.module_name)
+        plugin = PluginInfoMemoryCache.get_by_module_path(matcher.plugin.module_name)
+        if not plugin:
+            plugin = await PluginInfo.get_plugin(module_path=matcher.plugin.module_name)
+            if plugin:
+                PluginInfoMemoryCache.set_plugin(plugin)
         plugin_type = plugin.plugin_type if plugin else None
         if plugin_type == PluginType.NORMAL:
             logger.debug(f"提交调用记录: {matcher.plugin_name}...", session=session)
diff --git a/zhenxun/models/bot_console.py b/zhenxun/models/bot_console.py
index f57a326d..f16e39f5 100644
--- a/zhenxun/models/bot_console.py
+++ b/zhenxun/models/bot_console.py
@@ -62,8 +62,9 @@ class BotConsole(Model):
             list[tuple[str, bool]] | bool: bot状态
         """
         if not bot_id:
-            return await cls.all().values_list("bot_id", "status")
-        result = await cls.get_or_none(bot_id=bot_id)
+            data = await BotMemoryCache.get_all()
+            return [(bot_id, snapshot.status) for bot_id, snapshot in data.items()]
+        result = await BotMemoryCache.get(bot_id)
         return result.status if result else False
 
     @overload
@@ -95,12 +96,13 @@ class BotConsole(Model):
             list[tuple[str, str]] | str: 被动技能
         """
         if not bot_id:
-            task_field: Literal["available_tasks", "block_tasks"] = (
-                "available_tasks" if status else "block_tasks"
-            )
-            data_list = await cls.all().values_list("bot_id", task_field)
-            return {k: cls.convert_module_format(v) for k, v in data_list}
-        result = await cls.get_or_none(bot_id=bot_id)
+            data = await BotMemoryCache.get_all()
+            task_attr = "available_tasks" if status else "block_tasks"
+            return {
+                bot_id: cls.convert_module_format(getattr(snapshot, task_attr))
+                for bot_id, snapshot in data.items()
+            }
+        result = await BotMemoryCache.get(bot_id)
         if result:
             tasks = result.available_tasks if status else result.block_tasks
             return cls.convert_module_format(tasks)
@@ -135,11 +137,14 @@ class BotConsole(Model):
             list[tuple[str, str]] | str: 插件
         """
         if not bot_id:
-            plugin_field = "available_plugins" if status else "block_plugins"
-            data_list = await cls.all().values_list("bot_id", plugin_field)
-            return {k: cls.convert_module_format(v) for k, v in data_list}
+            data = await BotMemoryCache.get_all()
+            plugin_attr = "available_plugins" if status else "block_plugins"
+            return {
+                bot_id: cls.convert_module_format(getattr(snapshot, plugin_attr))
+                for bot_id, snapshot in data.items()
+            }
 
-        result = await cls.get_or_none(bot_id=bot_id)
+        result = await BotMemoryCache.get(bot_id)
         if result:
             plugins = result.available_plugins if status else result.block_plugins
             return cls.convert_module_format(plugins)
@@ -419,7 +424,9 @@ class BotConsole(Model):
         返回:
             bool: 是否被禁用
         """
-        bot_data, _ = await cls.get_or_create(bot_id=bot_id)
+        bot_data = await BotMemoryCache.get(bot_id)
+        if not bot_data:
+            return False
         return cls.format(plugin_name) in bot_data.block_plugins
 
     @classmethod
@@ -434,7 +441,9 @@ class BotConsole(Model):
         返回:
             bool: 是否被禁用
         """
-        bot_data, _ = await cls.get_or_create(bot_id=bot_id)
+        bot_data = await BotMemoryCache.get(bot_id)
+        if not bot_data:
+            return False
         return cls.format(task_name) in bot_data.block_tasks
 
     @classmethod
diff --git a/zhenxun/models/level_user.py b/zhenxun/models/level_user.py
index 764a1f31..3d0ddbb2 100644
--- a/zhenxun/models/level_user.py
+++ b/zhenxun/models/level_user.py
@@ -40,7 +40,7 @@ class LevelUser(Model):
         """
         if not group_id:
             return 0
-        if user := await cls.get_or_none(user_id=user_id, group_id=group_id):
+        if user := await LevelUserMemoryCache.get(user_id, group_id):
             return user.user_level
         return 0
 
@@ -103,11 +103,11 @@ class LevelUser(Model):
         if level == 0:
             return True
         if group_id:
-            if user := await cls.get_or_none(user_id=user_id, group_id=group_id):
+            if user := await LevelUserMemoryCache.get(user_id, group_id):
                 return user.user_level >= level
-        elif user_list := await cls.filter(user_id=user_id).all():
-            user = max(user_list, key=lambda x: x.user_level)
-            return user.user_level >= level
+            return False
+        max_level = await LevelUserMemoryCache.get_max_level(user_id)
+        return max_level >= level
         return False
 
     @classmethod
@@ -121,7 +121,7 @@ class LevelUser(Model):
         返回:
             bool: 是否会被自动更新权限刷新
         """
-        if user := await cls.get_or_none(user_id=user_id, group_id=group_id):
+        if user := await LevelUserMemoryCache.get(user_id, group_id):
             return user.group_flag == 1
         return False
 
diff --git a/zhenxun/models/task_info.py b/zhenxun/models/task_info.py
index d80d7bac..9148a6c0 100644
--- a/zhenxun/models/task_info.py
+++ b/zhenxun/models/task_info.py
@@ -1,5 +1,6 @@
 from tortoise import fields
 
+from zhenxun.services.cache.runtime_cache import TaskInfoMemoryCache
 from zhenxun.services.db_context import Model
 
 
@@ -25,6 +26,27 @@ class TaskInfo(Model):
         table = "task_info"
         table_description = "被动技能基本信息"
 
+    @classmethod
+    async def create(cls, *args, **kwargs):
+        result = await super().create(*args, **kwargs)
+        await TaskInfoMemoryCache.upsert_from_model(result)
+        return result
+
+    @classmethod
+    async def update_or_create(cls, *args, **kwargs):
+        result = await super().update_or_create(*args, **kwargs)
+        await TaskInfoMemoryCache.upsert_from_model(result[0])
+        return result
+
+    async def save(self, *args, **kwargs):
+        await super().save(*args, **kwargs)
+        await TaskInfoMemoryCache.upsert_from_model(self)
+
+    async def delete(self, *args, **kwargs):
+        module = self.module
+        await super().delete(*args, **kwargs)
+        await TaskInfoMemoryCache.remove(module)
+
     @classmethod
     async def _run_script(cls):
         return [
diff --git a/zhenxun/services/cache/__init__.py b/zhenxun/services/cache/__init__.py
index 79088dd6..c4e29e44 100644
--- a/zhenxun/services/cache/__init__.py
+++ b/zhenxun/services/cache/__init__.py
@@ -1092,8 +1092,11 @@ class Cache(Generic[T]):
 
 @driver.on_startup
 async def _():
-    CacheRoot.enabled = True
-    logger.info("缓存系统已启用", LOG_COMMAND)
+    CacheRoot.enabled = cache_config.cache_mode != CacheMode.NONE
+    if CacheRoot.enabled:
+        logger.info("缓存系统已启用", LOG_COMMAND)
+    else:
+        logger.info("缓存系统已禁用", LOG_COMMAND)
 
 
 @driver.on_shutdown
diff --git a/zhenxun/services/cache/runtime_cache.py b/zhenxun/services/cache/runtime_cache.py
index f3bec2bc..55b1249d 100644
--- a/zhenxun/services/cache/runtime_cache.py
+++ b/zhenxun/services/cache/runtime_cache.py
@@ -22,19 +22,19 @@ LOG_COMMAND = "RuntimeCache"
 Config.add_plugin_config(
     "hook",
     "PLUGININFO_MEM_REFRESH_INTERVAL",
-    300,
+    1800,
     help="plugin info memory cache refresh seconds",
 )
 Config.add_plugin_config(
     "hook",
     "BAN_MEM_REFRESH_INTERVAL",
-    60,
+    900,
     help="ban memory cache full refresh seconds",
 )
 Config.add_plugin_config(
     "hook",
     "BAN_MEM_CLEAN_INTERVAL",
-    60,
+    900,
     help="ban memory cache cleanup seconds",
 )
 Config.add_plugin_config(
@@ -52,7 +52,7 @@ Config.add_plugin_config(
 Config.add_plugin_config(
     "hook",
     "BOT_MEM_REFRESH_INTERVAL",
-    60,
+    900,
     help="bot memory cache refresh seconds",
 )
 Config.add_plugin_config(
@@ -64,7 +64,7 @@ Config.add_plugin_config(
 Config.add_plugin_config(
     "hook",
     "GROUP_MEM_REFRESH_INTERVAL",
-    60,
+    900,
     help="group memory cache refresh seconds",
 )
 Config.add_plugin_config(
@@ -76,7 +76,7 @@ Config.add_plugin_config(
 Config.add_plugin_config(
     "hook",
     "LEVEL_MEM_REFRESH_INTERVAL",
-    120,
+    900,
     help="level memory cache refresh seconds",
 )
 Config.add_plugin_config(
@@ -87,8 +87,20 @@ Config.add_plugin_config(
 )
 Config.add_plugin_config(
     "hook",
-    "LIMIT_MEM_REFRESH_INTERVAL",
+    "TASK_MEM_REFRESH_INTERVAL",
+    900,
+    help="task info memory cache refresh seconds",
+)
+Config.add_plugin_config(
+    "hook",
+    "TASK_MEM_NEGATIVE_TTL",
     60,
+    help="task info negative cache ttl seconds",
+)
+Config.add_plugin_config(
+    "hook",
+    "LIMIT_MEM_REFRESH_INTERVAL",
+    900,
     help="plugin limit memory cache refresh seconds",
 )
 Config.add_plugin_config(
@@ -440,6 +452,40 @@ class PluginLimitSnapshot:
         )
 
 
+@dataclass(frozen=True)
+class TaskInfoSnapshot:
+    module: str
+    status: bool
+    load_status: bool
+    default_status: bool
+
+    @classmethod
+    def from_model(cls, model) -> "TaskInfoSnapshot":
+        return cls(
+            module=str(model.module),
+            status=bool(getattr(model, "status", True)),
+            load_status=bool(getattr(model, "load_status", True)),
+            default_status=bool(getattr(model, "default_status", True)),
+        )
+
+    def to_payload(self) -> dict[str, Any]:
+        return {
+            "module": self.module,
+            "status": self.status,
+            "load_status": self.load_status,
+            "default_status": self.default_status,
+        }
+
+    @classmethod
+    def from_payload(cls, payload: dict[str, Any]) -> "TaskInfoSnapshot":
+        return cls(
+            module=str(payload.get("module", "")),
+            status=bool(payload.get("status", True)),
+            load_status=bool(payload.get("load_status", True)),
+            default_status=bool(payload.get("default_status", True)),
+        )
+
+
 class RuntimeCacheSync:
     _redis: ClassVar[Any | None] = None
     _pubsub: ClassVar[Any | None] = None
@@ -583,6 +629,8 @@ class RuntimeCacheSync:
             await BanMemoryCache.apply_sync_event(action, data)
         elif cache_type == "level":
             await LevelUserMemoryCache.apply_sync_event(action, data)
+        elif cache_type == "task":
+            await TaskInfoMemoryCache.apply_sync_event(action, data)
         elif cache_type == "plugin_limit":
             await PluginLimitMemoryCache.apply_sync_event(action, data)
 
@@ -740,6 +788,12 @@ class BotMemoryCache:
         cls._mark_negative(bot_id)
         return None
 
+    @classmethod
+    async def get_all(cls) -> dict[str, BotSnapshot]:
+        if not cls._loaded:
+            await cls.ensure_loaded()
+        return dict(cls._by_id)
+
     @classmethod
     async def update_status(cls, bot_id: str | None, status: bool) -> None:
         bot_id = cls._normalize(bot_id)
@@ -1000,6 +1054,7 @@ class GroupMemoryCache:
 class LevelUserMemoryCache:
     _lock: ClassVar[asyncio.Lock] = asyncio.Lock()
     _by_key: ClassVar[dict[tuple[str, str], LevelUserSnapshot]] = {}
+    _by_user_max: ClassVar[dict[str, int]] = {}
     _negative: ClassVar[dict[tuple[str, str], float]] = {}
     _loaded: ClassVar[bool] = False
     _refresh_task: ClassVar[asyncio.Task | None] = None
@@ -1048,12 +1103,17 @@ class LevelUserMemoryCache:
         async with cls._lock:
             records = await LevelUser.all()
             by_key: dict[tuple[str, str], LevelUserSnapshot] = {}
+            by_user_max: dict[str, int] = {}
             for record in records:
                 entry = LevelUserSnapshot.from_model(record)
                 key = cls._key(entry.user_id, entry.group_id)
                 if key:
                     by_key[key] = entry
+                    current = by_user_max.get(entry.user_id, 0)
+                    if entry.user_level > current:
+                        by_user_max[entry.user_id] = entry.user_level
             cls._by_key = by_key
+            cls._by_user_max = by_user_max
             cls._negative = {}
             cls._loaded = True
             cls._last_refresh = time.time()
@@ -1112,6 +1172,15 @@ class LevelUserMemoryCache:
                 group_user = cls._by_key.get(group_key)
         return global_user, group_user
 
+    @classmethod
+    async def get_max_level(cls, user_id: str | None) -> int:
+        user_id = cls._normalize(user_id)
+        if not user_id:
+            return 0
+        if not cls._loaded:
+            await cls.ensure_loaded()
+        return cls._by_user_max.get(user_id, 0)
+
     @classmethod
     async def upsert_from_model(cls, record) -> None:
         entry = LevelUserSnapshot.from_model(record)
@@ -1119,8 +1188,14 @@ class LevelUserMemoryCache:
         if not key:
             return
         async with cls._lock:
+            prev = cls._by_key.get(key)
             cls._by_key[key] = entry
             cls._negative.pop(key, None)
+            current = cls._by_user_max.get(entry.user_id, 0)
+            if entry.user_level >= current:
+                cls._by_user_max[entry.user_id] = entry.user_level
+            elif prev and prev.user_level == current and entry.user_level < current:
+                cls._recalc_user_max(entry.user_id)
         RuntimeCacheSync.publish_event("level", "upsert", entry.to_payload())
 
     @classmethod
@@ -1130,8 +1205,14 @@ class LevelUserMemoryCache:
         if not key:
             return
         async with cls._lock:
+            prev = cls._by_key.get(key)
             cls._by_key[key] = entry
             cls._negative.pop(key, None)
+            current = cls._by_user_max.get(entry.user_id, 0)
+            if entry.user_level >= current:
+                cls._by_user_max[entry.user_id] = entry.user_level
+            elif prev and prev.user_level == current and entry.user_level < current:
+                cls._recalc_user_max(entry.user_id)
 
     @classmethod
     async def remove(cls, user_id: str | None, group_id: str | None) -> None:
@@ -1139,11 +1220,24 @@ class LevelUserMemoryCache:
         if not key:
             return
         async with cls._lock:
-            cls._by_key.pop(key, None)
+            removed = cls._by_key.pop(key, None)
+            if removed and cls._by_user_max.get(removed.user_id) == removed.user_level:
+                cls._recalc_user_max(removed.user_id)
         RuntimeCacheSync.publish_event(
             "level", "delete", {"user_id": key[0], "group_id": key[1] or None}
         )
 
+    @classmethod
+    def _recalc_user_max(cls, user_id: str) -> None:
+        max_level = 0
+        for entry in cls._by_key.values():
+            if entry.user_id == user_id and entry.user_level > max_level:
+                max_level = entry.user_level
+        if max_level:
+            cls._by_user_max[user_id] = max_level
+        else:
+            cls._by_user_max.pop(user_id, None)
+
     @classmethod
     async def apply_sync_event(cls, action: str, data: dict[str, Any]) -> None:
         if action == "upsert":
@@ -1180,6 +1274,145 @@ class LevelUserMemoryCache:
         cls._refresh_task = None
 
 
+class TaskInfoMemoryCache:
+    _lock: ClassVar[asyncio.Lock] = asyncio.Lock()
+    _by_module: ClassVar[dict[str, TaskInfoSnapshot]] = {}
+    _negative: ClassVar[dict[str, float]] = {}
+    _loaded: ClassVar[bool] = False
+    _refresh_task: ClassVar[asyncio.Task | None] = None
+
+    @classmethod
+    def _normalize(cls, module: str | None) -> str | None:
+        if module is None:
+            return None
+        module = module.strip()
+        return module if module else None
+
+    @classmethod
+    def _negative_ttl(cls) -> int:
+        return _coerce_int(Config.get_config("hook", "TASK_MEM_NEGATIVE_TTL", 60), 60)
+
+    @classmethod
+    def _is_negative(cls, module: str) -> bool:
+        expire_at = cls._negative.get(module)
+        if not expire_at:
+            return False
+        if expire_at <= time.time():
+            cls._negative.pop(module, None)
+            return False
+        return True
+
+    @classmethod
+    def _mark_negative(cls, module: str) -> None:
+        ttl = cls._negative_ttl()
+        if ttl <= 0:
+            return
+        cls._negative[module] = time.time() + ttl
+
+    @classmethod
+    async def refresh(cls) -> None:
+        from zhenxun.models.task_info import TaskInfo
+
+        async with cls._lock:
+            records = await TaskInfo.all()
+            cls._by_module = {r.module: TaskInfoSnapshot.from_model(r) for r in records}
+            cls._negative = {}
+            cls._loaded = True
+            logger.debug(
+                f"task info cache refreshed: {len(cls._by_module)} entries",
+                LOG_COMMAND,
+            )
+
+    @classmethod
+    async def ensure_loaded(cls) -> None:
+        if cls._loaded:
+            return
+        await cls.refresh()
+
+    @classmethod
+    async def get(cls, module: str | None) -> TaskInfoSnapshot | None:
+        module = cls._normalize(module)
+        if not module:
+            return None
+        if not cls._loaded:
+            await cls.ensure_loaded()
+        entry = cls._by_module.get(module)
+        if entry:
+            return entry
+        if cls._is_negative(module):
+            return None
+        cls._mark_negative(module)
+        return None
+
+    @classmethod
+    async def is_disabled(cls, module: str | None) -> bool:
+        entry = await cls.get(module)
+        if not entry:
+            return False
+        return not entry.status
+
+    @classmethod
+    async def upsert_from_model(cls, record) -> None:
+        entry = TaskInfoSnapshot.from_model(record)
+        async with cls._lock:
+            cls._by_module[entry.module] = entry
+            cls._negative.pop(entry.module, None)
+        RuntimeCacheSync.publish_event("task", "upsert", entry.to_payload())
+
+    @classmethod
+    async def upsert_from_payload(cls, payload: dict[str, Any]) -> None:
+        entry = TaskInfoSnapshot.from_payload(payload)
+        if not entry.module:
+            return
+        async with cls._lock:
+            cls._by_module[entry.module] = entry
+            cls._negative.pop(entry.module, None)
+
+    @classmethod
+    async def remove(cls, module: str | None) -> None:
+        module = cls._normalize(module)
+        if not module:
+            return
+        async with cls._lock:
+            cls._by_module.pop(module, None)
+        RuntimeCacheSync.publish_event("task", "delete", {"module": module})
+
+    @classmethod
+    async def apply_sync_event(cls, action: str, data: dict[str, Any]) -> None:
+        if action == "upsert":
+            await cls.upsert_from_payload(data)
+        elif action == "delete":
+            await cls.remove(data.get("module"))
+        elif action == "refresh":
+            await cls.refresh()
+
+    @classmethod
+    async def _refresh_loop(cls, interval: int) -> None:
+        while True:
+            await asyncio.sleep(interval)
+            try:
+                await cls.refresh()
+            except Exception as exc:
+                logger.error("task info cache refresh failed", LOG_COMMAND, e=exc)
+
+    @classmethod
+    def start_tasks(cls) -> None:
+        interval = _coerce_int(
+            Config.get_config("hook", "TASK_MEM_REFRESH_INTERVAL", 300), 300
+        )
+        if interval <= 0:
+            return
+        if cls._refresh_task and not cls._refresh_task.done():
+            return
+        cls._refresh_task = asyncio.create_task(cls._refresh_loop(interval))
+
+    @classmethod
+    def stop_tasks(cls) -> None:
+        if cls._refresh_task and not cls._refresh_task.done():
+            cls._refresh_task.cancel()
+        cls._refresh_task = None
+
+
 class PluginLimitMemoryCache:
     _lock: ClassVar[asyncio.Lock] = asyncio.Lock()
     _by_id: ClassVar[dict[int, PluginLimitSnapshot]] = {}
@@ -1700,6 +1933,10 @@ async def _init_runtime_cache():
         await LevelUserMemoryCache.refresh()
     except Exception as exc:
         logger.error("level cache init failed", LOG_COMMAND, e=exc)
+    try:
+        await TaskInfoMemoryCache.refresh()
+    except Exception as exc:
+        logger.error("task info cache init failed", LOG_COMMAND, e=exc)
     try:
         await PluginLimitMemoryCache.refresh()
     except Exception as exc:
@@ -1712,6 +1949,7 @@ async def _init_runtime_cache():
     BotMemoryCache.start_tasks()
     GroupMemoryCache.start_tasks()
     LevelUserMemoryCache.start_tasks()
+    TaskInfoMemoryCache.start_tasks()
     PluginLimitMemoryCache.start_tasks()
     BanMemoryCache.start_tasks()
     _CACHE_READY_EVENT.set()
@@ -1723,6 +1961,7 @@ async def _stop_runtime_cache():
     BotMemoryCache.stop_tasks()
     GroupMemoryCache.stop_tasks()
     LevelUserMemoryCache.stop_tasks()
+    TaskInfoMemoryCache.stop_tasks()
     PluginLimitMemoryCache.stop_tasks()
     BanMemoryCache.stop_tasks()
     await RuntimeCacheSync.stop()
diff --git a/zhenxun/utils/common_utils.py b/zhenxun/utils/common_utils.py
index d7b14efc..14c8f91d 100644
--- a/zhenxun/utils/common_utils.py
+++ b/zhenxun/utils/common_utils.py
@@ -5,10 +5,13 @@ from nonebot.adapters import Bot
 from nonebot_plugin_uninfo import Session, SupportScope, Uninfo, get_interface
 
 from zhenxun.configs.config import BotConfig
-from zhenxun.models.ban_console import BanConsole
-from zhenxun.models.bot_console import BotConsole
 from zhenxun.models.group_console import GroupConsole
-from zhenxun.models.task_info import TaskInfo
+from zhenxun.services.cache.runtime_cache import (
+    BanMemoryCache,
+    BotMemoryCache,
+    GroupMemoryCache,
+    TaskInfoMemoryCache,
+)
 from zhenxun.services.log import logger
 
 
@@ -39,26 +42,27 @@ class CommonUtils:
             return False
         if not group_id and isinstance(session, Session):
             group_id = session.group.id if session.group else None
-        if task := await TaskInfo.get_or_none(module=module):
+        if await TaskInfoMemoryCache.is_disabled(module):
             """被动全局状态"""
-            if not task.status:
-                return True
-        if not await BotConsole.get_bot_status(session.self_id):
+            return True
+        bot_snapshot = await BotMemoryCache.get(session.self_id)
+        if bot_snapshot and not bot_snapshot.status:
             """bot是否休眠"""
             return True
-        block_tasks = await BotConsole.get_tasks(session.self_id, False)
-        if module in block_tasks:
-            """bot是否禁用被动"""
-            return True
+        if bot_snapshot:
+            block_tasks = cls.convert_module_format(bot_snapshot.block_tasks)
+            if module in block_tasks:
+                """bot是否禁用被动"""
+                return True
         if group_id:
             if await GroupConsole.is_block_task(group_id, module):
                 """群组是否禁用被动"""
                 return True
-            if g := await GroupConsole.get_group(group_id=group_id):
+            if g := GroupMemoryCache.get_if_ready(group_id, None):
                 """群组权限是否小于0"""
                 if g.level < 0:
                     return True
-            if await BanConsole.is_ban(None, group_id):
+            if BanMemoryCache.is_banned(None, group_id):
                 """群组是否被ban"""
                 return True
         return False