执行摘要

• 一句话：KV 卸载事件新增 ownership 字段以区分事件来源
• 推荐动作：值得快速精读，尤其是 kv_events.py 与 offloading/events.py 的配套改动：它展示了在 msgspec Struct 事件协议上做向前兼容扩展的完整模式（默认值字段 + hash 纳入 + 生成路径透传 + 测试断言），也展示了该项目对代码自解释性的 review 要求。建议结合 #52068（事件元数据保留到最终驱逐）与 #48281（ownership 元数据讨论源头）一起阅读，能更完整理解 KV offload 事件体系的演进脉络。

功能与动机

PR body 明确说明目的：Add optional ownership information to KV cache events so events emitted by a KV secondary tier can be distinguished from framework-owned events。即 KV 事件流需要区分框架自身（primary tier）与二级卸载 tier 发布的事件，避免下游消费者把二级 tier 的存储/移除行为误当成框架主路径行为。Change72 在评论中确认这是 #48281 中讨论的 ownership 元数据的后续落地；orozery 则要求代码内把 ownership 的语义与取值说明清楚，使代码自解释，作者随后补充了 docstring 并指出 #53624 是更好的参考示例。

实现拆解

实现分为四步，全部围绕事件协议字段扩展与透传：

1. 事件协议扩展（vllm/distributed/kv_events.py）：在 BlockStored 与 BlockRemoved 两个 msgspec Struct 上新增 ownership: str | None 字段，docstring 说明为 “Secondary offloading tier identifier, if generated by one”，None 表示框架 primary tier；同时把两个类的 hash 纳入 ownership，使 KVEventAggregator 基于 Counter 的聚合能够区分不同来源的同一 block 事件。

2. 事件源携带字段（vllm/v1/kv_offload/base.py）：给 OffloadingEvent dataclass 增加 ownership: str | None 字段（默认 None），作为调度器传入 OffloadingEventsTracker 的原始事件属性，是透传链路的源头。

3. 事件生成路径透传（vllm/distributed/kv_transfer/kv_connector/v1/offloading/events.py）：OffloadingEventsTracker 的三条事件生成路径全部透传 event.ownership —— 正常 BlockStored、无元数据时的 placeholder BlockStored（_placeholder_stored 新增 ownership 参数）、以及 BlockRemoved；_OffloadEventMetadata 的 docstring 同步更新为 medium 与 ownership 均由 OffloadingEvent 转发。

4. 测试配套（tests/v1/kv_connector/unit/offloading_connector/test_events.py）：_stored_event 与 _removed_event 辅助函数新增 ownership 参数；test_secondary_stored_event_does_not_mutate_cpu_metadata 断言透传值且不污染元数据缓存；test_take_events_groups_removed_hashes_by_kv_group 断言按 group 分组时 ownership 原样保留。

无部署或配置改动，纯事件协议与内部透传变更。

关键文件：

• vllm/distributed/kv_events.py（模块 事件协议；类别 source；类型 core-logic；符号 BlockStored, BlockRemoved）: 事件协议核心文件：BlockStored 与 BlockRemoved 新增 ownership 字段并纳入 hash，是本次变更的协议层落点，直接影响 KVEventAggregator 的事件去重与下游消费者可辨识性。
• vllm/distributed/kv_transfer/kv_connector/v1/offloading/events.py（模块 事件追踪；类别 source；类型 core-logic；符号 OffloadingEventsTracker, _placeholder_stored, _take_stored_event, _take_removed_event）: 事件生成与透传的核心实现：OffloadingEventsTracker 在正常 BlockStored、占位 BlockStored、BlockRemoved 三条路径上把 OffloadingEvent.ownership 原样透传到 KVCacheEvent，是本次功能真正生效的代码路径。
• vllm/v1/kv_offload/base.py（模块 卸载管理；类别 source；类型 data-contract；符号 OffloadingEvent）: 透传链路的源头：OffloadingEvent dataclass 新增 ownership 字段，是调度器事件进入连接器的数据契约扩展点。
• tests/v1/kv_connector/unit/offloading_connector/test_events.py（模块 单元测试；类别 test；类型 test-coverage；符号 _stored_event, _removed_event, test_secondary_stored_event_does_not_mutate_cpu_metadata, test_take_events_groups_removed_hashes_by_kv_group）: 配套单元测试：验证 ownership 在 store/removed 事件中的透传与按 group 分组时的保留，是本次协议扩展的回归保障。

关键符号：_placeholder_stored, _take_stored_event, _take_removed_event, BlockStored.hash, BlockRemoved.hash

关键源码片段

vllm/distributed/kv_events.py

事件协议核心文件：BlockStored 与 BlockRemoved 新增 ownership 字段并纳入 hash，是本次变更的协议层落点，直接影响 KVEventAggregator 的事件去重与下游消费者可辨识性。

class BlockStored(KVCacheEvent):
    # 存储事件：携带完整 block hash 与 KV cache spec 元数据，
    # 便于外部消费者重建 block hash 并对学习到的 group 分类、过滤。
    block_hashes: list[ExternalBlockHash]
    parent_block_hash: ExternalBlockHash | None
    token_ids: list[int]
    block_size: int
​
    """Deprecated: use `lora_name` for KV block key hash.
    Retained for backward compatibility.
    """
    lora_id: int | None
​
    medium: str | None
    lora_name: str | None
​
    """Extra keys used in block hash computation, one entry per block in
    block_hashes. Each entry contains MM identifiers, LoRA name, cache_salt,
    prompt embedding hashes, etc. for that specific block. Exposed for external
    KV cache consumers to reconstruct block hashes.
    """
    extra_keys: list[tuple[Any, ...] | None] | None = None
​
    """Store events carry cache-spec metadata so consumers can classify and
    filter groups as they are learned. Remove events only need group_idx+hash.
    """
    group_idx: int | None = None
    kv_cache_spec_kind: str | None = None
    kv_cache_spec_sliding_window: int | None = None
​
    """LOCAL or REMOTE relative to the publisher; None means unspecified."""
    locality: str | None = None
​
    # PR #52067 新增：标识发布该事件的二级卸载 tier；None 表示框架 primary tier。
    ownership: str | None = None
​
    def __hash__(self) -> int:
        # ownership 纳入 hash：不同来源的同一 block 在 KVEventAggregator 中
        # 会被视为不同事件，从而允许按发布方分别统计或过滤。
        return hash(
            (
                tuple(self.block_hashes),
                self.parent_block_hash,
                tuple(self.token_ids),
                self.block_size,
                self.lora_id,
                self.medium,
                tuple(self.extra_keys) if self.extra_keys else None,
                self.group_idx,
                self.kv_cache_spec_kind,
                self.kv_cache_spec_sliding_window,
                self.locality,
                self.ownership,
            )
        )

vllm/distributed/kv_transfer/kv_connector/v1/offloading/events.py

事件生成与透传的核心实现：OffloadingEventsTracker 在正常 BlockStored、占位 BlockStored、BlockRemoved 三条路径上把 OffloadingEvent.ownership 原样透传到 KVCacheEvent，是本次功能真正生效的代码路径。

def _take_stored_event(self, event: OffloadingEvent) -> Iterable[KVCacheEvent]:
    # 元数据只读、不弹出：条目必须存活到 eviction 事件，
    # 这样 BlockRemoved 才能扇出到相同的 hashes。事件自带 parent，key 顺序自由。
    locality = event.locality.value if event.locality is not None else None
    for key in event.keys:
        meta = self._pending_event_metadata.get(key)
        if meta is None:
            if self.self_describing_enabled:
                # 非 full-attention 分组或 promotion 未被 primary-tier 观察到时
                # 属于预期情况，只告警一次并发出占位事件。
                logger.warning_once(
                    "OffloadingEventsTracker: no event metadata for "
                    "offload key during BlockStored emission; emitting a "
                    "placeholder payload."
                )
            # 占位事件同样透传 ownership，保证二级 tier 事件可辨识。
            yield self._placeholder_stored(
                key, event.medium, locality, event.ownership
            )
            continue
​
        yield BlockStored(
            block_hashes=list(
                maybe_convert_block_hash(h) for h in meta.block_hashes
            ),
            parent_block_hash=(
                maybe_convert_block_hash(meta.parent_block_hash)
                if meta.parent_block_hash is not None
                else None
            ),
            token_ids=list(meta.token_ids),
            block_size=meta.block_size,
            lora_id=meta.lora_id,
            medium=_MEDIUM_TO_EVENT_STR[event.medium],
            lora_name=meta.lora_name,
            extra_keys=(
                list(meta.extra_keys) if meta.extra_keys is not None else None
            ),
            group_idx=meta.group_idx,
            kv_cache_spec_kind=meta.kv_cache_spec.kv_cache_spec_kind,
            kv_cache_spec_sliding_window=(
                meta.kv_cache_spec.kv_cache_spec_sliding_window
            ),
            locality=locality,
            ownership=event.ownership, # 透传事件来源
        )
​
​
def _take_removed_event(self, event: OffloadingEvent) -> Iterable[KVCacheEvent]:
    # 按 kv group 分组，避免 manager batch 跨组时 group_idx 歧义；元数据此刻弹出。
    locality = event.locality.value if event.locality is not None else None
    by_group: dict[int, list] = {}
    for key in event.keys:
        meta = self._pending_event_metadata.pop(key, None)
        if meta is not None:
            group_idx = meta.group_idx
            by_group.setdefault(group_idx, []).extend(
                maybe_convert_block_hash(h) for h in meta.block_hashes
            )
        else:
            if self.self_describing_enabled:
                logger.warning_once(
                    "OffloadingEventsTracker: no event metadata for "
                    "offload key during BlockRemoved emission; emitting a "
                    "placeholder removal."
                )
            group_idx = get_offload_group_idx(key)
            by_group.setdefault(group_idx, []).append(
                maybe_convert_block_hash(BlockHash(get_offload_block_hash(key)))
            )
​
    for group_idx, hashes in by_group.items():
        yield BlockRemoved(
            block_hashes=hashes,
            medium=_MEDIUM_TO_EVENT_STR[event.medium],
            group_idx=group_idx,
            locality=locality,
            ownership=event.ownership, # 移除事件同样透传来源
        )

评论区精华

Review 讨论集中在代码自解释性与注释风格：

• orozery 指出仅看代码无法理解 ownership 的含义与可能取值：just looking at the code it is not clear what "ownership" means and what are its possible values. Only #48281 has some context, though I still think is lacking. Can we add some comments to the code to make this self-explaining？作者随后在 BlockStored/BlockRemoved 上补充了 docstring，并回应 #53624 是更好的示例。
• 注释风格约定：orozery 要求把 # 注释统一改为 """ docstring 并放在字段上方（同时修正 locality 的注释位置），作者起初迟疑（group_idx 等旧字段用的是 # 且注释在成员下方），最终接受统一约定，并连带调整了 BlockStored 全部字段的注释顺序。

• 作者在提交历史中通过多次 “Comment reordering” commit 落实该约定，最终 orozery 回复 LGTM. Thanks @mkhazraee！并 approve。

• Ownership 字段语义不清晰，需补充注释 (documentation): 在 BlockStored / BlockRemoved 上补充 docstring：ownership 是二级卸载 tier 标识，None 表示 primary tier。

• 字段注释风格统一（docstring 与位置） (style): 统一为 docstring 风格、注释位于字段上方；BlockStored 字段注释顺序全部调整。

风险与影响

• 风险：
  1. 事件 hash 语义变化：ownership 被纳入 BlockStored/BlockRemoved 的 hash，KVEventAggregator 依赖 Counter 统计“所有 worker 都发出的事件才返回”。若同一逻辑 block 在不同 worker 上被标记为不同 ownership，聚合将被拆分；但 ownership 是 tier 级标识，全局一致，实际风险低。
  2. 序列化兼容性：KVEventBatch 使用 msgspec.Struct 且 omit_defaults=True，新字段默认 None，旧版本消费者可正常反序列化，属于向后兼容扩展；但任何外部 KV cache 消费者若直接用旧的完整字段列表做严格校验，可能遇到新字段。
  3. 覆盖不完整：GPU 主路径的事件发出方未设置 ownership（保持 None），当前只有 offloading 连接器路径透传，若后续需要区分多个二级 tier 之间的来源，还需各 tier 分别填充。
  4. 测试面较窄：仅有单元测试覆盖透传，未覆盖 KVEventAggregator 对 ownership 不同事件的聚合行为，建议后续补充。

• 影响：影响范围集中在 KV 卸载事件子系统（vllm/distributed、vllm/v1/kv_offload）：

  • 对下游/外部 KV cache 消费者：事件流新增来源标识，可据此过滤或路由 secondary tier 的事件，这是后续多 tier KV cache 拓扑中事件治理的基础。
  • 对框架内部：GPU 主路径事件不变（ownership 为 None），offloading 路径事件自动携带来源，行为可观测性提升。
  • 对团队：确立了一个事件协议向前兼容扩展的套路 —— 新增带默认值字段、纳入 hash、在生成路径透传、同步单元测试断言，后续类似元数据扩展可直接复用。
    影响程度：低到中，协议消费者需感知新字段，但无破坏性变更。

• 风险标记：事件协议字段扩展, hash 语义变更, 聚合器去重行为受影响, 测试覆盖较窄

关联脉络

• PR #52068 [KV Offload] Preserve KV event metadata until final residency removal: 与本次 PR 改动完全相同的三个文件（offloading/events.py、test_events.py、vllm/v1/kv_offload/base.py），是同一功能线的前置改动：先保证事件元数据存活到最终驱逐，再透传 ownership。
• PR #52912 [Bugfix][KVOffload] P2P tier declares REQUEST_LEVEL on the producer leg: 同属 KV offload tiering 子系统，涉及 P2P tier 的供给声明，与 ownership 机制共同服务于多 tier KV 卸载拓扑的正确区分与调度。