fix: resolve type warnings across core modules

astrbot/core/agent/run_context.py

@@ -1,4 +1,4 @@
-from typing import Any, Generic
+from typing import Any, Generic, cast
 
 from pydantic import Field
 from pydantic.dataclasses import dataclass
@@ -13,7 +13,7 @@ TContext = TypeVar("TContext", default=Any)
 class ContextWrapper(Generic[TContext]):
     """A context for running an agent, which can be used to pass additional data or state."""
 
-    context: TContext | None = None
+    context: TContext = cast(TContext, None)
     messages: list[Message] = Field(default_factory=list)
     """This field stores the llm message context for the agent run, agent runners will maintain this field automatically."""
     tool_call_timeout: int = 120  # Default tool call timeout in seconds

astrbot/core/agent/runners/tool_loop_agent_runner.py

@@ -1,5 +1,6 @@
 import asyncio
 import copy
+import os
 import time
 import traceback
 import typing as T
@@ -372,6 +373,7 @@ class ToolLoopAgentRunner(BaseAgentRunner[TContext]):
         overflow_dir = await anyio.Path(
             self.tool_result_overflow_dir,
         ).resolve(strict=False)
+        overflow_dir_path = os.fspath(overflow_dir)
         safe_tool_call_id = (
             "".join(
                 ch if ch.isalnum() or ch in {"-", "_", "."} else "_"
@@ -380,12 +382,13 @@ class ToolLoopAgentRunner(BaseAgentRunner[TContext]):
             or "tool_call"
         )
         file_name = f"{safe_tool_call_id}_{uuid.uuid4().hex[:8]}.txt"
-        overflow_path = overflow_dir / file_name
+        overflow_path = os.path.join(overflow_dir_path, file_name)
 
         def _run() -> str:
-            overflow_dir.mkdir(parents=True, exist_ok=True)
-            overflow_path.write_text(content, encoding="utf-8")
-            return str(overflow_path)
+            os.makedirs(overflow_dir_path, exist_ok=True)
+            with open(overflow_path, "w", encoding="utf-8") as f:
+                f.write(content)
+            return overflow_path
 
         return await asyncio.to_thread(_run)
 
@@ -965,6 +968,8 @@ class ToolLoopAgentRunner(BaseAgentRunner[TContext]):
     ) -> T.AsyncGenerator[_HandleFunctionToolsResult, None]:
         """处理函数工具调用。"""
         tool_call_result_blocks: list[ToolCallMessageSegment] = []
+        last_func_tool_name = "unknown"
+        last_func_tool_id = "unknown"
         logger.info(f"Agent 使用工具: {llm_response.tools_call_name}")
 
         def _append_tool_call_result(tool_call_id: str, content: str) -> None:
@@ -983,6 +988,8 @@ class ToolLoopAgentRunner(BaseAgentRunner[TContext]):
             llm_response.tools_call_ids,
             strict=False,
         ):
+            last_func_tool_name = func_tool_name
+            last_func_tool_id = func_tool_id
             tool_call_streak = self._track_tool_call_streak(func_tool_name)
             yield _HandleFunctionToolsResult.from_message_chain(
                 MessageChain(
@@ -1212,7 +1219,7 @@ class ToolLoopAgentRunner(BaseAgentRunner[TContext]):
                     chain=[
                         Json(
                             data={
-                                "id": func_tool_id,
+                                "id": last_func_tool_id,
                                 "ts": time.time(),
                                 "result": last_tcr_content,
                             },
@@ -1220,7 +1227,7 @@ class ToolLoopAgentRunner(BaseAgentRunner[TContext]):
                     ],
                 ),
             )
-            logger.info(f"Tool `{func_tool_name}` Result: {last_tcr_content}")
+            logger.info(f"Tool `{last_func_tool_name}` Result: {last_tcr_content}")
 
         # 处理函数调用响应
         if tool_call_result_blocks:
@@ -1237,7 +1244,7 @@ class ToolLoopAgentRunner(BaseAgentRunner[TContext]):
         contexts: list[dict[str, T.Any]] = []
         for msg in self.run_context.messages:
             if hasattr(msg, "model_dump"):
-                contexts.append(msg.model_dump())  # type: ignore[call-arg]
+                contexts.append(msg.model_dump())
             elif isinstance(msg, dict):
                 contexts.append(copy.deepcopy(msg))
         instruction = self.SKILLS_LIKE_REQUERY_INSTRUCTION_TEMPLATE.format(

astrbot/core/computer/booters/base.py

@@ -55,10 +55,7 @@ class ComputerBooter(abc.ABC):
     async def shutdown(self, **kwargs) -> None:
         """Shut down the computer sandbox.
 
-        Subclasses may accept extra keyword arguments for
-        type-specific cleanup (e.g. ``delete_sandbox`` for
-        ShipyardNeoBooter).  The default implementation ignores
-        them.
+        Subclasses may accept type-specific keyword arguments.
         """
 
     async def upload_file(self, path: str, file_name: str) -> dict:

astrbot/core/computer/booters/boxlite.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 import asyncio
 import functools
 import random
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, cast
 
 import aiohttp
 import anyio
@@ -15,7 +15,7 @@ from shipyard.shell import ShellComponent as ShipyardShellComponent
 from astrbot.api import logger
 
 if TYPE_CHECKING:
-    from astrbot.core.agent.tool import FunctionTool, ToolSchema
+    from astrbot.core.agent.tool import ToolSchema
 
 from astrbot.core.computer.olayer import (
     FileSystemComponent,
@@ -24,6 +24,7 @@ from astrbot.core.computer.olayer import (
 )
 
 from .base import ComputerBooter
+from .shipyard import ShipyardFileSystemWrapper, ShipyardShellWrapper
 
 
 class MockShipyardSandboxClient:
@@ -176,7 +177,8 @@ class BoxliteBooter(ComputerBooter):
             session_id,
         )
         random_port = random.randint(20000, 30000)
-        self.box = boxlite.SimpleBox(  # type: ignore
+        SimpleBox = vars(boxlite)["SimpleBox"]
+        self.box = SimpleBox(
             image="soulter/shipyard-ship",
             memory_mib=512,
             cpus=1,
@@ -196,25 +198,27 @@ class BoxliteBooter(ComputerBooter):
         self.mocked = MockShipyardSandboxClient(
             sb_url=f"http://127.0.0.1:{random_port}",
         )
-        self._fs = ShipyardFileSystemComponent(
-            client=self.mocked,
+        raw_fs = ShipyardFileSystemComponent(
+            client=cast(Any, self.mocked),
             ship_id=self.box.id,
             session_id=session_id,
         )
         self._python = ShipyardPythonComponent(
-            client=self.mocked,
+            client=cast(Any, self.mocked),
             ship_id=self.box.id,
             session_id=session_id,
         )
-        self._shell = ShipyardShellComponent(
-            client=self.mocked,
+        raw_shell = ShipyardShellComponent(
+            client=cast(Any, self.mocked),
             ship_id=self.box.id,
             session_id=session_id,
         )
+        self._shell = ShipyardShellWrapper(cast(Any, raw_shell))
+        self._fs = ShipyardFileSystemWrapper(cast(Any, raw_fs), self._shell)
 
         await self.mocked.wait_healthy(self.box.id, session_id)
 
-    async def shutdown(self) -> None:
+    async def shutdown(self, **kwargs) -> None:
         logger.info(
             "[Computer] booter_shutdown booter=boxlite ship_id=%s status=starting",
             self.box.id,
@@ -225,6 +229,9 @@ class BoxliteBooter(ComputerBooter):
             self.box.id,
         )
 
+    async def available(self) -> bool:
+        return hasattr(self, "box")
+
     @property
     def fs(self) -> FileSystemComponent:
         return self._fs
@@ -243,7 +250,7 @@ class BoxliteBooter(ComputerBooter):
 
     @classmethod
     @functools.cache
-    def _default_tools(cls) -> tuple[FunctionTool, ...]:
+    def _default_tools(cls) -> tuple[ToolSchema, ...]:
         from astrbot.core.computer.tools import (
             ExecuteShellTool,
             FileDownloadTool,
@@ -251,7 +258,7 @@ class BoxliteBooter(ComputerBooter):
             PythonTool,
         )
 
-        return (  # type: ignore
+        return (
             ExecuteShellTool(),
             PythonTool(),
             FileUploadTool(),

astrbot/core/computer/booters/bwrap.py

@@ -122,7 +122,10 @@ class BwrapShellComponent(ShellComponent):
         timeout: int | None = 30,
         shell: bool = True,
         background: bool = False,
+        session_id: str | None = None,
     ) -> dict[str, Any]:
+        _ = session_id
+
         def _run() -> dict[str, Any]:
             run_env = os.environ.copy()
             if env:
@@ -235,7 +238,14 @@ class HostBackedFileSystemComponent(FileSystemComponent):
         await asyncio.to_thread(os.chmod, p, mode)
         return {"success": True, "path": p}
 
-    async def read_file(self, path: str, encoding: str = "utf-8") -> dict[str, Any]:
+    async def read_file(
+        self,
+        path: str,
+        encoding: str = "utf-8",
+        offset: int | None = None,
+        limit: int | None = None,
+    ) -> dict[str, Any]:
+        _ = offset, limit
         p = self._safe_path(path)
         try:
             content = await asyncio.to_thread(_read_file_sync, p, encoding)
@@ -400,7 +410,7 @@ class BwrapBooter(ComputerBooter):
 {}""".format(test_py["stderr"]),
             )
 
-    async def shutdown(self) -> None:
+    async def shutdown(self, **kwargs) -> None:
         config = self.config
         if config is None:
             return

astrbot/core/computer/booters/cua.py

@@ -2,8 +2,10 @@ from __future__ import annotations
 
 import asyncio
 import base64
+import importlib
 import inspect
 import shlex
+from collections.abc import Callable
 from dataclasses import asdict, dataclass, is_dataclass
 from pathlib import Path
 from typing import Any
@@ -85,12 +87,14 @@ def _maybe_model_dump(value: Any) -> dict[str, Any]:
         return value
     if is_dataclass(value) and not isinstance(value, type):
         return asdict(value)
-    if hasattr(value, "model_dump"):
-        dumped = value.model_dump()
+    model_dump = getattr(value, "model_dump", None)
+    if callable(model_dump):
+        dumped = model_dump()
         if isinstance(dumped, dict):
             return dumped
-    if hasattr(value, "dict"):
-        dumped = value.dict()
+    dict_method = getattr(value, "dict", None)
+    if callable(dict_method):
+        dumped = dict_method()
         if isinstance(dumped, dict):
             return dumped
     attr_payload = {
@@ -139,13 +143,15 @@ def _normalize_process_result(raw: Any) -> ProcessResult:
 
     stdout = first_text("stdout", "output")
     stderr = first_text("stderr", "error")
-    exit_code = payload.get("exit_code")
-    if exit_code is None:
-        exit_code = payload.get("returncode")
-    if exit_code is None:
-        exit_code = payload.get("return_code")
-    if exit_code is None:
+    raw_exit_code = payload.get("exit_code")
+    if raw_exit_code is None:
+        raw_exit_code = payload.get("returncode")
+    if raw_exit_code is None:
+        raw_exit_code = payload.get("return_code")
+    if raw_exit_code is None:
         exit_code = 0 if not stderr else 1
+    else:
+        exit_code = int(raw_exit_code)
     success = bool(payload.get("success", not stderr and exit_code in (0, None)))
     return ProcessResult(
         stdout=stdout,
@@ -284,9 +290,10 @@ class CuaShellComponent(ShellComponent):
         self._sandbox = sandbox
         self._os_type = os_type.lower()
         shell = sandbox.shell
-        self._exec_raw = getattr(shell, "exec", None) or getattr(shell, "run", None)
-        if self._exec_raw is None:
+        exec_raw = getattr(shell, "exec", None) or getattr(shell, "run", None)
+        if exec_raw is None:
             raise RuntimeError("CUA sandbox shell must provide `.exec` or `.run`.")
+        self._exec_raw: Callable[..., Any] = exec_raw
 
     async def exec(
         self,
@@ -296,7 +303,9 @@ class CuaShellComponent(ShellComponent):
         timeout: int | None = 30,
         shell: bool = True,
         background: bool = False,
+        session_id: str | None = None,
     ) -> dict[str, Any]:
+        _ = session_id
         if not shell:
             return {
                 "stdout": "",
@@ -404,11 +413,19 @@ def _write_result(path: str, result: dict[str, Any]) -> dict[str, Any]:
     return {"success": True, "path": path, **result}
 
 
+CUA_DEFAULT_IMAGE = str(CUA_DEFAULT_CONFIG["image"])
+CUA_DEFAULT_OS_TYPE = str(CUA_DEFAULT_CONFIG["os_type"])
+CUA_DEFAULT_TTL = int(CUA_DEFAULT_CONFIG["ttl"])
+CUA_DEFAULT_TELEMETRY_ENABLED = bool(CUA_DEFAULT_CONFIG["telemetry_enabled"])
+CUA_DEFAULT_LOCAL = bool(CUA_DEFAULT_CONFIG["local"])
+CUA_DEFAULT_API_KEY = str(CUA_DEFAULT_CONFIG["api_key"])
+
+
 class CuaFileSystemComponent(FileSystemComponent):
     def __init__(
         self,
         sandbox: Any,
-        os_type: str = CUA_DEFAULT_CONFIG["os_type"],
+        os_type: str = CUA_DEFAULT_OS_TYPE,
     ) -> None:
         self._shell = CuaShellComponent(sandbox, os_type=os_type)
         self._fs_components = _resolve_files_components(sandbox)
@@ -546,6 +563,26 @@ class _PosixShellFileSystem(FileSystemComponent):
             return None
         return _non_posix_filesystem_result(path, self._os_type)
 
+    async def create_file(
+        self,
+        path: str,
+        content: str = "",
+        mode: int = 0o644,
+    ) -> dict[str, Any]:
+        write_result = await self.write_file(path, content)
+        if not write_result.get("success"):
+            return {**write_result, "mode": mode, "mode_applied": False}
+        chmod_result = await self._shell.exec(f"chmod {mode:o} {shlex.quote(path)}")
+        if chmod_result.get("stderr"):
+            return {
+                "success": True,
+                "path": path,
+                "mode": mode,
+                "mode_applied": False,
+                "mode_error": chmod_result["stderr"],
+            }
+        return {"success": True, "path": path, "mode": mode, "mode_applied": True}
+
     async def read_file(
         self,
         path: str,
@@ -623,6 +660,36 @@ class _PosixShellFileSystem(FileSystemComponent):
             before_context=before_context,
         )
 
+    async def edit_file(
+        self,
+        path: str,
+        old_string: str,
+        new_string: str,
+        replace_all: bool = False,
+        encoding: str = "utf-8",
+    ) -> dict[str, Any]:
+        read_result = await self.read_file(path, encoding=encoding)
+        if not read_result.get("success"):
+            return read_result
+        content = str(read_result.get("content", ""))
+        occurrences = content.count(old_string)
+        if occurrences == 0:
+            return {
+                "success": False,
+                "path": path,
+                "error": "old string not found in file",
+                "replacements": 0,
+            }
+        updated = content.replace(old_string, new_string, -1 if replace_all else 1)
+        write_result = await self.write_file(path, updated, encoding=encoding)
+        if not write_result.get("success"):
+            return write_result
+        return {
+            "success": True,
+            "path": path,
+            "replacements": occurrences if replace_all else 1,
+        }
+
 
 async def _list_dir_via_shell(
     shell: CuaShellComponent,
@@ -734,12 +801,12 @@ class _CuaRuntime:
 class CuaBooter(ComputerBooter):
     def __init__(
         self,
-        image: str = CUA_DEFAULT_CONFIG["image"],
-        os_type: str = CUA_DEFAULT_CONFIG["os_type"],
-        ttl: int = CUA_DEFAULT_CONFIG["ttl"],
-        telemetry_enabled: bool = CUA_DEFAULT_CONFIG["telemetry_enabled"],
-        local: bool = CUA_DEFAULT_CONFIG["local"],
-        api_key: str = CUA_DEFAULT_CONFIG["api_key"],
+        image: str = CUA_DEFAULT_IMAGE,
+        os_type: str = CUA_DEFAULT_OS_TYPE,
+        ttl: int = CUA_DEFAULT_TTL,
+        telemetry_enabled: bool = CUA_DEFAULT_TELEMETRY_ENABLED,
+        local: bool = CUA_DEFAULT_LOCAL,
+        api_key: str = CUA_DEFAULT_API_KEY,
     ) -> None:
         self.image = image
         self.os_type = os_type
@@ -752,13 +819,16 @@ class CuaBooter(ComputerBooter):
     async def boot(self, session_id: str) -> None:
         _ = session_id
         try:
-            from cua import Image, Sandbox
+            cua_module = importlib.import_module("cua")
         except ImportError as exc:
             raise RuntimeError(
                 "CUA sandbox support requires the optional `cua` package. "
                 "Install it with `pip install cua` in the AstrBot environment.",
             ) from exc
 
+        Image = vars(cua_module)["Image"]
+        Sandbox = vars(cua_module)["Sandbox"]
+
         image_obj = self._build_image(Image)
         ephemeral_kwargs = self._build_ephemeral_kwargs(Sandbox.ephemeral)
         sandbox_cm = Sandbox.ephemeral(image_obj, **ephemeral_kwargs)
@@ -808,7 +878,7 @@ class CuaBooter(ComputerBooter):
             kwargs["api_key"] = self.api_key
         return kwargs
 
-    async def shutdown(self) -> None:
+    async def shutdown(self, **kwargs) -> None:
         if self._runtime is not None:
             await self._runtime.sandbox_cm.__aexit__(None, None, None)
             self._runtime = None

astrbot/core/computer/booters/local.py

@@ -333,7 +333,7 @@ class LocalBooter(ComputerBooter):
     async def boot(self, session_id: str) -> None:
         logger.info(f"Local computer booter initialized for session: {session_id}")
 
-    async def shutdown(self) -> None:
+    async def shutdown(self, **kwargs) -> None:
         await LocalShellComponent.shutdown_all()
         logger.info("Local computer booter shutdown complete.")
 

astrbot/core/computer/booters/shipyard.py

@@ -44,7 +44,9 @@ class ShipyardShellWrapper:
         timeout: int | None = 300,
         shell: bool = True,
         background: bool = False,
+        session_id: str | None = None,
     ) -> dict[str, Any]:
+        _ = session_id
         if not shell:
             return {
                 "stdout": "",
@@ -246,7 +248,7 @@ class ShipyardBooter(ComputerBooter):
         self._shell = ShipyardShellWrapper(self._ship.shell)  # type: ignore[arg-type]
         self._fs = ShipyardFileSystemWrapper(self._ship.fs, self._shell)  # type: ignore[arg-type]
 
-    async def shutdown(self) -> None:
+    async def shutdown(self, **kwargs) -> None:
         logger.info("[Computer] booter_shutdown booter=shipyard status=done")
 
     @property
@@ -259,7 +261,7 @@ class ShipyardBooter(ComputerBooter):
 
     @property
     def shell(self) -> ShellComponent:
-        return self._shell  # type: ignore[return-value]
+        return self._shell
 
     async def upload_file(self, path: str, file_name: str) -> dict:
         """Upload file to sandbox"""

astrbot/core/computer/booters/shipyard_neo.py

@@ -21,6 +21,7 @@ from astrbot.core.computer.olayer import (
 )
 
 from .shell_background import build_detached_shell_command
+from .shipyard_search_file_util import search_files_via_shell
 
 try:
     from shipyard_neo import BayClient  # noqa: F401
@@ -91,7 +92,9 @@ class NeoShellComponent(ShellComponent):
         timeout: int | None = 300,  # noqa: ASYNC109
         shell: bool = True,
         background: bool = False,
+        session_id: str | None = None,
     ) -> dict[str, Any]:
+        _ = session_id
         if not shell:
             return {
                 "stdout": "",
@@ -163,10 +166,22 @@ class NeoFileSystemComponent(FileSystemComponent):
         await self._sandbox.filesystem.write_file(path, content)
         return {"success": True, "path": path}
 
-    async def read_file(self, path: str, encoding: str = "utf-8") -> dict[str, Any]:
+    async def read_file(
+        self,
+        path: str,
+        encoding: str = "utf-8",
+        offset: int | None = None,
+        limit: int | None = None,
+    ) -> dict[str, Any]:
         _ = encoding
         content = await self._sandbox.filesystem.read_file(path)
-        return {"success": True, "path": path, "content": content}
+        text = str(content)
+        if offset is not None or limit is not None:
+            lines = text.splitlines(keepends=True)
+            start = 0 if offset is None else offset
+            selected = lines[start:] if limit is None else lines[start : start + limit]
+            text = "".join(selected)
+        return {"success": True, "path": path, "content": text}
 
     async def write_file(
         self,
@@ -198,6 +213,57 @@ class NeoFileSystemComponent(FileSystemComponent):
             data.append(item)
         return {"success": True, "path": path, "entries": data}
 
+    async def search_files(
+        self,
+        pattern: str,
+        path: str | None = None,
+        glob: str | None = None,
+        after_context: int | None = None,
+        before_context: int | None = None,
+    ) -> dict[str, Any]:
+        if self._shell is None:
+            raise RuntimeError(
+                "NeoFileSystemComponent requires a shell for search_files."
+            )
+        return await search_files_via_shell(
+            self._shell,
+            pattern=pattern,
+            path=path,
+            glob=glob,
+            after_context=after_context,
+            before_context=before_context,
+        )
+
+    async def edit_file(
+        self,
+        path: str,
+        old_string: str,
+        new_string: str,
+        replace_all: bool = False,
+        encoding: str = "utf-8",
+    ) -> dict[str, Any]:
+        read_result = await self.read_file(path, encoding=encoding)
+        if not read_result.get("success"):
+            return read_result
+        content = str(read_result.get("content", ""))
+        occurrences = content.count(old_string)
+        if occurrences == 0:
+            return {
+                "success": False,
+                "path": path,
+                "error": "old string not found in file",
+                "replacements": 0,
+            }
+        updated = content.replace(old_string, new_string, -1 if replace_all else 1)
+        write_result = await self.write_file(path, updated, encoding=encoding)
+        if not write_result.get("success"):
+            return write_result
+        return {
+            "success": True,
+            "path": path,
+            "replacements": occurrences if replace_all else 1,
+        }
+
 
 class NeoBrowserComponent(BrowserComponent):
     def __init__(self, sandbox: Any) -> None:
@@ -500,7 +566,8 @@ class ShipyardNeoBooter(ComputerBooter):
             )
         return chosen
 
-    async def shutdown(self, *, delete_sandbox: bool = False) -> None:
+    async def shutdown(self, **kwargs) -> None:
+        delete_sandbox = bool(kwargs.get("delete_sandbox", False))
         if self._client is not None:
             sandbox_id = getattr(self._sandbox, "id", "unknown")
 
@@ -622,7 +689,7 @@ class ShipyardNeoBooter(ComputerBooter):
 
     @classmethod
     @functools.cache
-    def _base_tools(cls):
+    def _base_tools(cls) -> tuple[ToolSchema, ...]:
         """4 base + 11 Neo lifecycle = 15 tools (all Neo profiles)."""
         from astrbot.core.computer.tools import (
             AnnotateExecutionTool,
@@ -662,7 +729,7 @@ class ShipyardNeoBooter(ComputerBooter):
 
     @classmethod
     @functools.cache
-    def _browser_tools(cls):
+    def _browser_tools(cls) -> tuple[ToolSchema, ...]:
         from astrbot.core.computer.tools import (
             BrowserBatchExecTool,
             BrowserExecTool,
@@ -681,7 +748,7 @@ class ShipyardNeoBooter(ComputerBooter):
         caps = self.capabilities
         if caps is None:
             return self.__class__.get_default_tools()
-        tools = list(self._base_tools())
+        tools: list[ToolSchema] = list(self._base_tools())
         if "browser" in caps:
             tools.extend(self._browser_tools())
         return tools

astrbot/core/cron/events.py

@@ -5,7 +5,7 @@ from typing import Any
 from astrbot.core.message.components import Plain
 from astrbot.core.message.message_event_result import MessageChain
 from astrbot.core.platform.astr_message_event import AstrMessageEvent
-from astrbot.core.platform.astrbot_message import AstrBotMessage, MessageMember
+from astrbot.core.platform.astrbot_message import AstrBotMessage, Group, MessageMember
 from astrbot.core.platform.message_session import MessageSession
 from astrbot.core.platform.message_type import MessageType
 from astrbot.core.platform.platform_metadata import PlatformMetadata
@@ -63,5 +63,20 @@ class CronMessageEvent(AstrMessageEvent):
         async for chain in generator:
             await self.send(chain)
 
+    async def send_typing(self) -> None:
+        return None
+
+    async def stop_typing(self) -> None:
+        return None
+
+    async def _pre_send(self) -> None:
+        return None
+
+    async def _post_send(self) -> None:
+        return None
+
+    async def get_group(self, group_id: str | None = None, **kwargs) -> Group | None:
+        return None
+
 
 __all__ = ["CronMessageEvent"]

astrbot/core/cron/manager.py

@@ -336,8 +336,6 @@ class CronJobManager:
             cron_event.role = "admin" if sender_id in admin_ids else "member"
         if cron_payload.get("origin", "tool") == "api":
             cron_event.role = "admin"
-        from astrbot.core.computer.computer_tool_provider import ComputerToolProvider
-
         tool_call_timeout = cfg.get("provider_settings", {}).get(
             "tool_call_timeout",
             120,
@@ -346,7 +344,6 @@ class CronJobManager:
             tool_call_timeout=tool_call_timeout,
             llm_safety_mode=False,
             streaming_response=False,
-            tool_providers=[ComputerToolProvider()],
         )
         req = ProviderRequest()
         conv = await _get_session_conv(event=cron_event, plugin_context=self.ctx)

astrbot/core/db/migration/migra_3_to_4.py

@@ -255,6 +255,13 @@ async def migration_persona_data(
             logger.error(f"解析 Persona 配置失败:{e}")
 
 
+def _get_dict_preference(key: str) -> dict[str, object]:
+    value = sp_v3.get(key, default={})
+    if not isinstance(value, dict):
+        raise TypeError(f"旧偏好设置 {key} 应为 dict, 实际为 {type(value).__name__}")
+    return value
+
+
 async def migration_preferences(
     db_helper: BaseDatabase,
     platform_id_map: dict[str, dict[str, str]],
@@ -272,7 +279,7 @@ async def migration_preferences(
         if value is not None:
             await sp.put_async("global", "global", key, value)
             logger.info(f"迁移全局偏好设置 {key} 成功,值: {value}")
-    session_conversation = sp_v3.get("session_conversation", default={})
+    session_conversation = _get_dict_preference("session_conversation")
     for umo, conversation_id in session_conversation.items():
         if not umo or not conversation_id:
             continue
@@ -284,7 +291,7 @@ async def migration_preferences(
             logger.info(f"迁移会话 {umo} 的对话数据到新表成功,平台 ID: {platform_id}")
         except Exception as e:
             logger.error(f"迁移会话 {umo} 的对话数据失败: {e}", exc_info=True)
-    session_service_config = sp_v3.get("session_service_config", default={})
+    session_service_config = _get_dict_preference("session_service_config")
     for umo, config in session_service_config.items():
         if not umo or not config:
             continue
@@ -296,7 +303,7 @@ async def migration_preferences(
             logger.info(f"迁移会话 {umo} 的服务配置到新表成功,平台 ID: {platform_id}")
         except Exception as e:
             logger.error(f"迁移会话 {umo} 的服务配置失败: {e}", exc_info=True)
-    session_variables = sp_v3.get("session_variables", default={})
+    session_variables = _get_dict_preference("session_variables")
     for umo, variables in session_variables.items():
         if not umo or not variables:
             continue
@@ -307,7 +314,7 @@ async def migration_preferences(
             await sp.put_async("umo", str(session), "session_variables", variables)
         except Exception as e:
             logger.error(f"迁移会话 {umo} 的变量失败: {e}", exc_info=True)
-    session_provider_perf = sp_v3.get("session_provider_perf", default={})
+    session_provider_perf = _get_dict_preference("session_provider_perf")
     for umo, perf in session_provider_perf.items():
         if not umo or not perf:
             continue
@@ -315,8 +322,12 @@ async def migration_preferences(
             session = MessageSesion.from_str(session_str=umo)
             platform_id = get_platform_id(platform_id_map, session.platform_name)
             session.platform_id = platform_id
-            perf_dict = perf
-            for provider_type, provider_id in perf_dict.items():
+            if not isinstance(perf, dict):
+                raise TypeError(
+                    f"旧偏好设置 session_provider_perf.{umo} 应为 dict, "
+                    f"实际为 {type(perf).__name__}",
+                )
+            for provider_type, provider_id in perf.items():
                 await sp.put_async(
                     "umo",
                     str(session),

astrbot/core/star/star_manager.py

@@ -177,6 +177,20 @@ async def _install_requirements_with_precheck(
         )
 
 
+async def _get_global_list_preference(key: str) -> list[Any]:
+    value = await sp.global_get(key, [])
+    if not isinstance(value, list):
+        raise TypeError(f"全局偏好设置 {key} 应为 list, 实际为 {type(value).__name__}")
+    return value
+
+
+async def _get_global_dict_preference(key: str) -> dict[Any, Any]:
+    value = await sp.global_get(key, {})
+    if not isinstance(value, dict):
+        raise TypeError(f"全局偏好设置 {key} 应为 dict, 实际为 {type(value).__name__}")
+    return value
+
+
 class PluginManager:
     def __init__(self, context: Context, config: AstrBotConfig) -> None:
         self.updator = PluginUpdator()
@@ -470,6 +484,7 @@ class PluginManager:
         Notes: 旧版本 AstrBot 插件可能使用的是 info() 函数来获取元数据。
         """
         metadata = None
+        raw_metadata: object | None = None
 
         if not os.path.exists(plugin_path):
             raise Exception("插件不存在。")
@@ -479,52 +494,52 @@ class PluginManager:
                 os.path.join(plugin_path, "metadata.yaml"),
                 encoding="utf-8",
             ) as f:
-                metadata = yaml.safe_load(f)
+                raw_metadata = yaml.safe_load(f)
         elif plugin_obj and hasattr(plugin_obj, "info"):
             # 使用 info() 函数
-            metadata = plugin_obj.info()
+            raw_metadata = plugin_obj.info()
 
-        if isinstance(metadata, dict):
-            if "desc" not in metadata and "description" in metadata:
-                metadata["desc"] = metadata["description"]
+        if isinstance(raw_metadata, dict):
+            if "desc" not in raw_metadata and "description" in raw_metadata:
+                raw_metadata["desc"] = raw_metadata["description"]
 
             if (
-                "name" not in metadata
-                or "desc" not in metadata
-                or "version" not in metadata
-                or "author" not in metadata
+                "name" not in raw_metadata
+                or "desc" not in raw_metadata
+                or "version" not in raw_metadata
+                or "author" not in raw_metadata
             ):
                 raise Exception(
                     "插件元数据信息不完整。name, desc, version, author 是必须的字段。",
                 )
             metadata = StarMetadata(
-                name=metadata["name"],
-                author=metadata["author"],
-                desc=metadata["desc"],
+                name=raw_metadata["name"],
+                author=raw_metadata["author"],
+                desc=raw_metadata["desc"],
                 short_desc=(
-                    metadata["short_desc"]
-                    if isinstance(metadata.get("short_desc"), str)
+                    raw_metadata["short_desc"]
+                    if isinstance(raw_metadata.get("short_desc"), str)
                     else None
                 ),
-                version=metadata["version"],
-                repo=metadata["repo"] if "repo" in metadata else None,
-                display_name=metadata.get("display_name", None),
+                version=raw_metadata["version"],
+                repo=raw_metadata["repo"] if "repo" in raw_metadata else None,
+                display_name=raw_metadata.get("display_name", None),
                 support_platforms=(
                     [
                         platform_id
-                        for platform_id in metadata["support_platforms"]
+                        for platform_id in raw_metadata["support_platforms"]
                         if isinstance(platform_id, str)
                     ]
-                    if isinstance(metadata.get("support_platforms"), list)
+                    if isinstance(raw_metadata.get("support_platforms"), list)
                     else []
                 ),
                 astrbot_version=(
-                    metadata["astrbot_version"]
-                    if isinstance(metadata.get("astrbot_version"), str)
+                    raw_metadata["astrbot_version"]
+                    if isinstance(raw_metadata.get("astrbot_version"), str)
                     else None
                 ),
-                pages=metadata["pages"]
-                if isinstance(metadata.get("pages"), list)
+                pages=raw_metadata["pages"]
+                if isinstance(raw_metadata.get("pages"), list)
                 else [],
                 i18n=PluginManager._load_plugin_i18n(plugin_path),
             )
@@ -888,9 +903,11 @@ class PluginManager:
                 - error_message (str|None): 错误信息，成功时为 None
 
         """
-        inactivated_plugins = await sp.global_get("inactivated_plugins", [])
-        inactivated_llm_tools = await sp.global_get("inactivated_llm_tools", [])
-        alter_cmd = await sp.global_get("alter_cmd", {})
+        inactivated_plugins = await _get_global_list_preference("inactivated_plugins")
+        inactivated_llm_tools = await _get_global_list_preference(
+            "inactivated_llm_tools",
+        )
+        alter_cmd = await _get_global_dict_preference("alter_cmd")
 
         plugin_modules = self._get_plugin_modules()
         if plugin_modules is None:
@@ -1145,6 +1162,8 @@ class PluginManager:
                     star_map[path] = metadata
                     star_registry.append(metadata)
 
+                assert metadata.module_path, f"插件 {metadata.name} 模块路径为空"
+
                 # 禁用/启用插件
                 if metadata.module_path in inactivated_plugins:
                     metadata.activated = False
@@ -1153,8 +1172,6 @@ class PluginManager:
                 if await asyncio.to_thread(os.path.exists, logo_path):
                     metadata.logo_path = logo_path
 
-                assert metadata.module_path, f"插件 {metadata.name} 模块路径为空"
-
                 full_names = []
                 for handler in star_handlers_registry.get_handlers_by_module_name(
                     metadata.module_path,
@@ -1163,7 +1180,8 @@ class PluginManager:
 
                     # 检查并且植入自定义的权限过滤器（alter_cmd）
                     if (
-                        metadata.name in alter_cmd
+                        metadata.name is not None
+                        and metadata.name in alter_cmd
                         and handler.handler_name in alter_cmd[metadata.name]
                     ):
                         cmd_type = alter_cmd[metadata.name][handler.handler_name].get(
@@ -1696,12 +1714,14 @@ class PluginManager:
             await self._terminate_plugin(plugin)
 
             # 加入到 shared_preferences 中
-            inactivated_plugins: list = await sp.global_get("inactivated_plugins", [])
+            inactivated_plugins = await _get_global_list_preference(
+                "inactivated_plugins",
+            )
             if plugin.module_path not in inactivated_plugins:
                 inactivated_plugins.append(plugin.module_path)
 
-            inactivated_llm_tools: list = list(
-                set(await sp.global_get("inactivated_llm_tools", [])),
+            inactivated_llm_tools = list(
+                set(await _get_global_list_preference("inactivated_llm_tools")),
             )  # 后向兼容
 
             # 禁用插件启用的 llm_tool
@@ -1769,12 +1789,19 @@ class PluginManager:
             except Exception:
                 logger.error(traceback.format_exc())
 
+    async def cleanup_loaded_plugins(self) -> None:
+        """Terminate all currently loaded plugin instances."""
+        for plugin in self.context.get_all_stars():
+            await self._terminate_plugin(plugin)
+
     async def turn_on_plugin(self, plugin_name: str) -> None:
         plugin = self.context.get_registered_star(plugin_name)
         if plugin is None:
             raise Exception(f"插件 {plugin_name} 不存在。")
-        inactivated_plugins: list = await sp.global_get("inactivated_plugins", [])
-        inactivated_llm_tools: list = await sp.global_get("inactivated_llm_tools", [])
+        inactivated_plugins = await _get_global_list_preference("inactivated_plugins")
+        inactivated_llm_tools = await _get_global_list_preference(
+            "inactivated_llm_tools",
+        )
         if plugin.module_path in inactivated_plugins:
             inactivated_plugins.remove(plugin.module_path)
         await sp.global_put("inactivated_plugins", inactivated_plugins)

pyproject.toml

@@ -152,6 +152,22 @@ exclude = [
     "**/__pycache__",
 ]
 
+[tool.ty.src]
+include = ["astrbot"]
+exclude = [
+    "dashboard",
+    "node_modules",
+    "dist",
+    "data",
+    "tests",
+    "tests/**",
+    "**/tests/**",
+    "**/__pycache__",
+]
+
+[tool.ty.rules]
+all = "ignore"
+
 [tool.mypy]
 plugins = ["pydantic.mypy"]
 mypy_path = "types"