fix(transmission): confirm torrent state around episode selection

app/modules/transmission/__init__.py

@@ -254,7 +254,7 @@ class TransmissionModule(_DownloaderModuleBase[Transmission]):
             server: Transmission, torrent_hash: str, episodes: Set[int], custom_words: Optional[str],
     ) -> Tuple[bool, str]:
         """
-        先确认任务暂停再按来源集数映射选集，兼容 RPC 返回正在运行的重复任务。
+        先确认任务暂停再按来源集数映射选集，确认启动状态后才返回成功。
 
         :param server: 已配置的 Transmission 客户端
         :param torrent_hash: 需要暂停、选集并恢复的任务哈希
@@ -262,7 +262,7 @@ class TransmissionModule(_DownloaderModuleBase[Transmission]):
         :param custom_words: 来源识别词，按行参与物理文件名识别
         :return: 是否确认选集及启动成功，以及失败时的下载器核查说明
         """
-        if not server.stop_torrents(torrent_hash):
+        if not server.ensure_torrent_state(torrent_hash, paused=True):
             return False, f"选集失败，任务 {torrent_hash} 无法确认暂停，未执行选集，请在下载器核实"
         failure = f"选集失败，任务 {torrent_hash} 保持暂停，请在下载器检查"
         torrent_files = server.get_files(torrent_hash)
@@ -286,8 +286,8 @@ class TransmissionModule(_DownloaderModuleBase[Transmission]):
             return False, f"{failure}：未找到目标集数 {sorted(set(episodes) - matched)}"
         if not server.set_files(torrent_hash, wanted) or not server.set_unwanted_files(torrent_hash, unwanted):
             return False, f"{failure}：文件选择设置未确认"
-        if not server.start_torrents(torrent_hash):
-            if server.stop_torrents(torrent_hash):
+        if not server.ensure_torrent_state(torrent_hash, paused=False):
+            if server.ensure_torrent_state(torrent_hash, paused=True):
                 return False, f"{failure}：启动任务未确认，未记录成功"
             return False, f"选集失败，任务 {torrent_hash} 启动任务未确认，请在下载器核实，未记录成功"
         return True, "添加下载任务成功"

app/modules/transmission/transmission.py

@@ -2,6 +2,7 @@
 Transmission 客户端连接与下载任务操作封装。
 """
 
+import time
 from typing import List, Optional, Sequence, Tuple, Union
 
 import transmission_rpc
@@ -13,6 +14,14 @@ from app.runtime.log import logger
 
 TorrentIds = Union[str, int, Sequence[Union[str, int]]]
 
+# 每次启停的状态确认窗口与轮询间隔，单位均为秒。
+_TORRENT_STATE_TIMEOUT = 5.0
+_TORRENT_STATE_POLL_INTERVAL = 0.5
+# 启动后允许等待校验、下载或做种队列，不要求已经产生下载流量。
+_TORRENT_STARTED_STATES = {
+    "check pending", "checking", "download pending", "downloading", "seed pending", "seeding",
+}
+
 
 class Transmission:
     """
@@ -293,6 +302,43 @@ class Transmission:
             logger.error(f"停止种子出错：{str(err)}")
             return False
 
+    def ensure_torrent_state(self, tid: str, *, paused: bool) -> bool:
+        """
+        在选集边界确认任务实际暂停或启动，避免异步停止覆盖后续启动。
+
+        已暂停的任务不再发送停止请求。启动后只在仍停止时重试启动，
+        校验、下载、做种及对应排队状态均视为已接管，不强制跳过队列。
+
+        :param tid: 需要确认状态的单个种子哈希
+        :param paused: 为 True 时等待实际停止，为 False 时请求启动并确认状态
+        :return: 状态已确认时返回 True；未连接、RPC 失败或五秒内未确认时返回 False
+        """
+        if not self.trc:
+            return False
+        if not paused and not self.start_torrents(tid):
+            return False
+        expected = {"stopped"} if paused else _TORRENT_STARTED_STATES
+        deadline = time.monotonic() + _TORRENT_STATE_TIMEOUT
+        stop_requested = False
+        while (remaining := deadline - time.monotonic()) > 0:
+            try:
+                status = self.trc.get_torrent(tid, arguments=["status"], timeout=remaining).status
+            except Exception as err:
+                logger.error(f"确认种子 {tid} 启停状态出错：{str(err)}")
+                return False
+            if status in expected:
+                return True
+            if paused:
+                if not stop_requested and not self.stop_torrents(tid):
+                    return False
+                stop_requested = True
+            elif status == "stopped" and not self.start_torrents(tid):
+                return False
+            time.sleep(min(_TORRENT_STATE_POLL_INTERVAL, max(0, deadline - time.monotonic())))
+        action = "暂停" if paused else "启动"
+        logger.warning(f"种子 {tid} 在 {_TORRENT_STATE_TIMEOUT} 秒内未确认{action}状态")
+        return False
+
     def delete_torrents(self, delete_file: Optional[bool], ids: TorrentIds) -> bool:
         """
         删除指定种子任务，并按选项删除下载文件。

tests/test_downloader_submission.py

@@ -10,6 +10,7 @@ from unittest.mock import MagicMock, Mock
 import pytest
 from torrentool.bencode import Bencode
 
+import app.modules.transmission.transmission as transmission_client
 from app.chain.download.facade import DownloadChain
 from app.domain.context import Context, MediaInfo, TorrentInfo
 from app.domain.metainfo import MetaInfo
@@ -68,6 +69,10 @@ def downloader(request):
     )
     for method in ("set_files", "set_unwanted_files", "start_torrents", "stop_torrents"):
         getattr(server, method).return_value = True
+    server.ensure_torrent_state.side_effect = (
+        lambda torrent_hash, *, paused:
+        server.stop_torrents(torrent_hash) if paused else server.start_torrents(torrent_hash)
+    )
     module = request.param.__new__(request.param)
     module.get_instance = Mock(return_value=server)
     module.normalize_path = Mock(side_effect=lambda path, _name: path)
@@ -151,12 +156,140 @@ def test_transmission_all_files_wanted_starts_without_unwanted_rpc(downloader, f
     server.start_torrents.assert_called_once_with(torrent_hash)
 
 
+def _use_transmission_rpc(module, server) -> Transmission:
+    """
+    保留真实客户端的启停逻辑，仅替换 RPC 和种子文件读取边界。
+    """
+    client = Transmission.__new__(Transmission)
+    client.trc = Mock()
+    client.add_torrent = server.add_torrent
+    client.get_files = server.get_files
+    module.get_instance.return_value = client
+    return client
+
+
+@pytest.fixture
+def transmission_clock(monkeypatch):
+    """
+    在真实模块和客户端的联动测试中推进虚拟轮询时间。
+    """
+    clock = SimpleNamespace(now=0.0)
+
+    def advance(seconds: float) -> None:
+        """
+        按客户端要求推进等待时间。
+        """
+        clock.now += seconds
+
+    monkeypatch.setattr(transmission_client, "time", SimpleNamespace(monotonic=lambda: clock.now, sleep=advance))
+    return clock
+
+
+def test_transmission_new_paused_task_starts_without_stop_rpc(downloader) -> None:
+    """
+    已暂停的新任务无需再次停止，防止停止请求覆盖选集后的启动请求。
+    """
+    module, server, content, torrent_hash = downloader
+    if not isinstance(module, TransmissionModule):
+        pytest.skip("Transmission 的异步启停状态确认")
+    client = _use_transmission_rpc(module, server)
+    state = {"status": "stopped"}
+    client.trc.get_torrent.side_effect = lambda *_args, **_kwargs: SimpleNamespace(status=state["status"])
+    client.trc.start_torrent.side_effect = lambda **_kwargs: state.update(status="download pending")
+
+    result = module.download(content, Path("/downloads"), "", episodes={170}, custom_words=_CUSTOM_WORDS)
+
+    assert result[1] == torrent_hash
+    assert client.add_torrent.call_args.kwargs["is_paused"] is True
+    client.trc.stop_torrent.assert_not_called()
+    client.trc.start_torrent.assert_called_once_with(ids=torrent_hash)
+
+
+def test_transmission_stopped_after_accepted_start_never_settles_success(downloader, transmission_clock) -> None:
+    """
+    启动 RPC 成功而任务持续停止时，下载链必须失败结算，不能记录订阅下载成功。
+    """
+    module, server, content, _torrent_hash = downloader
+    if not isinstance(module, TransmissionModule):
+        pytest.skip("Transmission 的异步启停状态确认")
+    client = _use_transmission_rpc(module, server)
+    client.trc.get_torrent.return_value = SimpleNamespace(status="stopped")
+    chain = DownloadChain.__new__(DownloadChain)
+    chain.run_module = lambda method, **kwargs: getattr(module, method)(**kwargs)
+    chain._settle_accepted_download = Mock()
+    chain._record_rejected_download = Mock()
+
+    result, message = _submit(chain, content, {170}, _CUSTOM_WORDS)
+
+    assert result is None
+    assert "启动任务未确认" in message and "未记录成功" in message
+    assert transmission_clock.now == transmission_client._TORRENT_STATE_TIMEOUT
+    chain._settle_accepted_download.assert_not_called()
+    chain._record_rejected_download.assert_called_once()
+
+
+@pytest.mark.parametrize("pause_completes", [True, False])
+def test_transmission_async_pause_guards_file_selection(downloader, transmission_clock, pause_completes) -> None:
+    """
+    重复任务异步暂停前不得选集；暂停超时也不能启动或记录订阅成功。
+    """
+    module, server, content, torrent_hash = downloader
+    if not isinstance(module, TransmissionModule):
+        pytest.skip("Transmission 的异步启停状态确认")
+    client = _use_transmission_rpc(module, server)
+    state = {"status": "downloading", "stop_queries": 0}
+    list_files = client.get_files.side_effect
+
+    def read_status(*_args, **_kwargs):
+        """
+        模拟暂停 RPC 返回后，任务仍需两次查询才能实际停止。
+        """
+        if client.trc.stop_torrent.called:
+            state["stop_queries"] += 1
+            if pause_completes and state["stop_queries"] >= 2:
+                state["status"] = "stopped"
+        if client.trc.start_torrent.called:
+            state["status"] = "download pending"
+        return SimpleNamespace(status=state["status"])
+
+    def read_paused_files(*args, **kwargs):
+        """
+        文件选择只能发生在实际暂停和下一次启动之间。
+        """
+        assert state["status"] == "stopped"
+        client.trc.start_torrent.assert_not_called()
+        return list_files(*args, **kwargs)
+
+    client.trc.get_torrent.side_effect = read_status
+    client.get_files.side_effect = read_paused_files
+    chain = DownloadChain.__new__(DownloadChain)
+    chain.run_module = lambda method, **kwargs: getattr(module, method)(**kwargs)
+    chain._settle_accepted_download = Mock()
+    chain._record_rejected_download = Mock()
+
+    result, message = _submit(chain, content, {170}, _CUSTOM_WORDS)
+
+    client.trc.stop_torrent.assert_called_once_with(ids=torrent_hash)
+    assert transmission_clock.now > 0
+    if pause_completes:
+        assert result == torrent_hash
+        chain._settle_accepted_download.assert_called_once()
+        client.trc.start_torrent.assert_called_once_with(ids=torrent_hash)
+    else:
+        assert result is None and "无法确认暂停" in message
+        client.get_files.assert_not_called()
+        client.trc.change_torrent.assert_not_called()
+        client.trc.start_torrent.assert_not_called()
+        chain._settle_accepted_download.assert_not_called()
+
+
 def test_transmission_empty_episode_selection_never_starts_task() -> None:
     """
     选集入口没有目标集数时应保留暂停状态，不把空数组转换为全部文件。
     """
     server = Mock()
     server.stop_torrents.return_value = True
+    server.ensure_torrent_state.return_value = True
     server.get_files.return_value = [SimpleNamespace(id=0, name="Show.S01E01.mkv")]
 
     selected, message = TransmissionModule._select_download_files(server, "hash", set(), None)

tests/test_transmission_client_state.py

@@ -2,14 +2,34 @@
 Transmission 未配置状态和 SDK 参数转发的兼容边界。
 """
 
+from types import SimpleNamespace
 from unittest.mock import Mock
 
 import pytest
+from transmission_rpc import Torrent
 
 import app.modules.transmission.transmission as client_module
 from app.modules.transmission.transmission import Transmission
 
 
+@pytest.fixture
+def state_clock(monkeypatch):
+    """
+    使用虚拟时钟验证启停等待上限，不让回归测试实际等待五秒。
+    """
+    clock = SimpleNamespace(now=0.0)
+
+    def advance(seconds: float) -> None:
+        """
+        记录每次轮询等待的时间。
+        """
+        assert seconds > 0
+        clock.now += seconds
+
+    monkeypatch.setattr(client_module, "time", SimpleNamespace(monotonic=lambda: clock.now, sleep=advance))
+    return clock
+
+
 @pytest.mark.parametrize("host", [None, "", "invalid-address"])
 @pytest.mark.parametrize("rename_partial_files", [None, False, True])
 def test_missing_address_keeps_state_and_reconnect_safe(monkeypatch, host, rename_partial_files):
@@ -113,3 +133,126 @@ def test_file_selection_without_connection_keeps_failure(method, file_ids) -> No
     client.trc = None
 
     assert getattr(client, method)("hash", file_ids) is False
+
+
+@pytest.mark.parametrize("paused", [False, True])
+def test_state_confirmation_without_connection_fails(paused) -> None:
+    """
+    未连接时不能把启停状态解释为成功。
+    """
+    client = Transmission.__new__(Transmission)
+    client.trc = None
+
+    assert client.ensure_torrent_state("hash", paused=paused) is False
+
+
+@pytest.mark.parametrize("status", [1, 2, 3, 4, 5, 6])
+def test_started_state_accepts_sdk_active_and_queued_statuses(state_clock, status) -> None:
+    """
+    启动请求后校验、下载及做种的活动或排队状态均可确认，不要求实际流量。
+    """
+    client = Transmission.__new__(Transmission)
+    client.trc = Mock()
+    client.trc.get_torrent.return_value = Torrent(fields={"id": 1, "status": status})
+
+    assert client.ensure_torrent_state("hash", paused=False) is True
+
+    client.trc.start_torrent.assert_called_once_with(ids="hash")
+    client.trc.stop_torrent.assert_not_called()
+    assert state_clock.now == 0
+    assert client.trc.get_torrent.call_args.kwargs["arguments"] == ["status"]
+
+
+def test_pause_waits_for_async_stop_to_complete(state_clock) -> None:
+    """
+    停止请求已接受但仍在运行时继续等待，确认暂停前不能返回成功。
+    """
+    client = Transmission.__new__(Transmission)
+    client.trc = Mock()
+    client.trc.get_torrent.side_effect = [
+        SimpleNamespace(status=status) for status in ["downloading", "downloading", "stopped"]
+    ]
+
+    assert client.ensure_torrent_state("hash", paused=True) is True
+
+    client.trc.stop_torrent.assert_called_once_with(ids="hash")
+    client.trc.start_torrent.assert_not_called()
+    assert state_clock.now > 0
+
+
+def test_start_retries_stopped_task_until_queue_accepts_it(state_clock) -> None:
+    """
+    启动后仍停止时重发启动，进入下载队列后立即确认，保留正常排队语义。
+    """
+    client = Transmission.__new__(Transmission)
+    client.trc = Mock()
+    client.trc.get_torrent.side_effect = [
+        SimpleNamespace(status=status) for status in ["stopped", "download pending"]
+    ]
+
+    assert client.ensure_torrent_state("hash", paused=False) is True
+
+    assert client.trc.start_torrent.call_count == 2
+    assert state_clock.now > 0
+
+
+def test_start_retry_rpc_failure_never_confirms_success(state_clock) -> None:
+    """
+    首次启动已接受但重试断连时，仍须返回失败而非沿用首次 RPC 的成功。
+    """
+    client = Transmission.__new__(Transmission)
+    client.trc = Mock()
+    client.trc.get_torrent.return_value = SimpleNamespace(status="stopped")
+    client.trc.start_torrent.side_effect = [None, ConnectionError("重试启动失败")]
+
+    assert client.ensure_torrent_state("hash", paused=False) is False
+    assert state_clock.now == 0
+
+
+@pytest.mark.parametrize("paused,status", [(True, "downloading"), (False, "stopped")])
+def test_state_confirmation_timeout_is_bounded(state_clock, paused, status) -> None:
+    """
+    RPC 接受命令但状态始终未改变时，在确认窗口内结束并报告失败。
+    """
+    client = Transmission.__new__(Transmission)
+    client.trc = Mock()
+    client.trc.get_torrent.return_value = SimpleNamespace(status=status)
+
+    assert client.ensure_torrent_state("hash", paused=paused) is False
+
+    assert state_clock.now == client_module._TORRENT_STATE_TIMEOUT
+    assert all(0 < call.kwargs["timeout"] <= client_module._TORRENT_STATE_TIMEOUT
+               for call in client.trc.get_torrent.call_args_list)
+    if paused:
+        client.trc.stop_torrent.assert_called_once_with(ids="hash")
+    else:
+        assert client.trc.start_torrent.call_count > 1
+
+
+@pytest.mark.parametrize("paused", [False, True])
+@pytest.mark.parametrize("error", [KeyError("任务不存在"), ConnectionError("RPC 查询失败")])
+def test_state_query_failure_never_confirms_success(state_clock, paused, error) -> None:
+    """
+    查询失败或任务已消失时返回失败，不能把 RPC 接受命令当成状态确认。
+    """
+    client = Transmission.__new__(Transmission)
+    client.trc = Mock()
+    client.trc.get_torrent.side_effect = error
+
+    assert client.ensure_torrent_state("hash", paused=paused) is False
+    assert state_clock.now == 0
+
+
+@pytest.mark.parametrize("paused", [False, True])
+def test_state_change_rpc_failure_never_confirms_success(state_clock, paused) -> None:
+    """
+    启停命令失败时立即返回，暂停失败不应继续查询或重发启动。
+    """
+    client = Transmission.__new__(Transmission)
+    client.trc = Mock()
+    client.trc.get_torrent.return_value = SimpleNamespace(status="downloading")
+    command = client.trc.stop_torrent if paused else client.trc.start_torrent
+    command.side_effect = ConnectionError("RPC 启停失败")
+
+    assert client.ensure_torrent_state("hash", paused=paused) is False
+    assert state_clock.now == 0