P02 · Orchestrator v1：线性多 stage 路由

例 1：S0→S1→S2，final_stage_ids={0,2}，forward_fn = lambda nxt, prompt, prev: prev。

await orch.add_request("r1", "hi there")
S0.submit("r1", "hi there")          -> S0 吐出 "hi", "there"(finished)
S1.submit("r1", ["hi", "there"])     -> S1 吐出 ["HI","THERE"](finished)（只有一块）
S2.submit("r1", [["HI", "THERE"]])   -> S2 吐出 "wav:HI", "wav:THERE"(finished)

stage 1 的输出一条也不会出现在用户侧。结束后 request_states == {}，dropped_outputs == 0。

例 2：只有最后一个 stage 是 final（final_stage_ids={2}，prompt "a b c"）——用户只收到 wav:A(F)、wav:B(F)、wav:C(T)。S0 不是 final，但它的完成仍要触发向 S1 的转发——“给不给用户”和“要不要转发”是两件独立的事。

例 3：未知 request。add 之前 S1 先冒出 StageOutput("ghost", "?", True)；正常跑完 r1；r1 终态之后 S2 又迟到 StageOutput("r1", "wav:LATE", True)。结果：用户侧没有多余消息；request_states == {}；dropped_outputs == 2。

关键问题

1. S0 的最后一块 "there" 自带 finished=True。发给用户的那条 OrchOutput 的 finished 应该是什么？由什么决定？
2. S0 的最后一块既要给用户，又要转发给 S1。两件事都做吗？先做哪件有关系吗？
3. S2 拿到的输入为什么是 [["HI","THERE"]] 而不是 ["HI","THERE"]？“stage s 产出的全部数据”是哪个列表？
4. 为什么必须在 await stages[0].submit() 之前把 ReqState 放进 request_states？
5. 请求清理之后又来一条它的输出，会发生什么？用 setdefault 顺手建 ReqState 会怎样？
6. reader 为什么不自己处理输出，而要丢进 _ready_q 交给唯一 dispatcher？正确性靠什么保证？
7. final_stage_ids={0} 时 last_stage 是多少？S1、S2 会被 submit 吗？

细节

• StageOutput(request_id, data, finished) 是 stage→orchestrator，get_output() 一次返回 list[StageOutput]（可能混着多个请求），没有 stage_id——哪个 stage 来的只有读它的 reader 知道。OrchOutput(request_id, stage_id, data, finished, error) 是 orchestrator→用户，只为 final stage 生成。
• last_stage = max(final_stage_ids)：管线只需跑到最后一个 final stage。
• ReqState：submitted 已提交的 stage；finished 已报告完成的 stage；stage_data[s] stage s 的全部 data（按到达顺序）；pending_final P02 不用。
• 三个队列：_req_q（add → handler）、_ready_q（readers → dispatcher）、_out_q（dispatcher → next_output）。next_output(timeout=None) 表示一直等。

• shutdown()：cancel 所有 task，再 gather(..., return_exceptions=True)（否则 “Task was destroyed but it is pending”）。
• reader 带上 sid：一个 stage 的输出经过“一个 reader → 一个 FIFO 队列 → 一个 dispatcher”，同一 stage 内顺序天然保持。
• _handle_output 的决策顺序（先写成注释再填代码）：① 查状态，没有就 drop 计数并返回；② 追加 stage_data[sid]；③ 若 o.finished，sid 加入 finished；④ 若 sid 是 final：算出请求级 done，发一条 OrchOutput；done 的话清理并立刻 return；⑤ 若 o.finished 且 sid < last_stage：forward_fn 构造输入、标记 submitted、submit 给 sid+1。
• 先改状态再 await：dispatcher 里的 await _out_q.put(...) 和 await stages[nxt].submit(...) 都是让出点；在任何 await 之前把 stage_data / finished / submitted 更新好。
• add 路径：add_request 只把 ("add", rid, prompt) 放进 _req_q；真正的注册+提交在 handler 里做，注册排在第一个 await 之前。
• P03 会继承这个类并只覆盖 _handle_add / _handle_output——把三个循环和 cleanup 写成独立方法。

骨架要点：_handle_add = 建 ReqState → 注册 → submitted.add(0) → await stages[0].submit；_handle_output 按 L2 的 ①–⑤；_cleanup(rid) = request_states.pop(rid, None)。