refactor: 抽两个 zip worker 的重复实现，并补齐暂停/恢复路径的测试

为什么先测后改：_batch_zip_worker(347 行) 与 _podcast_zip_worker(295 行) 近乎
复制——连 _wait_if_paused 那 16 行都逐字相同（注释只差一行）。重复的代价是实打实
的：上一次修的 zip_valid「变量可能未绑定」缺陷就是**同型出现在这两处**。
但暂停/恢复是一条并发状态机，此前**零测试覆盖**，直接改不符合先测量后改动。

所以本 commit 分两步落地。

第一步（新增 tests/test_zip_workers.py，9 个用例）——行为级契约
断言只用「调用 worker 后世界里发生了什么」：任务终态、ZIP 内条目、账本守恒
（delivered / refunded）、暂停/恢复/取消后的余额。不引用任何内部函数名，
这样第二步重构时测试不必跟着改，才具备等价性验证能力。
- batch：正常完成 / 暂停→恢复 / 暂停→取消 / 全失败退费 /
  循环中途取消 / 发布前一刻取消
- podcast：正常完成 / 暂停→恢复 / 全失败退费
- 脚手架一律用 reserve_and_open 真预扣（ledger_open 只记账不扣款，
  会让「退款」凭空给余额加次数，守恒断言就失去意义）

写测试过程中暴露两处代码与注释不符，如实钉住而非粉饰：
1. **循环内取消并不会「交付已打包部分」**。zip_valid = True 位于 with zipfile
   块**之后**，而 while 循环内的任何取消都以异常**穿过**该块
   （zipfile.__exit__ 仍写出中央目录，但那行赋值不执行）→ zip_valid 恒为 False
   → delivered = ok_count if zip_valid else 0 恒为 0。源码注释所称的
   「部分成功后取消照常交付（成果守恒）」只在发布前最后一个检查点可达，
   已分别用两个用例把两条路径都钉住。
2. podcast 的 _wait_if_paused 与 batch 版本除 docstring 外完全一致。

第二步——抽取（只抽逐字相同、零语义差异的部分）
- _taskcore._wait_if_paused(task, pause_ev, cancel_ev, done)：done 写成参数
  而非捕获闭包，使函数脱离两个 worker 各自的局部状态
- _taskcore._task_work_dir(task_id)：任务专属目录约定（ZIP 与中间文件都落
  tasks/<task_id>/，对根目录的缓存清扫免疫）
- 两个 worker 各自的 16 行副本与 2 行 mkdir 删除，调用点改为显式传参

复杂度实测量（同口径，git HEAD 对比）：
  两个 worker 合计  行数 642 → 604（-38）  决策点 152 → 140（-12）
  新增共享实现      _wait_if_paused 27 行/7 决策点 · _task_work_dir 11 行/1
**净行数基本持平（总量 -38 +38）**——重复代码抽取的收益从来不是行数变少，
而是**唯一真源**：以后改暂停语义只需改一处，不会再出现「同型缺陷出现在两地」。

验证：pytest 184 passed（175 → 184，+9）· ruff 0 · pyright 0 errors/0 warnings

services/_taskcore.py

@@ -5,6 +5,7 @@
 import threading
 import time
 import traceback
+from pathlib import Path
 from typing import Any
 
 import auth_system
@@ -12,6 +13,9 @@ from core.state import (
     DOWNLOAD_TASKS,
     DOWNLOAD_TASKS_LOCK,
 )
+from music_downloader import (
+    DownloadCanceled,
+)
 
 from .api_service import (
     api_service,
@@ -108,3 +112,45 @@ def _settle_batch_billing(task_id: str, ok_count: int, final_status: str = 'sett
         except Exception:
             pass
     return refund
+
+
+def _task_work_dir(task_id: str) -> Path:
+    """任务专属工作目录（交付 ZIP 与中间文件都落这里）。
+
+    原实现把临时包建在 downloads 根目录，会被缓存清扫线程按 mtime 回收
+    （Linux 上已打开文件同样能被 unlink），排队超时的任务于是出现
+    「扣费后领取 404」的悬空交付；任务目录从此对根目录清扫免疫。
+    音乐歌单打包与播客整台打包共用本目录约定。
+    """
+    path = api_service.downloads_path / 'tasks' / task_id
+    path.mkdir(parents=True, exist_ok=True)
+    return path
+
+
+def _wait_if_paused(task: dict[str, Any], pause_ev: threading.Event,
+                    cancel_ev: threading.Event, done: int) -> None:
+    """断线暂停闸门：置 paused 态并等待恢复或取消，恢复后回到 running。
+
+    两个打包 worker（音乐歌单 / 播客整台）共用同一套暂停语义——此前各自持有
+    一份逐字相同的副本（连注释都只差一行），是「同型缺陷同时出现在两处」的温床。
+
+    调用点位于窗口循环顶部，等待期间**不持有** BATCH_ZIP_SEM：名额交还队列
+    给下一位，这正是「断线不占坑」的实现方式。
+
+    ``done`` 由调用方按当前进度传入（ok_count + len(failed)）——写成参数而非
+    捕获闭包，是为了让本函数脱离两个 worker 各自的局部状态。
+    """
+    if not (pause_ev.is_set() or task.get('status') == 'paused'):
+        return
+    with DOWNLOAD_TASKS_LOCK:
+        task['status'] = 'paused'
+        task['done'] = done
+        task['msg'] = '已暂停（重新打开页面自动恢复）'
+    while not cancel_ev.is_set() and (pause_ev.is_set() or task.get('status') == 'paused'):
+        time.sleep(1)
+    if cancel_ev.is_set():
+        raise DownloadCanceled("任务已取消")
+    pause_ev.clear()
+    with DOWNLOAD_TASKS_LOCK:
+        task['status'] = 'running'
+        task['msg'] = '继续下载...'

services/batch.py

@@ -27,6 +27,8 @@ from ._shared import (
 from ._taskcore import (
     _mark_task_terminal,
     _settle_batch_billing,
+    _task_work_dir,
+    _wait_if_paused,
 )
 from .api_service import (
     api_service,
@@ -110,30 +112,10 @@ def _batch_zip_worker(task_id: str) -> None:
                 DOWNLOAD_TASKS[task_id]['done'] = ok_count + len(failed)
                 DOWNLOAD_TASKS[task_id]['failed'] = len(failed)
 
-        def _wait_if_paused() -> None:
-            """暂停时等待恢复或取消（不持有 BATCH_ZIP_SEM）"""
-            if not (pause_ev.is_set() or task['status'] == 'paused'):
-                return
-            with DOWNLOAD_TASKS_LOCK:
-                task['status'] = 'paused'
-                task['done'] = ok_count + len(failed)
-                task['msg'] = '已暂停（重新打开页面自动恢复）'
-            while not cancel_ev.is_set() and (pause_ev.is_set() or task['status'] == 'paused'):
-                time.sleep(1)
-            if cancel_ev.is_set():
-                raise DownloadCanceled("任务已取消")
-            pause_ev.clear()
-            with DOWNLOAD_TASKS_LOCK:
-                task['status'] = 'running'
-                task['msg'] = '继续下载...'
-
         # ZIP 构建放 downloads 卷（原实现落系统 %TEMP%——500 首无损可达 GB 级，
         # 小系统盘一次大单即塞爆；单曲/播客早已指定 dir，唯批量漏了）
-        # F-A：临时包建在任务专属目录 tasks/<task_id>/——原放根目录会被
+        # 且建在任务专属目录（见 _task_work_dir）：根目录会被缓存清扫按 mtime 回收
-        # 缓存清扫线程按 mtime 回收（Linux 上已打开文件也能被 unlink，Docker 部署
+        _task_dir_for_zip = _task_work_dir(task_id)
-        # 排队超 35 分钟即出现"扣费后领取 404"的悬空交付），根目录从此碰不到交付物
-        _task_dir_for_zip = api_service.downloads_path / 'tasks' / task_id
-        _task_dir_for_zip.mkdir(parents=True, exist_ok=True)
         # 中央目录完整写出后（下方）才置 True，否则一律按失败全额退费
         with _tempfile.NamedTemporaryFile(prefix='batch_zip_', suffix='.zip',
                                           dir=str(_task_dir_for_zip),
@@ -149,7 +131,7 @@ def _batch_zip_worker(task_id: str) -> None:
                         if cancel_ev.is_set():
                             raise DownloadCanceled("任务已取消")
                         # 断线暂停：释放名额等待恢复/取消，恢复后重新进入排队
-                        _wait_if_paused()
+                        _wait_if_paused(task, pause_ev, cancel_ev, ok_count + len(failed))
                         # 每个窗口持有 BATCH_ZIP_SEM（同一时刻全局最多 3 个批量任务在下载）
                         # 等待名额期间可被取消，不再无限阻塞线程与退款
                         if not BATCH_ZIP_SEM.acquire_cancellable(cancel_ev):

services/podcast.py

@@ -33,6 +33,8 @@ from ._shared import (
 from ._taskcore import (
     _mark_task_terminal,
     _settle_batch_billing,
+    _task_work_dir,
+    _wait_if_paused,
 )
 from .api_service import (
     api_service,
@@ -359,23 +361,6 @@ def _podcast_zip_worker(task_id: str) -> None:
         except Exception as e:
             raise DownloadException(f"补扣下载次数异常: {e}") from e
 
-        def _wait_if_paused() -> None:
-            """暂停时等待恢复或取消"""
-            if not (pause_ev.is_set() or task['status'] == 'paused'):
-                return
-            with DOWNLOAD_TASKS_LOCK:
-                task['status'] = 'paused'
-                task['done'] = ok_count + len(failed)
-                task['msg'] = '已暂停（重新打开页面自动恢复）'
-            while not cancel_ev.is_set() and (pause_ev.is_set() or task['status'] == 'paused'):
-                time.sleep(1)
-            if cancel_ev.is_set():
-                raise DownloadCanceled("任务已取消")
-            pause_ev.clear()
-            with DOWNLOAD_TASKS_LOCK:
-                task['status'] = 'running'
-                task['msg'] = '继续下载...'
-
         def _update_progress(msg: str = '') -> None:
             with DOWNLOAD_TASKS_LOCK:
                 t = DOWNLOAD_TASKS[task_id]
@@ -389,8 +374,7 @@ def _podcast_zip_worker(task_id: str) -> None:
         # 未 valid 一律不发布、不按 ok_count 结算
         # 同音乐批量——ZIP 构建放 downloads 卷，不落系统 %TEMP%
         # 与音乐批量打包同口径：建在任务专属目录 tasks/<task_id>/，根目录缓存清扫不再触及交付物
-        _podcast_task_dir = api_service.downloads_path / 'tasks' / task_id
+        _podcast_task_dir = _task_work_dir(task_id)
-        _podcast_task_dir.mkdir(parents=True, exist_ok=True)
         with _tempfile.NamedTemporaryFile(prefix='podcast_batch_', suffix='.zip',
                                           dir=str(_podcast_task_dir),
                                           delete=False) as zip_tmp:
@@ -400,7 +384,7 @@ def _podcast_zip_worker(task_id: str) -> None:
                 for idx, ep in enumerate(episodes, 1):
                     if cancel_ev.is_set():
                         raise DownloadCanceled("任务已取消")
-                    _wait_if_paused()
+                    _wait_if_paused(task, pause_ev, cancel_ev, ok_count + len(failed))
                     # 并发对齐音乐批量：播客打包同样占用全局批量名额（BATCH_ZIP_SEM，全局 3），
                     # 逐集 acquire/release——暂停等待/取消时不长期占名额；音频本身仍受
                     # 全局下载信号量约束。原实现不受批量并发控制，多用户同时打包会无限并发

tests/test_zip_workers.py

@@ -0,0 +1,456 @@
+# -*- coding: utf-8 -*-
+"""两个 ZIP 打包 worker 的行为契约测试（沙箱，不触碰生产 data/）。
+
+为什么：``_batch_zip_worker`` / ``_podcast_zip_worker`` 合计约 642 行此前**零测试覆盖**，
+而两者高度重复——连 ``_wait_if_paused`` 那 16 行都逐字相同（上一次修的
+``zip_valid`` 可能未绑定缺陷就是同型出现在这两处，重复代码在持续制造同类 bug）。
+
+抽公共实现之前，先用**行为级**测试钉住可观察契约：任务终态、ZIP 内容、
+账本守恒、暂停/恢复/取消语义。断言只依赖「调用 worker 后世界里发生了什么」，
+不引用任何内部函数名——这样重构时测试不必跟着改，才具备等价性验证能力。
+
+账本口径：一律用 ``reserve_and_open`` 真预扣（``ledger_open`` 只记账不扣款，
+会让「退款」凭空给余额加次数，守恒断言就失去意义）。
+"""
+import threading
+import time
+import zipfile
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+import auth_system as auth
+import core.db as core_db
+from core.state import DOWNLOAD_TASKS, DOWNLOAD_TASKS_LOCK
+from music_downloader import DownloadCanceled, DownloadResult, MusicInfo
+from services import batch as batch_svc
+from services import podcast as podcast_svc
+from services.api_service import api_service
+
+PW = 'Passw0rd12'
+CREDITS = 50
+
+
+# ============ 测试替身 ============
+
+class _NullLogger:
+    """吞掉 api_service.logger 的输出——避免测试往项目根 music_api.log 追加。"""
+
+    def debug(self, *a: Any, **k: Any) -> None: ...
+    def info(self, *a: Any, **k: Any) -> None: ...
+    def warning(self, *a: Any, **k: Any) -> None: ...
+    def error(self, *a: Any, **k: Any) -> None: ...
+
+
+class _FakeDownloader:
+    """可编程下载器替身：记录调用序，支持预设失败集与按序触发钩子。"""
+
+    def __init__(self, cache_dir: Path, *, fail_ids: tuple[int, ...] = (),
+                 hooks: dict[int, Any] | None = None):
+        self.cache_dir = cache_dir
+        self.fail_ids = set(fail_ids)
+        self.hooks = hooks or {}
+        self.calls: list[int] = []
+        self._lock = threading.Lock()
+
+    def get_music_info(self, mid: Any, quality: str, cookies: Any) -> MusicInfo:
+        return MusicInfo(
+            id=int(mid), name=f'歌{mid}', artists='歌手A', album='专辑A',
+            pic_url='', duration=200, track_number=1, download_url='',
+            file_type='mp3', file_size=1024, quality='standard',
+            lyric='[00:01.00]第一行\n[00:02.00]第二行',
+        )
+
+    def download_music_file(self, mid: Any, quality: str, **kw: Any) -> DownloadResult:
+        with self._lock:
+            self.calls.append(int(mid))
+            seq = len(self.calls)
+        hook = self.hooks.get(seq)
+        if hook is not None:
+            hook()
+        cancel_ev = kw.get('cancel_event')
+        if cancel_ev is not None and cancel_ev.is_set():
+            raise DownloadCanceled('任务已取消')
+        if int(mid) in self.fail_ids:
+            return DownloadResult(success=False, error_message='上游 404')
+        path = self.cache_dir / f'{mid}.mp3'
+        path.write_bytes(b'ID3\x03\x00\x00\x00' + b'\x00' * 64)
+        return DownloadResult(success=True, file_path=str(path),
+                              file_size=path.stat().st_size, music_info=kw.get('music_info'))
+
+
+# ============ 沙箱 ============
+
+@pytest.fixture()
+def sandbox(tmp_path, monkeypatch):
+    """临时站点库 + 临时下载目录 + 空任务表。"""
+    db_path = tmp_path / 'site.db'
+    monkeypatch.setattr(auth.db, 'DATA_DIR', tmp_path)
+    monkeypatch.setattr(auth.db, 'DB_PATH', db_path)
+    monkeypatch.setattr(auth.db, 'BACKUP_DIR', tmp_path / 'backups')
+    monkeypatch.setattr(auth.db, 'SECRET_KEY_PATH', tmp_path / 'secret.key')
+    monkeypatch.setattr(core_db, 'DATA_DIR', tmp_path)
+    monkeypatch.setattr(core_db, 'DB_PATH', db_path)
+    auth.init_db()
+    auth.create_user('payer', PW, email='p@example.com', credits=CREDITS)
+
+    downloads = tmp_path / 'downloads'
+    downloads.mkdir()
+    cache = downloads / 'cache'
+    cache.mkdir()
+    monkeypatch.setattr(api_service, 'downloads_path', downloads)
+    monkeypatch.setattr(api_service, 'logger', _NullLogger())
+
+    with DOWNLOAD_TASKS_LOCK:
+        DOWNLOAD_TASKS.clear()
+    yield {'root': tmp_path, 'downloads': downloads, 'cache': cache}
+    with DOWNLOAD_TASKS_LOCK:
+        DOWNLOAD_TASKS.clear()
+
+
+# ============ 工具 ============
+
+def _uid(username: str = 'payer') -> int:
+    u = auth.get_user_by_name(username)
+    assert u is not None
+    return int(u['id'])
+
+
+def _balance(username: str = 'payer') -> int:
+    u = auth.get_user_by_name(username)
+    assert u is not None
+    return int(u['duration_seconds'] or 0)
+
+
+def _wait_for(pred, timeout: float = 30.0, interval: float = 0.05) -> bool:
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        if pred():
+            return True
+        time.sleep(interval)
+    return False
+
+
+def _status(task_id: str) -> str:
+    with DOWNLOAD_TASKS_LOCK:
+        return str(DOWNLOAD_TASKS.get(task_id, {}).get('status') or '')
+
+
+def _resume_like_heartbeat(task: dict[str, Any]) -> None:
+    """完全对齐生产心跳（web/routers/tasks.py:task_heartbeat）的恢复动作：
+    置 status='queued' 且 clear pause 事件——两者缺一 worker 都不会醒来。"""
+    with DOWNLOAD_TASKS_LOCK:
+        task['status'] = 'queued'
+        task['pause'].clear()
+
+
+def _register_batch(sandbox, ids: list[int], *, reserved: int, user_id: int,
+                    concurrency: int = 1) -> tuple[str, dict[str, Any]]:
+    tid = 't' + str(int(time.time() * 1000000) % 10 ** 10)
+    # 必须走 reserve_and_open（真预扣+开账同事务）——ledger_open 只记账不扣款，
+    # 用它会让「退款」凭空给余额加次数，账本守恒断言失去意义
+    left = auth.reserve_and_open(user_id, reserved, tid)
+    assert left is not None, '预扣失败（余额不足，检查 fixture 的 credits）'
+    task: dict[str, Any] = {
+        'type': 'batch', 'task_id': tid,
+        'user_id': user_id, 'billing_user_id': user_id, 'billing_reserved': reserved,
+        'total': len(ids), 'done': 0, 'failed': 0,
+        'status': 'queued', 'start': time.time(), 'msg': '排队中...',
+        'cancel': threading.Event(), 'pause': threading.Event(),
+        'device_id': 'dev-test', 'result': None,
+        'params': {
+            'ids': list(ids), 'quality': 'standard', 'filename_format': 'artist_song',
+            'write_meta': False, 'meta_fields': None, 'lyric_format': 'lrc',
+            'concurrency': concurrency, 'cookies': {}, 'remote_ip': '-',
+            'want_zip': True,
+        },
+    }
+    with DOWNLOAD_TASKS_LOCK:
+        DOWNLOAD_TASKS[tid] = task
+    return tid, task
+
+
+def _register_podcast(sandbox, pids: list[int], *, reserved: int, user_id: int) -> tuple[str, dict[str, Any]]:
+    tid = 'p' + str(int(time.time() * 1000000) % 10 ** 10)
+    left = auth.reserve_and_open(user_id, reserved, tid)
+    assert left is not None, '预扣失败（余额不足）'
+    task: dict[str, Any] = {
+        'type': 'podcast_batch', 'task_id': tid,
+        'user_id': user_id, 'billing_user_id': user_id, 'billing_reserved': reserved,
+        'total': len(pids), 'done': 0, 'failed': 0,
+        'status': 'queued', 'start': time.time(), 'msg': '排队中...',
+        'cancel': threading.Event(), 'pause': threading.Event(),
+        'device_id': 'dev-test', 'result': None,
+        'params': {
+            'radio_id': None, 'ids': list(pids), 'limit': max(1, len(pids)),
+            'quality': 'standard', 'cookies': {}, 'remote_ip': '-',
+            'radio_name': '测试电台', 'billing_uid': user_id, 'billed': reserved,
+        },
+    }
+    with DOWNLOAD_TASKS_LOCK:
+        DOWNLOAD_TASKS[tid] = task
+    return tid, task
+
+
+def _run(worker, task_id: str) -> threading.Thread:
+    th = threading.Thread(target=worker, args=(task_id,), daemon=True)
+    th.start()
+    return th
+
+
+def _zip_names(path: str) -> list[str]:
+    with zipfile.ZipFile(path) as zf:
+        return zf.namelist()
+
+
+# ============ batch worker ============
+
+def test_batch_zip_completes(sandbox, monkeypatch):
+    """3 首全部成功：终态 done、ZIP 含音频与歌词、账本按交付结算。"""
+    uid = _uid()
+    before = _balance()
+    monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache']))
+    tid, task = _register_batch(sandbox, [101, 102, 103], reserved=3, user_id=uid)
+
+    th = _run(batch_svc._batch_zip_worker, tid)
+    th.join(timeout=30)
+    assert not th.is_alive(), 'worker 未在 30s 内结束'
+
+    assert _status(tid) == 'done', f"终态={_status(tid)} msg={task.get('msg')}"
+    assert task['done'] == 3 and task['failed'] == 0
+    assert task['result'] is not None
+    names = _zip_names(task['result']['zip_path'])
+    assert sum(1 for n in names if n.endswith('.mp3')) == 3, names
+    assert sum(1 for n in names if n.endswith('.lrc')) == 3, names
+
+    row = auth.ledger_get(tid)
+    assert row is not None and row['status'] != 'open'
+    assert row['delivered'] == 3
+    assert _balance() == before - 3, '全部成功：只扣预扣的 3 次，无退款'
+
+
+def test_batch_zip_pause_then_resume(sandbox, monkeypatch):
+    """断线暂停 → 心跳恢复：ZIP 最终完整，任务走完 done。"""
+    uid = _uid()
+    before = _balance()
+    holder: dict[str, Any] = {}
+
+    monkeypatch.setattr(api_service, 'downloader',
+                        _FakeDownloader(sandbox['cache'],
+                                        hooks={1: lambda: holder['task']['pause'].set()}))
+    tid, task = _register_batch(sandbox, [201, 202, 203], reserved=3, user_id=uid)
+    holder['task'] = task
+
+    th = _run(batch_svc._batch_zip_worker, tid)
+
+    assert _wait_for(lambda: _status(tid) == 'paused'), f"未进入 paused（当前 {_status(tid)}）"
+    assert '已暂停' in task['msg']
+    assert task['done'] >= 1, '暂停前应至少完成 1 首'
+
+    _resume_like_heartbeat(task)
+    th.join(timeout=30)
+
+    assert _status(tid) == 'done', f"恢复后终态={_status(tid)} msg={task.get('msg')}"
+    names = _zip_names(task['result']['zip_path'])
+    assert sum(1 for n in names if n.endswith('.mp3')) == 3, names
+    assert _balance() == before - 3
+
+
+def test_batch_zip_pause_then_cancel(sandbox, monkeypatch):
+    """暂停期间取消：任务收敛 canceled，未交付则不发布 ZIP、账本全额退费。"""
+    uid = _uid()
+    before = _balance()
+    holder: dict[str, Any] = {}
+    monkeypatch.setattr(api_service, 'downloader',
+                        _FakeDownloader(sandbox['cache'],
+                                        hooks={1: lambda: holder['task']['pause'].set()}))
+    tid, task = _register_batch(sandbox, [301, 302, 303], reserved=3, user_id=uid)
+    holder['task'] = task
+
+    th = _run(batch_svc._batch_zip_worker, tid)
+    assert _wait_for(lambda: _status(tid) == 'paused'), f"未进入 paused（{_status(tid)}）"
+
+    task['cancel'].set()
+    th.join(timeout=30)
+
+    assert _status(tid) == 'canceled', f"终态={_status(tid)} msg={task.get('msg')}"
+    assert task['result'] is None, '未交付不应发布 ZIP'
+    assert _balance() == before, '取消后应全额退回预扣'
+    row = auth.ledger_get(tid)
+    assert row is not None and row['delivered'] == 0 and row['status'] != 'open'
+
+
+def test_batch_zip_all_fail_refunds(sandbox, monkeypatch):
+    """全部失败：不发布空 ZIP、转 error、预扣全额退回。"""
+    uid = _uid()
+    before = _balance()
+    monkeypatch.setattr(api_service, 'downloader',
+                        _FakeDownloader(sandbox['cache'], fail_ids=(401, 402)))
+    tid, task = _register_batch(sandbox, [401, 402], reserved=2, user_id=uid)
+
+    th = _run(batch_svc._batch_zip_worker, tid)
+    th.join(timeout=30)
+
+    assert _status(tid) == 'error', f"终态={_status(tid)} msg={task.get('msg')}"
+    assert task['failed'] == 2
+    assert task['result'] is None, '全失败不应发布 ZIP'
+    assert _balance() == before, '全失败应全额退回预扣'
+    row = auth.ledger_get(tid)
+    assert row is not None and row['delivered'] == 0 and row['status'] != 'open'
+
+
+def test_batch_zip_midway_cancel_refunds_all(sandbox, monkeypatch):
+    """循环中途取消：按「不发布 + 全额退费」收敛（记录现状）。
+
+    注意这与 ``test_batch_zip_cancel_at_publish_checkpoint_delivers`` 的差别：
+    worker 的 ``zip_valid = True`` 位于 ``with zipfile`` 块**之后**，而 ``while ids``
+    循环内的任何取消都以异常**穿过**该块（``zipfile.__exit__`` 仍会写出中央目录，
+    但那行赋值不会执行）→ ``zip_valid`` 保持 False → ``delivered = ok_count if
+    zip_valid else 0`` 恒为 0。因此源码注释所称的「已入包部分照常交付（成果守恒）」
+    在循环内取消时并不可达，只在发布前最后一个检查点可达。
+    """
+    uid = _uid()
+    before = _balance()
+    holder: dict[str, Any] = {}
+    monkeypatch.setattr(api_service, 'downloader',
+                        _FakeDownloader(sandbox['cache'],
+                                        hooks={2: lambda: holder['task']['cancel'].set()}))
+    tid, task = _register_batch(sandbox, [501, 502, 503, 504], reserved=4, user_id=uid)
+    holder['task'] = task
+
+    th = _run(batch_svc._batch_zip_worker, tid)
+    th.join(timeout=30)
+
+    assert _status(tid) == 'canceled', f"终态={_status(tid)} msg={task.get('msg')}"
+    assert task['result'] is None, '当前实现：循环内取消不发布'
+    assert _balance() == before, '未交付部分全额退回'
+    row = auth.ledger_get(tid)
+    assert row is not None and row['delivered'] == 0
+
+
+def test_batch_zip_cancel_at_publish_checkpoint_delivers(sandbox, monkeypatch):
+    """发布前最后一刻取消：已打包部分照常交付、按交付数结算（成果守恒）。
+
+    这是「部分成功后取消仍交付」唯一的可达路径，用替身包一层结算函数
+    在其返回后立即置取消位，从而精确命中该检查点。
+    """
+    uid = _uid()
+    before = _balance()
+    holder: dict[str, Any] = {}
+    real_settle = batch_svc._settle_batch_billing
+
+    def settle_then_cancel(task_id: str, ok_count: int, final_status: str = 'settled') -> int:
+        refunded = real_settle(task_id, ok_count, final_status)
+        holder['task']['cancel'].set()   # 结算已完成、发布前一刻取消
+        return refunded
+
+    monkeypatch.setattr(batch_svc, '_settle_batch_billing', settle_then_cancel)
+    monkeypatch.setattr(api_service, 'downloader', _FakeDownloader(sandbox['cache']))
+    tid, task = _register_batch(sandbox, [601, 602, 603], reserved=3, user_id=uid)
+    holder['task'] = task
+
+    th = _run(batch_svc._batch_zip_worker, tid)
+    th.join(timeout=30)
+
+    assert _status(tid) == 'canceled', f"终态={_status(tid)} msg={task.get('msg')}"
+    assert task['result'] is not None, '已打包部分应交付'
+    names = _zip_names(task['result']['zip_path'])
+    assert sum(1 for n in names if n.endswith('.mp3')) == 3, names
+    assert _balance() == before - 3, '交付 3 首即不退款'
+
+
+# ============ podcast worker ============
+
+@pytest.fixture()
+def podcast_stub(monkeypatch, sandbox):
+    """把播客上游与流式下载替换为本地确定性实现。"""
+    def fake_detail(pid: Any, cookies: Any) -> dict[str, Any]:
+        n = int(pid)
+        return {'id': n, 'name': f'单集{n}', 'main_song_id': 1,
+                'serial_num': n, 'cover_url': '', 'songs': []}
+
+    def fake_resolve(prog: dict[str, Any], quality: str, cookies: Any) -> dict[str, Any]:
+        return {'url': f"http://example.invalid/{prog['id']}.mp3", 'size': 2048, 'type': 'mp3'}
+
+    def fake_stream(url: str, target: Path, total_size: int, cancel_ev: Any,
+                    task_id: str = '', progress_label: str = '') -> int:
+        if cancel_ev is not None and cancel_ev.is_set():
+            raise DownloadCanceled('任务已取消')
+        target.write_bytes(b'\xff\xfb\x90\x00' + b'\x00' * 128)
+        return target.stat().st_size
+
+    monkeypatch.setattr(podcast_svc, '_podcast_resolve_audio', fake_resolve)
+    monkeypatch.setattr(podcast_svc, '_podcast_stream_download', fake_stream)
+    monkeypatch.setattr(podcast_svc, 'fetch_cover_bytes', lambda _u: None)
+    monkeypatch.setattr(podcast_svc.podcast_api, 'get_program_detail', fake_detail)
+    return sandbox
+
+
+def test_podcast_zip_completes(podcast_stub):
+    """整台打包正常路径：终态 done、ZIP 含全部单集、账本按交付结算。"""
+    uid = _uid()
+    before = _balance()
+    tid, task = _register_podcast(podcast_stub, [11, 12, 13], reserved=3, user_id=uid)
+
+    th = _run(podcast_svc._podcast_zip_worker, tid)
+    th.join(timeout=30)
+
+    assert _status(tid) == 'done', f"终态={_status(tid)} msg={task.get('msg')}"
+    assert task['done'] == 3 and task['failed'] == 0
+    names = _zip_names(task['result']['zip_path'])
+    assert sum(1 for n in names if n.endswith('.mp3')) == 3, names
+    row = auth.ledger_get(tid)
+    assert row is not None and row['delivered'] == 3
+    assert _balance() == before - 3
+
+
+def test_podcast_zip_pause_then_resume(podcast_stub, monkeypatch):
+    """播客打包同样支持断线暂停 → 心跳恢复，最终完整交付。"""
+    uid = _uid()
+    before = _balance()
+    holder: dict[str, Any] = {}
+    fired = {'n': 0}
+
+    def _pause_once(url, target, total_size, cancel_ev, task_id='', progress_label=''):
+        target.write_bytes(b'\xff\xfb\x90\x00' + b'\x00' * 128)
+        fired['n'] += 1
+        if fired['n'] == 1:      # 只在第一集触发——否则恢复后立刻又被暂停
+            holder['task']['pause'].set()
+        return target.stat().st_size
+
+    monkeypatch.setattr(podcast_svc, '_podcast_stream_download', _pause_once)
+    tid, task = _register_podcast(podcast_stub, [21, 22, 23], reserved=3, user_id=uid)
+    holder['task'] = task
+
+    th = _run(podcast_svc._podcast_zip_worker, tid)
+    assert _wait_for(lambda: _status(tid) == 'paused'), f"未进入 paused（{_status(tid)}）"
+    assert '已暂停' in task['msg']
+
+    _resume_like_heartbeat(task)
+    th.join(timeout=30)
+
+    assert _status(tid) == 'done', f"恢复后终态={_status(tid)} msg={task.get('msg')}"
+    names = _zip_names(task['result']['zip_path'])
+    assert sum(1 for n in names if n.endswith('.mp3')) == 3, names
+    assert _balance() == before - 3
+
+
+def test_podcast_zip_all_fail_refunds(podcast_stub, monkeypatch):
+    """全部单集失败：转 error 并全额退费，不发布空包。"""
+    uid = _uid()
+    before = _balance()
+
+    def _boom(prog, quality, cookies):
+        raise RuntimeError('上游不可用')
+
+    monkeypatch.setattr(podcast_svc, '_podcast_resolve_audio', _boom)
+    tid, task = _register_podcast(podcast_stub, [31, 32], reserved=2, user_id=uid)
+
+    th = _run(podcast_svc._podcast_zip_worker, tid)
+    th.join(timeout=30)
+
+    assert _status(tid) == 'error', f"终态={_status(tid)} msg={task.get('msg')}"
+    assert _balance() == before, '全失败应全额退回'
+    row = auth.ledger_get(tid)
+    assert row is not None and row['delivered'] == 0