[xiaocheny214]

Summary

• 动作 i2v 不再占用 worker 槽位 sleep 等成片：建单后写入 Redis ZSET 延迟队列，到期以 character_action_poll 投进现有 generation Stream，只做一次 inspect。
• 抽帧改走内存、ONNX 启动预热；上传 / 参考图下载 / 多图 gen_image 走共享 IO 线程池，降低单机磁盘抖动。
• executor 改为短 session，图/动作默认并发上调；轮询并发单独配置，不和新开 Stream。

Closes #570

Test plan

• uv run pytest tests/test_mq_delayed.py tests/test_mq_worker.py tests/test_mq_catalog.py tests/test_generation_orchestration.py tests/test_gateway_video.py tests/test_extract_streaming.py tests/test_sufy_video_download.py
• 提交动作生成：建单 worker 应立即 ACK；成片前不应占满 character_action 槽位
• 轮询到期后任务继续推进，完成后抽帧/上传/结清与现网一致
• worker 重启后未完成的 i2v 轮询能从延迟队列 / recover 接回
• 任务已失败或无开放冻结时，poll 消息跳过且不重复扣费

[vercel]

The latest updates on your projects. Learn more about Vercel for GitHub.

Project	Deployment	Actions	Updated (UTC)
windup	Ready	Preview	Aug 24, 2026 7:57am

[fennoai]

审查结论

核对了延迟队列到 generation Stream 的投递、动作建单/单次轮询/恢复、worker 并发隔离、视频网关拆分及内存抽帧链路。当前两处实现会分别破坏独立轮询并发和“轮询消息只 inspect 一次”的契约：共享线程池可被单一消息类型占满，且每次未完成轮询都会重复下载母版。

验证方面，python3 -m compileall -q packages tests 与固定 SHA 范围的 git diff --check 通过；按 PR 声明的 pytest 集合未能运行，因为执行环境没有 uv、pytest 及项目依赖。

View job run

[codecov]

Codecov Report

❌ Patch coverage is 73.90572% with 155 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
...app/src/windup_app/server/orchestrator/executor.py	78.80%	46 Missing ⚠️
...e/src/windup_ai_engine/impl/character_generator.py	22.58%	24 Missing ⚠️
...end/packages/app/src/windup_app/worker/handlers.py	21.73%	18 Missing ⚠️
...i_engine/src/windup_ai_engine/strategy/concrete.py	41.66%	14 Missing ⚠️
...end/packages/app/src/windup_app/worker/consumer.py	61.76%	13 Missing ⚠️
...es/framework/src/windup_framework/gateway/video.py	71.42%	10 Missing ⚠️
...packages/app/src/windup_app/server/mq/i2v_state.py	64.00%	9 Missing ⚠️
...app/src/windup_app/server/orchestrator/i2v_poll.py	89.28%	6 Missing ⚠️
...kages/framework/src/windup_framework/mq/delayed.py	91.07%	5 Missing ⚠️
...s/framework/src/windup_framework/providers/sufy.py	90.69%	4 Missing ⚠️
... and 3 more

📢 Thoughts on this report? Let us know!

[minorcell]

@fennoai

为何变更会这么大？

是否因为架构模块模糊的原因，导致修改队列相关的内容时存在大量耦合情况？

[fennoai]

结论

变更大，不完全是“队列模块耦合严重”造成的。更主要的原因是这个 PR 同时打包了 4 类改造：

1. Redis ZSET 延迟队列基础设施；
2. i2v 从同步等待改为“建单 → 持久化状态 → 延迟 poll → 恢复执行”的任务状态机；
3. worker 的 poll 独立并发、启动扫描、重启恢复与 ACK/幂等处理；
4. 短 DB session、共享 IO 池、内存抽帧、ONNX 预热、上传/下载/多图生成并行化。

因此它实际上不只是一个 MQ 改动，也是一次生成链路的并发与资源模型重构。

规模拆解

固定比较 17b25a7...bf1a9e2 后，共有 1806 行新增、243 行删除：

• 测试约 630 行变更（约 31%），用于覆盖延迟队列、worker、恢复、编排和视频下载；
• 生产代码约 1419 行变更；
• orchestrator/executor.py 单文件就有 570 行变更，从 751 行增长到 1101 行；
• 延迟队列及其 MQ/worker/recover 接线约 400 行；
• 抽帧、视频网关、供应商实现、生成并行化等 IO 优化约 435 行。

也就是说，直接属于“新增延迟队列”的代码并没有占据全部变更，IO 并行化和编排器重写贡献了相近甚至更大的规模。

是否存在架构耦合

存在，主要集中在生成编排层，而不是 Redis 队列实现本身。

• executor.py 同时负责任务状态、计费结算、模型调用、i2v 生命周期、Redis poll 状态、延迟调度、抽帧、质量检查、上传和线程池，职责明显偏多；改变 i2v 的等待方式会自然放大到整个文件。
• 新增一种 character_action_poll 消息，需要同时修改 catalog、consumer、handler、bootstrap、recover 和 executor。消息类型、并发策略、执行器选择与恢复策略分散在多个模块，扩展点不够集中。
• 任务生命周期横跨 Postgres 任务状态/冻结、Redis i2v 状态、ZSET 延迟项、Stream/outbox 和 worker ACK。为了保证重启恢复、幂等和不重复扣费，部分跨模块改动是业务一致性要求，不能简单归为坏耦合。
• windup_framework.mq.delayed 本身相对独立、通用；真正耦合的是“动作任务如何进入等待态、如何恢复和结算”的应用编排。

建议拆分

• PR 维度：延迟队列基础设施、i2v 状态机、worker 并发、生成 IO 优化可拆成 3–4 个独立 PR，降低审阅面和回归定位成本。
• 代码维度：从 executor.py 抽出 action_workflow、i2v_polling、generation_io，让执行器只协调明确的阶段结果，例如 Completed / AwaitingVideo / Failed。
• MQ 扩展点：用统一的 handler/spec 注册描述消息处理器、并发池、信号量和恢复策略，避免每增加消息类型都同步修改 4–5 个文件。
• 状态边界：把 i2v 等待状态及其 schedule/reschedule/cancel 行为封装为一个应用服务，减少 executor、recover 和 handler 对 Redis 细节的共同认知。

总体判断：约一半是本 PR 范围过宽造成的，另一部分确实暴露了 executor.py 的职责聚集和消息扩展点分散；但并非整个 MQ 架构都高度耦合。

本次仅做固定提交范围的只读分析，未修改代码或运行测试。

View job run

[xiaocheny214]

@minorcell 变更大，主要不是 Redis 队列模块糊在一起，而是这个 PR 同时打了四件事：延迟队列、i2v 状态机、poll 独立并发、生成 IO 并行化。胀在编排层：executor.py 原先同时管等待态、Redis/ZSET 和结算。

本 PR 已把 i2v 等待抽到 orchestrator/i2v_poll.py（schedule / inspect / recover），executor 只协调 Completed / AwaitingVideo / Failed。IO 池和 MQ 扩展点不塞进这次，避免审阅面再胀：

• refactor(orchestrator): 从 executor 抽出 generation_io #597 从 executor 抽出 generation_io
• refactor(mq): 用 handler/spec 注册扩展消息类型 #598 用 handler/spec 注册扩展 msg_type

#571 合入后再跟这两条。

[johnnyzhang-eng]

LGTM