refactor(offload): let the shared scheduler name where a load starts

main's #2332 made ChunkedOffloadSchedulerBase set
seq.offload_load_start_tokens = hbm right before it creates each load
operation, so every chunked offload scheduler publishes a restored prefix
into the HBM prefix index. mooncake_store's _new_load_operation override
set the same value (seq.num_cached_tokens, which is that hbm) and is now
redundant. The dense-scheduler test no longer asserts that the attribute
is absent, since main sets it there on purpose.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: Jasen2201

atom/kv_transfer/offload/mooncake_store/scheduler.py

@@ -481,19 +481,6 @@ def _build_load_request(
             chunk_hashes=hashes[: (transfer_end // self.chunk_size) * DIGEST_BYTES],
         )
 
-    def _new_load_operation(self, seq: Any) -> LoadOperationId:
-        operation = super()._new_load_operation(seq)
-        # The worker restores KV straight into allocated blocks, which never
-        # pass through `hash_blocks()`. Naming where the load starts lets
-        # `Scheduler._mark_offload_load_ready` publish the loaded prefix into
-        # the HBM prefix index once the load succeeds, so a later turn hits it
-        # in HBM instead of loading it again; without it the suffix prefill
-        # finds its parent block unhashed and cannot register its own blocks
-        # either. Loads are dispatched only from a chunk-aligned post-allocate
-        # HBM frontier, which is always a hash-block boundary.
-        seq.offload_load_start_tokens = int(seq.num_cached_tokens)
-        return operation
-
     def save_abandon_timeout_s(self) -> float:
         """Seconds a deferred save may sit before the engine reclaims it.
 

tests/test_mooncake_store_offload.py

@@ -1740,7 +1740,6 @@ def test_dense_scheduler_load_request_is_unchanged(monkeypatch):
     assert request.token_ids == list(range(32))
     assert request.chunk_hashes is None
     assert request.block_ids == seq.block_table
-    assert not hasattr(seq, "offload_load_start_tokens")
 
 
 def test_scheduler_settings(cluster):