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

Self-describing KV cache events for the offloading connector.

The OffloadingManager identifies an offloaded chunk only by its OffloadKey, so its raw events carry no token ids, parent hash, or block size. :class:OffloadingEventsTracker snapshots each chunk's full BlockStored payload while the Request is alive and publishes stores as block-granular payloads: a chunk event may carry multiple constituent per-block hashes, and evictions fan out to the same hashes. Chunks overlapping a non-chunk-aligned shared prefix re-announce the shared hashes once per chunk; consumers are expected to deduplicate (reference-count) repeated store/remove announcements of the same hash. Opt-in via kv_connector_extra_config["self_describing_kv_events"]; inert unless KV cache events are enabled. See the PR description for the full design.

Recurrent (Mamba / linear-attention) state groups announce state checkpoints, not token ranges: an offloaded chunk holds the recurrent state at the chunk's last hash only. Their BlockStored therefore carries that single hash, its parent, an empty token_ids and kv_cache_spec_kind "mamba". A consumer aligns it by hash with the attention group's blocks, and a hybrid prefix is reusable only up to a boundary where every group has a residency. Windowed attention groups keep the legacy placeholder payload.

Classes:

• OffloadingEventsTracker –

  Tracks offloaded chunks' KV event payloads from store to eviction.

Functions:

• describes_group –

  Whether the tracker builds self-describing payloads for this group.

OffloadingEventsTracker

Tracks offloaded chunks' KV event payloads from store to eviction.

The scheduler calls :meth:record_store from _build_store_jobs and :meth:record_lookup for allocated primary-tier hits while the Request is available. Unallocated and missing lookups add no state. Entries remain until the final observed residency removal or :meth:reset.

Methods:

• record_lookup –

  Snapshot metadata for a ready or write-pending primary-tier hit.

• record_partial_lookup –

  Backfill metadata for a partial recurrent tail lookup hit.

• record_partial_store –

  Snapshot metadata for a newly stored partial recurrent tail.

• record_store –

  Snapshot the KV cache event payload for one offloaded chunk.

• reset –

  Drop all tracked state; pending payloads are stale after a

• take_events –

  Translate raw OffloadingEvents into self-describing KV events.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/offloading/events.py

class OffloadingEventsTracker:
    """Tracks offloaded chunks' KV event payloads from store to eviction.

    The scheduler calls :meth:`record_store` from ``_build_store_jobs`` and
    :meth:`record_lookup` for allocated primary-tier hits while the ``Request`` is
    available. Unallocated and missing lookups add no state. Entries remain until
    the final observed residency removal or :meth:`reset`.
    """

    def __init__(
        self,
        config: OffloadingKVEventsConfig,
        group_specs: dict[int, OffloadingEventGroupSpec] | None = None,
        num_hash_lookahead_tokens: int = 0,
    ):
        self.config = config
        self.num_hash_lookahead_tokens = num_hash_lookahead_tokens
        # Static classification survives resets and never asserts prefix residency.
        self._group_specs = group_specs or {}
        self.self_describing_enabled = (
            config.enable_kv_cache_events and config.self_describing_kv_events
        )

        # OffloadKey -> payload snapshot, kept until final removal or reset.
        self._pending_event_metadata: dict[OffloadKey, _OffloadEventMetadata] = {}
        # (medium, ownership) -> its bit in active_residencies.
        self._residency_bits: dict[tuple[Medium, str | None], int] = {
            (Medium.CPU, None): _PRIMARY_RESIDENCY_BIT
        }

    def record_store(
        self,
        req: Request,
        group_config: "GroupOffloadConfig",
        chunk_idx: int,
        offload_key: OffloadKey,
    ) -> None:
        """Snapshot the KV cache event payload for one offloaded chunk.

        No-op when self-describing event capture is disabled or for windowed
        attention groups, which keep the legacy placeholder payload.
        """
        if not self.self_describing_enabled:
            return
        if not describes_group(group_config):
            return
        meta = self._build_event_metadata(req, group_config, chunk_idx)
        if existing := self._pending_event_metadata.get(offload_key):
            meta.active_residencies |= existing.active_residencies
        self._pending_event_metadata[offload_key] = meta

    def record_lookup(
        self,
        req: Request,
        group_config: "GroupOffloadConfig",
        chunk_idx: int,
        offload_key: OffloadKey,
    ) -> None:
        """Snapshot metadata for a ready or write-pending primary-tier hit."""
        if not self.self_describing_enabled:
            return
        if not describes_group(group_config):
            return
        meta = self._pending_event_metadata.get(offload_key)
        if meta is None:
            meta = self._build_event_metadata(req, group_config, chunk_idx)
            self._pending_event_metadata[offload_key] = meta
        # Primary residency is registered here rather than by the manager's
        # prepare_write(): the payload needs the Request, which only the
        # scheduler holds at lookup time, and the manager reaches this tracker
        # only through OffloadingEvents.
        meta.active_residencies |= _PRIMARY_RESIDENCY_BIT

    def record_partial_store(
        self,
        req: Request,
        group_config: "GroupOffloadConfig",
        boundary_tokens: int,
        offload_key: OffloadKey,
    ) -> None:
        """Snapshot metadata for a newly stored partial recurrent tail."""
        if not describes_group(group_config):
            return
        self._record_partial_tail(req, group_config, boundary_tokens, offload_key)

    def record_partial_lookup(
        self,
        req: Request,
        group_config: "GroupOffloadConfig",
        boundary_tokens: int,
        offload_key: OffloadKey,
    ) -> None:
        """Backfill metadata for a partial recurrent tail lookup hit."""
        if not describes_group(group_config):
            return
        if offload_key not in self._pending_event_metadata:
            self._record_partial_tail(req, group_config, boundary_tokens, offload_key)

    def _record_partial_tail(
        self,
        req: Request,
        group_config: "GroupOffloadConfig",
        boundary_tokens: int,
        offload_key: OffloadKey,
    ) -> None:
        """Build metadata for the valid prefix of one physical cache block.

        A partial recurrent tail ends on a hash boundary but before its
        physical cache block is full. The event describes only the valid
        hashes and tokens, not the unused remainder of that physical block.
        """
        if not self.self_describing_enabled:
            return

        tokens_per_hash = group_config.tokens_per_chunk // group_config.hashes_per_chunk
        # Subtract one so the boundary token itself cannot select the next
        # physical block when the boundary lies exactly on a block edge.
        chunk_start = (
            (boundary_tokens - 1) // group_config.tokens_per_chunk
        ) * group_config.tokens_per_chunk
        first_hash_idx = chunk_start // tokens_per_hash
        last_hash_idx = boundary_tokens // tokens_per_hash
        assert chunk_start < boundary_tokens
        assert boundary_tokens % tokens_per_hash == 0
        assert last_hash_idx <= len(req.block_hashes)

        # Unlike a complete chunk, a partial tail contains only the hashes
        # between its physical block start and its valid token boundary.
        maybe_block_hashes = req.block_hashes[first_hash_idx:last_hash_idx]
        block_hashes = tuple(
            block_hash for block_hash in maybe_block_hashes if block_hash is not None
        )
        assert block_hashes and len(block_hashes) == len(maybe_block_hashes)
        parent_block_hash = (
            req.block_hashes[first_hash_idx - 1] if first_hash_idx > 0 else None
        )
        assert first_hash_idx == 0 or parent_block_hash is not None

        lora_id = req.lora_request.adapter_id if req.lora_request is not None else None
        lora_name = req.lora_request.name if req.lora_request is not None else None
        meta = _OffloadEventMetadata(
            block_hashes=block_hashes,
            parent_block_hash=parent_block_hash,
            token_ids=tuple(req.all_token_ids[chunk_start:boundary_tokens]),
            lookahead_token_ids=self._lookahead_tokens(
                req, chunk_start, boundary_tokens, tokens_per_hash
            ),
            block_size=tokens_per_hash,
            lora_id=lora_id,
            lora_name=lora_name,
            extra_keys=None,
            group_idx=group_config.group_idx,
            kv_cache_spec=group_config.kv_event_group_spec,
            active_residencies=_PRIMARY_RESIDENCY_BIT,
        )
        if _is_recurrent_group(group_config):
            meta = _as_state_checkpoint(meta)
        if existing := self._pending_event_metadata.get(offload_key):
            meta.active_residencies |= existing.active_residencies
        self._pending_event_metadata[offload_key] = meta

    def take_events(self, events: Iterable[OffloadingEvent]) -> Iterable[KVCacheEvent]:
        """Translate raw OffloadingEvents into self-describing KV events.

        Complete metadata is available for full-attention and recurrent
        state groups when the tracker is enabled. Windowed attention groups
        retain the legacy placeholder payload so consumers can ignore them.

        Yields:
            ``BlockStored`` or ``BlockRemoved`` events corresponding to
            the underlying :class:`OffloadingEvent` stream.

        """
        removed_keys: set[OffloadKey] = set()
        for event in events:
            if event.removed:
                if self.self_describing_enabled:
                    removed_keys.update(event.keys)
                yield from self._take_removed_event(event)
            else:
                yield from self._take_stored_event(event)

        # A primary removal can precede a queued secondary store in this batch.
        # Keep its payload until all stores have registered their residencies.
        for key in removed_keys:
            meta = self._pending_event_metadata.get(key)
            if meta is not None and not meta.active_residencies:
                self._pending_event_metadata.pop(key)

    def reset(self) -> None:
        """Drop all tracked state; pending payloads are stale after a
        manager cache reset."""
        self._pending_event_metadata.clear()

    def _lookahead_tokens(
        self, req: Request, start: int, end: int, block_size: int
    ) -> tuple[tuple[int, ...], ...] | None:
        length = self.num_hash_lookahead_tokens
        if not length:
            return None
        assert end + length <= req.num_tokens
        return tuple(
            tuple(req.all_token_ids[i : i + length])
            for i in range(start + block_size, end + 1, block_size)
        )

    def _build_event_metadata(
        self,
        req: Request,
        group_config: "GroupOffloadConfig",
        chunk_idx: int,
    ) -> _OffloadEventMetadata:
        """Build the payload snapshot for one offloaded chunk: its
        constituent per-block hashes, the whole chunk's tokens, and the
        per-block ``block_size``. A recurrent group's chunk is reduced to
        its state checkpoint (see :func:`_as_state_checkpoint`)."""
        hashes_per_chunk = group_config.hashes_per_chunk
        assert hashes_per_chunk > 0
        assert chunk_idx >= 0
        tokens_per_hash = group_config.tokens_per_chunk // hashes_per_chunk
        # Each chunk's final raw hash is its OffloadKey.
        first_hash_idx = chunk_idx * hashes_per_chunk
        last_hash_idx = first_hash_idx + hashes_per_chunk
        assert first_hash_idx >= 0
        assert last_hash_idx <= len(req.block_hashes)
        raw_chunk_hashes = req.block_hashes[first_hash_idx:last_hash_idx]
        chunk_hashes = resolve_block_hashes(
            raw_chunk_hashes,
            tokens_per_hash,
            group_config.tokens_per_block,
        )
        for block_hash in chunk_hashes:
            assert block_hash is not None
        assert len(chunk_hashes) == (
            group_config.tokens_per_chunk // group_config.tokens_per_block
        )

        if not describes_group(group_config):
            # The recording methods filter these out before calling this helper.
            raise AssertionError(
                "self-describing events only support full-attention and "
                "recurrent state groups"
            )

        parent_block_hash: BlockHash | None
        if first_hash_idx == 0:
            parent_block_hash = None
        else:
            parent_block_hash = req.block_hashes[first_hash_idx - 1]
            assert parent_block_hash is not None

        tok_start = chunk_idx * group_config.tokens_per_chunk
        tok_end = tok_start + group_config.tokens_per_chunk
        assert tok_end <= len(req.all_token_ids)
        # A recurrent chunk's payload carries no tokens; skip the copy.
        token_ids: tuple[int, ...] = (
            ()
            if _is_recurrent_group(group_config)
            else tuple(req.all_token_ids[tok_start:tok_end])
        )

        lora_id: int | None = None
        lora_name: str | None = None
        if req.lora_request is not None:
            lora_id = req.lora_request.adapter_id
            lora_name = req.lora_request.name

        meta = _OffloadEventMetadata(
            block_hashes=tuple(chunk_hashes),
            parent_block_hash=parent_block_hash,
            token_ids=token_ids,
            lookahead_token_ids=self._lookahead_tokens(
                req, tok_start, tok_end, group_config.tokens_per_block
            ),
            block_size=group_config.tokens_per_block,
            lora_id=lora_id,
            lora_name=lora_name,
            extra_keys=None,
            group_idx=group_config.group_idx,
            kv_cache_spec=group_config.kv_event_group_spec,
            active_residencies=_PRIMARY_RESIDENCY_BIT,
        )
        if _is_recurrent_group(group_config):
            meta = _as_state_checkpoint(meta)
        return meta

    def _placeholder_stored(
        self,
        key: OffloadKey,
        medium: Medium,
        locality: str | None,
        ownership: str | None,
    ) -> BlockStored:
        group_idx = get_offload_group_idx(key)
        # Static classification is part of the self-describing contract only;
        # legacy placeholders keep carrying just group_idx and the hash.
        spec = (
            self._group_specs.get(group_idx, OffloadingEventGroupSpec(None, None))
            if self.self_describing_enabled
            else OffloadingEventGroupSpec(None, None)
        )
        return BlockStored(
            block_hashes=[
                maybe_convert_block_hash(BlockHash(get_offload_block_hash(key)))
            ],
            parent_block_hash=None,
            token_ids=[],
            lora_id=None,
            block_size=0,
            medium=_MEDIUM_TO_EVENT_STR[medium],
            lora_name=None,
            group_idx=group_idx,
            kv_cache_spec_kind=spec.kv_cache_spec_kind,
            kv_cache_spec_sliding_window=spec.kv_cache_spec_sliding_window,
            locality=locality,
            ownership=ownership,
        )

    def _take_stored_event(self, event: OffloadingEvent) -> Iterable[KVCacheEvent]:
        # Metadata is read, NOT popped: the entry must survive until the
        # eviction event so BlockRemoved can fan out to the same hashes.
        # Events are self-contained (own parent), so key order is free.
        locality = event.locality.value if event.locality is not None else None
        residency_bit = 0
        if event.removal_expected:
            residency_bit = self._residency_bits.setdefault(
                (event.medium, event.ownership), 1 << len(self._residency_bits)
            )
        for key in event.keys:
            meta = self._pending_event_metadata.get(key)
            if meta is None:
                if self.self_describing_enabled:
                    # Expected for unsupported shapes; warn once only.
                    logger.warning_once(
                        "OffloadingEventsTracker: no event metadata for "
                        "offload key during BlockStored emission; emitting a "
                        "placeholder payload. Expected for windowed attention "
                        "groups and promotions not observed as a primary-tier "
                        "hit before translation."
                    )
                yield self._placeholder_stored(
                    key, event.medium, locality, event.ownership
                )
                continue

            meta.active_residencies |= residency_bit
            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),
                lookahead_token_ids=(
                    [list(tokens) for tokens in meta.lookahead_token_ids]
                    if meta.lookahead_token_ids is not None
                    else None
                ),
                block_size=meta.block_size,
                lora_id=meta.lora_id,
                medium=_MEDIUM_TO_EVENT_STR[event.medium],
                lora_name=meta.lora_name,
                extra_keys=to_event_extra_keys(meta.extra_keys),
                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]:
        # Keep group_idx unambiguous if a manager batch spans groups.
        locality = event.locality.value if event.locality is not None else None
        by_group: dict[int, list] = {}
        residency_mask = ~self._residency_bits.get((event.medium, event.ownership), 0)
        for key in event.keys:
            meta = self._pending_event_metadata.get(key)
            if event.metadata_only:
                if meta is not None:
                    meta.active_residencies &= residency_mask
                continue
            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
                )
                meta.active_residencies &= residency_mask
            else:
                if self.self_describing_enabled:
                    logger.warning_once(
                        "OffloadingEventsTracker: no event metadata for "
                        "offload key during BlockRemoved emission; emitting a "
                        "placeholder removal. Expected if the matching store "
                        "used the legacy placeholder payload; otherwise "
                        "indicates missing store metadata."
                    )
                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,
            )

_build_event_metadata(req, group_config, chunk_idx)

Build the payload snapshot for one offloaded chunk: its constituent per-block hashes, the whole chunk's tokens, and the per-block block_size. A recurrent group's chunk is reduced to its state checkpoint (see :func:_as_state_checkpoint).

Source code in vllm/distributed/kv_transfer/kv_connector/v1/offloading/events.py

def _build_event_metadata(
    self,
    req: Request,
    group_config: "GroupOffloadConfig",
    chunk_idx: int,
) -> _OffloadEventMetadata:
    """Build the payload snapshot for one offloaded chunk: its
    constituent per-block hashes, the whole chunk's tokens, and the
    per-block ``block_size``. A recurrent group's chunk is reduced to
    its state checkpoint (see :func:`_as_state_checkpoint`)."""
    hashes_per_chunk = group_config.hashes_per_chunk
    assert hashes_per_chunk > 0
    assert chunk_idx >= 0
    tokens_per_hash = group_config.tokens_per_chunk // hashes_per_chunk
    # Each chunk's final raw hash is its OffloadKey.
    first_hash_idx = chunk_idx * hashes_per_chunk
    last_hash_idx = first_hash_idx + hashes_per_chunk
    assert first_hash_idx >= 0
    assert last_hash_idx <= len(req.block_hashes)
    raw_chunk_hashes = req.block_hashes[first_hash_idx:last_hash_idx]
    chunk_hashes = resolve_block_hashes(
        raw_chunk_hashes,
        tokens_per_hash,
        group_config.tokens_per_block,
    )
    for block_hash in chunk_hashes:
        assert block_hash is not None
    assert len(chunk_hashes) == (
        group_config.tokens_per_chunk // group_config.tokens_per_block
    )

    if not describes_group(group_config):
        # The recording methods filter these out before calling this helper.
        raise AssertionError(
            "self-describing events only support full-attention and "
            "recurrent state groups"
        )

    parent_block_hash: BlockHash | None
    if first_hash_idx == 0:
        parent_block_hash = None
    else:
        parent_block_hash = req.block_hashes[first_hash_idx - 1]
        assert parent_block_hash is not None

    tok_start = chunk_idx * group_config.tokens_per_chunk
    tok_end = tok_start + group_config.tokens_per_chunk
    assert tok_end <= len(req.all_token_ids)
    # A recurrent chunk's payload carries no tokens; skip the copy.
    token_ids: tuple[int, ...] = (
        ()
        if _is_recurrent_group(group_config)
        else tuple(req.all_token_ids[tok_start:tok_end])
    )

    lora_id: int | None = None
    lora_name: str | None = None
    if req.lora_request is not None:
        lora_id = req.lora_request.adapter_id
        lora_name = req.lora_request.name

    meta = _OffloadEventMetadata(
        block_hashes=tuple(chunk_hashes),
        parent_block_hash=parent_block_hash,
        token_ids=token_ids,
        lookahead_token_ids=self._lookahead_tokens(
            req, tok_start, tok_end, group_config.tokens_per_block
        ),
        block_size=group_config.tokens_per_block,
        lora_id=lora_id,
        lora_name=lora_name,
        extra_keys=None,
        group_idx=group_config.group_idx,
        kv_cache_spec=group_config.kv_event_group_spec,
        active_residencies=_PRIMARY_RESIDENCY_BIT,
    )
    if _is_recurrent_group(group_config):
        meta = _as_state_checkpoint(meta)
    return meta

_record_partial_tail(req, group_config, boundary_tokens, offload_key)

Build metadata for the valid prefix of one physical cache block.

A partial recurrent tail ends on a hash boundary but before its physical cache block is full. The event describes only the valid hashes and tokens, not the unused remainder of that physical block.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/offloading/events.py

def _record_partial_tail(
    self,
    req: Request,
    group_config: "GroupOffloadConfig",
    boundary_tokens: int,
    offload_key: OffloadKey,
) -> None:
    """Build metadata for the valid prefix of one physical cache block.

    A partial recurrent tail ends on a hash boundary but before its
    physical cache block is full. The event describes only the valid
    hashes and tokens, not the unused remainder of that physical block.
    """
    if not self.self_describing_enabled:
        return

    tokens_per_hash = group_config.tokens_per_chunk // group_config.hashes_per_chunk
    # Subtract one so the boundary token itself cannot select the next
    # physical block when the boundary lies exactly on a block edge.
    chunk_start = (
        (boundary_tokens - 1) // group_config.tokens_per_chunk
    ) * group_config.tokens_per_chunk
    first_hash_idx = chunk_start // tokens_per_hash
    last_hash_idx = boundary_tokens // tokens_per_hash
    assert chunk_start < boundary_tokens
    assert boundary_tokens % tokens_per_hash == 0
    assert last_hash_idx <= len(req.block_hashes)

    # Unlike a complete chunk, a partial tail contains only the hashes
    # between its physical block start and its valid token boundary.
    maybe_block_hashes = req.block_hashes[first_hash_idx:last_hash_idx]
    block_hashes = tuple(
        block_hash for block_hash in maybe_block_hashes if block_hash is not None
    )
    assert block_hashes and len(block_hashes) == len(maybe_block_hashes)
    parent_block_hash = (
        req.block_hashes[first_hash_idx - 1] if first_hash_idx > 0 else None
    )
    assert first_hash_idx == 0 or parent_block_hash is not None

    lora_id = req.lora_request.adapter_id if req.lora_request is not None else None
    lora_name = req.lora_request.name if req.lora_request is not None else None
    meta = _OffloadEventMetadata(
        block_hashes=block_hashes,
        parent_block_hash=parent_block_hash,
        token_ids=tuple(req.all_token_ids[chunk_start:boundary_tokens]),
        lookahead_token_ids=self._lookahead_tokens(
            req, chunk_start, boundary_tokens, tokens_per_hash
        ),
        block_size=tokens_per_hash,
        lora_id=lora_id,
        lora_name=lora_name,
        extra_keys=None,
        group_idx=group_config.group_idx,
        kv_cache_spec=group_config.kv_event_group_spec,
        active_residencies=_PRIMARY_RESIDENCY_BIT,
    )
    if _is_recurrent_group(group_config):
        meta = _as_state_checkpoint(meta)
    if existing := self._pending_event_metadata.get(offload_key):
        meta.active_residencies |= existing.active_residencies
    self._pending_event_metadata[offload_key] = meta

record_lookup(req, group_config, chunk_idx, offload_key)

Snapshot metadata for a ready or write-pending primary-tier hit.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/offloading/events.py

def record_lookup(
    self,
    req: Request,
    group_config: "GroupOffloadConfig",
    chunk_idx: int,
    offload_key: OffloadKey,
) -> None:
    """Snapshot metadata for a ready or write-pending primary-tier hit."""
    if not self.self_describing_enabled:
        return
    if not describes_group(group_config):
        return
    meta = self._pending_event_metadata.get(offload_key)
    if meta is None:
        meta = self._build_event_metadata(req, group_config, chunk_idx)
        self._pending_event_metadata[offload_key] = meta
    # Primary residency is registered here rather than by the manager's
    # prepare_write(): the payload needs the Request, which only the
    # scheduler holds at lookup time, and the manager reaches this tracker
    # only through OffloadingEvents.
    meta.active_residencies |= _PRIMARY_RESIDENCY_BIT

record_partial_lookup(req, group_config, boundary_tokens, offload_key)

Backfill metadata for a partial recurrent tail lookup hit.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/offloading/events.py

def record_partial_lookup(
    self,
    req: Request,
    group_config: "GroupOffloadConfig",
    boundary_tokens: int,
    offload_key: OffloadKey,
) -> None:
    """Backfill metadata for a partial recurrent tail lookup hit."""
    if not describes_group(group_config):
        return
    if offload_key not in self._pending_event_metadata:
        self._record_partial_tail(req, group_config, boundary_tokens, offload_key)

record_partial_store(req, group_config, boundary_tokens, offload_key)

Snapshot metadata for a newly stored partial recurrent tail.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/offloading/events.py

def record_partial_store(
    self,
    req: Request,
    group_config: "GroupOffloadConfig",
    boundary_tokens: int,
    offload_key: OffloadKey,
) -> None:
    """Snapshot metadata for a newly stored partial recurrent tail."""
    if not describes_group(group_config):
        return
    self._record_partial_tail(req, group_config, boundary_tokens, offload_key)

record_store(req, group_config, chunk_idx, offload_key)

Snapshot the KV cache event payload for one offloaded chunk.

No-op when self-describing event capture is disabled or for windowed attention groups, which keep the legacy placeholder payload.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/offloading/events.py

def record_store(
    self,
    req: Request,
    group_config: "GroupOffloadConfig",
    chunk_idx: int,
    offload_key: OffloadKey,
) -> None:
    """Snapshot the KV cache event payload for one offloaded chunk.

    No-op when self-describing event capture is disabled or for windowed
    attention groups, which keep the legacy placeholder payload.
    """
    if not self.self_describing_enabled:
        return
    if not describes_group(group_config):
        return
    meta = self._build_event_metadata(req, group_config, chunk_idx)
    if existing := self._pending_event_metadata.get(offload_key):
        meta.active_residencies |= existing.active_residencies
    self._pending_event_metadata[offload_key] = meta

reset()

Drop all tracked state; pending payloads are stale after a manager cache reset.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/offloading/events.py

def reset(self) -> None:
    """Drop all tracked state; pending payloads are stale after a
    manager cache reset."""
    self._pending_event_metadata.clear()

take_events(events)

Translate raw OffloadingEvents into self-describing KV events.

Complete metadata is available for full-attention and recurrent state groups when the tracker is enabled. Windowed attention groups retain the legacy placeholder payload so consumers can ignore them.

Yields:

• Iterable[KVCacheEvent] –

  BlockStored or BlockRemoved events corresponding to

• Iterable[KVCacheEvent] –

  the underlying :class:OffloadingEvent stream.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/offloading/events.py

def take_events(self, events: Iterable[OffloadingEvent]) -> Iterable[KVCacheEvent]:
    """Translate raw OffloadingEvents into self-describing KV events.

    Complete metadata is available for full-attention and recurrent
    state groups when the tracker is enabled. Windowed attention groups
    retain the legacy placeholder payload so consumers can ignore them.

    Yields:
        ``BlockStored`` or ``BlockRemoved`` events corresponding to
        the underlying :class:`OffloadingEvent` stream.

    """
    removed_keys: set[OffloadKey] = set()
    for event in events:
        if event.removed:
            if self.self_describing_enabled:
                removed_keys.update(event.keys)
            yield from self._take_removed_event(event)
        else:
            yield from self._take_stored_event(event)

    # A primary removal can precede a queued secondary store in this batch.
    # Keep its payload until all stores have registered their residencies.
    for key in removed_keys:
        meta = self._pending_event_metadata.get(key)
        if meta is not None and not meta.active_residencies:
            self._pending_event_metadata.pop(key)

_OffloadEventMetadata

Bases: Struct

BlockStored payload snapshot for one OffloadKey, captured while the Request is available and kept until the final matching removal event. medium and ownership are forwarded from the OffloadingEvent.

One is kept per offloaded chunk, so it is untracked by the cyclic GC: keep fields free of tracked containers (active_residencies is a bitmask).

Source code in vllm/distributed/kv_transfer/kv_connector/v1/offloading/events.py

class _OffloadEventMetadata(
    msgspec.Struct,
    gc=False,  # type: ignore[call-arg]
):
    """BlockStored payload snapshot for one OffloadKey, captured while the
    Request is available and kept until the final matching removal event.
    ``medium`` and ``ownership`` are forwarded from the OffloadingEvent.

    One is kept per offloaded chunk, so it is untracked by the cyclic GC: keep
    fields free of tracked containers (``active_residencies`` is a bitmask)."""

    # The chunk's constituent block hashes; the last one is the OffloadKey.
    block_hashes: tuple[BlockHash, ...]
    parent_block_hash: BlockHash | None
    token_ids: tuple[int, ...]
    block_size: int
    lora_id: int | None
    lora_name: str | None
    # Deferred: needs the same incremental curr_mm_idx handling as GPU events.
    extra_keys: tuple[tuple[Any, ...] | None, ...] | None
    group_idx: int
    kv_cache_spec: OffloadingEventGroupSpec
    active_residencies: int
    lookahead_token_ids: tuple[tuple[int, ...], ...] | None = None

_as_state_checkpoint(meta)

Reduce a chunk payload to the recurrent state checkpoint it holds.

A recurrent group's offloaded chunk is one state, valid at the chunk's last hash. Announcing the whole hash range would let a consumer treat every hash in it as a resumable position, so the payload keeps only the last hash (the OffloadKey), chained to the hash before it, and no tokens: the consumer aligns it with the attention group's blocks by hash.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/offloading/events.py

def _as_state_checkpoint(meta: _OffloadEventMetadata) -> _OffloadEventMetadata:
    """Reduce a chunk payload to the recurrent state checkpoint it holds.

    A recurrent group's offloaded chunk is one state, valid at the chunk's
    last hash. Announcing the whole hash range would let a consumer treat
    every hash in it as a resumable position, so the payload keeps only the
    last hash (the OffloadKey), chained to the hash before it, and no tokens:
    the consumer aligns it with the attention group's blocks by hash.
    """
    assert meta.block_hashes
    if len(meta.block_hashes) > 1:
        meta.parent_block_hash = meta.block_hashes[-2]
    meta.block_hashes = meta.block_hashes[-1:]
    meta.token_ids = ()
    return meta

describes_group(group_config)

Whether the tracker builds self-describing payloads for this group.

Full-attention groups and recurrent state groups (MambaSpec) do; windowed attention groups (sliding window, chunked local) keep the legacy placeholder because a window's hashes do not identify a reusable prefix.

Source code in vllm/distributed/kv_transfer/kv_connector/v1/offloading/events.py

def describes_group(group_config: "GroupOffloadConfig") -> bool:
    """Whether the tracker builds self-describing payloads for this group.

    Full-attention groups and recurrent state groups (``MambaSpec``) do;
    windowed attention groups (sliding window, chunked local) keep the legacy
    placeholder because a window's hashes do not identify a reusable prefix.
    """
    if group_config.sliding_window_size_in_chunks is None:
        return True
    return _is_recurrent_group(group_config)