From 6c334f7b9b50bae4da96d62ebd747d89b7a1b093 Mon Sep 17 00:00:00 2001 From: InfinityPacer <160988576+InfinityPacer@users.noreply.github.com> Date: Tue, 25 Aug 2026 06:43:37 +0800 Subject: [PATCH] =?UTF-8?q?fix(runtime):=20=E6=94=B6=E6=95=9B=E8=B7=A8?= =?UTF-8?q?=E7=BA=BF=E7=A8=8B=E4=BB=BB=E5=8A=A1=E6=8F=90=E4=BA=A4=E7=BB=88?= =?UTF-8?q?=E6=80=81=20(#6446)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/chain/_transfer.py | 53 +++-- app/runtime/tasks.py | 112 +++++++--- scripts/perf/README.md | 15 ++ scripts/perf/task_registry_ab.py | 123 +++++++++++ tests/test_agent_message_routing.py | 22 +- tests/test_server_statistic_lifecycle.py | 18 ++ tests/test_task_registry.py | 226 ++++++++++++++++++-- tests/test_transfer_failed_retry_buttons.py | 105 ++++++++- 8 files changed, 609 insertions(+), 65 deletions(-) create mode 100644 scripts/perf/task_registry_ab.py diff --git a/app/chain/_transfer.py b/app/chain/_transfer.py index b15ce0366..139317c0f 100644 --- a/app/chain/_transfer.py +++ b/app/chain/_transfer.py @@ -1404,19 +1404,6 @@ class FailedRetryMixin: redo_prompt = build_manual_redo_prompt(history) - self.post_message( - Message( - channel=channel, - source=source, - userid=userid, - username=username, - title=f"已将整理记录 #{history_id} 交给智能助手处理", - text="处理完成后会在这里回复结果。", - link=self.runtime_config.history_url, - save_history=False, - ) - ) - async def _run_ai_takeover(): final_output = "" @@ -1425,6 +1412,18 @@ class FailedRetryMixin: final_output = text_output or "" try: + await self.async_post_message( + Message( + channel=channel, + source=source, + userid=userid, + username=username, + title=f"已将整理记录 #{history_id} 交给智能助手处理", + text="处理完成后会在这里回复结果。", + link=self.runtime_config.history_url, + save_history=False, + ) + ) manager = get_running_agent_manager() if manager is None: raise RuntimeError("智能助手服务未运行") @@ -1462,11 +1461,29 @@ class FailedRetryMixin: ) ) - get_task_registry().submit_threadsafe( - _run_ai_takeover(), - loop=global_vars.loop, - owner="chain.transfer.ai_takeover", - ) + try: + registry = get_task_registry() + loop = global_vars.loop + registry.submit_threadsafe( + _run_ai_takeover(), + loop=loop, + owner="chain.transfer.ai_takeover", + ) + except RuntimeError as error: + logger.warning("智能助手整理任务提交失败：%s", error) + self.post_message( + Message( + channel=channel, + source=source, + userid=userid, + username=username, + title="智能助手整理失败", + text="系统正在关闭，无法提交处理任务，请稍后重试。", + link=self.runtime_config.history_url, + save_history=False, + ) + ) + return def _re_transfer( self, diff --git a/app/runtime/tasks.py b/app/runtime/tasks.py index e58d66b45..000b1fa07 100644 --- a/app/runtime/tasks.py +++ b/app/runtime/tasks.py @@ -4,6 +4,7 @@ from __future__ import annotations import asyncio import concurrent.futures +import threading from collections.abc import Coroutine from dataclasses import dataclass from functools import partial @@ -19,6 +20,14 @@ class TaskRecord: cancel_on_shutdown: bool +@dataclass(slots=True) +class _ThreadsafeSubmission: + """持有跨线程提交从排队到真实 Task 发布之间的生命周期。""" + + coroutine: Coroutine[Any, Any, Any] + task: asyncio.Task[Any] | None = None + + class TaskRegistry: """管理由宿主创建的进程内后台任务，并提供统一取消与等待入口。""" @@ -27,6 +36,10 @@ class TaskRegistry: 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._threadsafe_submissions: dict[ + concurrent.futures.Future[Any], _ThreadsafeSubmission + ] = {} + self._state_lock = threading.Lock() self._accepting = True @property @@ -77,37 +90,61 @@ class TaskRegistry: owner: str, cancel_on_shutdown: bool = True, ) -> concurrent.futures.Future[Any]: - """从宿主线程提交协程，并在目标循环内原子登记 owner 后执行。""" + """从宿主线程提交协程，并持有排队阶段直至发布真实 Task。""" completion: concurrent.futures.Future[Any] = concurrent.futures.Future() - task_holder: dict[str, asyncio.Task[Any]] = {} + submission = _ThreadsafeSubmission(coroutine=coroutine) def mirror_completion(task: asyncio.Task[Any]) -> None: """把登记任务的真实终态镜像给跨线程调用方。""" if completion.done(): return - if task.cancelled(): - completion.cancel() + try: + if task.cancelled(): + completion.cancel() + return + exception = task.exception() + if exception is not None: + completion.set_exception(exception) + else: + completion.set_result(task.result()) + except concurrent.futures.InvalidStateError: + # completion 可由调用线程同时取消，真实 Task 仍由 Registry 观察。 return - exception = task.exception() - if exception is not None: - completion.set_exception(exception) - else: - completion.set_result(task.result()) def submit_on_loop() -> None: - """在目标循环内完成 accepting 检查、任务创建和 owner 登记。""" - if completion.cancelled(): + """在目标循环内把 pending submission 原子移交给真实 Task。""" + close_coroutine = False + task: asyncio.Task[Any] | None = None + error: Exception | None = None + with self._state_lock: + current = self._threadsafe_submissions.get(completion) + if current is not submission: + return + if completion.cancelled(): + self._threadsafe_submissions.pop(completion, None) + close_coroutine = True + else: + try: + task = self.create( + coroutine, + owner=owner, + cancel_on_shutdown=cancel_on_shutdown, + ) + except Exception as caught: + error = caught + close_coroutine = True + else: + submission.task = task + finally: + self._threadsafe_submissions.pop(completion, None) + + if close_coroutine: coroutine.close() - return - try: - task = self.create( - coroutine, - owner=owner, - cancel_on_shutdown=cancel_on_shutdown, - ) - except Exception as error: - if not completion.done(): + if error is not None: + try: completion.set_exception(error) + except concurrent.futures.InvalidStateError: + pass loop.call_exception_handler( { "message": "MoviePilot 跨线程后台任务提交失败", @@ -116,7 +153,8 @@ class TaskRegistry: } ) return - task_holder["task"] = task + if task is None: + return task.add_done_callback(mirror_completion) if completion.cancelled() and not task.done(): task.cancel() @@ -127,7 +165,17 @@ class TaskRegistry: """调用方取消 completion 时，把取消请求转交目标循环中的真实任务。""" if not submitted.cancelled(): return - task = task_holder.get("task") + close_coroutine = False + with self._state_lock: + task = submission.task + if ( + task is None + and self._threadsafe_submissions.get(submitted) is submission + ): + self._threadsafe_submissions.pop(submitted, None) + close_coroutine = True + if close_coroutine: + coroutine.close() if task is not None and not task.done(): try: loop.call_soon_threadsafe(task.cancel) @@ -135,10 +183,20 @@ class TaskRegistry: pass completion.add_done_callback(cancel_registered_task) + with self._state_lock: + if not self._accepting: + coroutine.close() + raise RuntimeError("后台任务登记器正在关闭，不能再创建新任务") + self._threadsafe_submissions[completion] = submission try: loop.call_soon_threadsafe(submit_on_loop) except RuntimeError: - coroutine.close() + with self._state_lock: + close_coroutine = ( + self._threadsafe_submissions.pop(completion, None) is submission + ) + if close_coroutine: + coroutine.close() raise return completion @@ -182,7 +240,13 @@ class TaskRegistry: async def shutdown(self, *, timeout_seconds: float = 10.0) -> bool: """停止接收并有限等待存量任务，返回全部 owner 是否真实收敛。""" - self._accepting = False + with self._state_lock: + self._accepting = False + pending_submissions = tuple(self._threadsafe_submissions.items()) + self._threadsafe_submissions.clear() + for completion, submission in pending_submissions: + submission.coroutine.close() + completion.cancel() records = self.records tasks = [record.task for record in records] for record in records: diff --git a/scripts/perf/README.md b/scripts/perf/README.md index 6c23c5d02..bc8775ad8 100644 --- a/scripts/perf/README.md +++ b/scripts/perf/README.md @@ -224,6 +224,21 @@ preflight 会验证 Python 3.14、GIL 状态、`thread_inherit_context`、MovieP 依赖、语义、驱动、启动或样本完整性不成立。该工具只用于隔离的本地长 A/B，不接真实凭据、用户数据库、 媒体目录或外网，也不加入常规 CI。 +## TaskRegistry 跨线程提交 A/B + +`task_registry_ab.py` 验证目标事件循环尚未分发 callback 时执行 shutdown，pending completion 与原始 +coroutine 是否取得明确终态，同时采集跨线程提交最小协程的提交和完成耗时。分别在 Before/After revision +运行相同参数并保留两份 JSON，即可比较正确性与固定负载开销： + +```bash +../.venv-test/bin/python scripts/perf/task_registry_ab.py \ + --iterations 2000 \ + --samples 7 +``` + +该探针不访问数据库、配置或网络。吞吐结果用于识别可重复回退，不作为跨机器性能阈值；pending completion +取消且 coroutine 关闭属于正确性门禁。 + ### PostgreSQL 同步驱动三方案 `postgresql_driver_ab.py` 在同一 PostgreSQL 容器中比较标准 V3/psycopg2、标准 diff --git a/scripts/perf/task_registry_ab.py b/scripts/perf/task_registry_ab.py new file mode 100644 index 000000000..3ad984b16 --- /dev/null +++ b/scripts/perf/task_registry_ab.py @@ -0,0 +1,123 @@ +"""测量 TaskRegistry 跨线程 pending submission 的关停终态与提交吞吐。""" + +from __future__ import annotations + +import argparse +import asyncio +import inspect +import json +import statistics +import sys +import time +from pathlib import Path + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +from app.runtime.tasks import TaskRegistry + + +async def _value(value: int) -> int: + """返回输入值，提供最小可完成协程。""" + return value + + +def pending_shutdown_probe() -> dict[str, object]: + """在目标 loop 尚未分发 callback 时关闭 Registry，并报告提交终态。""" + registry = TaskRegistry() + target_loop = asyncio.new_event_loop() + coroutine = _value(1) + completion = registry.submit_threadsafe( + coroutine, + loop=target_loop, + owner="probe.pending", + ) + shutdown_result = asyncio.run(registry.shutdown(timeout_seconds=0.001)) + result = { + "shutdown_result": shutdown_result, + "completion_done": completion.done(), + "completion_cancelled": completion.cancelled(), + "coroutine_state": inspect.getcoroutinestate(coroutine), + } + completion.cancel() + coroutine.close() + target_loop.close() + return result + + +async def throughput_sample(iterations: int) -> dict[str, float]: + """从工作线程提交一组最小协程，并测量提交和完整完成时间。""" + registry = TaskRegistry() + loop = asyncio.get_running_loop() + started = time.perf_counter() + + def submit_all(): + """在同一宿主线程连续提交，保持各轮工作负载一致。""" + return [ + registry.submit_threadsafe( + _value(index), + loop=loop, + owner="probe.throughput", + ) + for index in range(iterations) + ] + + completions = await asyncio.to_thread(submit_all) + submitted = time.perf_counter() + values = await asyncio.gather( + *(asyncio.wrap_future(completion) for completion in completions) + ) + finished = time.perf_counter() + assert sum(values) == iterations * (iterations - 1) // 2 + assert await registry.shutdown(timeout_seconds=1.0) is True + return { + "submit_ms": (submitted - started) * 1000, + "total_ms": (finished - started) * 1000, + } + + +async def run_samples(iterations: int, samples: int) -> list[dict[str, float]]: + """顺序执行多轮样本，避免并行样本互相争抢事件循环。""" + return [await throughput_sample(iterations) for _ in range(samples)] + + +def parse_args() -> argparse.Namespace: + """解析探针负载规模。""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--iterations", type=int, default=2000) + parser.add_argument("--samples", type=int, default=7) + return parser.parse_args() + + +def main() -> None: + """运行终态探针和吞吐样本并输出 JSON。""" + args = parse_args() + if args.iterations < 1 or args.samples < 1: + raise SystemExit("iterations 和 samples 必须大于 0") + samples = asyncio.run(run_samples(args.iterations, args.samples)) + print( + json.dumps( + { + "pending": pending_shutdown_probe(), + "throughput": { + "iterations": args.iterations, + "samples": args.samples, + "submit_ms": [sample["submit_ms"] for sample in samples], + "total_ms": [sample["total_ms"] for sample in samples], + "submit_median_ms": statistics.median( + sample["submit_ms"] for sample in samples + ), + "total_median_ms": statistics.median( + sample["total_ms"] for sample in samples + ), + }, + }, + ensure_ascii=False, + indent=2, + ) + ) + + +if __name__ == "__main__": + main() diff --git a/tests/test_agent_message_routing.py b/tests/test_agent_message_routing.py index dc91c3813..2f7fc3415 100644 --- a/tests/test_agent_message_routing.py +++ b/tests/test_agent_message_routing.py @@ -3,7 +3,7 @@ from concurrent.futures import Future from dataclasses import replace from unittest.mock import AsyncMock, Mock, patch -from app.agent import MoviePilotAgent +from app.agent.orchestrator import MoviePilotAgent from app.agent.orchestrator import AgentManagerQueueFullError from app.agent.tools.impl.ask_user_choice import ( AskUserChoiceTool, @@ -12,13 +12,14 @@ from app.agent.tools.impl.ask_user_choice import ( from app.agent.tools.impl.send_message import SendMessageTool from app.chain.message import MessageChain from app.runtime.config import global_vars, settings -from app.db import SessionFactory +from app.db.session import SessionFactory from app.db.oper.message import MessageOper from app.db.models.message import Message from app.application.messaging.agent import AgentInteractionOption, agent_interaction_manager from app.application.messaging.interaction import InteractionContext from app.application.messaging.media import media_interaction_manager from app.schemas.types import NotificationChannel, MessageType +from app.runtime.tasks import TaskRegistry def _clear_messages() -> None: @@ -61,6 +62,23 @@ def test_agent_session_clear_uses_owned_threadsafe_submission(): } +def test_agent_session_clear_handles_closed_task_registry(): + """宿主已停止接收任务时，同步消息清理链应记录拒绝而不是泄漏协程。""" + registry = TaskRegistry() + asyncio.run(registry.shutdown(timeout_seconds=0.01)) + manager = Mock(clear_session=AsyncMock()) + + with patch.object(global_vars, "CURRENT_EVENT_LOOP", _running_loop_stub()), patch( + "app.chain.message.get_running_agent_manager", return_value=manager + ), patch("app.chain.message.get_task_registry", return_value=registry), patch( + "app.chain.message.logger" + ) as logger: + MessageChain._schedule_agent_session_clear("session-1", "10001") + + logger.warning.assert_called_once() + assert "正在关闭" in logger.warning.call_args.args[0] + + def test_remote_session_clear_reuses_owned_clear_scheduler(): """远程清理命令应复用唯一 Agent 会话清理入口。""" chain = MessageChain() diff --git a/tests/test_server_statistic_lifecycle.py b/tests/test_server_statistic_lifecycle.py index 25aaacd4b..be13684fc 100644 --- a/tests/test_server_statistic_lifecycle.py +++ b/tests/test_server_statistic_lifecycle.py @@ -1,9 +1,12 @@ """服务端统计兼容入口的后台任务生命周期回归。""" +import asyncio + from unittest.mock import Mock, patch from app.adapters.external.server import MoviePilotServerHelper from app.runtime.config import global_vars +from app.runtime.tasks import TaskRegistry def test_legacy_subscription_reports_use_owned_threadsafe_tasks() -> None: @@ -40,3 +43,18 @@ def test_legacy_subscription_report_rejects_without_runtime_loop() -> None: """宿主生命周期不可用时应拒绝提交，并保持布尔返回合同。""" with patch.object(global_vars, "CURRENT_EVENT_LOOP", None): assert MoviePilotServerHelper.sub_done_async({"media_id": "1"}) is False + + +def test_legacy_subscription_report_handles_closed_task_registry() -> None: + """运行 loop 尚在但宿主已封口时，兼容入口应保持布尔失败合同。""" + registry = TaskRegistry() + asyncio.run(registry.shutdown(timeout_seconds=0.01)) + loop = Mock(**{"is_running.return_value": True, "is_closed.return_value": False}) + + with patch.object(global_vars, "CURRENT_EVENT_LOOP", loop), patch( + "app.adapters.external.server.get_task_registry", return_value=registry + ), patch("app.adapters.external.server.logger") as logger: + assert MoviePilotServerHelper.sub_done_async({"media_id": "1"}) is False + + logger.warning.assert_called_once() + assert "正在关闭" in logger.warning.call_args.args[0] diff --git a/tests/test_task_registry.py b/tests/test_task_registry.py index 677f99de4..eb9aebd89 100644 --- a/tests/test_task_registry.py +++ b/tests/test_task_registry.py @@ -1,13 +1,39 @@ """进程内后台任务登记与关停语义测试。""" import asyncio +import concurrent.futures +import inspect import threading +import weakref +from collections.abc import Coroutine +from typing import Any import pytest from app.runtime.tasks import TaskRegistry +class _CloseCountingCoroutine(Coroutine[Any, Any, None]): + """记录 pending submission 的协程关闭所有权。""" + + def __init__(self) -> None: + self.close_count = 0 + + def __await__(self): + return self + + def send(self, _value): + raise StopIteration + + def throw(self, typ, val=None, tb=None): + if val is None: + raise typ + raise val.with_traceback(tb) + + def close(self) -> None: + self.close_count += 1 + + def test_task_registry_removes_completed_task() -> None: """正常完成的任务应自动退出登记表，避免长期持有请求对象。""" @@ -124,37 +150,209 @@ def test_task_registry_owns_threadsafe_submission_until_shutdown() -> None: def test_task_registry_rejects_threadsafe_submission_after_shutdown() -> None: - """关停先赢得竞态时应关闭协程并通过 completion 报告拒绝原因。""" + """关停先赢得竞态时应关闭协程并同步拒绝提交。""" async def scenario() -> None: registry = TaskRegistry() loop = asyncio.get_running_loop() - reports: list[dict[str, object]] = [] - previous_handler = loop.get_exception_handler() - loop.set_exception_handler(lambda _, context: reports.append(context)) async def late_worker() -> None: """模拟关停完成后从宿主线程到达的晚任务。""" - try: - assert await registry.shutdown(timeout_seconds=1.0) is True - completion = await asyncio.to_thread( + assert await registry.shutdown(timeout_seconds=1.0) is True + coroutine = late_worker() + with pytest.raises(RuntimeError, match="正在关闭"): + await asyncio.to_thread( registry.submit_threadsafe, - late_worker(), + coroutine, loop=loop, owner="test.threadsafe-late", ) - with pytest.raises(RuntimeError, match="正在关闭"): - await asyncio.wrap_future(completion) - assert registry.records == () - assert reports[-1]["owner"] == "test.threadsafe-late" - finally: - loop.set_exception_handler(previous_handler) + assert inspect.getcoroutinestate(coroutine) == inspect.CORO_CLOSED + assert registry.records == () asyncio.run(scenario()) +def test_task_registry_shutdown_closes_submission_before_loop_dispatch() -> None: + """回调尚未执行时关停也必须关闭协程并结束 completion。""" + registry = TaskRegistry() + loop = asyncio.new_event_loop() + + async def worker() -> None: + """不应被迟到的 loop callback 启动。""" + + coroutine = worker() + completion = registry.submit_threadsafe( + coroutine, + loop=loop, + owner="test.threadsafe-pending", + ) + + assert asyncio.run(registry.shutdown(timeout_seconds=0.01)) is True + assert completion.cancelled() + assert inspect.getcoroutinestate(coroutine) == inspect.CORO_CLOSED + + loop.run_until_complete(asyncio.sleep(0)) + assert registry.records == () + loop.close() + + +def test_task_registry_caller_cancels_before_loop_dispatch() -> None: + """调用方在 callback 前取消时应关闭协程，迟到 callback 不得创建 Task。""" + registry = TaskRegistry() + loop = asyncio.new_event_loop() + + async def worker() -> None: + """不应在 completion 取消后启动。""" + + coroutine = worker() + completion = registry.submit_threadsafe( + coroutine, + loop=loop, + owner="test.threadsafe-cancel-before-dispatch", + ) + + assert completion.cancel() + assert inspect.getcoroutinestate(coroutine) == inspect.CORO_CLOSED + loop.run_until_complete(asyncio.sleep(0)) + + assert registry.records == () + loop.close() + + +def test_task_registry_closes_submission_when_loop_rejects_callback() -> None: + """目标 loop 已关闭时同步失败，并且不遗留未等待协程。""" + registry = TaskRegistry() + loop = asyncio.new_event_loop() + loop.close() + + async def worker() -> None: + """模拟无法进入目标 loop 的提交。""" + + coroutine = worker() + with pytest.raises(RuntimeError, match="closed"): + registry.submit_threadsafe( + coroutine, + loop=loop, + owner="test.threadsafe-closed-loop", + ) + + assert inspect.getcoroutinestate(coroutine) == inspect.CORO_CLOSED + + +def test_task_registry_shutdown_owns_close_before_loop_rejection() -> None: + """shutdown 已取走 pending 后，迟到的投递失败不得再次关闭同一协程。""" + registry = TaskRegistry() + dispatch_entered = threading.Event() + release_dispatch = threading.Event() + coroutine = _CloseCountingCoroutine() + + class BlockingClosedLoop: + """让 shutdown 稳定发生在投递抛错之前。""" + + @staticmethod + def call_soon_threadsafe(_callback) -> None: + dispatch_entered.set() + release_dispatch.wait() + raise RuntimeError("loop closed") + + with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor: + submission = executor.submit( + registry.submit_threadsafe, + coroutine, + loop=BlockingClosedLoop(), + owner="test.threadsafe-shutdown-before-rejection", + ) + assert dispatch_entered.wait(1.0) + assert asyncio.run(registry.shutdown(timeout_seconds=0.01)) is True + release_dispatch.set() + with pytest.raises(RuntimeError, match="loop closed"): + submission.result(timeout=1.0) + + assert coroutine.close_count == 1 + + +def test_task_registry_shutdown_rejects_late_successful_dispatch() -> None: + """shutdown 取走 pending 后，迟到的成功投递也不得发布真实 Task。""" + registry = TaskRegistry() + dispatch_entered = threading.Event() + release_dispatch = threading.Event() + callbacks = [] + coroutine = _CloseCountingCoroutine() + + class BlockingLoop: + """让 shutdown 稳定发生在 loop 接受 callback 之前。""" + + @staticmethod + def call_soon_threadsafe(callback) -> None: + dispatch_entered.set() + release_dispatch.wait() + callbacks.append(callback) + + with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor: + submission = executor.submit( + registry.submit_threadsafe, + coroutine, + loop=BlockingLoop(), + owner="test.threadsafe-shutdown-before-dispatch", + ) + assert dispatch_entered.wait(1.0) + assert asyncio.run(registry.shutdown(timeout_seconds=0.01)) is True + release_dispatch.set() + completion = submission.result(timeout=1.0) + + assert completion.cancelled() + callbacks[0]() + assert coroutine.close_count == 1 + assert registry.records == () + + +def test_task_registry_completion_does_not_retain_itself_after_cancellation() -> None: + """终态回调不得让已取消 completion 依赖循环 GC 才能释放。""" + registry = TaskRegistry() + loop = asyncio.new_event_loop() + completion = registry.submit_threadsafe( + _CloseCountingCoroutine(), + loop=loop, + owner="test.threadsafe-completion-release", + ) + completion_ref = weakref.ref(completion) + + assert completion.cancel() + loop.run_until_complete(asyncio.sleep(0)) + del completion + + assert completion_ref() is None + loop.close() + + +def test_task_registry_pending_non_cancellable_work_does_not_start_on_shutdown() -> None: + """尚未发布成 Task 的 non-cancellable 工作仍应在关停时直接关闭。""" + registry = TaskRegistry() + loop = asyncio.new_event_loop() + + async def worker() -> None: + """未开始的同步兼容工作不应拖延关停。""" + + coroutine = worker() + completion = registry.submit_threadsafe( + coroutine, + loop=loop, + owner="test.threadsafe-pending-non-cancellable", + cancel_on_shutdown=False, + ) + + assert asyncio.run(registry.shutdown(timeout_seconds=0.01)) is True + assert completion.cancelled() + assert inspect.getcoroutinestate(coroutine) == inspect.CORO_CLOSED + + loop.run_until_complete(asyncio.sleep(0)) + assert registry.records == () + loop.close() + + def test_task_registry_keeps_timed_out_sync_owner_until_real_completion() -> None: """同步线程超过关停预算后仍应保留 owner，不能把包装任务取消成伪完成。""" diff --git a/tests/test_transfer_failed_retry_buttons.py b/tests/test_transfer_failed_retry_buttons.py index ec9868f68..a3d6a3f94 100644 --- a/tests/test_transfer_failed_retry_buttons.py +++ b/tests/test_transfer_failed_retry_buttons.py @@ -16,6 +16,7 @@ from app.chain.message import MessageChain from app.chain.transfer import TransferChain from app.application.messaging.interaction import InteractionContext from app.runtime.config import global_vars, settings +from app.runtime.tasks import TaskRegistry from app.schemas.types import NotificationChannel @@ -128,10 +129,18 @@ class TestTransferFailedRetryButtons(unittest.TestCase): errmsg="未识别到媒体信息", ) - def _close_pending_coro(coro, *args, **kwargs): - """关闭被调度的协程：测试中事件循环未运行，不关闭会残留 never-awaited 警告。""" - coro.close() + async_messages = [] + def _run_pending_coro(coro, *args, **kwargs): + asyncio.run(coro) + + async def _capture_message(message): + async_messages.append(message) + + async def _finish_immediately(**kwargs): + kwargs["output_callback"]("ok") + + manager = SimpleNamespace(run_background_prompt=_finish_immediately) loop = Mock(**{"is_running.return_value": True, "is_closed.return_value": False}) with patch.object(global_vars, "CURRENT_EVENT_LOOP", loop), patch.object( settings, "AI_AGENT_ENABLE", True @@ -141,12 +150,14 @@ class TestTransferFailedRetryButtons(unittest.TestCase): ) as history_oper_cls, patch( "app.chain._transfer.build_manual_redo_prompt", return_value="retry transfer prompt", + ), patch( + "app.chain._transfer.get_running_agent_manager", return_value=manager ), patch("app.chain._transfer.get_task_registry") as get_registry: get_registry.return_value.submit_threadsafe.side_effect = ( - _close_pending_coro + _run_pending_coro ) history_oper_cls.return_value.get.return_value = history - with patch.object(chain, "post_message") as post_message: + with patch.object(chain, "async_post_message", side_effect=_capture_message): chain.handle_failed_transfer_callback( callback_data="transfer_ai_retry_34", channel=NotificationChannel.Telegram, @@ -160,10 +171,90 @@ class TestTransferFailedRetryButtons(unittest.TestCase): get_registry.return_value.submit_threadsafe.call_args.kwargs["owner"], "chain.transfer.ai_takeover", ) + self.assertEqual(len(async_messages), 2) + self.assertEqual( + async_messages[0].title, + "已将整理记录 #34 交给智能助手处理", + ) + self.assertEqual(async_messages[1].title, "智能助手整理完成") + + def test_transfer_ai_retry_callback_reports_closed_task_registry(self): + """宿主停止接收任务时，不得向用户报告智能助手已接管。""" + chain = TransferChain() + chain.runtime_config = replace(chain.runtime_config, ai_agent_enable=True) + history = SimpleNamespace(id=34) + registry = TaskRegistry() + asyncio.run(registry.shutdown(timeout_seconds=0.01)) + loop = Mock(**{"is_running.return_value": True, "is_closed.return_value": False}) + + with patch.object(global_vars, "CURRENT_EVENT_LOOP", loop), patch.object( + settings, "AI_AGENT_ENABLE", True + ), patch( + "app.chain._transfer.get_chain_transfer_history_port" + ) as history_port, patch( + "app.chain._transfer.build_manual_redo_prompt", + return_value="retry transfer prompt", + ), patch( + "app.chain._transfer.get_task_registry", return_value=registry + ), patch( + "app.chain._transfer.logger" + ) as logger, patch.object( + chain, "post_message" + ) as post_message: + history_port.return_value.get.return_value = history + + chain.handle_failed_transfer_callback( + callback_data="transfer_ai_retry_34", + channel=NotificationChannel.Telegram, + source="telegram-test", + userid="10001", + username="tester", + ) + + logger.warning.assert_called_once() self.assertEqual(post_message.call_count, 1) self.assertEqual( - post_message.call_args_list[0].args[0].title, - "已将整理记录 #34 交给智能助手处理", + post_message.call_args.args[0].title, + "智能助手整理失败", + ) + self.assertNotIn("已将", post_message.call_args.args[0].title) + + def test_transfer_ai_retry_callback_reports_unavailable_event_loop(self): + """主循环不可用时，应在创建后台协程前返回明确失败提示。""" + chain = TransferChain() + chain.runtime_config = replace(chain.runtime_config, ai_agent_enable=True) + history = SimpleNamespace(id=34) + + with patch.object(global_vars, "CURRENT_EVENT_LOOP", None), patch.object( + settings, "AI_AGENT_ENABLE", True + ), patch( + "app.chain._transfer.get_chain_transfer_history_port" + ) as history_port, patch( + "app.chain._transfer.build_manual_redo_prompt", + return_value="retry transfer prompt", + ), patch( + "app.chain._transfer.get_task_registry" + ) as get_registry, patch( + "app.chain._transfer.logger" + ) as logger, patch.object( + chain, "post_message" + ) as post_message: + history_port.return_value.get.return_value = history + + chain.handle_failed_transfer_callback( + callback_data="transfer_ai_retry_34", + channel=NotificationChannel.Telegram, + source="telegram-test", + userid="10001", + username="tester", + ) + + get_registry.return_value.submit_threadsafe.assert_not_called() + logger.warning.assert_called_once() + self.assertEqual(post_message.call_count, 1) + self.assertEqual( + post_message.call_args.args[0].title, + "智能助手整理失败", ) def test_transfer_ai_retry_callback_uses_successful_move_dest_as_source(self):