Async Task 生命周期#

本文描述 sub-agent 异步调用的设计：显式的 task 实体、后台 worker pool，以及 一套可查询、可取消的生命周期。接口形态对齐 Claude Code 的 TaskCreate / TaskList / TaskGet / TaskUpdate / TaskStop。

它所依托的底盘 WebUI 已经具备：每个 session 一个 worker thread （_execute_in_context）、一个 _cancel_events dict（_pause_stop.py）， 一个供 UI spinner 用的 _running_tasks dict。run_agent_turn （openprogram/agent/sub_agent_run.py）是那条同步路径，/task tool、 /spawn WS action、_merge.process_merge_turn 都复用它；task 抽象在其之上 叠加。

Part 1. Task 生命周期需要考虑的维度#

一套完整的 spawn / list / get / cancel 生命周期，下面 15 个点都得明确。Part 2 按同一张 checklist 逐个场景走一遍。

D1. Task 实体存什么#

Task 至少要带：

• task_id：spawn 立刻返回的稳定 id（独立于 user_msg_id / assistant_msg_id）
• subject：单行简介（用于 list / panel 显示）
• description：完整 prompt（agent 调用时复用为 prompt）
• agent_id：跑哪个 profile
• status：D2 的状态枚举
• created_at / queued_at / started_at / completed_at：时间戳
• parent_session_id：task 跑在哪个 session 上（task 永远绑定一个 session）
• parent_task_id：spawn 这个 task 的 task（顶层 user-spawn 时为 None）
• parent_msg_id：触发 spawn 的 user / assistant msg id（用于把 attach 卡片挂回去）
• context_mode：inherit / clean
• head_id：task 完成后 sub-agent 落地的 assistant msg id（运行中为 None）
• result_text：sub-agent 最终回复（运行中为 None）
• error：失败时的错误字符串
• cancel_requested_at：cancel 信号写入时间
• attempt：重试时的计数（初版固定 0）

cancel_event / future 这类 runtime 对象不进 entity；entity 只描述 "what"，"how to cancel / wait" 由 runner 内部 map 解决（见 D5）。

D2. 状态机#

pending → queued → running → completed
                          ↘ cancelled
                          ↘ errored

• pending：spawn API 收到、entity 写入持久层、还没排上 worker pool。
• queued：交给 ThreadPoolExecutor 但还没 pick up（worker 全忙）。
• running：worker pick up 并开始 process_user_turn。
• completed：sub-agent 正常返回，head_id 和 result_text 写入。
• cancelled：cancel 事件被消费且 worker 退出（部分输出可能落地）。
• errored：worker 抛异常 / 进程崩溃恢复时未完成的 task。

每次状态转移都更新对应时间戳，不可逆。pending → cancelled 直接跳过 queued / running 也允许（用户在 worker pick up 前就 stop）。

D3. Worker pool 模型#

pool 用 concurrent.futures.ThreadPoolExecutor，与现有 _execute_in_context 的 daemon thread 模型同构，且不需要把整个 codebase async-color。理由：

• process_user_turn 内部已经自己开 asyncio.new_event_loop() 跑 agent loop，外部包 thread 不会双重事件循环冲突。
• BashTool / 文件 IO 都是阻塞调用，线程模型对它们零成本。
• 并发上限按 OPENPROGRAM_TASK_WORKERS 环境变量配置，默认 4。超过上限 task 停在 queued，FIFO 公平。
• Backpressure：单 session 不限制（task 之间是 sibling，UI 显示串行的 spinner 不影响），全局 pool 上限就够了。priority 可留待后续按需加。

D4. 持久化#

持久化复用现有 session-git meta：session repo 的 meta.json 旁边放一个 tasks.json，结构 {task_id: TaskRow}。每个 task 状态转移调一次 commit_turn("task: ...")，落 git 历史。

理由：

• task 永远绑定单一 session（D6），存在 session 目录下天然分区。
• session 已经走 git-as-truth，task 状态进 git history 反而帮 debug "为什么 task 卡住"。
• 不需要新一张 SQLite 表或新一个 DB 文件。

不可恢复策略：进程 crash 后，所有 status=running / queued / pending 的 task 在 startup 时一律标记 errored（error="worker died before completion"）。 理由是 LLM call 已经发出去拿不回来；让用户重 spawn 是最干净的语义。

D5. Cancel 信号传播#

每个 running task 在 runner 内有一份 threading.Event（不存 entity）。 Cancel API 触发时：

1. 写 cancel_requested_at 到 entity，状态转 cancelled（如果还在 queued / pending），或保持 running 等 worker 自然退出。
2. cancel_event.set() — 通过 _pause_stop.register_cancel_event 已经定义的 contract，传到 process_user_turn(cancel_event=...)。
3. process_user_turn 已经把 cancel_event bridge 进 asyncio.Event（agent_loop 调用），LLM provider stream 会在下一个 chunk 检查到 cancel 然后中断。
4. BashTool / 其他 subprocess：复用 _pause_stop.kill_active_runtime（已有）。Tool 层的 cancel 是合作式：每个 @agentic_function 的 pre-invocation hook 检查 is_cancelled（已有），下一个 tool call 入口会 raise CancelledError。
5. 兜底 timeout：cancel 后 30 秒 worker 还没退出，runner 把 entity 标 cancelled（error="cancel timed out, worker may be stuck"），然后 detach worker thread（不强杀，等 GC）。

工具自身原子操作（比如一个 Write 写一半）不中断，等当前 atomic 完成再退出。

D6. Task ↔ session 的关系#

一个 task 永远跑在一个 parent session 上。跨 session 在概念上等于 merge / attach（已有），不在 task 范畴。

task.parent_session_id 就是 sub-agent process_user_turn(session_id=...) 用的那个——与 run_agent_turn 的行为一致。sub-agent 的输出落地为该 session 的 一个 branch（或新 root，看 context_mode），所以 session repo 既存了 task entity 又存了 task 产出，自洽。

D7. Task ↔ sub-agent 的关系#

所有 spawn 都走 task entity，因此 /task 不同步阻塞。语义如下：

• agent-facing tool agent(prompt, ...) 是 async 的——立刻返回 task_id，不阻塞主对话。LLM 拿到 id 后选择继续干别的，或者立刻调 task_output(task_id) 取得同步语义（D15）。
• 兼容旗子：/task --sync（或工具的前台默认 run_in_background=False）在 tool 层包一层 spawn → await，对 LLM 透明地同步返回结果。
• _task_impl 仍保留 run_agent_turn 调用，但走 runner 入口。

/spawn（用户在 chat 输入 /spawn label: prompt）走同一条 spawn API，区别只是 caller 是 WS handler 而非 LLM。

D8. Task ↔ ContextCommit 的关系#

attach pointer（function="attach" 节点）由 _run_spawn / _task_impl 写入：

• spawn 时立刻写一个 placeholder attach card（function="attach"，extra.attach.task_id = <task_id>，extra.attach.status = "running"），content="(running)"，source_commit_id 留空。
• task 完成时 runner update 同一个 attach card 节点：填 head_id / source_commit_id / 替换 content 为 final_text，status 改 completed / cancelled / errored。
• generator 看到 status="running" 的 attach 节点：跳过展开（不进 commit items），只在 UI 显示卡片占位。看到 status="completed" 走现有 attach 展开路径（见 context.md 场景 B）。

这样 LLM 在 task 跑完前再触发新 turn 不会看到半成品 attach 内容，但用户能看到 spinner。

D9. WS API#

四个 ws action（参考 ws_actions/ 现有命名）：

• spawn_task — {action, session_id, prompt, description, agent_id?, context?, wait?} → 立刻回 {task_id, status, parent_msg_id}。
• list_tasks — {action, session_id?, status_filter?, limit?} → {tasks: [...]}。无 session_id 等于全局列。
• get_task — {action, task_id} → 单条 entity + 当前 head_id / 部分 result（如果 running）。
• cancel_task — {action, task_id} → {task_id, status}（同步返回当前 status，不等 worker 退出）。

广播事件：

• task_created、task_status（queued / running / completed / cancelled / errored）、task_progress（可选，未来可加 token 进度）。
• 复用现有 running_task 广播并加 task_id 字段（UI 兼容）。

D10. agent-facing 工具#

agent 用三件套：

• agent(prompt, description, agent_id?, context?, run_in_background=False) → 前台默认返回最终 result，run_in_background=True 返回 task_id。包装 runner.submit(...)。
• task_output(task_id, block=True, timeout=30000) → 等待(timeout 毫秒,上限 600000,对齐 Claude Code 的 TaskOutput 参数形状)直到完成 / cancelled / 超时,返回回复文本+终态;block=false 立即窥探当前状态。list_tasks() 列出本会话的后台任务(id、状态、主题),LLM 用它掌握并行工作的全景。
• task_stop(task_id, reason?) → {ok, status}。

变体 await_tasks([id1, id2, ...], mode="all"|"any", timeout) 用于 plan mode 的 wait_all（D14）。三件套加 wait_all = 四个工具，但 plan-mode 不暴露 wait_all 给普通 agent——普通 agent 只能 await 单个 id，以免被误用。

D11. UI 表达#

• 右侧 panel 一个 Tasks tab（紧挨现有 Branches / Context Commits panel）。列出当前 session 的所有 task entity：spinner + subject + status + 创建时间。
• 点开 task：跳到对应的 attach card（chat 里已有 placeholder）→ checkout 它的 head_id branch（如果 completed）。
• 每个 attach card 自带 status badge：running / done / cancelled / error。完成后的行为与普通 attach card 一致。
• 全局 sidebar 显示一个总计数徽章（多少 running）：复用现有 running_task 机制。

D12. 错误恢复#

• worker 抛异常：runner 捕获 → 状态 errored → error 字段填 f"{type}: {msg}" → attach card status 同步更新 → 广播 task_status 事件。
• 主进程 crash：参考 D4。startup hook 扫 tasks.json，把所有非终止态记为 errored。
• pool shutdown：进程关闭时 wait 5 秒；超时的 task 标 errored（"worker pool shutdown"）。
• 同一 task_id 不允许重 spawn（spawn API idempotent on task_id 但默认是 mint 新 id）。

D13. 测试边界#

unit test 不跑真 LLM。需要的 seam：

• runner.submit(req, *, sync_fn=...) —— 把 process_user_turn 替成 fake 同步函数（接收 cancel_event，返回 fake TurnResult）。
• entity store 用 in-memory MockTaskStore（dict 而非 git，但同样接口）。
• 状态机测试矩阵：每对合法转移一个 case，每对非法转移一个 reject case。

集成测试覆盖：spawn → await → completed 一路、spawn → cancel mid-flight、 spawn → worker raise → errored、多 task FIFO 排队、crash recovery 把 running 标 errored。

D14. Plan mode 集成#

plan agent 产出一个 plan（一组 sub-task spec），exit_plan_mode 完成时：

• plan 工具 / executor agent 接收 spec list（每条含 description + prompt + 可选 agent_id）。
• 顺序调 spawn_task 拿 N 个 task_id。
• 调 await_tasks(ids, mode="all") 等全部完成。
• 拿到结果后由 executor 综合（写一条 user-visible summary，或自动触发 merge）。

并发上限就是 D3 的 pool size — plan 列 10 个 task、pool 4 个，会有 6 个停在 queued，UI 显示排队。

D15. 向后兼容#

/task 在 chat 里继续是用户输入入口，行为不变（用户看到的是 attach card + 完整结果），但底层走 task entity。

LLM-facing agent(prompt, ...) tool 提供两个语义：

• run_in_background=False（默认）：内部 spawn + await + 把 result_text 当 return value 给 LLM。LLM 视角与同步工具等同。
• run_in_background=True：返回 task_id，LLM 自己决定何时 await。新代码 / plan mode 用。

切换开关位于 tool 签名里，按同步工具写的老 prompt 不改也能跑。工具能力广播给 LLM 的描述里说明两种模式 + 推荐用法。

Part 2. 每种场景按维度过一遍#

场景 A：单个 sync /task#

最常见情况：LLM 调 agent(prompt="探查 X")，希望阻塞拿到结果。行为与同步工具 一致，但底层走 task entity。

场景 B：单个 async task（agent 主动选 async）#

LLM 决定干个长活，先 spawn 拿 id，回头再 await 或 cancel。

场景 C：并发 N 个 async task（plan mode）#

plan agent 列出 5 个调研任务，spawn 5 个 task，调 await_tasks(ids, mode="all") 等齐。

场景 D：长时间 task + cancel#

agent spawn 一个 30 分钟的 deep research task，10 分钟后用户在 UI 点 Stop 或 agent 自己想撤。

Part 3. 关键不变式#

1. 终止态唯一：每个 task entity 最终必有 completed / cancelled / errored 之一，不存在永久 running（pool shutdown / crash recovery 必转 errored）。
2. 状态单调：状态机每条边只走一次。一旦 completed 不会再 → cancelled，不允许 running → pending。
3. cancel 可达：cancel API 返回后，30 秒内 entity 必出现 cancelled / errored 终态（强制 timeout 兜底）。
4. placeholder 一致：spawn 写的 attach card 跟 entity 同步状态；entity 状态变化必同步 update card。
5. 持久化 idempotent：crash 后 reload，未完成 task 的状态确定为 errored；同 task_id 永不复活。
6. session 绑定不可变：task 创建后 parent_session_id 不变；同一 task 不跑两个 session。
7. 并发安全：runner 内 _tasks + _cancel_events map 全程持锁；状态读写不竞争。
8. 测试可控：runner 必须接受可注入的 sync_fn / store，不依赖真 LLM 也能跑全状态机。

Part 4. 不在本设计范围#

• 跨进程 task：所有 worker 都在主进程内。分布式 / multi-host 留给后续，需要 message broker。
• Task 优先级 / SLA：FIFO 即可，没有高优先级抢占。后续按需加 priority queue。
• Resume / 续跑：cancelled / errored task 不能"接着跑"。用户 retry 等于新 spawn 一个 task。
• Task 重试策略：runner 不自动 retry；上层 agent / plan 自己决定。
• 跨 session task：一个 task 只绑一个 session。跨 session 用 attach / merge。
• DAG-shaped task 依赖：await_tasks(mode="all"|"any") 已经够 plan mode；显式 DAG / pipeline 留给后续。
• Task 输出流式订阅：初版只在 task 完成时拿 final_text。中途订阅 stream（让父 agent 看到 sub-agent 边想边说）留给后续。
• 资源配额：单 user 同时 task 数 / token 上限不在本设计，需要先有 multi-tenant 模型。

附录：实现状态#

以上尚未落地。当前 runtime 没有 task entity——只有驱动 spinner 的内存 _running_tasks dict，refresh 即丢；没有显式状态机（函数返回就代表"完成"）； 没有查询接口；也没有 plan mode 并发，因为 /task 是同步串行的。Cancel 存在于 整 session 粒度（_cancel_events），没有单 task 粒度；worker 模型是每 session 一个 daemon thread（_execute_in_context）而非共享 pool；attach card 完成后才 写；agent 只有同步的 task 工具；UI 也没有 Tasks panel。

落地顺序按依赖排列：