30 的有界队列说明“send 已返回”不等于“任务已完成”。现在要给执行过程加一条进度流：如果函数返回 `Flow`，是不是任务已经启动、状态已经写入 Store？对 `flow {}` 构造器而言，答案可能正好相反：还没有任何人收集，里面的业务步骤便不会运行；同一个流对象被收集两次，步骤还可能执行两遍。要给这条进度链路做决策，先看**生产时刻、执行上下文和异常由谁接住**。

版本依旧是 Kotlin Gradle Plugin/stdlib 2.2.21、Gradle 8.14、JDK 21、language/API 2.2、jvmTarget 21、coroutines core/test 1.10.2（tag SHA 5f8900478a8e20c073145b1608fbc71fe3d7378b）。官方旧入口 https://kotlinlang.org/docs/flow.html 这次只返回 334 字节的 HTML 跳转页，因此实际依据是它指向的 Asynchronous Flow 正文、固定版本的 Flow、flowOn、buffer、catch API。下文的次数和线程名全部以实验为准，不拿滚动文档代替本地运行。

构造了流，究竟构造了什么

累计工程的最小进度入口只负责在收集时发出三个领域状态：

1
2
3
4
5
fun taskProgressFlow(id: TaskId, work: suspend () -> String): Flow<TaskState> = flow {
    emit(TaskState.Queued)
    emit(TaskState.Running)
    emit(TaskState.Succeeded(work()))
}

注意 work() 在 flow {} 中，而非返回 Flow 对象时调用。测试让 work 每次被调用都把计数加一并返回 "run-<次数>"。构造后计数是 0；第一次完整收集得到 Queued → Running → Succeeded("run-1")，计数是 1；第二次收集同一个对象得到 Succeeded("run-2")，计数是 2。XML 中记录 FLOW before_collect=0 first_collect=1 second_collect=2。这证明本例由 flow {} 建的流在每次收集时运行生产代码，不意味着所有 Flow 实现都具备冷特性；StateFlow/SharedFlow 的热状态与事件将在下一篇单独验收。

若把这个 work 换成“真正提交任务”的操作，两个不同订阅者可能触发两次提交；仅凭类型 Flow<TaskState> 不能充当幂等协议。当前实验的 TaskState.Succeeded 也是流中的一项合成值，没有调用 Store、没有持久化，更没有证明同一 HTTP 请求已经得到最终响应。通知订阅者、业务状态转移和对外提交必须由不同的 API 保持可追踪的完成点。

buffer：上游跑完不等于收集完成

定向测试发出整数 1、2，中间加 .buffer(1)。下游在收到第一项后完成“已进入”闸门，再等待测试线程放行；上游在发出第二项之后标记 upstreamFinished。测试确认下游第一项仍在等待时，上游已标记完成，而整条收集操作尚未完成；放闸后收集的顺序仍是 [1, 2]，输出 FLOW buffered_upstream_finished=true downstream_first_waiting=true collected=2。

一个槽位让上游和下游的进度可以交叠，不会魔法般把业务步骤变成“恰好处理一次”。该测试使用 runTest 调度器和受控闸门来确证先后关系，没有测量真实线程的吞吐率、无限容量队列的内存成本或网络慢订阅者的处理策略。与 30 的 Channel 完成点类似，上游 emit 返回也不保证下游业务回调已经结束；缓冲区满时上游会遇到背压，而关闭或取消处理还是订阅者的责任。

flowOn 换的是哪一侧的线程

第三组用两个真实单线程执行器分别命名 kotlin-31-upstream 与 kotlin-31-downstream。在下游执行器中进行收集，给 taskProgressFlow 上游加 .flowOn(upstreamDispatcher)；work() 记录自己的线程名，collect 在三个状态上各记录一次线程名。断言上游记录恰好一次且以 kotlin-31-upstream 开头，下游三个记录均以 kotlin-31-downstream 开头。JUnit 输出 REAL flowOn_upstream=kotlin-31-upstream downstream=kotlin-31-downstream ...。这条真实线程证据与前两个 runTest 虚拟调度实验分开保存。

为什么这里只比较线程名前缀？协程测试/调试环境可能给执行中的线程名附加 @coroutine#...；第 26 篇已有精确名字断言的失败记录。flowOn 调整其上游流的上下文，下游 collect 仍在调用方上下文；不能想当然在 flow {} 体里直接调用 withContext 来改变 emit 的执行上下文，也不能把 flowOn 当作自动限制业务阻塞 I/O 的手段。真实同步调用仍须按 26 的专用执行器、取消和资源协议单独验证。

catch 为什么没接住收集者抛的错

同一组测试再验证异常方向。上游 flow<Int> { emit(1); throw upstreamError } 后接 .catch { emit(-1) }，收集到 [1, -1]；这个 -1 是明确写出的替代数据，不说明原来的业务失败已被修复。另一条 flowOf(1).catch { emit(-1) }.collect { throw downstreamError } 则由下游收集 lambda 抛异常，外层能取得同消息异常，不会被位于收集之前的 catch 捕获。输出 upstream_caught=true downstream_escaped=true 对应两条分别核对的路径。

位置 此次结果 为什么不能说已完成业务任务
构造 flow {} 返回 work 执行 0 次 没有收集者，连合成处理都未发生
同一对象收集两遍 work 执行两次，得到 run-1/run-2 同一任务可能因二次收集重复产生副作用
带 buffer(1) 的第二项已 emit 上游完成，下游仍等第一项闸门 发出完成不是收集者完成
上游故障进入 catch [1,-1] 补救数据不等于事务已提交
下游 lambda 抛异常 外层收到异常 上游的 catch 不替下游确认成功

catch 也不应该把 CancellationException 当作普通业务失败随手换成成功通知；28 已单独演示 runCatching 会吞入取消。本篇只有指定上游 IllegalStateException 与指定下游异常路径，没测所有异常、重试或断电重放，不会把一条流的输出解释成完整的任务日志。

练习答案与验收

  1. 把 taskProgressFlow(id) { submitToStore() } 交给两个客户端收集，会自然去重吗？不会。 这次 flow {} 实测 0/1/2 次生产；去重应由独立的提交契约负责，不能假设返回 Flow 就是共享作业。
  2. .buffer(1) 上游已发送第二项但下游等在第一项时，能报告整条进度已被处理吗？不能。 upstreamFinished=true 与 collector.isCompleted=false 同时成立；要等收集者处理并明确任务副作用完成。
  3. catch 为何没捕获 collect { throw downstreamError }？它在该下游终端操作的上游，只处理向下流动的上游异常。 本次 [1,-1] 和下游抛异常有分别保存的测试输出，不能合并成“一处兜底”。

verification/31/ 保存 33 组累计 JUnit、真实线程与虚拟调度分账、官方实际跳转后的来源、运行命令/环境/退出码以及旧 CLI 正反例；同名素材 kotlin-language-31.tar.gz 是累计工程、Wrapper、RUN/VERSIONS 与证据快照，可按 README 解包复跑。下一篇换用热流分别验收最新状态与事件通知的遗漏/重放，而不是反推它们也会每次收集重做业务。