From 7bc5e01afd1f8b50fd618f0b3dc9982a84669233 Mon Sep 17 00:00:00 2001
From: jxxghp <jxxghp@gmail.com>
Date: Sat, 22 Aug 2026 13:07:32 +0800
Subject: [PATCH] refactor: expand async and type safety gates

---
 app/adapters/external/market.py               | 13 +++++-
 app/agent/middleware/activity_log.py          | 23 +++++++---
 app/api/dependencies/subscription.py          |  4 +-
 app/application/scheduling.py                 | 10 ++---
 app/runtime/event/errors.py                   |  2 +-
 app/runtime/extensions/module/dispatcher.py   |  4 +-
 .../backend-architecture-next-stage.md        |  6 +++
 mypy.ini                                      |  6 ++-
 scripts/architecture/async_blocking.py        | 42 +++++++++++++++----
 .../architecture/async-blocking-baseline.json |  4 +-
 tests/test_async_blocking_gate.py             | 16 ++++++-
 tests/test_type_gate.py                       |  6 ++-
 12 files changed, 103 insertions(+), 33 deletions(-)

diff --git a/app/adapters/external/market.py b/app/adapters/external/market.py
index c3d935180..34c8ddd71 100644
--- a/app/adapters/external/market.py
+++ b/app/adapters/external/market.py
@@ -2772,8 +2772,11 @@ class PluginHelper(metaclass=WeakSingleton):
                         await async_dest_path.mkdir(parents=True, exist_ok=True)
                         continue
                     await async_dest_path.parent.mkdir(parents=True, exist_ok=True)
-                    with zf.open(info, 'r') as src:
-                        data = src.read()
+                    data = await asyncio.to_thread(
+                        self.__read_release_zip_member,
+                        zf,
+                        info,
+                    )
                     async with aiofiles.open(dest_path, 'wb') as dst:
                         await dst.write(data)
                     wrote_any = True
@@ -2784,6 +2787,12 @@ class PluginHelper(metaclass=WeakSingleton):
             logger.error(f"解压 Release 压缩包失败：{e}")
             return False, f"解压 Release 压缩包失败：{e}"
 
+    @staticmethod
+    def __read_release_zip_member(zf: zipfile.ZipFile, info: zipfile.ZipInfo) -> bytes:
+        """在线程池读取并解压单个 Release 文件，避免阻塞事件循环。"""
+        with zf.open(info, "r") as source:
+            return source.read()
+
 
 # 公开 Release 查询的缓存管理统一指向仓库级分页缓存。
 PluginHelper.get_plugin_release_versions.cache_clear = PluginHelper._get_plugin_repo_releases.cache_clear
diff --git a/app/agent/middleware/activity_log.py b/app/agent/middleware/activity_log.py
index 3ea9ac2b0..d0a25a260 100644
--- a/app/agent/middleware/activity_log.py
+++ b/app/agent/middleware/activity_log.py
@@ -87,6 +87,17 @@ SUMMARY_PROMPT = """请判断以下 AI 助手与用户的对话是否值得写
 ACTIVITY_ENTRY_PATTERN = re.compile(r"^-\s+\*\*(?P<time>\d{2}:\d{2})\*\*\s+(?P<summary>.+)$")
 
 
+def _write_activity_log_exclusive(path: Path, content: str) -> bool:
+    """同步独占创建日志文件；调用方必须在线程池中执行本函数。"""
+    try:
+        fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o644)
+    except FileExistsError:
+        return False
+    with os.fdopen(fd, "w", encoding="utf-8") as stream:
+        stream.write(content)
+    return True
+
+
 class QueryActivityLogInput(BaseModel):
     """查询活动日志工具的输入参数模型。"""
 
@@ -565,18 +576,18 @@ class ActivityLogMiddleware(AgentMiddleware[ActivityLogState, ContextT, Response
                     await stream.write(entry)
             else:
                 header = f"# {today_str} 活动日志\n\n"
-                try:
-                    fd = os.open(log_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o644)
-                except FileExistsError:
+                created = await anyio.to_thread.run_sync(
+                    _write_activity_log_exclusive,
+                    Path(log_path),
+                    header + entry,
+                )
+                if not created:
                     async with await anyio.open_file(
                         log_path,
                         mode="a",
                         encoding="utf-8",
                     ) as stream:
                         await stream.write(entry)
-                else:
-                    with os.fdopen(fd, "w", encoding="utf-8") as stream:
-                        stream.write(header + entry)
             logger.debug(f"Activity logged: {summarize_result(summary, max_chars=80)}")
         except Exception as e:
             logger.warning(f"Failed to append activity log: {summarize_error(e)}")
diff --git a/app/api/dependencies/subscription.py b/app/api/dependencies/subscription.py
index e1fbc3662..906925ca6 100644
--- a/app/api/dependencies/subscription.py
+++ b/app/api/dependencies/subscription.py
@@ -16,7 +16,7 @@ from app.api.context import (
 from app.api.data import get_async_db, get_db
 from app.api.dependencies.data import repository
 from app.application.outbox import AsyncOutboxTransaction
-from app.application.scheduling import Scheduler
+from app.application.scheduling import start_scheduler_job
 from app.application.servarr import ServarrSubscriptionService
 from app.application.subscription.delete import (
     AsyncUnitOfWork as DeleteUnitOfWork,
@@ -99,7 +99,7 @@ def get_search_subscriptions_command(
     def schedule_search(subscribe_id: int | None, state: str | None) -> None:
         """按历史参数提交订阅搜索调度任务。"""
         background_tasks.add_task(
-            Scheduler().start,
+            start_scheduler_job,
             job_id="subscribe_search",
             sid=subscribe_id,
             state=state,
diff --git a/app/application/scheduling.py b/app/application/scheduling.py
index 5fb9cb352..97daf070e 100644
--- a/app/application/scheduling.py
+++ b/app/application/scheduling.py
@@ -12,7 +12,7 @@ Scheduler 实现由 startup 组合根在导入期注册，避免 application 层
 import asyncio
 from dataclasses import dataclass, field
 from enum import StrEnum
-from typing import Any, Awaitable, Callable, List, Optional
+from typing import Any, Awaitable, Callable, List, Optional, cast
 
 # Agent 自主定时任务在运行时调度器中的任务 ID 前缀。
 AGENT_TASK_JOB_PREFIX = "agent-task"
@@ -143,12 +143,12 @@ class Scheduler:
 
 def list_scheduler_jobs() -> List[Any]:
     """列出运行时调度器的全部任务。"""
-    return get_scheduler().list()
+    return cast(List[Any], get_scheduler().list())
 
 
-def start_scheduler_job(job_id: str) -> None:
+def start_scheduler_job(job_id: str, **kwargs: Any) -> None:
     """立即运行指定的运行时定时任务。"""
-    get_scheduler().start(job_id)
+    get_scheduler().start(job_id, **kwargs)
 
 
 def update_plugin_job(plugin_id: str) -> None:
@@ -163,7 +163,7 @@ def remove_plugin_job(plugin_id: str) -> None:
 
 def start_agent_task(task_id: int) -> bool:
     """立即执行 Agent 自主定时任务。"""
-    return get_scheduler().start_agent_task(task_id)
+    return cast(bool, get_scheduler().start_agent_task(task_id))
 
 
 def get_agent_task_next_run(task_id: int) -> Optional[Any]:
diff --git a/app/runtime/event/errors.py b/app/runtime/event/errors.py
index 5d2e9bf16..81b294455 100644
--- a/app/runtime/event/errors.py
+++ b/app/runtime/event/errors.py
@@ -20,7 +20,7 @@ class EventErrorPolicy:
         self,
         *,
         notifier: Callable[[], Optional[EventErrorNotifier]],
-        emit_system_error: Callable[[dict], object],
+        emit_system_error: Callable[[dict[str, Any]], object],
     ) -> None:
         """注入通知读取器和 SystemError 发送回调。"""
         self._notifier = notifier
diff --git a/app/runtime/extensions/module/dispatcher.py b/app/runtime/extensions/module/dispatcher.py
index ca5960d0f..dcf45de14 100644
--- a/app/runtime/extensions/module/dispatcher.py
+++ b/app/runtime/extensions/module/dispatcher.py
@@ -4,7 +4,7 @@ from __future__ import annotations
 
 import inspect
 from collections.abc import Callable, Mapping
-from typing import Any, Protocol
+from typing import Any, Protocol, cast
 
 from app.foundation.reflection import ObjectUtils
 from app.runtime.execution import run_in_threadpool
@@ -324,7 +324,7 @@ class ModuleInvocationDispatcher:
     def _module_name(module: Any, fallback: str) -> str:
         """读取模块展示名，失败时回退到稳定类名。"""
         try:
-            return module.get_name()
+            return cast(str, module.get_name())
         except Exception as err:
             logger.debug("获取模块名称出错：%s", str(err))
             return fallback
diff --git a/docs/refactor/backend-architecture-next-stage.md b/docs/refactor/backend-architecture-next-stage.md
index 87e120c03..6b51930f1 100644
--- a/docs/refactor/backend-architecture-next-stage.md
+++ b/docs/refactor/backend-architecture-next-stage.md
@@ -899,6 +899,9 @@ OTel 初始化只能位于 Startup/Adapter；Domain/Application 只依赖 no-op-
 Workflow 执行状态 UoW 切片将 `app/application/workflow.py` 与 `app/startup/workflow.py` 纳入 strict 清单，
 治理范围扩大到 22 个源文件；事务命令、仓储 Protocol 和短会话适配器保持零错误。
 
+异步安全与契约收口继续纳管 scheduling facade、Event error policy、Module dispatcher 和 async blocking
+scanner，strict 清单扩大到 26 个源文件；已登记范围保持零错误，未使用全文件 ignore 或 `cast(Any, ...)`。
+
 #### ARCH-271：复杂度和端点预算 ratchet
 
 **目标**：阻止大方法继续增长，并让拆分对应真实阶段，而不是机械 helper 化。
@@ -949,6 +952,9 @@ Workflow 执行状态 UoW 切片将 `app/application/workflow.py` 与 `app/start
   通过。同步第三方 Module 仍由 dispatcher 的 `app.runtime.execution.run_in_threadpool` 兼容。
 - 2026-08-22 扫描范围扩大到 `app/chain`、`app/modules`、`app/startup` 与 `app/scheduler.py`；扩大后未发现
   新存量，仍只保留 ActivityLog 的一处原子 `os.open` 精确债务，并由测试锁定扫描根目录。
+- 扫描进一步覆盖 `adapters/db/doctor/domain/foundation/monitor/runtime/schemas/workflow` 及 CLI、Command、
+  Factory、Main 顶层入口，明确排除插件源码和 SDK。AsyncPath 条件派生识别已修正；Release zip 解压读取和
+  ActivityLog `O_EXCL` 独占创建移入线程池，扩围后 async 阻塞 baseline 从 1 降为 0。
 
 ## 6. 推荐执行队列
 
diff --git a/mypy.ini b/mypy.ini
index 17d67989d..37d1d6a14 100644
--- a/mypy.ini
+++ b/mypy.ini
@@ -14,9 +14,12 @@ files =
     app/runtime/correlation.py,
     app/runtime/observability/__init__.py,
     app/runtime/event/contracts.py,
+    app/runtime/event/errors.py,
     app/runtime/extensions/module/contracts.py,
+    app/runtime/extensions/module/dispatcher.py,
     app/application/outbox.py,
     app/application/configuration.py,
+    app/application/scheduling.py,
     app/application/workflow.py,
     app/application/chain/context.py,
     app/application/chain/durable_events.py,
@@ -28,4 +31,5 @@ files =
     app/startup/download_failure.py,
     app/startup/workflow.py,
     app/api/context.py,
-    app/api/dependencies/subscription.py
+    app/api/dependencies/subscription.py,
+    scripts/architecture/async_blocking.py
diff --git a/scripts/architecture/async_blocking.py b/scripts/architecture/async_blocking.py
index 2bdae70ae..d66d5a8f1 100644
--- a/scripts/architecture/async_blocking.py
+++ b/scripts/architecture/async_blocking.py
@@ -6,17 +6,31 @@ import argparse
 import ast
 import json
 from collections import Counter
+from collections.abc import Iterator
 from pathlib import Path
 
 PROJECT_ROOT = Path(__file__).resolve().parents[2]
 DEFAULT_BASELINE = PROJECT_ROOT / "tests/fixtures/architecture/async-blocking-baseline.json"
 SCAN_ROOTS = (
+    "app/adapters",
     "app/api",
     "app/agent",
     "app/application",
     "app/chain",
+    "app/db",
+    "app/doctor",
+    "app/domain",
+    "app/foundation",
+    "app/monitor",
     "app/modules",
+    "app/runtime",
+    "app/schemas",
     "app/startup",
+    "app/workflow",
+    "app/cli.py",
+    "app/command.py",
+    "app/factory.py",
+    "app/main.py",
     "app/scheduler.py",
 )
 BLOCKING_EXACT = {
@@ -83,20 +97,28 @@ class _AsyncPathCollector(ast.NodeVisitor):
 
     def visit_Assign(self, node: ast.Assign) -> None:
         """识别 AsyncPath 构造和已知路径的 `/` 派生赋值。"""
-        is_async_path = (
-            isinstance(node.value, ast.Call)
-            and _call_name(node.value).endswith("AsyncPath")
-        ) or (
-            isinstance(node.value, ast.BinOp)
-            and isinstance(node.value.left, ast.Name)
-            and node.value.left.id in self.paths
-        )
+        is_async_path = self._is_async_path_expression(node.value)
         if is_async_path:
             for target in node.targets:
                 if isinstance(target, ast.Name):
                     self.paths.add(target.id)
         self.generic_visit(node)
 
+    def _is_async_path_expression(self, expression: ast.expr) -> bool:
+        """识别 AsyncPath 构造及任意层级的 `/` 路径派生表达式。"""
+        if isinstance(expression, ast.Call):
+            return _call_name(expression).endswith("AsyncPath")
+        if isinstance(expression, ast.Name):
+            return expression.id in self.paths
+        if isinstance(expression, ast.BinOp):
+            return self._is_async_path_expression(expression.left)
+        if isinstance(expression, ast.IfExp):
+            return (
+                self._is_async_path_expression(expression.body)
+                and self._is_async_path_expression(expression.orelse)
+            )
+        return False
+
     def visit_AnnAssign(self, node: ast.AnnAssign) -> None:
         """识别 `list[AsyncPath]` 等路径集合。"""
         if isinstance(node.target, ast.Name):
@@ -160,7 +182,9 @@ class _AsyncCallVisitor(ast.NodeVisitor):
         """嵌套异步函数由模块级收集器单独治理。"""
 
 
-def _async_functions(tree: ast.Module):
+def _async_functions(
+    tree: ast.Module,
+) -> Iterator[tuple[str, ast.AsyncFunctionDef]]:
     """产出模块顶层及类直接拥有的 async 函数限定名。"""
     for node in tree.body:
         if isinstance(node, ast.AsyncFunctionDef):
diff --git a/tests/fixtures/architecture/async-blocking-baseline.json b/tests/fixtures/architecture/async-blocking-baseline.json
index 89dbb3b6d..0967ef424 100644
--- a/tests/fixtures/architecture/async-blocking-baseline.json
+++ b/tests/fixtures/architecture/async-blocking-baseline.json
@@ -1,3 +1 @@
-{
-  "app/agent/middleware/activity_log.py:ActivityLogMiddleware._append_activity:os.open": 1
-}
+{}
diff --git a/tests/test_async_blocking_gate.py b/tests/test_async_blocking_gate.py
index 6e4f1009c..2a1d468ef 100644
--- a/tests/test_async_blocking_gate.py
+++ b/tests/test_async_blocking_gate.py
@@ -8,13 +8,27 @@ from scripts.architecture.async_blocking import SCAN_ROOTS, compare_async_blocki
 
 
 def test_async_blocking_scan_covers_runtime_entrypoints() -> None:
-    """扫描范围必须覆盖 API、Scheduler、Chain 及其启动组合根。"""
+    """扫描范围必须覆盖全部 canonical 宿主目录和顶层运行入口。"""
     assert {
+        "app/adapters",
         "app/api",
+        "app/agent",
         "app/application",
         "app/chain",
+        "app/db",
+        "app/doctor",
+        "app/domain",
+        "app/foundation",
+        "app/monitor",
         "app/modules",
+        "app/runtime",
+        "app/schemas",
         "app/startup",
+        "app/workflow",
+        "app/cli.py",
+        "app/command.py",
+        "app/factory.py",
+        "app/main.py",
         "app/scheduler.py",
     }.issubset({str(path) for path in SCAN_ROOTS})
 
diff --git a/tests/test_type_gate.py b/tests/test_type_gate.py
index 35a33a0bb..15ef77a15 100644
--- a/tests/test_type_gate.py
+++ b/tests/test_type_gate.py
@@ -23,12 +23,16 @@ def test_mypy_gate_has_explicit_strict_scope_without_global_ignore() -> None:
     assert settings.getboolean("strict") is True
     assert "app/runtime/event/contracts.py" in governed_files
     assert "app/runtime/extensions/module/contracts.py" in governed_files
+    assert "app/runtime/extensions/module/dispatcher.py" in governed_files
+    assert "app/runtime/event/errors.py" in governed_files
+    assert "app/application/scheduling.py" in governed_files
+    assert "scripts/architecture/async_blocking.py" in governed_files
     assert "app/startup/context.py" in governed_files
     assert "app/startup/download_failure.py" in governed_files
     assert "app/startup/workflow.py" in governed_files
     assert "app/application/workflow.py" in governed_files
     assert "app/api/context.py" in governed_files
-    assert len(governed_files) >= 22
+    assert len(governed_files) >= 26
     assert any(path.startswith("app/domain/") for path in governed_files)
     assert "ignore_errors" not in MYPY_CONFIG.read_text(encoding="utf-8")