[KVConnector][Tiering] Drive the tiering control plane from a dedicated thread

[liranschour]

Purpose

Addresses #55179. The P2P secondary tier only advances its control plane from the
per-step scheduler hooks: TieringOffloadingManager.on_schedule_end polls each
tier for finished jobs and then lets each one serve inbound peer requests. The
p2p tier's transports are non-blocking by design (ZmqTransport.poll() returns
immediately with no traffic), so nothing moves between steps and a peer's
control-plane round trip waits for a step boundary.

This moves that sweep onto a dedicated thread owned by the tiering manager.

Honest summary of what this does and does not buy

It does not, on its own, produce a measurable latency win. I benchmarked it
loaded (numbers below) and fetch RTT is parity-within-noise. I am submitting it
because the mechanism it removes is real and because the measurement localises
the actual bottleneck, not because it makes the benchmark faster.

The residual wait appears to be on the producer side: the fetch is not
waiting to be answered, it is waiting for the producer's scheduler thread to
reach complete_store → tier.submit_store and park the blocks. This change
deliberately leaves _flush_pending_promotions and the cascade submit_store
calls on the scheduler thread, so that wait is untouched. Getting the win
probably requires moving the producer-side block parking off the step boundary
as well.

Relationship to the existing PRs on this issue (per AGENTS.md)

This overlaps open work, and I want that on the record rather than buried:

• [Core][KV Offload] Service the P2P control plane while the engine waits on the model #57165 (@nilig, "Service the P2P control plane while the engine waits on
  the model"). Its TieringOffloadingManager.poll_pending_work body is the same
  two calls in the same order as this PR's _sweep_control_plane. The fix is the
  same; only the driver differs. [Core][KV Offload] Service the P2P control plane while the engine waits on the model #57165 adds
  KVConnectorBase_V1.poll_pending_work and drives it from EngineCore during
  the window where the engine thread blocks on future.result(), handing the
  connector between threads under a lock. [Core][KV Offload] Service the P2P control plane while the engine waits on the model #57165 also bundles
  add_remote_peer_async.
• [Perf] Keep P2P control requests moving during model steps #55962 (@Etelis, "Keep P2P control requests moving during model steps") —
  same goal, also via EngineCore + the connector base API + executor changes.
• [KVConnector][P2P] Register peers off the scheduler thread #57139 (mine) covers the separate peer-registration half.

In #55179 @nilig asked for a read on "a thread touching the connector from
inside EngineCore, even phase-exclusive, versus a control thread owned by the
tier". This PR is a working implementation of the second option, offered as
input to that question:

	#57165 / #55962	this PR
Driver	EngineCore poller thread	thread owned by TieringOffloadingManager
API surface	new KVConnectorBase_V1.poll_pending_work	none
Files touched	17	1 source + 3 test
Exclusion	lock hands connector between engine phases	lock held for a whole scheduler step

If the maintainers prefer either existing PR, close this one. It is up as a
concrete alternative and as a carrier for the measurement, not a land grab.

Design: why the lock is step-scoped

Per-call locking is not sufficient. The connector learns a chunk is a HIT from
lookup() (under get_num_new_matched_tokens) and only reads it in a later
prepare_load() hook (under update_state_after_alloc). In between the chunk
still has ref_cnt == 0, i.e. it is evictable. A promotion started while serving
a peer lookup in that window can evict it via
prepare_write → CPUOffloadingManager.prepare_store, and prepare_load then
trips assert chunk is not None.

So the lock is taken on the first scheduler-side call of a step and released at
the end of on_schedule_end. That places the control thread's window over model
execution. The read-mostly hooks (has_pending_work, get_stats,
take_events) are the exception: they hold the lock only for their own duration,
because the engine ticks them on every iteration including ones that schedule
nothing and run no model, and folding them into the step-wide hold leaves an idle
engine with no gap at all — the state a producer waiting for a consumer sits in.

Tiers re-enter through _SecondaryTierFacingParent, whose caller already holds
the lock, so the four ParentManager methods gained *_unlocked variants.
exclude_tier_idx moved onto those, which restores the public signatures to
match OffloadingManager. A sweep that raises is logged and re-raised on the
next scheduler-side call, matching the engine crash these failures produced when
they ran on the scheduler thread rather than leaving the control plane silently
dead.

No interface changes, and no changes under tiering/p2p/.

Tests

Unit, on this branch (upstream main base):

venv/bin/python -m pytest tests/v1/kv_offload/tiering/ -q
  -> 387 passed, 12 skipped   (includes the 215 untouched p2p tests)

venv/bin/python -m pytest tests/v1/kv_connector/unit/offloading_connector/ -q
  -> 462 passed

Also run with the change cherry-picked onto the #57139 base for the live tests
below: tests/v1/kv_offload/tiering/ -> 387 passed, 11 skipped.

Ten new tests in tests/v1/kv_offload/tiering/test_tiering_offloading.py cover:
sweeping with no further steps, sweeps running off the scheduler thread, the step
locking out the control thread, on_schedule_end not double-serving, a failed
sweep re-raising on the scheduler thread, shutdown() joining, reset_cache()
with the thread live, idle-engine hooks not starving the thread, and the
disabled/no-tier cases.

Two of those were verified by mutation rather than assertion, since they are the
substantive claims: reverting lookup to per-call locking fails
test_step_locks_out_the_control_thread, and reverting _observation_lock to a
step-wide hold fails
test_idle_engine_hooks_do_not_block_the_control_thread.

Existing step-driven suites pass control_plane_thread=False to stay
deterministic; that flag is the only new knob and it is constructor-only.

Notes

• Model evaluation on accuracy-sensitive benchmarks was not run; the change does
  not alter what is computed, only which thread services the tier control plane.
  The correctness checks above are the serving-level evidence.
• AI assistance was used to write this change, its tests, and this
  description. Per AGENTS.md it stays a draft until I have reviewed every changed
  line and can defend it end to end.

🤖 Generated with Claude Code

[mergify]

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

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