Jellyfish 任务异步化与取消能力建设：从同步 Agent 接口到「可恢复、可取消」的 script-processing 任务体系

本文是 Jellyfish 项目“任务计划”系列的技术方案解读，源文档见 task-async-cancellation-plan.md。它面向需要接入、维护或扩展脚本处理（script-processing）异步任务的开发者，完整说明了长耗时 LLM 智能体接口如何任务化、页面如何按业务实体恢复任务、以及“请求取消 + 协作式取消”的分阶段取消语义在 Jellyfish 中的真实落地形态。读完本文，你将掌握 divide-async 等接口的设计骨架、GenerationTask / GenerationTaskLink 复用方式、relation_type 约定，以及 worker 侧取消检查点与 Celery 执行链路的实现细节。

背景：长耗时智能体接口为什么必须任务化

Jellyfish 的核心生产链路是「剧本输入 → 结构化分镜 → 一致性管理 → 镜头准备 → 视频生成」。其中 script-processing 路由下聚集了一批依赖 LLM Agent 的长耗时接口。在任务化改造之前，这些接口是同步直接调用 Agent 的，最典型的就是：

• POST /api/v1/script-processing/divide

同步调用的直接后果是请求长时间阻塞在 Agent 推理上。结合页面体验，会暴露出一组非常具体的问题：

• 请求长时间阻塞，页面一直 loading；
• 页面刷新后上下文丢失，无法知道任务是否还在跑；
• 用户无法明确知道当前是否已有任务在运行，同一章节可能被重复触发提取；
• 任务体系虽然有 cancelled 状态，但缺少真正的取消入口和取消执行机制。

方案的目标是渐进式落地，原文给出了清晰的演进链：

长耗时智能体接口任务化
→ 页面刷新后可恢复
→ 任务状态可见
→ 先支持请求取消与协作式停止
→ 强终止能力后置

这一链条的每一环都在仓库中有对应实现（详见下文各节）。

三个核心结论

1. divide 应优先改为异步任务接口

建议新增异步接口，同时保留原同步接口：

POST /api/v1/script-processing/divide-async
POST /api/v1/script-processing/divide   （保留：兼容旧调用与调试）

这样同时兼顾三件事：前端主流程迁移、管理端/脚本调试、单测与回归验证。从源码看，script_processing.py 中 divide_script_async 已落地：先校验 chapter_id，再调 create_divide_task 创建任务，非复用场景下 spawn_divide_task 触发后台执行，最后把 task_id/status/reused/relation_type/relation_entity_id 封装成 AsyncTaskCreateRead 返回。

2. 页面恢复应复用现有任务体系

不建议为 divide 单独造一套任务模型，而是直接复用：

• GenerationTask
• GenerationTaskLink

并通过下面这组业务关联恢复页面状态：

relation_type = "chapter_division"
relation_entity_id = chapter_id

复用意味着不新增表、不重复造轮子，页面恢复逻辑可以统一抽象为“按业务实体查最新活跃任务”。

3. 取消能力要分阶段做

当前技术栈下，很多长任务本质上还是单次同步 Agent 调用，不一定暴露可中断句柄。因此第一阶段不承诺“立即强终止”，先做：

• 请求取消（登记取消意图）
• 页面显示“已请求取消”
• worker 在检查点协作式停止

如果未来要强终止，再升级到运行句柄注册表、独立 worker 进程 / 队列系统。

为什么 divide 要优先任务化

当前问题

/api/v1/script-processing/divide 当前的同步调用链是：

请求进入
→ ScriptDividerAgent.divide_script(...)
→ 结果返回

对应源码中的同步实现（script_processing.py）直接实例化 ScriptDividerAgent 并同步返回 ScriptDivisionResult，这条链一旦慢下来，用户会直接感受到页面一直 loading、刷新后不知道任务是否还在跑、同一章节被重复触发提取。

任务化后的目标

改成异步后流程变为：

POST divide-async
→ 创建任务
→ 创建 task_link
→ 立即返回 task_id
→ 后台执行 divide
→ 页面按 chapter_id 恢复任务状态

divide-async 设计

路由

POST /api/v1/script-processing/divide-async

请求体

建议复用当前同步接口的请求体（ScriptDividerRequest）：

{
  "chapter_id": "2a67890b-46a9-41ed-b220-18d18a3d300a",
  "script_text": "...",
  "write_to_db": true
}

字段说明：

• script_text：完整剧本文本，必填（min_length=1）；
• write_to_db：是否将分镜写入 AI Studio shots 表，默认 false；
• chapter_id：章节 ID，write_to_db=true 或 divide-async 场景下必填（异步接口在缺省时会直接返回 400，见 divide_script_async）。

返回体

{
  "success": true,
  "code": 200,
  "message": "Task created",
  "data": {
    "task_id": "uuid",
    "status": "pending",
    "reused": false,
    "relation_type": "chapter_division",
    "relation_entity_id": "2a67890b-46a9-41ed-b220-18d18a3d300a"
  }
}

其中 reused=false 表示本次是新建任务；若命中同章节活跃任务则 reused=true。

同章节活跃任务复用

建议增加业务约束：同一 chapter_id 同时只允许一个活跃的 chapter_division 任务。活跃状态包括：

• pending
• running
• streaming

如果已存在活跃任务，则不重复创建、直接返回已有 task_id、reused = true。这解决了重复点击、页面刷新后再次点击、多标签页并发触发三类场景。

仓库实现上，script_processing_tasks.py 的 _find_active_task 通过 GenerationTaskLink 的 relation_type + relation_entity_id 反查，再按 GenerationTask.status IN (pending/running/streaming) 过滤并 limit(1)；create_divide_task 先查活跃任务，命中即返回 reused=True，否则通过 TaskManager.create(mode=DeliveryMode.async_polling, task_kind=SCRIPT_DIVIDE_TASK_KIND) 创建任务并追加 GenerationTaskLink 关联。测试 test_script_processing_tasks.py 中 assert result.reused is False 与 assert second.reused is True 恰好验证了首建与复用两条路径。

页面刷新后的关联与恢复

业务关联写法

divide 任务创建时写入：

resource_type = "task"
relation_type = "chapter_division"
relation_entity_id = chapter_id

注意：仓库实际写入 GenerationTaskLink 时使用的 resource_type 为 "task_link"（见 create_divide_task），恢复逻辑的关键是 relation_type + relation_entity_id 这一对键。

页面恢复方式

当用户进入章节页、分镜列表页、项目工作台时，可按 chapter_id 查询最新活跃任务：

chapter_id
→ relation_type=chapter_division
→ latest active task

如果存在活跃任务，页面应显示“正在提取分镜”、轮询任务状态、任务完成后刷新分镜列表。任务中心（任务列表）接口也支持按 relation_type / relation_entity_id 过滤（见 task_status.py 的 GET /tasks），后台的 SqlAlchemyTaskStore.list_task_views 会 join GenerationTaskLink 并解析导航目标（chapter/shot/actor 等，见 stores.py），前端可直接获得跳转目标。

页面提示建议

统一展示四类状态文案：

• 正在提取分镜
• 已请求取消
• 提取失败
• 提取完成

取消能力设计（四阶段）

第一阶段：请求取消

新增接口：

POST /api/v1/film/tasks/{task_id}/cancel

并在 generation_tasks 表上新增字段：

• cancel_requested（是否已请求取消，TINYINT(1) NOT NULL DEFAULT 0）
• cancel_requested_at（请求取消时间，DATETIME NULL）
• cancel_reason（取消原因，VARCHAR(255) NULL）
• cancelled_at（实际取消完成时间，DATETIME NULL）

这些字段在迁移脚本 004-add-generation-task-cancel-fields.sql 中有完整的幂等建列语句，并附带 ix_generation_tasks_status_cancel_requested (status, cancel_requested) 索引；模型层定义见 task.py。

这一阶段先解决的是：用户可以发起取消、页面知道取消请求已存在、任务状态接口能显示取消请求状态。这不等于任务一定能立刻停下。

阶段边界的语义在 Store 层有明确区分（stores.py）：

• request_cancel(task_id, reason)：登记取消请求；若任务仍处于 pending，直接置为 cancelled 并写 cancelled_at/finished_at；
• mark_cancelled(task_id)：worker 在检查点检测到取消请求后，统一完成 cancelled 收尾；
• is_cancel_requested(task_id)：worker 检查点轮询取消标记。

路由层 cancel_task 会先 request_cancel，若任务未立即终止且 revoke_task_execution 成功（Celery 场景），再 mark_cancelled，返回体中的 effective_immediately 明确告知前端是否立即生效。

第二阶段：协作式取消

在 worker 中增加取消检查点，原文建议的检查点位置：

1. 任务启动前
2. 调用下一个 Agent 前
3. 多阶段流程的阶段边界
4. 写库前

如果检测到 cancel_requested = true，则停止后续步骤、将任务写成 cancelled、记录 cancelled_at。

仓库中抽出了统一收尾函数 _cancel_if_requested：在阶段边界调用 store.is_cancel_requested(task_id)，命中则 store.mark_cancelled(task_id) 并提交。以 run_merge_task 为例，可见同一任务内安排了多次检查点：进入任务前、拿到 run_args 后、Agent 调用返回后、写库前。测试 test_script_processing_tasks.py 中的 test_run_extract_task_marks_cancelled_when_cancel_requested_before_start 与 test_run_extract_task_marks_cancelled_at_stage_boundary 分别验证了“启动前取消”和“阶段边界取消”两条路径，断言任务最终为 cancelled 且 cancelled_at 已记录。

这种方式的限制

如果当前代码正卡在单次长时间同步模型调用或阻塞式 SDK 调用，第一阶段无法做到“立刻中断”，只能等待当前步骤结束后再停。所以产品与开发文档里都应明确：

第一阶段的取消能力属于“请求取消 + 协作式取消”，不是强终止。

第三阶段：运行句柄注册表

如果后续要增强取消能力，建议在 TaskManager 上新增运行句柄注册表：

task_id -> asyncio.Task

这样对真正的异步后台任务，可以尝试 task.cancel()。但要注意：即使有运行句柄，也不代表所有同步 LLM 调用都能立即停住。

当前 TaskManager（manager.py）已经预留了 request_cancel / is_cancel_requested / mark_cancelled 三个取消相关方法，运行句柄注册表可以平滑扩展进同一抽象。

第四阶段：强终止能力预留

如果未来业务明确要求“立即终止长任务”，需要进一步升级架构：

• 独立 worker 进程
• 队列系统
• 执行进程与 Web 进程隔离

到那时才能更可靠地支持 kill worker、回收阻塞任务、更强的取消语义。当前阶段不建议直接跳到这一步。值得一提的是，仓库在任务化过程中已经引入了 Celery 执行入口：execute_task.py 中 enqueue_task_execution 通过 run_task_celery.delay(task_id) 入队并回写 executor_type="celery" 与 Celery task id；revoke_task_execution 则可在取消接口中被调用，对尚未启动的 Celery 任务直接 AsyncResult.revoke(terminate=True)，这为强终止预留下了真实可行的执行通道。

当前落地状态

主线接口：已完成任务化并接入真实页面

以下接口均已进入「异步任务 → 页面按业务实体恢复任务 → 可请求取消 → 协作式取消」链路：

• /api/v1/script-processing/divide
• /api/v1/script-processing/extract
• /api/v1/script-processing/check-consistency
• /api/v1/script-processing/optimize-script
• /api/v1/script-processing/simplify-script
• /api/v1/script-processing/analyze-character-portrait
• /api/v1/script-processing/analyze-prop-info
• /api/v1/script-processing/analyze-scene-info
• /api/v1/script-processing/analyze-costume-info

预备能力：已任务化但当前无真实前端入口

• /api/v1/script-processing/merge-entities
• /api/v1/script-processing/analyze-variants

这两条接口当前不再列为前端主线整改项，处理策略是：保留后端实现、测试与 OpenAPI；在路由描述与代码注释中标明“预备能力”（如 merge_entities_async 的 summary 明确写着“当前保留为预备能力，尚无真实前端入口”）；等未来出现真实页面入口后，再按同一套任务恢复模型接入。

推荐的 relation_type 约定

建议统一使用这批值（与 script_processing_tasks.py 中的常量一一对应）：

chapter_division          （divide）
script_extraction         （extract）
entity_merge              （merge-entities，预备能力）
consistency_check         （check-consistency）
variant_analysis          （analyze-variants，预备能力）
character_portrait_analysis
prop_info_analysis
scene_info_analysis
costume_info_analysis
script_optimization       （optimize-script）
script_simplification     （simplify-script）

当前已落地全部 11 个值。统一的 relation_type 是页面恢复、任务列表过滤与任务中心导航（stores.py 中 chapter-like / shot-like / image-like 映射）共同依赖的契约，不要在业务代码里发明新值。

实施顺序回顾与后续建议

已完成阶段

1. generation_tasks 增加取消字段（004-add-generation-task-cancel-fields.sql）
2. TaskStore / TaskManager 增加取消请求能力（manager.py、stores.py）
3. 新增 script_processing_tasks.py（script_processing_tasks.py）
4. divide-async / extract-async 落地
5. 页面按业务实体恢复 divide / extract 任务
6. check-consistency / optimize-script / simplify-script 接入任务体系
7. 资产分析类接口接入任务体系
8. worker 协作式取消检查点补齐第一轮

当前收口重点

1. 清理未启用接口的前端接入预期
2. 在路由描述、代码注释、开发文档中明确：哪些是主线接口、哪些是预备能力、哪些同步接口只用于兼容与调试（同步接口的描述里已普遍注明“主要用于兼容旧调用与调试场景”）

后续建议

1. 继续观察是否会新增真实页面入口
2. 如果未来重新启用 merge-entities / analyze-variants，再按同一套任务恢复模型接入
3. 如果业务明确要求更强的终止能力，再推进运行句柄注册表或独立 worker

对前端的约束

页面不应只依赖本地缓存 taskId。正确做法是：

页面打开
→ 根据业务实体查最新活跃任务
→ 恢复轮询
→ 根据任务状态更新 UI

也就是说：章节页看 chapter_division、分镜编辑页看 script_extraction，不以“页面是否没刷新”为前提。任务轮询使用 GET /api/v1/film/tasks/{task_id}/status，其返回体（get_task_status）包含 status / progress / cancel_requested / started_at / finished_at / elapsed_ms 等字段，足够支撑进度条与取消态 UI。

对产品语义的建议

当前阶段建议统一对外说明：

用户可看到

• 任务进行中
• 已请求取消
• 已完成
• 已失败

用户暂时不要期待

• 点击取消后立即硬停止

更准确的提示文案应是：

已请求取消，系统会在当前步骤结束后停止。

最终建议

这轮最值得先做的，不是一步到位上强终止，而是：

divide 异步任务化
→ 页面刷新可恢复
→ 任务状态可见
→ 取消请求可登记
→ worker 协作式停止

这样能先解决绝大多数体验问题，同时为后续更强的任务控制能力（运行句柄注册表、独立 worker/队列、强终止）留好扩展点。仓库当前实现已完整覆盖这条链路：任务模型与取消字段（task.py）、可插拔 Store 与 TaskManager（task_manager）、异步任务编排（script_processing_tasks.py）、Celery 统一执行入口（execute_task.py）以及成体系的异步任务测试（test_script_processing_tasks.py），是后续继续演进时可对照参考的完整样例。