[Refactor][KV Offloading] Unify fence field and move fence check after _build_store_jobs

[Alex-ai-future]

Unify Offloading Connector Fence Field and Check Timing

Summary

Two changes to the offloading connector's fence mechanism
(_block_id_to_pending_jobs):

1. Unify the fence field. Merge sliding_window_block_ids and
   non_sliding_window_block_ids (introduced in [kv_offload+HMA][12/N]: Scheduler-side support for sliding window groups #41228) into a single
   block_ids field, and register all source blocks at store creation time.
   Block type no longer affects registration timing.

2. Move the fence check after _build_store_jobs. The unified check
   sees both pre-existing and newly created jobs in one pass and produces
   the same flush outcomes as [Bugfix][KV Offloading] Offload last block at request finish and prevent reuse race #48596 (see Motivation).

Motivation

This PR simplifies #48596's fence mechanism. The flush behavior — which
jobs are flushed when their blocks are reallocated — is unchanged.

#48596's pattern was correct but split across two locations and two
fields. The main fence check in build_connector_meta ran before
_build_store_jobs (seeing only pre-existing in-flight jobs), and
_build_store_jobs did an inline registration inside its if req.is_finished()
branch for the finished-request deferred store's NSW blocks. Combined with
the SW zeroing in _update_req_states, every flush case was already
covered:

• Pre-existing in-flight jobs — flushed unconditionally by the main
  check (which ran before _build_store_jobs).
• Finished-request deferred stores whose NSW blocks were reallocated —
  registered and inline-flushed inside _build_store_jobs.
• SW blocks of any deferred store that were reallocated — zeroed by
  _update_req_states so they never entered the new job's src_block_ids,
  removing them from the flush decision entirely.

This PR expresses the same behavior through a single block_ids field
registered at store creation time and a single fence check run after
_build_store_jobs. The check distinguishes old jobs (flush
unconditionally) from new jobs (flush only for finished requests). The
two-stage split, the second fence field, and the request_finished
registration loop all disappear.

This is a preparatory simplification — subsequent store-direction refactors
will touch this area, and a single field plus a single check is a smaller
surface to evolve than the two-field, two-stage shape #48596 left behind.

Unified Fence Check

The check runs after _build_store_jobs and distinguishes old vs. new jobs.
For SW blocks of running requests that were evicted and reallocated, the
zeroing in _update_req_states keeps them out of the new job entirely, so
the fence check has nothing to do for them — same outcome as the old
main-check-before-build path, which would have flushed the old in-flight
SW jobs in _block_id_to_pending_jobs.

new_job_ids = set(store_jobs)
for bid in self._current_batch_allocated_block_ids:
    for jid in self._block_id_to_pending_jobs.get(bid, ()):
        if (
            jid not in new_job_ids
            or self._req_status[self._jobs[jid].req_id].req.is_finished()
        ):
            self._current_batch_jobs_to_flush.add(jid)

• Old jobs (pre-existing): flushed unconditionally when their blocks are
  reallocated - the blocks were already freed by completion or SW eviction.
• New jobs: flushed only for finished requests. A finished request's
  blocks cannot still belong to it, so any appearing in
  _current_batch_allocated_block_ids were reallocated to another request.
  Active requests' new stores are left alone (their blocks are protected by
  ref_cnt).

Changes

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

• Merged sliding_window_block_ids + non_sliding_window_block_ids into a
  single block_ids field on TransferJobStatus.
• Removed the per-block-type split in _build_store_jobs and the
  request_finished registration loop; all source blocks now register at
  store creation time.
• Moved the fence flush into build_connector_meta after _build_store_jobs.
• Made _remove_pending_job cleanup unconditional on job completion,
  mirroring the unified registration.

tests/v1/kv_connector/unit/offloading_connector/test_scheduler.py

The three existing fence tests are kept (mechanical block_ids rename only).
Two shared helpers (_make_finished_req_status, _drive_finished_store_job)
remove repeated finished-request setup. Three new tests (a fourth SW-
while-running candidate was removed during review as redundant with
existing coverage of in-flight + reallocation flush):

Test	What it guards
test_finished_store_job_flushes_on_partial_reallocation	Core test for the unified check. A deferred finished-request store with blocks (1, 2) and _current_batch_allocated_block_ids = {2} is flushed even though only one source block was reallocated.
test_finished_store_job_no_overlap_does_not_flush	Negative guard: no false-positive flush when none of the store's blocks were reallocated.
test_completed_store_removes_fence_entries	The now-unconditional cleanup path removes fence entries on store completion for a non-finished (still active) request.

reset_cache and preemption flush are already covered by existing
test_reset_cache and
test_async_preempt_readmit_before_transfer_output_is_deferred; isolated
duplicates are intentionally not added.

Test Plan

.venv/bin/python -m pytest tests/v1/kv_connector/unit/offloading_connector/ -v

157 passed, 2 skipped in 152.82s

Ruff check / format: pass.

Why This Is Not Duplicate Work

Builds on #48596 (merged 2026-07-17), which introduced the two-stage NSW
registration pattern (main check before _build_store_jobs + inline NSW
registration inside _build_store_jobs for finished-request deferred stores).
This PR unifies that pattern into a single field and a single fence check —
a structural simplification that preserves #48596's flush behavior, not
a fix to a gap and not a duplicate.

AI Assistance Statement

Developed with AI assistance for code generation, testing, and documentation.
Design decisions (unified fence field, unified vs. in-loop fence check, test
coverage) were made by the human author based on analysis of #41228 and #48596.

[claude]

Claude Code Review

This pull request is from a fork — automated review is disabled. A repository maintainer can comment @claude review to run a one-time review.

[Alex-ai-future]

@orozery This is a simplification refactor, PTAL!

[orozery]

Thanks @Alex-ai-future for this suggestion!
The reason that we introduced separate handling for sliding window vs non-sliding window blocks is that they have a different lifecycle.
Non sliding window blocks (full attention) or secured from eviction while the request is alive, whereas sliding window blocks are not.
This distinction allows us a more optimized fence checking, which we lose if we unify them.

[Alex-ai-future]

Thanks @Alex-ai-future for this suggestion! The reason that we introduced separate handling for sliding window vs non-sliding window blocks is that they have a different lifecycle. Non sliding window blocks (full attention) or secured from eviction while the request is alive, whereas sliding window blocks are not. This distinction allows us a more optimized fence checking, which we lose if we unify them.

@orozery Thank you for the explanation - I really didn't think about it from this long-term perspective. I proposed unification from the observation that "the current two implementations behave the same on fence inspection", and did not take into account the "optimization space that may depend on split in the future".

[mergify]

This pull request has merge conflicts that must be resolved before it can be
merged. Please rebase the PR, @Alex-ai-future.

https://docs.github.com/en/pull-requests/collaborating-with-pull-requests/working-with-forks/syncing-a-fork