refactor: enforce background task ownership

.github/workflows/test.yml

@@ -59,6 +59,9 @@ jobs:
       - name: Check async blocking ratchet
         run: uv run --locked --no-sync python scripts/architecture/async_blocking.py
 
+      - name: Check background task ownership
+        run: uv run --locked --no-sync python scripts/architecture/task_ownership.py
+
       - name: Check startup performance contract
         run: >-
           uv run --locked --no-sync python

app/runtime/tasks.py

@@ -24,6 +24,8 @@ class TaskRegistry:
     def __init__(self) -> None:
         """初始化空任务登记表。"""
         self._records: dict[asyncio.Task[Any], TaskRecord] = {}
+        self._shutdown_cancel_requested: set[asyncio.Task[Any]] = set()
+        self._shutdown_timeout_reported: set[asyncio.Task[Any]] = set()
         self._accepting = True
 
     @property
@@ -89,6 +91,8 @@ class TaskRegistry:
     def _discard(self, task: asyncio.Task[Any]) -> None:
         """移除已结束任务，并把未处理异常交给事件循环统一报告。"""
         record = self._records.pop(task, None)
+        self._shutdown_cancel_requested.discard(task)
+        self._shutdown_timeout_reported.discard(task)
         if task.cancelled():
             return
         exception = task.exception()
@@ -103,18 +107,38 @@ class TaskRegistry:
             )
 
     async def shutdown(self, *, timeout_seconds: float = 10.0) -> None:
-        """取消并等待全部登记任务，超时后放弃等待但不影响其他关闭步骤。"""
+        """停止接收并有限等待存量任务，超时任务保留登记并报告责任域。"""
         self._accepting = False
         records = self.records
         tasks = [record.task for record in records]
         for record in records:
-            if record.cancel_on_shutdown:
+            if (
+                record.cancel_on_shutdown
+                and record.task not in self._shutdown_cancel_requested
+            ):
+                self._shutdown_cancel_requested.add(record.task)
                 record.task.cancel()
         if tasks:
-            _, pending = await asyncio.wait(tasks, timeout=timeout_seconds)
-            for task in pending:
-                task.cancel()
-        self._records.clear()
+            await asyncio.wait(tasks, timeout=timeout_seconds)
+
+        unfinished = tuple(
+            record
+            for record in records
+            if not record.task.done()
+            and record.task not in self._shutdown_timeout_reported
+        )
+        if unfinished:
+            self._shutdown_timeout_reported.update(
+                record.task for record in unfinished
+            )
+            asyncio.get_running_loop().call_exception_handler(
+                {
+                    "message": "MoviePilot 后台任务未在关停预算内结束",
+                    "owners": tuple(record.owner for record in unfinished),
+                    "tasks": tuple(record.task for record in unfinished),
+                    "timeout_seconds": timeout_seconds,
+                }
+            )
 
 
 _default_registry = TaskRegistry()

docs/refactor/backend-architecture-governance.md

@@ -1320,6 +1320,7 @@ done_when: []
 ./.venv/bin/python -m pytest tests/test_legacy_import_compat.py -q
 ./.venv/bin/python -m pytest tests/test_legacy_plugin_resource_imports.py -q
 ./.venv/bin/python -m pytest tests/test_plugin_sdk.py -q
+./.venv/bin/python scripts/architecture/task_ownership.py
 ```
 
 再运行本批次聚焦测试。涉及发布级公共行为时，使用仓库完整门禁：

docs/refactor/backend-architecture-next-stage.md

@@ -6,7 +6,7 @@
 > 审计范围：宿主后端；排除 `app/plugins/**` 运行时插件副本
 > 规范优先级：`AGENTS.md` 与 `docs/rules/` 高于本文
 > 相关文档：`docs/architecture-overview.md`、`docs/refactor/backend-architecture-governance.md`、`docs/refactor/backend-module-refactor-compatibility.md`
-> 实施进度：阶段 0～6 的宿主架构能力已完成收口；API/Application 公共复杂度基线已清零，启动组合根的 SystemConfigOper 构造点已由 14 降至 1；API 进程内后台任务已完成首批统一登记，插件仓适配、Outbox 外围扩展和 Model 查询兼容面仍按风险切片推进。2026-08-23 的长期整改阶段 0 已恢复宿主、启动性能、官方插件和 SDK 契约门禁的可信基线。
+> 实施进度：阶段 0～6 的宿主架构能力已完成收口；API/Application 公共复杂度基线已清零，启动组合根的 SystemConfigOper 构造点已由 14 降至 1；API 进程内后台任务已完成首批统一登记，插件仓适配、Outbox 外围扩展和 Model 查询兼容面仍按风险切片推进。2026-08-23 的长期整改阶段 0 已恢复宿主、启动性能、官方插件和 SDK 契约门禁的可信基线；阶段 1a 已补齐 TaskRegistry owner 零债务门禁和诚实的关停超时语义。
 
 ## 当前复核结论（2026-08-23）
 
@@ -23,6 +23,19 @@
 
 本阶段只修复治理信号和事实源，不把基线刷新当作业务重构完成。后台任务所有权、Module Contract V2、typed runtime、durable 副作用和质量规模化仍按下列 P1/P2 顺序推进。
 
+### 长期整改阶段 1a：TaskRegistry owner 与关停契约（2026-08-23）
+
+- 手工订阅搜索曾因 `create_sync()` 缺少必填 `owner` 在真实命令路径抛出 `TypeError`；现以
+  `api.subscription.search_schedule` 登记，并由命令级测试冻结既有 scheduler 参数和立即返回语义。
+- 新增符号感知的 `scripts/architecture/task_ownership.py`：宿主中所有可证明为 TaskRegistry 的
+  `create`、`create_sync`、`register` 调用必须显式传入非空字符串字面量 owner，当前债务为零；CI
+  只读执行该门禁。插件、SDK、`runtime/compat` 和测试运行时目录明确排除，不扩大插件 ABI 约束。
+- TaskRegistry 关停超过预算时不再取消不可中断的同步线程包装任务或清空其记录；尚未真正结束的任务
+  保留 owner 并通过事件循环异常处理器报告。可取消协程在整个关停周期只收到一次取消请求，重复或并发
+  关停不会再次打断其异步清理，超时诊断也按任务去重。
+- 本子阶段仍只覆盖 TaskRegistry。Transfer worker/replay、Agent blocking executor、Event handler
+  drain、通道线程和 E2/E3 durable 完成点继续作为阶段 1 后续切片，不能因 owner 门禁通过而宣称完成。
+
 ### 总体判断
 
 当前架构总体合理，已经从跨层混合的遗留单体收敛为**边界清晰的模块化单体**：

scripts/architecture/task_ownership.py

@@ -0,0 +1,318 @@
+"""检查宿主 TaskRegistry 调用是否声明稳定的任务 owner。"""
+
+from __future__ import annotations
+
+import ast
+from dataclasses import dataclass, field
+from pathlib import Path
+
+
+PROJECT_ROOT = Path(__file__).resolve().parents[2]
+TASK_METHODS = frozenset({"create", "create_sync", "register"})
+TASK_MODULE = "app.runtime.tasks"
+CONTEXT_MODULE = "app.api.context"
+TASK_FACTORIES = frozenset(
+    {
+        "get_task_registry",
+        "get_background_task_registry",
+        "get_background_task_registry_compat",
+        "resolve_background_task_registry",
+    }
+)
+EXCLUDED_ROOTS = (
+    "app/plugins",
+    "app/runtime/compat",
+    "app/sdk",
+    "app/testing",
+)
+
+
+@dataclass(frozen=True, order=True, slots=True)
+class TaskOwnerViolation:
+    """描述一处缺少稳定字符串 owner 的 TaskRegistry 调用。"""
+
+    path: str
+    line: int
+    method: str
+    reason: str
+
+    def render(self) -> str:
+        """返回适合 CI 输出的稳定诊断文本。"""
+        return f"{self.path}:{self.line}: TaskRegistry.{self.method} {self.reason}"
+
+
+@dataclass(slots=True)
+class _Scope:
+    """保存当前词法作用域内可确认来源的任务登记器符号。"""
+
+    task_classes: set[str] = field(default_factory=set)
+    task_factories: set[str] = field(default_factory=set)
+    module_aliases: dict[str, str] = field(default_factory=dict)
+    registry_names: set[str] = field(default_factory=set)
+
+    def child(self) -> _Scope:
+        """复制父级可见绑定，供嵌套函数或类独立追踪局部赋值。"""
+        return _Scope(
+            task_classes=set(self.task_classes),
+            task_factories=set(self.task_factories),
+            module_aliases=dict(self.module_aliases),
+            registry_names=set(self.registry_names),
+        )
+
+
+class _TaskOwnershipVisitor(ast.NodeVisitor):
+    """仅跟踪可由 import、类型注解或工厂调用确认的 TaskRegistry。"""
+
+    def __init__(self, relative_path: str) -> None:
+        """初始化源码位置、词法作用域和违规记录。"""
+        self._relative_path = relative_path
+        self._scopes = [_Scope()]
+        self.violations: list[TaskOwnerViolation] = []
+
+    @property
+    def _scope(self) -> _Scope:
+        """返回当前词法作用域。"""
+        return self._scopes[-1]
+
+    def _visit_nested_scope(
+        self,
+        statements: list[ast.stmt],
+        arguments: ast.arguments | None = None,
+    ) -> None:
+        """在继承可见符号的新作用域中访问函数、类或 lambda 主体。"""
+        self._scopes.append(self._scope.child())
+        try:
+            if arguments is not None:
+                self._bind_arguments(arguments)
+            for statement in statements:
+                self.visit(statement)
+        finally:
+            self._scopes.pop()
+
+    def _bind_arguments(self, arguments: ast.arguments) -> None:
+        """把明确标注为 TaskRegistry 的函数参数加入当前作用域。"""
+        positional = (*arguments.posonlyargs, *arguments.args, *arguments.kwonlyargs)
+        for argument in positional:
+            self._bind_name(
+                argument.arg,
+                self._is_registry_annotation(argument.annotation),
+            )
+        if arguments.vararg:
+            self._bind_name(
+                arguments.vararg.arg,
+                self._is_registry_annotation(arguments.vararg.annotation),
+            )
+        if arguments.kwarg:
+            self._bind_name(
+                arguments.kwarg.arg,
+                self._is_registry_annotation(arguments.kwarg.annotation),
+            )
+
+    def _bind_name(self, name: str, is_registry: bool) -> None:
+        """更新局部名称的 TaskRegistry 绑定，显式重赋值会清除旧绑定。"""
+        if is_registry:
+            self._scope.registry_names.add(name)
+        else:
+            self._scope.registry_names.discard(name)
+
+    def _bind_target(self, target: ast.expr, is_registry: bool) -> None:
+        """处理普通名称和解构赋值产生的局部绑定。"""
+        if isinstance(target, ast.Name):
+            self._bind_name(target.id, is_registry)
+        elif isinstance(target, (ast.List, ast.Tuple)):
+            for item in target.elts:
+                self._bind_target(item, False)
+
+    def _is_registry_annotation(self, annotation: ast.expr | None) -> bool:
+        """识别 TaskRegistry、联合类型和 Annotated 中的明确类型来源。"""
+        if annotation is None:
+            return False
+        if isinstance(annotation, ast.Constant) and isinstance(annotation.value, str):
+            try:
+                annotation = ast.parse(annotation.value, mode="eval").body
+            except SyntaxError:
+                return False
+        for node in ast.walk(annotation):
+            if isinstance(node, ast.Name) and node.id in self._scope.task_classes:
+                return True
+            if (
+                isinstance(node, ast.Attribute)
+                and isinstance(node.value, ast.Name)
+                and self._scope.module_aliases.get(node.value.id) == TASK_MODULE
+                and node.attr == "TaskRegistry"
+            ):
+                return True
+        return False
+
+    def _is_registry_factory(self, expression: ast.expr) -> bool:
+        """判断调用目标是否是明确导入的登记器构造器或解析工厂。"""
+        if isinstance(expression, ast.Name):
+            return expression.id in (
+                self._scope.task_classes | self._scope.task_factories
+            )
+        if isinstance(expression, ast.Attribute) and isinstance(
+            expression.value, ast.Name
+        ):
+            module = self._scope.module_aliases.get(expression.value.id)
+            if module == TASK_MODULE:
+                return expression.attr in {"TaskRegistry", "get_task_registry"}
+            if module == CONTEXT_MODULE:
+                return expression.attr in TASK_FACTORIES
+        return False
+
+    def _is_registry_expression(self, expression: ast.expr | None) -> bool:
+        """判断表达式是否确定返回或引用 TaskRegistry。"""
+        if isinstance(expression, ast.Name):
+            return expression.id in self._scope.registry_names
+        if isinstance(expression, ast.Call):
+            return self._is_registry_factory(expression.func)
+        if isinstance(expression, ast.IfExp):
+            return self._is_registry_expression(
+                expression.body
+            ) and self._is_registry_expression(expression.orelse)
+        return False
+
+    def _check_owner(self, node: ast.Call, method: str) -> None:
+        """要求 owner 以显式、非空字符串字面量传入。"""
+        owner = next(
+            (keyword.value for keyword in node.keywords if keyword.arg == "owner"),
+            None,
+        )
+        if owner is None:
+            reason = "缺少显式 owner"
+        elif not (
+            isinstance(owner, ast.Constant)
+            and isinstance(owner.value, str)
+            and owner.value.strip()
+        ):
+            reason = "的 owner 必须是非空字符串字面量"
+        else:
+            return
+        self.violations.append(
+            TaskOwnerViolation(
+                path=self._relative_path,
+                line=node.lineno,
+                method=method,
+                reason=reason,
+            )
+        )
+
+    def visit_Import(self, node: ast.Import) -> None:
+        """记录 TaskRegistry 与 API context 模块别名。"""
+        for alias in node.names:
+            if alias.name not in {TASK_MODULE, CONTEXT_MODULE}:
+                continue
+            local_name = alias.asname or alias.name.split(".")[0]
+            self._scope.module_aliases[local_name] = alias.name
+
+    def visit_ImportFrom(self, node: ast.ImportFrom) -> None:
+        """记录明确导入的 TaskRegistry 类和登记器工厂别名。"""
+        if node.module == TASK_MODULE:
+            for alias in node.names:
+                local_name = alias.asname or alias.name
+                if alias.name == "TaskRegistry":
+                    self._scope.task_classes.add(local_name)
+                elif alias.name == "get_task_registry":
+                    self._scope.task_factories.add(local_name)
+        elif node.module == CONTEXT_MODULE:
+            for alias in node.names:
+                if alias.name in TASK_FACTORIES:
+                    self._scope.task_factories.add(alias.asname or alias.name)
+
+    def _visit_function(
+        self,
+        node: ast.FunctionDef | ast.AsyncFunctionDef,
+    ) -> None:
+        """在隔离的函数作用域中追踪参数和局部登记器。"""
+        for expression in (*node.decorator_list, *node.args.defaults):
+            self.visit(expression)
+        for default in node.args.kw_defaults:
+            if default is not None:
+                self.visit(default)
+        self._visit_nested_scope(node.body, node.args)
+
+    def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
+        """分析同步函数作用域。"""
+        self._visit_function(node)
+
+    def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
+        """按与同步函数相同的规则追踪异步函数作用域。"""
+        self._visit_function(node)
+
+    def visit_ClassDef(self, node: ast.ClassDef) -> None:
+        """隔离类体局部名称，同时保留模块导入的符号来源。"""
+        for expression in (*node.decorator_list, *node.bases, *node.keywords):
+            self.visit(expression)
+        self._visit_nested_scope(node.body)
+
+    def visit_Assign(self, node: ast.Assign) -> None:
+        """传播由登记器构造或解析工厂建立的简单赋值。"""
+        self.visit(node.value)
+        is_registry = self._is_registry_expression(node.value)
+        for target in node.targets:
+            self._bind_target(target, is_registry)
+
+    def visit_AnnAssign(self, node: ast.AnnAssign) -> None:
+        """优先使用明确 TaskRegistry 注解，并兼容带初始值的赋值。"""
+        if node.value is not None:
+            self.visit(node.value)
+        self._bind_target(
+            node.target,
+            self._is_registry_annotation(node.annotation)
+            or self._is_registry_expression(node.value),
+        )
+
+    def visit_NamedExpr(self, node: ast.NamedExpr) -> None:
+        """传播海象表达式建立的登记器局部绑定。"""
+        self.visit(node.value)
+        self._bind_target(node.target, self._is_registry_expression(node.value))
+
+    def visit_Call(self, node: ast.Call) -> None:
+        """只校验接收者已被证明为 TaskRegistry 的目标方法调用。"""
+        if (
+            isinstance(node.func, ast.Attribute)
+            and node.func.attr in TASK_METHODS
+            and self._is_registry_expression(node.func.value)
+        ):
+            self._check_owner(node, node.func.attr)
+        self.generic_visit(node)
+
+
+def _is_excluded(relative_path: str) -> bool:
+    """排除插件、SDK、兼容层与扫描器实现自身等非宿主调用面。"""
+    if relative_path == "app/runtime/tasks.py":
+        return True
+    return any(
+        relative_path == root or relative_path.startswith(f"{root}/")
+        for root in EXCLUDED_ROOTS
+    )
+
+
+def collect_task_owner_violations(
+    root: Path = PROJECT_ROOT,
+) -> list[TaskOwnerViolation]:
+    """扫描 canonical 宿主源码并返回缺少稳定 owner 的调用。"""
+    violations: list[TaskOwnerViolation] = []
+    for path in sorted((root / "app").rglob("*.py")):
+        relative_path = path.relative_to(root).as_posix()
+        if _is_excluded(relative_path):
+            continue
+        tree = ast.parse(path.read_text(encoding="utf-8-sig"), filename=str(path))
+        visitor = _TaskOwnershipVisitor(relative_path)
+        visitor.visit(tree)
+        violations.extend(visitor.violations)
+    return sorted(violations)
+
+
+def main() -> int:
+    """执行 TaskRegistry owner 零债务门禁。"""
+    violations = collect_task_owner_violations()
+    if violations:
+        print("\n".join(violation.render() for violation in violations))
+        return 1
+    print("TaskRegistry owner 门禁通过")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_api_background_task_registry.py

@@ -4,6 +4,8 @@ import asyncio
 from types import SimpleNamespace
 
 from app.api.endpoints import anthropic, history, message, openai, site, subscribe, webhook
+from app.api.dependencies import subscription as subscription_dependencies
+from app.application.subscription.search import SubscribeSearchActor
 from app.runtime.tasks import TaskRegistry
 
 
@@ -186,6 +188,36 @@ def test_seerr_subscribe_uses_task_registry(monkeypatch) -> None:
     assert owner == "api.subscribe.seerr"
 
 
+def test_manual_subscription_search_uses_task_registry() -> None:
+    """手工订阅搜索命令应以稳定 owner 提交历史兼容的调度参数。"""
+    registry = _TaskRegistry()
+    repository = object()
+    runtime = SimpleNamespace(
+        subscription=SimpleNamespace(repository=lambda _db: repository)
+    )
+    command = subscription_dependencies.get_search_subscriptions_command(
+        task_registry=registry,
+        db=object(),
+        runtime=runtime,
+    )
+
+    found = asyncio.run(
+        command.execute(SubscribeSearchActor(username="admin", is_superuser=True))
+    )
+
+    function, args, kwargs, owner = registry.calls[0]
+    assert found is True
+    assert function is subscription_dependencies.start_scheduler_job
+    assert args == ()
+    assert kwargs == {
+        "job_id": "subscribe_search",
+        "sid": None,
+        "state": "R",
+        "manual": True,
+    }
+    assert owner == "api.subscription.search_schedule"
+
+
 def test_history_ai_redo_uses_task_registry() -> None:
     """单条历史 AI 重做应登记宿主任务并使用稳定 owner。"""
     registry = _TaskRegistry()

tests/test_task_ownership_gate.py

@@ -0,0 +1,71 @@
+"""TaskRegistry owner 静态门禁回归测试。"""
+
+import textwrap
+from pathlib import Path
+
+from scripts.architecture.task_ownership import collect_task_owner_violations
+
+
+def _scan_source(tmp_path: Path, source: str):
+    """构造最小宿主源码并通过公开扫描入口返回违规。"""
+    source_path = tmp_path / "app/api/sample.py"
+    source_path.parent.mkdir(parents=True)
+    source_path.write_text(textwrap.dedent(source), encoding="utf-8")
+    return collect_task_owner_violations(tmp_path)
+
+
+def test_host_task_registry_calls_use_literal_owner() -> None:
+    """当前 canonical 宿主的登记器调用必须保持 owner 零债务。"""
+    assert collect_task_owner_violations() == []
+
+
+def test_owner_gate_tracks_known_registry_without_matching_same_named_methods(
+    tmp_path: Path,
+) -> None:
+    """门禁只检查可证明的登记器接收者，并区分缺失、动态和空 owner。"""
+    violations = _scan_source(
+        tmp_path,
+        """
+        from app.api.context import resolve_background_task_registry as resolve_registry
+        from app.runtime.tasks import TaskRegistry, get_task_registry
+
+        def schedule(task_registry: TaskRegistry, unrelated, dynamic_owner):
+            unrelated.create(work())
+            task_registry.create(work())
+            resolve_registry(task_registry).create_sync(work, owner=dynamic_owner)
+            get_task_registry().register(task, owner="   ")
+        """,
+    )
+
+    assert [violation.method for violation in violations] == [
+        "create",
+        "create_sync",
+        "register",
+    ]
+    assert [violation.reason for violation in violations] == [
+        "缺少显式 owner",
+        "的 owner 必须是非空字符串字面量",
+        "的 owner 必须是非空字符串字面量",
+    ]
+
+
+def test_owner_gate_accepts_aliases_and_stable_literal_owners(tmp_path: Path) -> None:
+    """类、模块和工厂别名仍应被识别，稳定字符串 owner 可以通过。"""
+    violations = _scan_source(
+        tmp_path,
+        """
+        import app.runtime.tasks as runtime_tasks
+        from app.runtime.tasks import TaskRegistry as Registry
+
+        def schedule(task_registry: Registry):
+            local_registry = runtime_tasks.TaskRegistry()
+            task_registry.create(work(), owner="api.example.async")
+            local_registry.create_sync(work, owner="api.example.sync")
+            runtime_tasks.get_task_registry().register(
+                task,
+                owner="api.example.existing",
+            )
+        """,
+    )
+
+    assert violations == []

tests/test_task_registry.py

@@ -1,6 +1,7 @@
 """进程内后台任务登记与关停语义测试。"""
 
 import asyncio
+import threading
 
 import pytest
 
@@ -84,3 +85,97 @@ def test_task_registry_runs_sync_function_and_tracks_until_completion() -> None:
         release.set()
 
     asyncio.run(scenario())
+
+
+def test_task_registry_keeps_timed_out_sync_owner_until_real_completion() -> None:
+    """同步线程超过关停预算后仍应保留 owner，不能把包装任务取消成伪完成。"""
+
+    async def scenario() -> None:
+        registry = TaskRegistry()
+        started = threading.Event()
+        release = threading.Event()
+        reports: list[dict[str, object]] = []
+        loop = asyncio.get_running_loop()
+        previous_handler = loop.get_exception_handler()
+        loop.set_exception_handler(lambda _, context: reports.append(context))
+
+        def worker() -> None:
+            """模拟无法由 asyncio 取消、需要外部资源自行结束的同步工作。"""
+            started.set()
+            release.wait()
+
+        try:
+            task = registry.create_sync(worker, owner="test.sync-timeout")
+            assert await asyncio.to_thread(started.wait, 1.0)
+
+            await registry.shutdown(timeout_seconds=0.001)
+
+            assert not task.done()
+            assert [record.owner for record in registry.records] == [
+                "test.sync-timeout"
+            ]
+            assert reports[-1]["owners"] == ("test.sync-timeout",)
+
+            release.set()
+            await task
+            await asyncio.sleep(0)
+            assert registry.records == ()
+        finally:
+            release.set()
+            loop.set_exception_handler(previous_handler)
+
+    asyncio.run(scenario())
+
+
+def test_task_registry_keeps_stubborn_cancelled_task_visible() -> None:
+    """协程清理超过预算时只收一次取消，并在最终退出后自动清理。"""
+
+    async def scenario() -> None:
+        registry = TaskRegistry()
+        started = asyncio.Event()
+        cleanup_started = asyncio.Event()
+        release = asyncio.Event()
+        cancellation_count = 0
+        reports: list[dict[str, object]] = []
+        loop = asyncio.get_running_loop()
+        previous_handler = loop.get_exception_handler()
+        loop.set_exception_handler(lambda _, context: reports.append(context))
+
+        async def worker() -> None:
+            """模拟收到取消后仍必须完成的异步清理。"""
+            nonlocal cancellation_count
+            started.set()
+            try:
+                await asyncio.Event().wait()
+            except asyncio.CancelledError:
+                cancellation_count += 1
+                cleanup_started.set()
+                await release.wait()
+
+        task = registry.create(worker(), owner="test.stubborn")
+        await started.wait()
+        try:
+            await registry.shutdown(timeout_seconds=0.001)
+
+            assert cleanup_started.is_set()
+            assert not task.done()
+            assert cancellation_count == 1
+            assert [record.owner for record in registry.records] == [
+                "test.stubborn"
+            ]
+            assert reports[-1]["owners"] == ("test.stubborn",)
+
+            await registry.shutdown(timeout_seconds=0.001)
+            assert not task.done()
+            assert cancellation_count == 1
+            assert len(reports) == 1
+
+            release.set()
+            await task
+            await asyncio.sleep(0)
+            assert registry.records == ()
+        finally:
+            release.set()
+            loop.set_exception_handler(previous_handler)
+
+    asyncio.run(scenario())