planP4E4approved2026-09-19

R8-Fix-7：Realtime Consumer 有界 Burst 与 Device Conditions 作用域 / Closeout 修复方案

> 状态：implemented · gate-open / E2 completed / E4 pending > > 本方案是 R8-Fix-6 > 之后，针对 Build 68 两轮真机 E4 实测数据的定向修复批次。 > 它不是 FPS / FrameTime 的重新设计：Display 指标公式、Source Owner、 > FrameLifetime 白名单与解码器、MetricRegistry、CapturePlan、CoreProfile 策略、 > Pmd3 pump 公平性（32 events / 1ms）、ActivityTrace filter 公平性（64 messages / 1ms）、 > Swift 10s 终止预算、Python 8s 关闭预算、consumer drain 0.5s 窗口、 > 以及 Device Condition 20s 采样间隔全部保持不变。 > 本批次只做三件事——让 realtime Raw consumer 在有 backlog 时按有界 burst 排空、 > 让 Device Conditions 的启动由 canonical 指标作用域决定、 > 让 20s 空闲等待在 Stop 时立即退出。 > > 证据等级声明：本批次的代码与自动化测试验证为 E2；方案设计目标是为 > 下一次真机 E4 提供可验证的行为，因此在真机 E4 执行前，本批次 > 不得记为 validated，也不得自建发布包。本批次不构建 Build 69。

1. 背景：Build 68 真机 E4 的两个事实

Build 68 跑了两场真机 E4，一场全指标、一场严格收窄到纯 Display。两场都必须分开读。

1.1 好消息（R8-Fix-6 的分类修复已生效）

Build 67 那组自相矛盾的关闭诊断（dataCompleteness=complete 却被判 consumerDrainStatus=timed-out / overallCloseoutStatus=partial，而 closeoutFailureStage/Reason 均为 null）在 Build 68 中不再以"辅助 sleeper 被算进 consumer drain"的形态出现：超时被明确记在一个真正的 Source Consumer 上，而不是 sleep 750ms/1000ms 的辅助任务上。也就是说 R8-Fix-6 的 Part B（source_consumer_tasks / auxiliary_tasks 分离、 auxiliary-stop 登记为 advisory 阶段）在真机上按设计工作， 分类不再是问题。

1.2 问题 A：realtime Raw consumer 每事件 `await asyncio.sleep(0)`

全指标场（Session 544ea082…）应为 6 个 source consumer，实测 pumpQueueWait P95 = 463.987ms、> 500ms 计数 = 27。

把作用域收窄到纯 Display 后（Session 0e9da98e…，本应为 1 个 source consumer，实际 2 个 —— 见问题 B），pumpQueueWait 改善为 P50 = 123.471ms / P95 = 282.982ms / Max = 491.312ms、> 500ms 计数 = 0。

多源竞争被削掉之后仍然残留 100–500ms 的 queue wait，这说明剩下的嫌疑在 consumer 自身。读源码可以确认这一点：

# 旧行为：每处理完一条 Raw 就无条件让出事件循环
async for raw in source:
    normalized, derived_frames = await append_raw(raw)
    ...
    await asyncio.sleep(0)   # <- 每一条！

关键在于 Pmd3Subscription.__anext__ 的消费契约：它在有界队列仍有 backlog 时是同步返回的（get_nowait）。因此"每条都 sleep(0)"并不会让 consumer 更公平，只会把排空速度硬性锁死为每个调度 tick 一条。事件循环被其它任务占用时，一条 backlog 的排空就会被切成 N × 调度延迟，这正是 100–500ms queue wait 的来源，且它与"纯 Display 只有一个源"无关。

1.3 问题 B：纯 Display 也启动了 Device Conditions

Build 68 的纯 Display Session（0e9da98e…）读到：

execution metric ids          = 5 项 Display
sourceConsumerTaskCount       = 2               <- 期望 1
adapters                      = [... , pmd3.device-condition.raw]   <- 多出来的那个

Device Conditions 是一个 20s 周期的 diagnostics_relay 任务，并且在既有实现中它被注册为一个真正的 Raw Source Consumer（不是辅助任务）。于是用户的指标选择里一个 battery/thermal 指标都没有，却仍然拉起了一条 20s 的采集 cadence。

更糟的是它与关闭路径的相互作用：正常 Stop 时，这条 consumer 正处在 asyncio.sleep(20) 的空闲等待里，500ms 的 consumer drain 窗口理所当然覆盖不到它：

dataCompleteness          = complete
ingressUndrained          = 0
consumerDrainStatus       = timed-out     (500.853ms)
overallCloseoutStatus     = partial

再一次是"没有任何数据丢失，却被判 partial"。 但这一次的根因不是分类缺陷（R8-Fix-6 已经把分类修对了），而是：

作用域缺陷：不该启动的源被启动了；
停止语义缺陷：20s 的空闲等待被当成了"consumer 拒绝完成最后一次入队"。

关键判断：修法不是放宽 drain 窗口。把 0.5s 放大只会掩盖"不该启动的源" 并让真正停摆的 consumer 多等；也不是把 Device Conditions 降级成辅助任务—— 它是一个真实的 Raw 源，降级会破坏"只有真 Source Consumer 才能翻转判决"这条 R8-Fix-6 刚建立的语义。正确修法是缩短它的空闲等待并不启动它。

2. 目标与非目标

2.1 目标

Part A（有界 burst）：把 realtime Raw consumer 的"每事件 sleep(0)"换成

有界 burst —— 只在同步工作量为零、且尚未达到任何上限时才让出。只对同步工作量计费，上游 await 永不入账。

Part A（可观测）：为每个源输出有界的 realtimeConsumerFairness 诊断

（maxBurstEvents / maxActiveBurstMs / yieldCount / maxObservedBurstCount / maxObservedActiveBurstMs / perSource）。

Part B（canonical 作用域）：Device Conditions 的启动必须由

canonical 的指标 → 源映射（execution_source_ids_for / scoped_feature_flags）决定，而不是由控件层的布尔量单独决定。纯 Display（5 项）不得启动它； sourceConsumerTaskCount 必须从 2 回到 1。

Part C（可停止的空闲等待）：iter_device_conditions 接受 stop_event，

20s 空闲等待在 Stop 时立即退出；在途的 get_battery() 采样必须照常完成。

2.2 非目标（明确禁止）

本批次不允许：

改任何指标公式（FPS / FrameTime / P95 / Pacing Jitter / Refresh Rate 一行不改）；
改 Source Owner；
改 FrameLifetime 白名单与解码器（含 GAP_REFRESH_MULTIPLE_THRESHOLD = 12）；
改 MetricRegistry / CapturePlan 的既有映射语义；
放大 subscription 队列（max_queue_size）；
放宽 consumer drain 0.5s 窗口；
把 Device Conditions 重新归类为 auxiliary / 辅助任务；
给 FrameTime 的缺失值补数；
构建 Build 69。

3. Part A：realtime Raw consumer 的有界 burst

3.1 只对同步工作量计费

新增 RealtimeConsumerFairness（一个 consumer 一个账本）， __slots__ 化、无锁、无分配：

fairness = consumer_fairness_registry.for_source(source_id)   # 每条流解析一次，不每事件
while True:
    async for raw in source:
        fairness.begin_item()        # <- 在 await 之后才开始计费
        ...                          # 同步解码 / 投影 / 入队
        if fairness.end_item():      # True = 本 burst 已用完，让出
            await asyncio.sleep(0)

begin_item() 在 __anext__ 返回之后才打时间戳，因此 "等待上游"这段（本来就已经把控制权交回事件循环）永远不会被算成 consumer 的 CPU 占用。源空闲时账本不累积，consumer 完全靠自然的 await 让出—— 不会产生自旋。

3.2 两个上限，先到先让

REALTIME_CONSUMER_MAX_BURST_EVENTS = 8
REALTIME_CONSUMER_MAX_ACTIVE_NS    = 1_000_000      # 1.0ms

end_item() 每事件只做一次 time.monotonic_ns() 与几次整数运算，命中任一上限即 _close_burst(reason) 并返回 True；reason ∈ event-limit / active-time，会计入 yieldReasonCounts。两个上限都不是新引入的"公平性时钟"，而是consumer 自己的 burst 预算——pump 的 32/1ms 与 filter 的 64/1ms 一行未改，两套账本互不干扰。

三处 await asyncio.sleep(0) 全部改为条件让出（Display 快路径、 CoreProfile 关闭期快路径、普通 append 路径），语义从"每事件"变为"每 burst"。

3.3 有界诊断

RealtimeConsumerFairnessRegistry 按源数量设上限（REALTIME_CONSUMER_FAIRNESS_SOURCE_CAP = 64），超出部分并入单个 __overflow__ 桶，因此内存是 O(源数量) 而非 O(事件数)。写入 manifest 的 realtimeConsumerFairness 形态：

{
  "maxBurstEvents": 8,
  "maxActiveBurstMs": 1.0,
  "yieldCount": 0,
  "processedEventCount": 0,
  "maxObservedBurstCount": 0,
  "maxObservedActiveBurstMs": 0.0,
  "sourceCount": 1,
  "overflowSourceCount": 0,
  "perSource": { "pmd3.activity-trace.frame-lifetime.raw": { "...": "..." } }
}

全部为整数 / 有界小数值，可安全进入每个 manifest。

4. Part B：Device Conditions 由 canonical 作用域驱动

capture_plan.py 新增一个问源契约而不是再写一份指标清单的判据：

DEVICE_CONDITION_SOURCE_ID = "pmd3.device-condition.raw"

def device_conditions_required(metric_ids) -> bool:
    return DEVICE_CONDITION_SOURCE_ID in execution_source_ids_for(metric_ids)

execution_source_ids_for 就是运行时激活路径本来就在用的那条 canonical 链路，因此"要不要启动条件源"与"激活哪些源"从此是同一个答案，不存在第二份会漂移的硬编码列表。

realtime.py 里原本三处 (include_device_conditions and device_conditions_enabled) 门禁（两处 recorder adapter 列表、一处 consumer 启动）统一收敛到一个闭包：

def device_conditions_active() -> bool:
    if not include_device_conditions or not device_conditions_enabled:
        return False
    return device_conditions_required(execution_metric_ids)

include_device_conditions 仍是显式 opt-out（fixtures / 测试用）、 device_conditions_enabled 仍是控制面请求，但两者都不能再"增加"一个请求的指标并不需要的源。纯 Display 的 5 项指标只映射到 pmd3.activity-trace.frame-lifetime.raw，因此条件源不启动， sourceConsumerTaskCount 回到 1。

5. Part C：可停止的 20s 空闲等待

async def wait_for_stop_or_interval(stop_event, timeout_seconds) -> bool:
    """True = 完整间隔已过（可以取下一个样本）；False = 先观察到 Stop（应立即退出）。"""

stop_event is None → 保持 legacy 的 await asyncio.sleep(timeout)（测试与 fixture 不受影响）；
否则 await asyncio.wait_for(stop_event.wait(), timeout=...)，

超时即 TimeoutError → True；先被 set → False。用 wait_for 是因为它在超时时会取消挂起的 Event.wait()， 不会留下孤立的 waiter，也不会产生自旋。

iter_device_conditions 新增 stop_event 参数，并在两处检查：进入循环前 if stop_event.is_set(): return，以及每次 yield 之后 if not await wait_for_stop_or_interval(...): return。

在途采样不被丢弃：await service.get_battery() 一旦开始就让它跑完并 yield，因为一条已经解码的样本是真实 Raw，它应当在正常的 drain 窗口内落到 recorder。 stop 被允许提前中止的，只有两次采样之间的空闲等待。

DEVICE_CONDITION_INTERVAL_SECONDS 仍是 20.0，一行未改；这条 consumer 仍是一个 verdict-bearing 的 Source Consumer，没有被降级为辅助任务。

6. 实施记录（Implementation Update）

实际修改范围（R8-Fix-7 自有文件）：

文件	改动
`perfowl_collector_core/capture_plan.py`	新增 `DEVICE_CONDITION_SOURCE_ID` 与 `device_conditions_required()`（走 `execution_source_ids_for`），加入 `__all__`
`perfowl_collector_core/device_conditions.py`	新增 `wait_for_stop_or_interval()`；`iter_device_conditions()` 新增 `stop_event` 参数与两处提前返回；间隔常量保持 20.0
`perfowl_collector_core/realtime.py`	新增 `REALTIME_CONSUMER_MAX_BURST_EVENTS` / `REALTIME_CONSUMER_MAX_ACTIVE_NS` / `REALTIME_CONSUMER_FAIRNESS_SOURCE_CAP` 与 yield reason 常量；新增 `RealtimeConsumerFairness` / `RealtimeConsumerFairnessRegistry`；三处 `sleep(0)` 改为 `if fairness.end_item()`；新增 `device_conditions_active()` 闭包并替换三处门禁；`consume_conditions()` 传入 `stop_event`；manifest 新增 `realtimeConsumerFairness`
`perfowl_collector_core/__init__.py`	导出 `wait_for_stop_or_interval`

相对原方案的偏差与原因：

RealtimeConsumerFairnessRegistry 按源数量设上限，而不是"不设限"。

方案最初只写了"一个源一个账本"；实现加了 64 的容量上限与单个 __overflow__ 桶。原因：源标识虽然事实上有限，但不应该被信任；上限保证 manifest 大小与内存占用在任何输入下都可证明有界，与本批次其余"有界性"设计一致。

device_conditions_active() 收敛为闭包，而不是在每个门禁点各自判断。

原因：三处门禁必须同时改变，分散判断迟早漂移；闭包让"同一个答案"这件事在代码结构上显式化，也让 session 级测试可以只验证一个入口。

**device_conditions_required() 在源契约不可用时返回 False，由运行时叠加

resource-independent 的 scoped_feature_flags 兜底。** 原因：源码树探针（无 resources）不应该让"全 Registry 选择"被静默剥掉 battery/thermal 源。两层判断各自职责清晰：契约负责"这个选择需要哪些源"，控件标志负责"控制面是否允许"。

未触碰的文件：共享工作区中其它并行批次的未提交改动

（pmd3_backend.py / pmd3_deep_services.py / display_performance.py / frame_derivation.py / metric_engine.py / online_report.py / recording.py / Swift 源与构建脚本 / 若干测试）未被本批次触碰。其中 pmd3_backend.py（pump 32/1ms）与 pmd3_deep_services.py（filter 64/1ms） 完全不在本批次 diff 中，这就是"公平性时钟未变"的结构性证明。

当前状态：implemented · gate-open；E2 completed / E4 pending。

7. 测试与验证（Validation Update · 2026-09-19 · E2）

新增两个测试文件、扩展现有一个：

tests/test_realtime_consumer_fairness.py（17 项）
Test A（快 backlog）：A1 轻量 backlog 按 burst 而非按事件让出；

A2 新路径排空快于每事件让出；A3 竞争任务不被 burst 饿死。

Test B（active 上限）：B1 单条 0.4ms 的 "heavy" 事件先撞 active-time 上限；

B2 active burst 永不超事件上限；B3 两个上限就是声明的 8 / 1ms。

Test C（空闲上游）：C1 上游暂停不被计入 active

（源 stall 时 maxObservedActiveBurstMs 保持平坦）； C2 空闲源不会在上游等待上产生额外忙等；C3 源停住时不产生自旋（用 stall_after + 两次快照对比证明空闲稳定）。

Test D：D1 心跳延迟未退化。
Test E（多源）：E1 每个并发源都取得进展；E2 registry 由源数量界定

（7 个不同源 id、cap=3 → sourceCount=4、overflowSourceCount=4）。

FairnessWindowRegression：pump 32/1ms 与 filter 64/1ms 未变。
DisplayPipelineBenchmark：三种形态（见下）。
tests/test_device_condition_scope.py（16 项）
契约层（5 项）：纯 Display 不是条件源选择；每个 battery/thermal 指标都选择该通道；

其它 device 指标不选择；全 Registry 选择仍然选择（兜底不被剥掉）；间隔常量仍为 20.0。

Test H（6 项）：H1 20s 空闲等待被 Stop 在 < 0.5s 内打断；

H2 在途样本不被丢弃；H3 无 Stop 时间隔正常走完； H4 Stop 已置位时立即返回；H5 不传 stop_event 时保持 legacy 行为； H6 不留下孤立的 waiter。

session 级（5 项）：F1 纯 Display 从不打开条件源、sourceConsumerTaskCount == 1、

consumerDrainStatus == completed、dataCompleteness == complete、closeoutStatus == clean； F2 manifest 中 realtimeConsumerFairness（maxBurstEvents=8 / maxActiveBurstMs=1.0 / sourceCount=1）与 displayPipelineGapDiagnostics 共存； G1 选中 battery 指标时仍然打开条件源； I1 条件 consumer 是 Source Consumer 而非 auxiliary（两者同时选中时计数为 2）； I2 Stop 被传播进条件迭代器。

tests/test_closeout_task_classification.py（+4 项，共 15 项）
Test J：J1 空闲 cadence 等待能在 drain 窗口内自行解开

（consumer_status = completed，不再假超时）； J2 最后一次入队真的被阻塞时仍是 STATUS_TIMED_OUT 且 verdict-bearing； J3 空闲与阻塞混合时仍正确上报 consumer 失败； J4 consumer 窗口仍是 0.5s，未被本批次改动。

benchmark（新路径 vs legacy 每事件让出）：

[形态 1] backlog 1500 frames + 0.2ms/turn 占用者（Build 68 纯 Display 形态复现）
  legacy/per-event-yield   552.8ms   2714 ev/s  handoffs=1500  p50=268.054 p95=500.537 p99=520.849 max=526.647 >500ms=77  drops=0 undrained=0
  r8fix7/bounded-burst      91.4ms  16410 ev/s  handoffs=187   p50= 35.206 p95= 60.718 p99= 62.274 max= 63.082 >500ms= 0  drops=0 undrained=0
  -> p95 改善 87.9%；总排空耗时 -83.5%；吞吐 x6.0；>500ms 从 77 归零

[形态 2] backlog 尾部 1500 frames + 0.5ms/turn 占用者
  legacy/per-event-yield  1037.2ms   1446 ev/s  handoffs=1500  p50=513.038 p95=941.562 p99=981.975 max=994.754 >500ms=770 drops=0 undrained=0
  r8fix7/bounded-burst     149.7ms  10020 ev/s  handoffs=187   p50= 61.738 p95= 99.613 p99=102.532 max=103.871 >500ms=  0 drops=0 undrained=0
  -> p95 改善 89.4%；总排空耗时 -85.6%；>500ms 从 770 归零

[形态 3] 稳态 120Hz（时钟压缩 120x），7200 frames
  legacy/per-event-yield   910.3ms   7909 ev/s  handoffs=7200  p50= 59.233 p95=110.497 p99=112.023 max=112.712 >500ms=0 drops=0 undrained=0
  r8fix7/bounded-burst    1125.1ms   6400 ev/s  handoffs=900   p50=  1.066 p95=  1.336 p99=  1.904 max=  3.974 >500ms=0 drops=0 undrained=0
  -> 交接延迟 p95 改善 98.8%；但总排空耗时 +23.6%（见下方诚实说明）

关于形态 3 的诚实说明：形态 3 把时钟压缩 120 倍，7200 帧因此近似同时可用， "稳态"实际上退化成 backlog 形态。此时新路径总的排空墙钟比 legacy 长 23.6%： legacy 每条都让出，pump 因此获得更多调度机会把事件推入队列；新路径每 8 条才让出， consumer 占住循环更久，最后一个事件反而更晚到达。这是一个明确的取舍：用总排空时间换交接延迟的两个数量级改善（p95 110.5ms → 1.3ms）与 >500ms 的彻底归零。在真实（未压缩）120Hz 会话中帧间隔为 8.33ms，两条路径都不会饱和，因此该取舍不影响产品形态——但这是推断，必须由真机 E4 确认，不得写成已验证结论。

验证结果：

Python 3.12 全量（隔离 worktree，独立于共享脏工作区）   703 / 703 OK   (57.181s)
  - test_realtime_consumer_fairness                       17 / 17 OK
  - test_device_condition_scope                            16 / 16 OK
  - test_closeout_task_classification                      15 / 15 OK
专注 E2 集合（11 个模块）                                  225 / 225 PASS
Swift 全量                                                见下方门禁
站点（知识门禁 + VitePress build）                          见下方门禁
git diff --check                                          PASS（无空白错误）

静态不变性（逐条核对）：本批次未修改

FPS / FrameTime / P95 / Pacing Jitter / Refresh Rate 公式；
ActivityTrace Source Owner；
FrameLifetime 白名单与解码器（含 GAP_REFRESH_MULTIPLE_THRESHOLD = 12）；
MetricRegistry / CapturePlan 的既有映射语义 / CoreProfile 策略；
Swift Core gracefulTimeout = 10s；
Python collector 8s 关闭预算；
Pmd3 pump fairness 32 events / 1ms（pmd3_backend.py 不在 diff 中）；
ActivityTrace filter fairness 64 messages / 1ms（pmd3_deep_services.py 不在 diff 中）；
consumer drain 窗口 SOURCE_CONSUMER_DRAIN_TIMEOUT_SECONDS = 0.5；
Device Condition 采样间隔 DEVICE_CONDITION_INTERVAL_SECONDS = 20.0；
RawEventEnvelope.SCHEMA（仍为 perfowl.raw-event/1）。

8. 对 R8-Fix-6 的验证结论

见 R8-Fix-6 方案文末追加的 Validation Update 增量段。

9. 遗留项与下一阶段

本批次不构建 Build 69、不执行真机 E4、不声明 validated。
下一次真机 E4 需要验证：

纯 Display 会话 sourceConsumerTaskCount == 1，

且 pmd3.device-condition.raw 不在 adapter 列表中；

纯 Display 会话 pumpQueueWait 的 P95/P99 明显低于 Build 68 的 282.982ms，

且 > 500ms 计数为 0；

正常 Stop 时 consumerDrainStatus = completed、overallCloseoutStatus = clean、

不再是 Build 68 那种 dataCompleteness = complete + timed-out 的自相矛盾组合；

选中 battery/thermal 指标时条件源仍然启动，20s 间隔保持，

Stop → 条件迭代器退出的实测延迟远小于 20s（E2 上为 0.299s）；

realtimeConsumerFairness.maxObservedBurstCount <= 8 且

maxObservedActiveBurstMs 约在 1ms 量级（证明只对同步工作量计费）；

真实 120Hz 会话（未压缩）中总排空时间没有可测量的退化

——形态 3 的取舍只能在真机上判定，不得在此之前写成结论。

关联文档：R8-Fix-6 及更早批次的结论不回改，只在文末追加带绝对日期的

update 增量段。