feat(mq): 延迟队列拆出 i2v 轮询，并并行化生成 IO

.env.example

@@ -100,16 +100,25 @@ WINDUP_MQ_MAX_PUBLISH_ATTEMPTS=10
 WINDUP_MQ_MAX_CONSUME_ATTEMPTS=5
 WINDUP_MQ_CONSUME_LEASE_SECONDS=1800
 WINDUP_MQ_EMAIL_HANDLER_RETRIES=3
-# worker 内 handler 并行度（单进程内 ThreadPoolExecutor；默认合计 ≤ 连接池容量）
+# worker 内 handler 并行度（单进程内 ThreadPoolExecutor）
+# 生成期间不占 Postgres 连接，可按机器内存与上游配额上调。
 WINDUP_MQ_EMAIL_CONCURRENCY=8
-WINDUP_MQ_GENERATION_IMAGE_CONCURRENCY=6
-WINDUP_MQ_GENERATION_ACTION_CONCURRENCY=4
+WINDUP_MQ_GENERATION_IMAGE_CONCURRENCY=16
+WINDUP_MQ_GENERATION_ACTION_CONCURRENCY=8
+# poll 使用独立线程池,不与图/动作共享槽位
+WINDUP_MQ_GENERATION_POLL_CONCURRENCY=16
+# i2v 轮询走 ZSET 延迟队列,到期促进到 Stream;不用 Redis 过期事件
+WINDUP_MQ_DELAYED_ZSET=windup:zset:delayed
+WINDUP_MQ_DELAYED_CLAIM_LIMIT=50
+WINDUP_MQ_DELAYED_TICK_SECONDS=1
+# 上传 / 参考图下载 / 多图 gen_image 共用的 IO 线程池
+WINDUP_IO_POOL_SIZE=32
 # 生成任务排队 / 孤儿 RUNNING 超时（秒）
 WINDUP_GENERATION_PENDING_MAX_AGE=86400
 WINDUP_GENERATION_RUNNING_STALE_SECONDS=1800
 # SSE 跨进程桥接：worker publish → web subscribe → EventBus
 WINDUP_SSE_REDIS_CHANNEL=windup:pubsub:generation-task-events
-# worker 并发高时建议调大连接池（默认 pool_size=5, max_overflow=10）
+# web / worker 各有一份连接池。生成已短 session，不必再按 handler 并发配满。
 POSTGRES_POOL_SIZE=5
 POSTGRES_MAX_OVERFLOW=10
 

backend/packages/ai_engine/src/windup_ai_engine/impl/character_generator.py

@@ -128,6 +128,63 @@ def generate(
         )
         return self._finish(frames, action, route, progress, canvas)
 
+    def can_defer_i2v(self) -> bool:
+        strategy = self._by_route.get(GenRoute.VIDEO_I2V)
+        video = getattr(strategy, "_video", None)
+        return hasattr(strategy, "start_job") and hasattr(video, "start_i2v")
+
+    def start_video(
+        self,
+        card: CharacterCard,
+        action: ActionSpec,
+        master: bytes,
+        progress: ProgressPort,
+        canvas: tuple[int, int] | None = None,
+    ):
+        """花钱之前的预检 + 建 i2v 单,不跟单。"""
+        facts = check_master(master, canvas)
+        progress.step("precheck", _TICK_PRECHECK, _TOTAL, facts.note())
+        route = ROUTE_MATRIX[action.action]
+        progress.step("route", _TICK_ROUTE, _TOTAL, f"{action.action.value} → {route.value}")
+        strategy = self._pick(route, action)
+        start_job = getattr(strategy, "start_job", None)
+        if start_job is None:
+            raise TypeError(f"{route.value} 不支持异步建单")
+        return start_job(
+            card, action, master, _BandProgress(progress, _DERIVE_FROM, _DERIVE_TO)
+        )
+
+    def poll_video(self, job_id: str, *, route_id: str | None = None) -> bytes | None:
+        """walk/idle/attack/custom 共用 VIDEO_I2V 桶,按 job 探一次。"""
+        strategy = self._by_route.get(GenRoute.VIDEO_I2V)
+        if strategy is None:
+            raise NotImplementedError("未注入视频路线,无法轮询 i2v")
+        poll_job = getattr(strategy, "poll_job", None)
+        if poll_job is None:
+            raise TypeError("视频路线不支持 poll_job")
+        return poll_job(job_id, route_id=route_id)
+
+    def finish_video(
+        self,
+        video: bytes,
+        card: CharacterCard,
+        action: ActionSpec,
+        master: bytes,
+        progress: ProgressPort,
+        canvas: tuple[int, int] | None = None,
+    ) -> GeneratedAction:
+        """成片到位后走抽帧 / 抠图 / 最后一公里。"""
+        route = ROUTE_MATRIX[action.action]
+        progress.step("route", _TICK_ROUTE, _TOTAL, f"{action.action.value} → {route.value}")
+        strategy = self._pick(route, action)
+        frames_from = getattr(strategy, "frames_from_video", None)
+        if frames_from is None:
+            raise TypeError(f"{route.value} 不支持从成片续跑")
+        frames = frames_from(
+            video, card, action, master, _BandProgress(progress, _DERIVE_FROM, _DERIVE_TO)
+        )
+        return self._finish(frames, action, route, progress, canvas)
+
     def generate_rendered(
         self,
         card: CharacterCard,

backend/packages/ai_engine/src/windup_ai_engine/slicing/extract.py

@@ -8,6 +8,7 @@
 
 from __future__ import annotations
 
+import io
 import logging
 import os
 import tempfile
@@ -19,21 +20,18 @@
 
 __all__ = ["extract_frames_bytes", "extract_all_frames_bytes"]
 
+_PYAV = {"plugin": "pyav"}
+_PYAV_BYTES = {"plugin": "pyav", "extension": ".mp4"}
+
 
 def extract_frames_bytes(video: bytes, n: int) -> list[Image.Image]:
-    """从视频 bytes 均匀抽 ``n`` 帧（供后端 strategy 用，provider 返回的是 bytes）。"""
-    with tempfile.TemporaryDirectory() as tmp:
-        path = Path(tmp) / "source.mp4"
-        path.write_bytes(video)
-        return _extract_frames(str(path), n)
+    """从视频 bytes 均匀抽 ``n`` 帧。优先内存解码,不落盘。"""
+    return _extract_frames(video, n)
 
 
 def extract_all_frames_bytes(video: bytes, cap: int = 150) -> list[Image.Image]:
     """抽视频全部帧（至多 ``cap``，均匀降采样），供周期检测用。"""
-    with tempfile.TemporaryDirectory() as tmp:
-        path = Path(tmp) / "source.mp4"
-        path.write_bytes(video)
-        return _extract_frames(str(path), cap)
+    return _extract_frames(video, cap)
 
 
 def _uniform_indices(total: int, n: int) -> list[int]:
@@ -42,7 +40,19 @@ def _uniform_indices(total: int, n: int) -> list[int]:
     return [round(i * (total - 1) / max(1, m - 1)) for i in range(m)]
 
 
-def _frame_count(video_path: str) -> int:
+def _pyav_source(source: str | bytes) -> tuple[str | io.BytesIO, dict]:
+    if isinstance(source, (bytes, bytearray)):
+        return io.BytesIO(source), dict(_PYAV_BYTES)
+    return source, dict(_PYAV)
+
+
+def _rewind(handle: object) -> None:
+    seek = getattr(handle, "seek", None)
+    if seek is not None:
+        seek(0)
+
+
+def _frame_count(source, kw: dict | None = None) -> int:
     """帧数。先问容器元数据,不可信时退回逐帧计数(计数不保留帧,内存不涨)。
 
     元数据在 14 段真实 i2v 视频上与实际帧数全部一致(2026-08-10 实测),但不同容器/编码
@@ -51,32 +61,40 @@ def _frame_count(video_path: str) -> int:
     """
     import imageio.v3 as iio
 
+    opts = dict(kw or _PYAV)
     try:
-        shape = iio.improps(video_path, plugin="pyav").shape
+        shape = iio.improps(source, **opts).shape
         if shape and isinstance(shape[0], int) and shape[0] > 0:
             return shape[0]
     except Exception:
         pass
-    return sum(1 for _ in iio.imiter(video_path, plugin="pyav"))
+    _rewind(source)
+    total = sum(1 for _ in iio.imiter(source, **opts))
+    _rewind(source)
+    return total
 
 
-def _extract_frames(video_path: str, n: int) -> list[Image.Image]:
+def _extract_frames(source: str | bytes, n: int) -> list[Image.Image]:
     """从视频均匀抽 ``n`` 帧。优先 imageio(流式),回退系统 ffmpeg。
 
     **流式而不是一次性读整段**(2026-08-10,机器审 P2):原先走 ``iio.imread`` 会把
     ``(T, H, W, C)`` 整个 materialize 出来。实测 121 帧 720p 的真实 i2v 视频峰值
     319 MiB,而我们只要其中 8~16 帧;并发 worker 叠加时这是实打实的内存墙。
     现在峰值≈保留帧数 × 单帧,与视频长度无关。
+
+    bytes 入口走内存缓冲,避免把 mp4 写到 overlay/云盘再读回来。
     """
+    handle, kw = _pyav_source(source)
     try:
         import imageio.v3 as iio
 
-        total = _frame_count(video_path)
+        total = _frame_count(handle, kw)
         if total <= 0:
             raise RuntimeError("视频无可解码帧")
+        _rewind(handle)
         wanted = set(_uniform_indices(total, n))
         out: list[Image.Image] = []
-        for i, frame in enumerate(iio.imiter(video_path, plugin="pyav")):
+        for i, frame in enumerate(iio.imiter(handle, **kw)):
             if i in wanted:
                 # convert 之后原始 ndarray 就可以被回收;不持有 frame 本身。
                 out.append(Image.fromarray(frame).convert("RGBA"))
@@ -89,15 +107,26 @@ def _extract_frames(video_path: str, n: int) -> list[Image.Image]:
         # 两者都表现为悄悄换用 ffmpeg 分支、产出看着正常的帧。至少留一条日志。
         logger.warning("imageio 抽帧失败,回退系统 ffmpeg", exc_info=True)
 
+    return _ffmpeg_extract(source, n)
+
+
+def _ffmpeg_extract(source: str | bytes, n: int) -> list[Image.Image]:
+    """仅 pyav 失败时走。会把整段导出成 PNG,只应是冷路径。"""
     import glob
     import subprocess
     from imageio_ffmpeg import get_ffmpeg_exe
 
     with tempfile.TemporaryDirectory() as tmp:
+        if isinstance(source, str):
+            video_path = source
+        else:
+            video_path = str(Path(tmp) / "source.mp4")
+            Path(video_path).write_bytes(source)
         subprocess.run(
             [get_ffmpeg_exe(), "-y", "-i", video_path, "-vsync", "0",
              os.path.join(tmp, "f_%04d.png")],
-            capture_output=True, check=True,
+            capture_output=True,
+            check=True,
         )
         files = sorted(glob.glob(os.path.join(tmp, "f_*.png")))
         if not files:

backend/packages/ai_engine/src/windup_ai_engine/strategy/concrete.py

@@ -130,17 +130,57 @@ def derive(
         master: bytes,
         progress: ProgressPort,
     ) -> list[bytes]:
-        # 帧数直接读契约字段:缺省值已收进 ActionSpec(DEFAULT_N_FRAMES),不再由本层
-        # 用 `or 8` 兜底 —— 那等于把契约的缺省值写在实现里,换条 strategy 就换个默认值。
-        n = action.n_frames
         # 进度文案里的枚举一律取 .value:Python 3.11+ 改了 str-mixin 枚举的 __format__,
         # f"{action.action}" 现在给的是 "ActionType.WALK" 而不是 "walk"(3.12.13 实测),
         # 而这串字会经 server 变成用户看到的 SSE 进度。
         progress.step("derive", 0, 3, f"{action.action.value}: i2v 生成视频")
-        # 母版按动作预处理:jump 要在顶部补空间,否则角色腾空时头顶顶出视频画面被裁
+        video = self._submit_i2v(card, action, master)
+        return self.frames_from_video(video, card, action, master, progress)
+
+    def start_job(
+        self,
+        card: CharacterCard,
+        action: ActionSpec,
+        master: bytes,
+        progress: ProgressPort,
+    ):
+        """只建 i2v 单,把 job 交给延迟队列。"""
+        progress.step("derive", 0, 3, f"{action.action.value}: i2v 提交")
         framed = prepare_master(master, action.action.value)
-        video = self._video.i2v(framed, self._build_prompt(action, card.stance), seconds=5)
+        start = getattr(self._video, "start_i2v", None)
+        if start is None:
+            raise TypeError("当前 VideoProvider 不支持 start_i2v")
+        return start(framed, self._build_prompt(action, card.stance), seconds=5)
+
+    def poll_job(self, job_id: str, *, route_id: str | None = None) -> bytes | None:
+        """探一次。``None`` = 仍在跑;bytes = 成片;失败抛错。"""
+        poll = getattr(self._video, "poll_i2v", None)
+        if poll is None:
+            raise TypeError("当前 VideoProvider 不支持 poll_i2v")
+        snap = poll(job_id, route_id=route_id)
+        if snap.ok and snap.body:
+            return snap.body
+        if getattr(snap, "error_type", None) is None:
+            return None
+        raise RuntimeError(
+            f"i2v 轮询失败(HTTP {snap.http_status} {snap.error_type}): "
+            f"{snap.edge_fingerprint or snap.job_status or ''}"
+        )
 
+    def _submit_i2v(self, card: CharacterCard, action: ActionSpec, master: bytes) -> bytes:
+        framed = prepare_master(master, action.action.value)
+        return self._video.i2v(framed, self._build_prompt(action, card.stance), seconds=5)
+
+    def frames_from_video(
+        self,
+        video: bytes,
+        card: CharacterCard,
+        action: ActionSpec,
+        master: bytes,
+        progress: ProgressPort,
+    ) -> list[bytes]:
+        del card
+        n = action.n_frames
         dense = extract_all_frames_bytes(video)
         # 跨动作一致性:用视频首帧(=母版姿态)的角色高当共同定标基准。各动作都从同一母版
         # 起手,故此值一致 —— 否则各动作按自己最高帧定标,切状态时角色会忽大忽小。

backend/packages/app/src/windup_app/bootstrap/worker.py

@@ -9,9 +9,13 @@
 
 from windup_app.server.mq.catalog import all_stream_specs, email_stream_spec, generation_stream_spec
 from windup_app.server.orchestrator import task_repo
-from windup_app.server.orchestrator.executor import run_action_task, run_image_task
+from windup_app.server.orchestrator.executor import (
+    resume_action_poll,
+    run_action_task,
+    run_image_task,
+)
 from windup_app.server.orchestrator.recover import recover_orphaned_generation_tasks
-from windup_app.worker.consumer import StreamConsumer, start_relay_loop
+from windup_app.worker.consumer import StreamConsumer, start_delayed_loop, start_relay_loop
 from windup_app.worker.pending_timeout import release_stale_pending_tasks
 from windup_framework.db import Base, SessionLocal, engine
 from windup_framework.mq.model import MqMessage  # noqa: F401 — register metadata
@@ -48,6 +52,7 @@ def main() -> None:
     _recover_on_start(publisher)
     relay_pending_messages()
     release_stale_pending_tasks()
+    _warmup_local_inference()
 
     stop_event = threading.Event()
 
@@ -70,10 +75,12 @@ def _handle_signal(_signum, _frame) -> None:
             run_image_task=run_image_task,
             run_action_task=run_action_task,
             stop_event=stop_event,
+            resume_action_poll=resume_action_poll,
         ),
     ]
     threads = [consumer.start() for consumer in consumers]
     relay_thread = start_relay_loop(stop_event)
+    delayed_thread = start_delayed_loop(stop_event)
 
     pending_thread = threading.Thread(
         target=_pending_timeout_loop,
@@ -95,10 +102,31 @@ def _handle_signal(_signum, _frame) -> None:
         for thread in threads:
             thread.join(timeout=5)
         relay_thread.join(timeout=5)
+        delayed_thread.join(timeout=5)
         pending_thread.join(timeout=5)
         logger.info("windup worker 已停止")
 
 
+def _warmup_local_inference() -> None:
+    """顺序预热抽帧 / 抠图依赖,把 .so 和 onnx 一次读进页缓存。
+
+    首个动作任务再惰性加载时,会和并发 handler、同一块云盘上的 Postgres/Redis
+    叠成 IOPS 脉冲。预热失败不挡启动,只记一条日志。
+    """
+    try:
+        import imageio.v3  # noqa: F401
+        import av  # noqa: F401
+    except Exception:
+        logger.warning("抽帧后端预热失败", exc_info=True)
+    try:
+        from windup_framework.providers import OnnxU2NetMatteProvider
+
+        OnnxU2NetMatteProvider().warmup()
+        logger.info("ONNX 抠图会话已预热")
+    except Exception:
+        logger.warning("ONNX 预热失败,首个抠图任务会再加载", exc_info=True)
+
+
 def _pending_timeout_loop(stop_event: threading.Event) -> None:
     while not stop_event.wait(timeout=300):
         try: