diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5389de3..be8c41d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,10 @@
 > 所以不是每个版本号都有条目。条目格式 `## <版本> — <日期>`,新条目加在最上面。
 > 工程口径的完整记录见 `PROGRESS.md` / git log。
 
+## 0.60.27 — 2026-08-03
+
+- 加强服务端命令执行、第三方依赖和文件处理链路的安全防护，并补齐持续类型检查，降低已知依赖漏洞与边界类型错误导致服务异常的风险。
+
 ## 0.60.26 — 2026-08-03
 
 - 默认 DeepSeek Flash 已使用官方 0731 API 升级，继续展示思考过程并显式控制推理强度；不同模型的思考开关不再依赖服务端默认值，切换模型时行为更稳定。
diff --git a/PROGRESS.md b/PROGRESS.md
index 8309eb9..cedb341 100644
--- a/PROGRESS.md
+++ b/PROGRESS.md
@@ -2,7 +2,7 @@
 
 > 配合 `DESIGN.md`。本文件只记 phase 状态、决策偏差、文件量、下一步。每条 1-2 句:做了啥 + 关键判断;细节查 `git log` / `git diff` / `DESIGN §7.9`。
 
-最后更新:2026-08-03(DeepSeek Flash-0731 + thinking 参数统一,bump 0.60.26)
+最后更新:2026-08-03(工程类型门禁 + 安全审计清零,bump 0.60.27)
 
 ---
 
@@ -23,6 +23,7 @@
 
 ### 2026-08-03
 
+- **08-03 / 0.60.27 / 工程类型门禁 + 安全审计清零**:修复 141 个源文件中的 203 项 mypy 错误，新增按第三方无类型依赖与动态工具签名精确收口的 mypy 配置；外部协议规定的弱哈希显式标注非本地密码用途，host shell 前后台执行均改用明确解释器 argv，移除隐式 `shell=True`。提高 aiohttp、Pillow、cryptography、Starlette、python-multipart、pydantic-settings 与 pip 的安全版本下限，`pip check` 无冲突、`pip-audit` 无已知漏洞；新增显式 shell 回归测试，完整审计 448 项 unittest 全绿(17 skip)，Ruff/mypy/Bandit/pip-audit 通过，engineering 80/100 PASS、security 100/100 PASS。无 schema/migration/API 变化，未连接生产 DB。
 - **08-03 / 0.60.26 / DeepSeek Flash-0731 + thinking 参数统一**:默认 `deepseek-v4-flash` 无需换模型 ID 即接入官方 0731 后训练升级；Flash 改为显式开启 thinking 并透传 `reasoning_effort=high`，校准当前常规时段 token 成本但保留 8K 稳定输出预算。模型档案统一用 `thinking_enabled`(开关)+`thinking_transport`(协议)+`reasoning_effort`(强度)，DeepSeek/GLM/方舟共用纯函数请求构造，移除 family 分支与旧 `thinking_mode` 字段；方舟保持既有思考开启，GLM 保持生产验证过的显式关闭，未验证网关标记 `none`。`/v1/models` 同步只返回新字段；445 项 unittest 全绿(17 skip)，Ruff 与 diff 检查通过；未连生产 DB、未发真实模型请求，无 schema/migration/依赖变化。
 - **08-03 / 0.60.25 / 交互式 HTML 预览 + 对话内嵌**:文件预览将 HTML 从普通源码提升为可切换“预览 / 源文件”的 sandbox iframe，允许脚本与 HTTPS CDN/接口但保持 opaque origin，禁止宿主权限、表单和顶层跳转；助手最终答复中的 HTML 产物改为进入可视区才加载的内嵌卡片，并可放大复用完整预览，Markdown 同步补源文件切换。Node 14 项、Python 27 项、JavaScript 语法及 diff 检查通过；当前环境无可用浏览器实例，真实页面点击/截图留部署后冒烟；无 schema、migration、HTTP API 或依赖变化。
 - **08-03 / 0.60.24 / Web Mermaid 直出 + Markdown 围栏容错**:模型偶发用同长度围栏嵌套 Markdown/Mermaid 示例，CommonMark 会把后续正文吞进未闭合代码块；新增仅针对该明确形态的前后端确定性修复，提示词统一要求外层使用更长异类围栏，历史上下文加载时同样修正且不批量回写生产数据。聊天页本地 vendoring Mermaid 11.16.0，仅在助手文字段定稿后顺序渲染 `language-mermaid`，采用 strict 安全级别、文本/边数上限，语法错误或组件不可用时保留源码并提示；真实 Edge 冒烟确认中文流程图与 XYChart 柱线组合图可生成 SVG。Python 27 项、Node 9 项、Ruff、JS/Python 语法及 diff 检查通过；无 schema、migration、HTTP API 或 Python 依赖变化。
diff --git a/core/__init__.py b/core/__init__.py
index f9862a4..39ccda7 100644
--- a/core/__init__.py
+++ b/core/__init__.py
@@ -1,3 +1,3 @@
 # zcbot 版本号单一事实源:web/app.py 的 FastAPI version、/healthz 返回、前端展示都引这里。
 # 改版本只动这一行。
-__version__ = "0.60.26"
+__version__ = "0.60.27"
diff --git a/core/asr_lfasr.py b/core/asr_lfasr.py
index 33bac28..d72f747 100644
--- a/core/asr_lfasr.py
+++ b/core/asr_lfasr.py
@@ -77,7 +77,11 @@ def _load_credentials() -> tuple[str, str]:
 def _auth_params() -> dict:
     appid, secret = _load_credentials()
     ts = str(int(time.time()))
-    md5hex = hashlib.md5((appid + ts).encode()).hexdigest()
+    # 讯飞 LFASR 协议固定要求 MD5(appid+ts) 后再做 HMAC-SHA1；此处不能
+    # 替换算法，且 MD5 仅作为协议输入，不承担本地密码学安全用途。
+    md5hex = hashlib.md5(
+        (appid + ts).encode(), usedforsecurity=False
+    ).hexdigest()
     signa = base64.b64encode(
         hmac.new(secret.encode(), md5hex.encode(), hashlib.sha1).digest()
     ).decode()
@@ -180,7 +184,8 @@ def format_transcript(segments: list[tuple[str, str]]) -> str:
     if len(roles) < 2:
         return "".join(t for _, t in segments).strip()
     lines: list[str] = []
-    cur_role, buf = None, []
+    cur_role: Optional[str] = None
+    buf: list[str] = []
     for role, text in segments:
         if role != cur_role:
             if buf:
@@ -232,7 +237,8 @@ def transcribe_file(
         status = info.get("status")
         if status == -1:
             ft = info.get("failType")
-            hint = _FAIL_HINTS.get(ft, f"转写失败(failType={ft})")
+            fail_type = ft if isinstance(ft, int) else -1
+            hint = _FAIL_HINTS.get(fail_type, f"转写失败(failType={ft})")
             raise LfasrError(f"讯飞录音转写失败:{hint}")
         if status != 4:
             continue  # 0 已创建 / 3 处理中
diff --git a/core/asr_xfyun.py b/core/asr_xfyun.py
index 886b9bb..df2d6b8 100644
--- a/core/asr_xfyun.py
+++ b/core/asr_xfyun.py
@@ -21,6 +21,7 @@ import json
 import os
 import time
 from contextlib import suppress
+from typing import Any
 from urllib.parse import urlencode
 from wsgiref.handlers import format_date_time
 
@@ -109,7 +110,7 @@ class XfyunStream:
     def __init__(self, on_text, *, language: str = "zh_cn"):
         self._on_text = on_text
         self._language = language
-        self._ws = None
+        self._ws: Any = None
         self._recv_task: asyncio.Task | None = None
         self._segs: dict[int, str] = {}   # sn → 该段文本;pgs=rpl 按 rg 区间删旧段
         self._done = asyncio.Event()
diff --git a/core/context.py b/core/context.py
index 038ec26..105ebbb 100644
--- a/core/context.py
+++ b/core/context.py
@@ -212,7 +212,7 @@ def prepare_messages_with_stats(
 
     # 未到上下文压力门槛 → 原样发,零压缩(缓存全暖 + 不丢信息)。压缩是"放不下"才做的事。
     if original_chars < compact_threshold_chars:
-        prepared = [deepcopy(m) for m in messages]
+        unchanged = [deepcopy(m) for m in messages]
         stats = {
             "original_chars": original_chars,
             "sent_chars": original_chars,
@@ -222,7 +222,7 @@ def prepare_messages_with_stats(
             "compaction_skipped": 1,
             "repaired_tool_calls": repaired_tool_calls,
         }
-        return prepared, stats
+        return unchanged, stats
 
     recent_start = max(0, len(messages) - keep_recent)
     prepared: List[dict[str, Any]] = []
diff --git a/core/export_docx.py b/core/export_docx.py
index 92e64a5..f2d6f1b 100644
--- a/core/export_docx.py
+++ b/core/export_docx.py
@@ -23,7 +23,8 @@ from uuid import UUID
 
 from core.task import TaskState
 
-from docx import Document
+from docx import Document as create_document
+from docx.document import Document
 from docx.enum.text import WD_ALIGN_PARAGRAPH
 from docx.oxml.ns import qn
 from docx.shared import Cm, Pt, RGBColor
@@ -56,7 +57,7 @@ def _preserve_spaces(run) -> None:
 # ───────────────────────── 文档骨架 ─────────────────────────
 
 def _init_doc() -> Document:
-    doc = Document()
+    doc = create_document()
     section = doc.sections[0]
     section.page_height = Cm(29.7)
     section.page_width = Cm(21)
diff --git a/core/file_store.py b/core/file_store.py
index fe7b9ed..1449b61 100644
--- a/core/file_store.py
+++ b/core/file_store.py
@@ -83,7 +83,11 @@ def _try_lock(f) -> bool:
     import fcntl
 
     try:
-        fcntl.flock(f.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+        flock = getattr(fcntl, "flock")
+        flock(
+            f.fileno(),
+            getattr(fcntl, "LOCK_EX") | getattr(fcntl, "LOCK_NB"),
+        )
         return True
     except BlockingIOError:
         return False
@@ -99,7 +103,7 @@ def _unlock(f) -> None:
 
     import fcntl
 
-    fcntl.flock(f.fileno(), fcntl.LOCK_UN)
+    getattr(fcntl, "flock")(f.fileno(), getattr(fcntl, "LOCK_UN"))
 
 
 @contextmanager
diff --git a/core/kb_ingest.py b/core/kb_ingest.py
index 8d8de57..40da6b9 100644
--- a/core/kb_ingest.py
+++ b/core/kb_ingest.py
@@ -216,7 +216,9 @@ def _doc_name_for(d: Path, source_name: str) -> str:
     doc_rel = f"docs/{stem}.md"
     for e in entries:
         if e["doc"] == doc_rel and e["source"] != f"sources/{source_name}":
-            suffix = hashlib.md5(source_name.encode("utf-8")).hexdigest()[:6]
+            suffix = hashlib.md5(
+                source_name.encode("utf-8"), usedforsecurity=False
+            ).hexdigest()[:6]
             return f"{stem}_{suffix}.md"
     return f"{stem}.md"
 
@@ -301,8 +303,8 @@ def _run_ingest_locked(
                 print(f"[kb_ingest] {kb_name}/{name} failed: {msg}", flush=True)
         return True
     finally:
-        st = _status.get(key)
-        if st is not None:
-            st["running"] = False
-            st["current"] = None
-            st["finished_at"] = datetime.now().isoformat(timespec="seconds")
+        final_state = _status.get(key)
+        if final_state is not None:
+            final_state["running"] = False
+            final_state["current"] = None
+            final_state["finished_at"] = datetime.now().isoformat(timespec="seconds")
diff --git a/core/loop.py b/core/loop.py
index d3444ab..397b385 100644
--- a/core/loop.py
+++ b/core/loop.py
@@ -162,7 +162,9 @@ class _RepeatGuard:
         一步里只要有一次净产出就算在推进。
         """
         st = self._state(name, args)
-        h = hashlib.sha1(result.encode("utf-8", "replace")).hexdigest()
+        h = hashlib.sha1(
+            result.encode("utf-8", "replace"), usedforsecurity=False
+        ).hexdigest()
         esig = _tool_error_signature(name, result)
         is_err = esig is not None
         dup = h in st["hashes"]
@@ -303,6 +305,7 @@ class AgentLoop:
                 self._emit({"type": "cancelled"})
                 return "[cancelled]"
 
+            assert response is not None
             msg = response.choices[0].message
             tool_calls = getattr(msg, "tool_calls", None) or []
             asst_msg_id = self.session.append(
@@ -624,7 +627,9 @@ class AgentLoop:
                     self._emit({"type": "reasoning", "delta": delta_reasoning})
         finally:
             # generator 提前 break 时 GeneratorExit 触发 chat_stream finally → close 底层连接
-            stream.close()
+            close = getattr(stream, "close", None)
+            if callable(close):
+                close()
 
         if cancelled:
             return None, True
diff --git a/core/markdown_guard.py b/core/markdown_guard.py
index aa94cd0..bb3e3a6 100644
--- a/core/markdown_guard.py
+++ b/core/markdown_guard.py
@@ -92,6 +92,7 @@ def normalize_markdown_fences(text: str) -> MarkdownFenceResult:
         ):
             idx += 1
             continue
+        assert inner_idx is not None
 
         inner_close_idx = None
         for candidate in range(inner_idx + 1, len(lines)):
@@ -118,6 +119,7 @@ def normalize_markdown_fences(text: str) -> MarkdownFenceResult:
         ):
             idx += 1
             continue
+        assert outer_close_idx is not None
 
         repaired_length = max(outer[2], inner[2]) + 1
         lines[idx] = _replace_fence(lines[idx], repaired_length)
diff --git a/core/pptx_guard.py b/core/pptx_guard.py
index b740ddd..a013a46 100644
--- a/core/pptx_guard.py
+++ b/core/pptx_guard.py
@@ -15,11 +15,16 @@ svg_to_pptx.py,用外部渲染器把每页渲成整页 PNG 贴进幻灯片交付
 from __future__ import annotations
 
 from pathlib import Path
-from typing import Iterator, List, Optional
+from typing import Any, Iterator, List, Optional
+
+Presentation: Any
+MSO_SHAPE_TYPE: Any
 
 try:
-    from pptx import Presentation
-    from pptx.enum.shapes import MSO_SHAPE_TYPE
+    from pptx import Presentation as _Presentation
+    from pptx.enum.shapes import MSO_SHAPE_TYPE as _MSO_SHAPE_TYPE
+    Presentation = _Presentation
+    MSO_SHAPE_TYPE = _MSO_SHAPE_TYPE
 except Exception:  # pragma: no cover - 宿主缺 python-pptx 时静默降级为不检
     Presentation = None
     MSO_SHAPE_TYPE = None
diff --git a/core/proc_wrapper.py b/core/proc_wrapper.py
index c4dd21e..d4a6b91 100644
--- a/core/proc_wrapper.py
+++ b/core/proc_wrapper.py
@@ -23,6 +23,13 @@ from pathlib import Path
 MAX_LOG_BYTES = 10 * 1024 * 1024
 
 
+def _shell_argv(command: str) -> list[str]:
+    """兼容升级前已落盘的 shell_cmd 元数据，不使用 subprocess 隐式 shell。"""
+    if os.name == "nt":
+        return [os.environ.get("COMSPEC") or "cmd.exe", "/d", "/s", "/c", command]
+    return ["/bin/sh", "-c", command]
+
+
 def _read_meta(d: Path) -> dict:
     return json.loads((d / "proc.json").read_text(encoding="utf-8"))
 
@@ -37,7 +44,7 @@ def _kill_tree(child: subprocess.Popen) -> None:
     if os.name == "posix":
         import signal
         try:
-            os.killpg(child.pid, signal.SIGKILL)
+            getattr(os, "killpg")(child.pid, getattr(signal, "SIGKILL"))
         except (ProcessLookupError, PermissionError, OSError):
             try:
                 child.kill()
@@ -87,7 +94,7 @@ def main() -> int:
         if meta.get("argv"):
             child = subprocess.Popen(meta["argv"], **spawn_kw)
         else:
-            child = subprocess.Popen(meta["shell_cmd"], shell=True, **spawn_kw)
+            child = subprocess.Popen(_shell_argv(meta["shell_cmd"]), **spawn_kw)
     except Exception as e:  # 启动失败也要写 exit_code,否则状态永远悬着
         log.write(f"[proc_wrapper] spawn failed: {type(e).__name__}: {e}\n".encode("utf-8"))
         log.close()
diff --git a/core/procs.py b/core/procs.py
index 9cec635..5423667 100644
--- a/core/procs.py
+++ b/core/procs.py
@@ -225,6 +225,12 @@ def finished_at(d: Path) -> Optional[float]:
 
 # ───────────── host 模式启动 / 终止 ─────────────
 
+def shell_argv(command: str) -> List[str]:
+    """用明确的系统解释器执行 shell 语法，避免 subprocess 隐式 shell。"""
+    if os.name == "nt":
+        return [os.environ.get("COMSPEC") or "cmd.exe", "/d", "/s", "/c", command]
+    return ["/bin/sh", "-c", command]
+
 def launch_host(
     anchor: Path,
     task_id: str,
@@ -310,10 +316,10 @@ def kill_proc(meta: Dict[str, Any], d: Path) -> str:
             for pid in (child_pid, wrapper_pid):
                 if pid > 0:
                     try:
-                        os.killpg(pid, signal.SIGKILL)
+                        getattr(os, "killpg")(pid, getattr(signal, "SIGKILL"))
                     except (ProcessLookupError, PermissionError, OSError):
                         try:
-                            os.kill(pid, signal.SIGKILL)
+                            os.kill(pid, getattr(signal, "SIGKILL"))
                         except OSError:
                             pass
         else:
diff --git a/core/sandbox/pool.py b/core/sandbox/pool.py
index 45a6397..fb10e8b 100644
--- a/core/sandbox/pool.py
+++ b/core/sandbox/pool.py
@@ -141,7 +141,9 @@ class SandboxPool:
         """
         self.user_root_base = user_root_base
         self.repo_root = repo_root
-        self.image = image or os.getenv("ZCBOT_SANDBOX_IMAGE", DEFAULT_IMAGE)
+        self.image: str = (
+            image or os.getenv("ZCBOT_SANDBOX_IMAGE") or DEFAULT_IMAGE
+        )
         self.runtime = runtime or os.getenv("ZCBOT_SANDBOX_RUNTIME") or ""
         self.idle_ttl = idle_ttl if idle_ttl is not None else int(
             os.getenv("ZCBOT_SANDBOX_IDLE_TTL", str(DEFAULT_IDLE_TTL_SECONDS))
@@ -370,12 +372,16 @@ def setup_pool(
     dns_cfg = cfg.get("dns") or []
     if not isinstance(dns_cfg, list):
         dns_cfg = []
+    memory = cfg.get("memory")
+    cpus = cfg.get("cpus")
+    pids_limit = cfg.get("pids_limit")
+    shm_size = cfg.get("shm_size")
     return SandboxPool(
         user_root_base=user_root_base,
         repo_root=repo_root,
-        memory=cfg.get("memory") if isinstance(cfg.get("memory"), str) else None,
-        cpus=str(cfg["cpus"]) if cfg.get("cpus") is not None else None,
-        pids_limit=int(cfg["pids_limit"]) if cfg.get("pids_limit") is not None else None,
-        shm_size=cfg.get("shm_size") if isinstance(cfg.get("shm_size"), str) else None,
+        memory=memory if isinstance(memory, str) else None,
+        cpus=str(cpus) if cpus is not None else None,
+        pids_limit=int(str(pids_limit)) if pids_limit is not None else None,
+        shm_size=shm_size if isinstance(shm_size, str) else None,
         dns=[str(x) for x in dns_cfg],
     )
diff --git a/core/sandbox/tool_runner.py b/core/sandbox/tool_runner.py
index 4777018..d2d5597 100644
--- a/core/sandbox/tool_runner.py
+++ b/core/sandbox/tool_runner.py
@@ -20,6 +20,7 @@ import os
 import sys
 import traceback
 from pathlib import Path
+from typing import Any
 
 
 # 镜像里 /sandbox/ 下放了 tools/ 的拷贝,让 import 走 /sandbox/
@@ -66,7 +67,8 @@ def main() -> int:
         )
         return 2
 
-    tool = cls(base_dir=Path(os.getcwd()), user_root=Path("/workspace"))
+    tool_cls: Any = cls
+    tool = tool_cls(base_dir=Path(os.getcwd()), user_root=Path("/workspace"))
     try:
         result = tool.execute(**args)
     except TypeError as e:
diff --git a/core/sinks.py b/core/sinks.py
index 39f1587..8c9273e 100644
--- a/core/sinks.py
+++ b/core/sinks.py
@@ -20,6 +20,7 @@ from typing import Optional
 
 from rich.console import Console
 from rich.markdown import Markdown
+from rich.status import Status
 
 
 class ConsoleEventSink:
@@ -28,7 +29,7 @@ class ConsoleEventSink:
 
     def __init__(self, console: Console) -> None:
         self.console = console
-        self._status = None
+        self._status: Optional[Status] = None
         self._stop: Optional[threading.Event] = None
         self._thread: Optional[threading.Thread] = None
         self._start = 0.0
@@ -59,19 +60,21 @@ class ConsoleEventSink:
 
     def _spinner_start(self) -> None:
         self._start = time.monotonic()
-        self._stop = threading.Event()
+        stop = threading.Event()
+        self._stop = stop
 
         def fmt() -> str:
             elapsed = time.monotonic() - self._start
             return f"[muted]thinking... {elapsed:.1f}s[/muted]"
 
-        self._status = self.console.status(fmt(), spinner="dots")
-        self._status.__enter__()
+        status = self.console.status(fmt(), spinner="dots")
+        self._status = status
+        status.__enter__()
 
         def tick() -> None:
-            while not self._stop.wait(0.1):
+            while not stop.wait(0.1):
                 try:
-                    self._status.update(fmt())
+                    status.update(fmt())
                 except Exception:
                     return
 
diff --git a/core/storage/utils.py b/core/storage/utils.py
index 7e6a417..29f1867 100644
--- a/core/storage/utils.py
+++ b/core/storage/utils.py
@@ -101,7 +101,7 @@ def upsert_task(
     """
     values = {"task_id": task_id, "user_id": user_id, **fields}
     stmt = insert(Task).values(**values)
-    update_cols = {k: stmt.excluded[k] for k in fields}
+    update_cols: dict[str, Any] = {k: stmt.excluded[k] for k in fields}
     if update_cols:
         # ORM 的 onupdate=func.now() 只在 ORM-level UPDATE 触发,DO UPDATE 是 raw DML
         # 不会自动刷 updated_at —— 这里显式追加。
@@ -126,7 +126,7 @@ def update_task(task_id: UUID, **fields: Any) -> int:
         result = s.execute(
             update(Task).where(Task.task_id == task_id).values(**fields)
         )
-        return result.rowcount or 0
+        return int(getattr(result, "rowcount", 0) or 0)
 
 
 def get_task(task_id: UUID) -> Optional[Task]:
diff --git a/core/task_title.py b/core/task_title.py
index cbf4064..eecba5a 100644
--- a/core/task_title.py
+++ b/core/task_title.py
@@ -121,7 +121,7 @@ def generate_task_title(
             )
             .values(**values)
         )
-        applied = bool(result.rowcount)
+        applied = bool(getattr(result, "rowcount", 0))
     if response is not None and caps is not None:
         usage = getattr(response, "usage", None)
         try:
diff --git a/core/tool_registry.py b/core/tool_registry.py
index f9b0021..a179c41 100644
--- a/core/tool_registry.py
+++ b/core/tool_registry.py
@@ -63,7 +63,7 @@ class ToolContext:
     tool_base: Path            # fs/shell 类工具的 base_dir(cwd / task 目录)
     ur_path: Path              # user_root(输出渲染相对路径 + host-side 落点)
     working_dir_path: Path     # 该 task 的宿主工作目录绝对路径
-    task_id: str
+    task_id: UUID
     uid: UUID
     cfg: dict                  # config/agent.yaml(quotas 段)
     caps: Any                  # ModelCapabilities(enable_run_python)
@@ -115,8 +115,8 @@ def build_tools(ctx: ToolContext) -> dict[str, Any]:
             AskUserTool(**base),
             ReadTool(**base), WriteTool(**base), EditTool(**base),
             GlobTool(**base), GrepTool(**base),
-            ShellTool(task_id=ctx.task_id, **base),
-            CheckProcessTool(task_id=ctx.task_id, **base),
+            ShellTool(task_id=str(ctx.task_id), **base),
+            CheckProcessTool(task_id=str(ctx.task_id), **base),
             WebFetchTool(**base),
         ]
 
@@ -183,7 +183,7 @@ def build_tools(ctx: ToolContext) -> dict[str, Any]:
         return [WechatPushTool(ctx.uid, task_id=ctx.task_id, **wd_base)]
 
     def _run_python() -> list:
-        return [RunPythonTool(task_id=ctx.task_id, **base)]
+        return [RunPythonTool(task_id=str(ctx.task_id), **base)]
 
     def _office_to_pdf() -> list:
         # LibreOffice 只装在 backend host；Docker agent 通过 typed tool 转换用户目录内
@@ -195,7 +195,7 @@ def build_tools(ctx: ToolContext) -> dict[str, Any]:
         # 媒体段同源);本次 run 锁定该 variant,下一条消息可重选。
         if ctx.img_cfg is None:
             return []
-        cls_kwargs = dict(
+        cls_kwargs: dict[str, Any] = dict(
             image_variant_cfg=ctx.img_cfg, variant_key=ctx.img_key,
             working_dir=ctx.working_dir_path, task_id=ctx.task_id, user_id=ctx.uid,
             daily_limit=images_per_day, **base,
@@ -235,7 +235,9 @@ def build_tools(ctx: ToolContext) -> dict[str, Any]:
         )]
 
     def _web_search() -> list:
-        return [WebSearchTool(cfg=BochaConfig.load())]
+        cfg = BochaConfig.load()
+        assert cfg is not None
+        return [WebSearchTool(cfg=cfg)]
 
     # ── 注册表:(组名, gate, factory)。gate 判定统一零参 bool;新工具在此加行 ──
     registry: list[tuple[str, Callable[[], bool], Callable[[], list]]] = [
diff --git a/core/toolfail.py b/core/toolfail.py
index f35abee..51886d1 100644
--- a/core/toolfail.py
+++ b/core/toolfail.py
@@ -351,8 +351,8 @@ def scan_tool_wire_health(days: int = 7) -> Dict[str, Any]:
             {"cutoff": cutoff, "cutoff_24h": cutoff_24h},
         ).fetchall()
 
-    out = []
-    totals = {
+    out: list[dict[str, Any]] = []
+    counts: dict[str, int] = {
         "salvaged": 0,
         "malformed": 0,
         "salvaged_24h": 0,
@@ -377,7 +377,7 @@ def scan_tool_wire_health(days: int = 7) -> Dict[str, Any]:
             ("salvaged_24h", saved_24h),
             ("malformed_24h", residual_24h),
         ):
-            totals[key] += value
+            counts[key] += value
         out.append({
             "model_profile": model_profile or "?",
             "tool": tool or "?",
@@ -399,9 +399,10 @@ def scan_tool_wire_health(days: int = 7) -> Dict[str, Any]:
         ),
         reverse=True,
     )
-    totals["recovery_rate"] = _rate(totals["salvaged"], totals["malformed"])
+    totals: dict[str, int | float | None] = dict(counts)
+    totals["recovery_rate"] = _rate(counts["salvaged"], counts["malformed"])
     totals["recovery_rate_24h"] = _rate(
-        totals["salvaged_24h"], totals["malformed_24h"]
+        counts["salvaged_24h"], counts["malformed_24h"]
     )
     return {"days": days, "rows": out, "total": totals}
 
diff --git a/core/wechat/ilink.py b/core/wechat/ilink.py
index f52673a..0fd54ae 100644
--- a/core/wechat/ilink.py
+++ b/core/wechat/ilink.py
@@ -333,7 +333,9 @@ class ILinkClient:
     # —— 发文件(getuploadurl → AES-128-ECB → CDN → file_item)——
     def _upload_file(self, to_user_id: str, data: bytes) -> dict[str, Any]:
         rawsize = len(data)
-        rawmd5 = hashlib.md5(data).hexdigest()
+        # iLink 上传协议字段名即 rawfilemd5，算法由服务端契约固定；这里只做
+        # 传输完整性字段，不用于本地认证或密码存储。
+        rawmd5 = hashlib.md5(data, usedforsecurity=False).hexdigest()
         aeskey = os.urandom(16)
         filekey = os.urandom(16).hex()
         ciphertext = _aes_ecb_pkcs7(data, aeskey)
diff --git a/core/wechat/wecom_crypto.py b/core/wechat/wecom_crypto.py
index 7ccb457..6841b30 100644
--- a/core/wechat/wecom_crypto.py
+++ b/core/wechat/wecom_crypto.py
@@ -42,7 +42,10 @@ def _aes_key() -> bytes:
 
 def _signature(timestamp: str, nonce: str, encrypt: str) -> str:
     arr = sorted([callback_token(), timestamp, nonce, encrypt])
-    return hashlib.sha1("".join(arr).encode("utf-8")).hexdigest()
+    # 企业微信回调协议固定使用 SHA-1；算法不可单方面升级。
+    return hashlib.sha1(
+        "".join(arr).encode("utf-8"), usedforsecurity=False
+    ).hexdigest()
 
 
 def _aes_decrypt(encrypt_b64: str) -> bytes:
diff --git a/evaluation/scoring.py b/evaluation/scoring.py
index a82cc80..f97b1d7 100644
--- a/evaluation/scoring.py
+++ b/evaluation/scoring.py
@@ -87,9 +87,11 @@ def evaluate_assertion(
             text = observation.artifact(spec.path).decode("utf-8")
             passed = spec.value in text
         elif spec.type == "max_cost_cny":
+            assert spec.max_value is not None
             passed = observation.cost_cny <= float(spec.max_value)
             detail = f"{observation.cost_cny:.6f} CNY"
         elif spec.type == "max_duration_s":
+            assert spec.max_value is not None
             passed = observation.duration_s <= float(spec.max_value)
             detail = f"{observation.duration_s:.3f}s"
     except (FileNotFoundError, UnicodeDecodeError, zipfile.BadZipFile, OSError) as exc:
diff --git a/mypy.ini b/mypy.ini
new file mode 100644
index 0000000..be3ebfb
--- /dev/null
+++ b/mypy.ini
@@ -0,0 +1,24 @@
+[mypy]
+python_version = 3.12
+
+# Tool implementations expose heterogeneous, JSON-schema-defined signatures.
+# They are invoked dynamically by ExecutorHost from JSON arguments, so the
+# uniform Tool.execute(**kwargs) declaration is an interface marker rather than
+# a substitutable Python call signature.
+[mypy-tools.*]
+disable_error_code = override
+
+# These runtime dependencies do not publish typing metadata. Keep the exception
+# scoped to their import namespaces instead of suppressing missing imports for
+# application modules.
+[mypy-yaml.*]
+ignore_missing_imports = True
+
+[mypy-markdown.*]
+ignore_missing_imports = True
+
+[mypy-pilk.*]
+ignore_missing_imports = True
+
+[mypy-croniter.*]
+ignore_missing_imports = True
diff --git a/rendering/docx_brief.py b/rendering/docx_brief.py
index e09fa5a..5715804 100644
--- a/rendering/docx_brief.py
+++ b/rendering/docx_brief.py
@@ -11,7 +11,8 @@ import re
 import sys
 from pathlib import Path
 
-from docx import Document
+from docx import Document as create_document
+from docx.document import Document
 from docx.enum.text import WD_ALIGN_PARAGRAPH
 from docx.opc.constants import RELATIONSHIP_TYPE as RT
 from docx.oxml import OxmlElement
@@ -141,7 +142,7 @@ def add_external_link(paragraph, url: str, text: str, *, size_pt: float) -> None
 # ───────────────────────── 文档初始化 ─────────────────────────
 
 def init_doc(color: bool) -> Document:
-    doc = Document()
+    doc = create_document()
     section = doc.sections[0]
     section.page_height = Cm(29.7)
     section.page_width = Cm(21)
diff --git a/rendering/docx_manuscript.py b/rendering/docx_manuscript.py
index 8a51426..810c037 100644
--- a/rendering/docx_manuscript.py
+++ b/rendering/docx_manuscript.py
@@ -13,7 +13,8 @@ import re
 import sys
 from pathlib import Path
 
-from docx import Document
+from docx import Document as create_document
+from docx.document import Document
 from docx.enum.text import WD_ALIGN_PARAGRAPH
 from docx.oxml import OxmlElement
 from docx.oxml.ns import qn
@@ -79,7 +80,7 @@ PROFILES = {
 # ───────────────────────── 文档初始化 ─────────────────────────
 
 def init_doc(prof: dict) -> Document:
-    doc = Document()
+    doc = create_document()
 
     section = doc.sections[0]
     section.page_height = Cm(29.7)
diff --git a/requirements.txt b/requirements.txt
index bb2db01..e5799ca 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,4 +1,9 @@
+# 打包工具也纳入 pip-audit；安全下限需随部署一起升级，避免应用依赖干净但
+# 虚拟环境自带的 pip 仍触发已知漏洞。
+pip>=26.1.2
+
 litellm>=1.83.0    # zai provider(GLM)要 ≥1.83;PR #17307 merge 后才内置
+aiohttp>=3.14.1    # litellm 传递依赖；3.14.1 修复 2026-08 pip-audit 命中项
 pyyaml>=6.0
 click>=8.1.0
 rich>=13.7.0
@@ -7,7 +12,7 @@ rich>=13.7.0
 python-pptx>=0.6.21
 python-docx>=1.1.0
 matplotlib>=3.8.0
-Pillow>=9.0.0    # ppt skill(SVG-first)svg_finalize:配图裁切/内嵌
+Pillow>=12.3.0   # ppt skill(SVG-first)svg_finalize:配图裁切/内嵌 + 安全修复
 # ppt skill 可选 —— 老版 Office(<2019)的 SVG→PNG 兜底;现代 PowerPoint 直接渲 SVG 无需,核心不依赖:
 # svglib>=1.5.0
 # reportlab>=4.0.0
@@ -40,7 +45,7 @@ croniter>=2.0
 
 # 微信接入(§8.7 ClawBot):segno 渲绑定二维码;cryptography 做凭据列加密 + 文件 AES-128-ECB
 segno>=1.6
-cryptography>=42.0
+cryptography>=48.0.1
 
 # broker 外置(§7.0,蓝绿双实例跨进程 SSE/cancel):ZCBOT_REDIS_URL 设了才用,
 # 不设走进程内 broker(dev 不需要起 redis);fakeredis 供单测(无需真 redis 服务)
@@ -54,8 +59,9 @@ alembic>=1.13.0
 
 # §7 Phase G / D: 纯 JSON API(FastAPI + 原生 SSE),前端由 platform 提供
 fastapi>=0.111.0
+starlette>=1.3.1      # FastAPI 底层 ASGI；显式固定安全下限
 uvicorn[standard]>=0.30.0
-python-multipart>=0.0.9   # files upload multipart 解析
+python-multipart>=0.0.31  # files upload multipart 解析 + 安全修复
 pyjwt>=2.8.0              # /v1/auth/login HS256 token mint/verify(§7 D' 过渡形态)
 bcrypt>=4.1.0             # /v1/auth/login_password 密码哈希(users.password_hash)
 
@@ -63,6 +69,7 @@ bcrypt>=4.1.0             # /v1/auth/login_password 密码哈希(users.password_
 # pymatgen skill: 无机材料计算(晶体结构/XRD/相图/Materials Project)
 pymatgen>=2024.0
 mp-api>=0.41.0
+pydantic-settings>=2.14.2  # mp-api 传递依赖；固定安全下限
 # stats_ml skill: 统计建模与 ML(sklearn 必装,statsmodels 必装,PyMC 可选)
 scikit-learn>=1.4.0
 statsmodels>=0.14.0
diff --git a/tests/test_shell_security.py b/tests/test_shell_security.py
new file mode 100644
index 0000000..be10818
--- /dev/null
+++ b/tests/test_shell_security.py
@@ -0,0 +1,45 @@
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from core import procs
+from core.proc_wrapper import _shell_argv as wrapper_shell_argv
+from tools.shell import ShellTool
+
+
+class ShellSecurityTests(unittest.TestCase):
+    def test_shell_argv_uses_explicit_interpreter(self) -> None:
+        argv = procs.shell_argv("echo ok")
+        wrapper_argv = wrapper_shell_argv("echo ok")
+        self.assertEqual(argv, wrapper_argv)
+        self.assertEqual(argv[-1], "echo ok")
+        if os.name == "nt":
+            self.assertEqual(argv[1:4], ["/d", "/s", "/c"])
+        else:
+            self.assertEqual(argv[:2], ["/bin/sh", "-c"])
+
+    def test_foreground_shell_still_executes_command(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            result = ShellTool(base_dir=Path(tmp)).execute("echo shell-ok")
+        self.assertIn("shell-ok", result)
+        self.assertIn("[exit 0]", result)
+
+    def test_background_shell_persists_explicit_argv(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp, patch(
+            "tools.shell.procs.count_running", return_value=0
+        ), patch("tools.shell.procs.launch_host", return_value=("proc-1", Path(tmp))) as launch:
+            result = ShellTool(base_dir=Path(tmp)).execute(
+                "echo background-ok", background=True
+            )
+        self.assertIn("proc_id=proc-1", result)
+        kwargs = launch.call_args.kwargs
+        self.assertEqual(kwargs["argv"], procs.shell_argv("echo background-ok"))
+        self.assertNotIn("shell_cmd", kwargs)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tools/look_at_image.py b/tools/look_at_image.py
index 3b69a10..febbeb1 100644
--- a/tools/look_at_image.py
+++ b/tools/look_at_image.py
@@ -126,6 +126,7 @@ class LookAtImageTool(Tool):
         )
         if api_err:
             return api_err
+        assert resp is not None
 
         answer, _truncated = extract_chat_answer(resp)
         if not answer:
diff --git a/tools/materials_project.py b/tools/materials_project.py
index 8c45cd0..28eeffa 100644
--- a/tools/materials_project.py
+++ b/tools/materials_project.py
@@ -140,7 +140,11 @@ class MaterialsProjectSearchSummaryTool(Tool):
                 f"[Error] 一次最多 {self._MAX_BATCH} 个化学式(收到 {len(formulas)});请分批调用。"
             )
         seen: set[str] = set()
-        uniq = [f for f in formulas if not (f in seen or seen.add(f))]
+        uniq = []
+        for formula in formulas:
+            if formula not in seen:
+                seen.add(formula)
+                uniq.append(formula)
         try:
             session = _mpr()
         except Exception as e:
@@ -172,7 +176,8 @@ class MaterialsProjectSearchSummaryTool(Tool):
         )
         if self.working_dir is not None:
             h = hashlib.sha1(
-                ",".join(a["formula"] for a in agg).encode("utf-8")
+                ",".join(a["formula"] for a in agg).encode("utf-8"),
+                usedforsecurity=False,
             ).hexdigest()[:8]
             dest = self.working_dir / "materials" / f"mp_search_batch_{h}.json"
             try:
diff --git a/tools/read_document.py b/tools/read_document.py
index 2155908..4dfcace 100644
--- a/tools/read_document.py
+++ b/tools/read_document.py
@@ -168,6 +168,7 @@ class ReadDocumentTool(Tool):
         )
         if api_err:
             return api_err
+        assert resp is not None
 
         answer, truncated = extract_chat_answer(resp)
         if not answer:
diff --git a/tools/shell.py b/tools/shell.py
index 4bf16da..103f553 100644
--- a/tools/shell.py
+++ b/tools/shell.py
@@ -102,7 +102,7 @@ class ShellTool(Tool):
                 anchor, self.task_id,
                 kind="shell",
                 command_display=command,
-                shell_cmd=command,
+                argv=procs.shell_argv(command),
                 cwd=self.base_dir, timeout_s=timeout_s, env=None,
             )
             return (
@@ -113,8 +113,7 @@ class ShellTool(Tool):
 
         try:
             result = subprocess.run(
-                command,
-                shell=True,
+                procs.shell_argv(command),
                 cwd=str(self.base_dir),
                 capture_output=True,
                 timeout=timeout,
diff --git a/web/admin.py b/web/admin.py
index 64c2d0a..5936f1e 100644
--- a/web/admin.py
+++ b/web/admin.py
@@ -12,6 +12,8 @@ app.state 内存(轻);其余走 DB 聚合(GROUP BY,无 N+1)。指标只读、不
 """
 from __future__ import annotations
 
+import importlib
+from types import ModuleType
 from datetime import datetime, timedelta, timezone
 from typing import Any
 from uuid import UUID
@@ -26,8 +28,9 @@ from core.storage.models import Task, UsageEvent, User, UserDiskUsage
 
 from .broker import broker
 
+resource: ModuleType | None
 try:
-    import resource  # Unix only;Windows dev 无此模块,RSS 监控降级跳过
+    resource = importlib.import_module("resource")
 except ImportError:  # pragma: no cover - Windows
     resource = None
 
@@ -302,6 +305,6 @@ def register_admin_routes(app: FastAPI, require_admin) -> None:
             result = s.execute(
                 update(User).where(User.user_id == target).values(plan=plan or None)
             )
-            if result.rowcount == 0:
+            if getattr(result, "rowcount", 0) == 0:
                 raise HTTPException(404, f"user not found: {uid}")
         return {"user_id": str(target), "plan": plan}
diff --git a/web/background.py b/web/background.py
index 7798993..8db08ed 100644
--- a/web/background.py
+++ b/web/background.py
@@ -8,7 +8,9 @@ bg proc 清扫 / 孤儿 run 收割 / 优雅 drain。每个 start_* 返回 asynci
 from __future__ import annotations
 
 import asyncio
+import importlib
 import os
+from types import ModuleType
 from typing import Optional
 
 from sqlalchemy import or_, update
@@ -19,8 +21,9 @@ from core.storage.models import Task
 from .broker import broker
 from .common import INSTANCE
 
+resource: ModuleType | None
 try:
-    import resource  # Unix only;Windows dev 无此模块,RSS 监控自动降级跳过
+    resource = importlib.import_module("resource")
 except ImportError:  # pragma: no cover - Windows
     resource = None
 
@@ -55,8 +58,9 @@ def reap_stale_runs() -> None:
                 run_error="server restarted before run finished",
             )
         )
-        if result.rowcount:
-            print(f"[startup] reaped {result.rowcount} stale active run(s)")
+        rowcount = int(getattr(result, "rowcount", 0) or 0)
+        if rowcount:
+            print(f"[startup] reaped {rowcount} stale active run(s)")
 
 
 def start_disk_scanner(cfg: dict) -> asyncio.Task:
diff --git a/web/broker.py b/web/broker.py
index 2defb65..afa3329 100644
--- a/web/broker.py
+++ b/web/broker.py
@@ -170,7 +170,7 @@ class RedisRunBroker:
         self._aredis = async_client if async_client is not None else _aredis.from_url(
             url, decode_responses=True,
         )
-        self._pubsub = None
+        self._pubsub: Any = None
         self._reader_task: Optional[asyncio.Task] = None
         self._subs: dict[UUID, set[asyncio.Queue]] = defaultdict(set)
         self._loop: Optional[asyncio.AbstractEventLoop] = None
@@ -231,11 +231,12 @@ class RedisRunBroker:
         断线 1s 退避重连,并把 _subs 里所有活跃 channel 重挂上(订阅状态不丢)。"""
         while True:
             try:
-                self._pubsub = self._aredis.pubsub()
+                pubsub = self._aredis.pubsub()
+                self._pubsub = pubsub
                 channels = [self._CTL_CHANNEL] + [self._chan(t) for t in self._subs]
-                await self._pubsub.subscribe(*channels)
+                await pubsub.subscribe(*channels)
                 while True:
-                    msg = await self._pubsub.get_message(
+                    msg = await pubsub.get_message(
                         ignore_subscribe_messages=True, timeout=5.0
                     )
                     if msg is None or msg.get("type") != "message":
diff --git a/web/pptx_render.py b/web/pptx_render.py
index 933eb9b..5ceccf0 100644
--- a/web/pptx_render.py
+++ b/web/pptx_render.py
@@ -63,7 +63,7 @@ def _cache_pdf_path(pptx_path: Path) -> Path:
     """
     st = pptx_path.stat()
     sig = f"{st.st_mtime_ns}-{st.st_size}".encode("utf-8")
-    digest = hashlib.sha1(sig).hexdigest()[:12]
+    digest = hashlib.sha1(sig, usedforsecurity=False).hexdigest()[:12]
     return pptx_path.parent / _PREVIEW_DIRNAME / f"{pptx_path.stem}.{digest}.pdf"
 
 
diff --git a/web/routers/messages.py b/web/routers/messages.py
index 0b42f81..8b4e5d4 100644
--- a/web/routers/messages.py
+++ b/web/routers/messages.py
@@ -7,7 +7,7 @@ from __future__ import annotations
 
 import asyncio
 import time
-from typing import Any
+from typing import Any, Optional
 from uuid import UUID
 
 from fastapi import Depends, HTTPException
@@ -76,9 +76,9 @@ def register_message_routes(app, *, require_user) -> None:
     @app.get("/v1/tasks/{task_id}/messages", tags=["messages"])
     def list_messages(
         task_id: str,
-        limit: int = None,
-        before_idx: int = None,
-        after_idx: int = None,
+        limit: Optional[int] = None,
+        before_idx: Optional[int] = None,
+        after_idx: Optional[int] = None,
         user_id: UUID = Depends(require_user),
     ):
         """task 历史消息(idx 升序);LiteLLM 原 payload 透传给前端,自行渲染。
diff --git a/web/routers/schedules.py b/web/routers/schedules.py
index caf070e..f2befb9 100644
--- a/web/routers/schedules.py
+++ b/web/routers/schedules.py
@@ -84,14 +84,16 @@ def register_schedule_routes(app, *, require_user) -> None:
                 .limit(page_size).offset(offset)
             ).scalars().all()
             tids = [r.task_id for r in rows]
-            msg_counts = (
-                dict(s.execute(
+            msg_counts: dict[UUID, int] = {}
+            if tids:
+                count_rows = s.execute(
                     select(Message.task_id, func.count())
                     .where(Message.task_id.in_(tids))
                     .group_by(Message.task_id)
-                ).all())
-                if tids else {}
-            )
+                ).all()
+                msg_counts = {
+                    task_id: int(count) for task_id, count in count_rows
+                }
             usage = usage_aggregates(s, tids)
 
         return {
diff --git a/web/routers/tasks.py b/web/routers/tasks.py
index 394ff3e..bac76d0 100644
--- a/web/routers/tasks.py
+++ b/web/routers/tasks.py
@@ -168,14 +168,16 @@ def register_task_routes(app, *, require_user) -> None:
             ).scalars().all()
 
             tids = [r.task_id for r in rows]
-            msg_counts = (
-                dict(s.execute(
+            msg_counts: dict[UUID, int] = {}
+            if tids:
+                count_rows = s.execute(
                     select(Message.task_id, func.count())
                     .where(Message.task_id.in_(tids))
                     .group_by(Message.task_id)
-                ).all())
-                if tids else {}
-            )
+                ).all()
+                msg_counts = {
+                    task_id: int(count) for task_id, count in count_rows
+                }
             usage = usage_aggregates(s, tids)
 
         return {
@@ -227,13 +229,16 @@ def register_task_routes(app, *, require_user) -> None:
                         )
                     ).scalars().all()
                 }
-                msg_counts = dict(
-                    s.execute(
+                msg_counts: dict[UUID, int] = {}
+                if rows:
+                    count_rows = s.execute(
                         select(Message.task_id, func.count())
                         .where(Message.task_id.in_(list(rows.keys())))
                         .group_by(Message.task_id)
                     ).all()
-                ) if rows else {}
+                    msg_counts = {
+                        task_id: int(count) for task_id, count in count_rows
+                    }
                 usage = usage_aggregates(s, list(rows.keys()))
                 for kind, tid in tids.items():
                     row = rows.get(tid) if tid else None
@@ -456,7 +461,7 @@ def register_task_routes(app, *, require_user) -> None:
                 .where(Task.task_id == tid, Task.user_id == user_id)
                 .values(**updates)
             )
-            if result.rowcount == 0:
+            if getattr(result, "rowcount", 0) == 0:
                 raise HTTPException(404, f"task not found: {tid}")
             row = s.execute(select(Task).where(Task.task_id == tid)).scalar_one()
             n = s.execute(