sgl-router: experimental Rust HTTP router for SGLang worker pools (#25851)

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

experimental/sgl-router/tests/component/discovery/mod.rs

@@ -0,0 +1,4 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 The SGLang Authors
+// SPDX-License-Identifier: Apache-2.0
+
+mod static_urls;

experimental/sgl-router/tests/component/discovery/static_urls.rs

@@ -0,0 +1,169 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 The SGLang Authors
+// SPDX-License-Identifier: Apache-2.0
+
+use sgl_router::config::StaticUrlsDiscoveryConfig;
+use sgl_router::discovery::{DiscoveryEvent, WorkerMode};
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::mpsc;
+
+#[tokio::test]
+async fn emits_one_added_per_url_with_plain_seed() {
+    let cfg = StaticUrlsDiscoveryConfig {
+        urls: vec!["http://x:30000".into(), "http://y:30000".into()],
+    };
+    let (tx, mut rx) = mpsc::channel(16);
+    let _h = sgl_router::discovery::static_urls::spawn(cfg, tx)
+        .await
+        .unwrap();
+
+    let mut seen = std::collections::HashSet::new();
+    for _ in 0..2 {
+        let event = tokio::time::timeout(Duration::from_secs(2), rx.recv())
+            .await
+            .unwrap()
+            .unwrap();
+        match event {
+            DiscoveryEvent::Added(spec) => {
+                // mode / model_ids / bootstrap_port are seeded as Plain/empty/None;
+                // the worker manager fills them from /server_info post-discovery.
+                assert_eq!(spec.mode, WorkerMode::Plain);
+                assert!(spec.model_ids.is_empty());
+                assert_eq!(spec.bootstrap_port, None);
+                // The URL doubles as the worker id — strings already have to
+                // be unique (rejected at config-load otherwise).
+                assert_eq!(spec.id.0, spec.url);
+                seen.insert(spec.url);
+            }
+            other => panic!("unexpected event: {other:?}"),
+        }
+    }
+    assert_eq!(
+        seen,
+        ["http://x:30000".to_string(), "http://y:30000".to_string()].into(),
+    );
+}
+
+/// Single-URL list — the common dev deployment shape. The producer
+/// emits exactly one event and then parks until the receiver is
+/// dropped. Earlier versions exited as soon as fan-out completed,
+/// which tripped `server::supervisor::supervise_critical_tasks` →
+/// `mark_unready` → `/readyz` 503; the lib-side
+/// `stays_alive_after_fanout_until_receiver_dropped` pins that
+/// invariant in isolation, while this test pins the same contract
+/// through the public `spawn` entry point used by the binary.
+#[tokio::test]
+async fn emits_one_event_and_parks_until_receiver_dropped() {
+    let cfg = StaticUrlsDiscoveryConfig {
+        urls: vec!["http://x:30000".into()],
+    };
+    let (tx, mut rx) = mpsc::channel(16);
+    let h = sgl_router::discovery::static_urls::spawn(cfg, tx)
+        .await
+        .unwrap();
+    let event = rx.recv().await.unwrap();
+    assert!(matches!(event, DiscoveryEvent::Added(_)));
+    assert!(rx.try_recv().is_err(), "exactly one event expected");
+
+    // Drop the receiver → producer's `tx.closed()` resolves → task
+    // exits cleanly.
+    drop(rx);
+    tokio::time::timeout(Duration::from_secs(2), h)
+        .await
+        .expect("static_urls task should exit after receiver is dropped")
+        .expect("join handle should not panic");
+}
+
+/// Spin up a fake worker that advertises
+/// `disaggregation_mode = "prefill"` + `disaggregation_bootstrap_port`,
+/// pipe it through `spawn_discovery` (StaticUrls backend) into
+/// `manager::run_with_config`, and assert the worker lands in the
+/// registry with `WorkerMode::Prefill` + the disclosed port.
+///
+/// This is the load-bearing end-to-end assertion for the refactor's
+/// central claim — "prefill, decode, and plain workers can all appear
+/// in the same `urls` list and end up classified correctly" — exercised
+/// against the full discovery → introspect → registry pipeline rather
+/// than just the in-isolation `register_one` unit test.
+#[tokio::test]
+async fn static_urls_pd_role_resolved_end_to_end() {
+    use axum::{routing::get, Json, Router};
+    use serde_json::json;
+    use sgl_router::config::{
+        ActiveLoadConfig, Config, DiscoveryBackend, DiscoveryConfig, ObservabilityConfig,
+        ProxyConfig, ServerConfig,
+    };
+    use sgl_router::discovery::{spawn_discovery, WorkerId};
+    use sgl_router::workers::{manager, WorkerRegistry};
+    use tokio::net::TcpListener;
+    use tokio::sync::oneshot;
+
+    // Fake worker advertising a prefill role + bootstrap port.
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let port = listener.local_addr().unwrap().port();
+    let url = format!("http://127.0.0.1:{port}");
+    let app = Router::new().route(
+        "/server_info",
+        get(|| async {
+            Json(json!({
+                "served_model_name": "tiny",
+                "disaggregation_mode": "prefill",
+                "disaggregation_bootstrap_port": 8998,
+            }))
+        }),
+    );
+    let (shutdown_tx, shutdown_rx) = oneshot::channel::<()>();
+    tokio::spawn(async move {
+        let _ = axum::serve(listener, app)
+            .with_graceful_shutdown(async move {
+                let _ = shutdown_rx.await;
+            })
+            .await;
+    });
+
+    let cfg = Config {
+        server: ServerConfig {
+            host: "127.0.0.1".into(),
+            port: 0,
+        },
+        observability: ObservabilityConfig::default(),
+        models: vec![],
+        discovery: DiscoveryConfig {
+            backend: DiscoveryBackend::StaticUrls(StaticUrlsDiscoveryConfig {
+                urls: vec![url.clone()],
+            }),
+        },
+        proxy: ProxyConfig::default(),
+        active_load: ActiveLoadConfig::default(),
+    };
+
+    let registry = Arc::new(WorkerRegistry::default());
+    let (event_rx, _disc) = spawn_discovery(&cfg).await.unwrap();
+    let _mgr = tokio::spawn(manager::run_with_config(
+        event_rx,
+        registry.clone(),
+        Some(Arc::new(cfg)),
+        None,
+        None,
+    ));
+
+    let id = WorkerId(url);
+    let resolved = tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            if let Some(w) = registry.get(&id) {
+                if w.mode() == WorkerMode::Prefill && w.bootstrap_port() == Some(8998) {
+                    return true;
+                }
+            }
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        }
+    })
+    .await;
+    assert!(
+        resolved.is_ok(),
+        "expected mode=Prefill bootstrap_port=Some(8998); got {:?}",
+        registry.get(&id).map(|w| (w.mode(), w.bootstrap_port()))
+    );
+
+    let _ = shutdown_tx.send(());
+}

experimental/sgl-router/tests/component/health/circuit_breaker.rs

@@ -0,0 +1,163 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 The SGLang Authors
+// SPDX-License-Identifier: Apache-2.0
+
+use sgl_router::health::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig};
+use std::time::Duration;
+
+fn cb() -> CircuitBreaker {
+    CircuitBreaker::with_config(CircuitBreakerConfig {
+        threshold: std::num::NonZeroU32::new(3).unwrap(),
+        cool_down: Duration::from_millis(100),
+    })
+}
+
+#[test]
+fn starts_closed_and_allows() {
+    let b = cb();
+    assert!(b.allow());
+}
+
+#[test]
+fn three_failures_open_the_breaker() {
+    let b = cb();
+    b.record_failure();
+    b.record_failure();
+    assert!(b.allow(), "still closed before threshold");
+    b.record_failure();
+    assert!(!b.allow(), "open after threshold reached");
+}
+
+#[test]
+fn intermittent_success_resets_failure_count() {
+    let b = cb();
+    b.record_failure();
+    b.record_failure();
+    b.record_success(); // resets
+    b.record_failure();
+    b.record_failure();
+    assert!(b.allow(), "should still be closed (2 failures since reset)");
+}
+
+#[tokio::test(start_paused = true)]
+async fn open_breaker_recovers_via_half_open() {
+    let b = cb();
+    b.record_failure();
+    b.record_failure();
+    b.record_failure();
+    assert!(!b.allow());
+
+    // Wait past cool_down.
+    tokio::time::advance(Duration::from_millis(150)).await;
+
+    // Half-open: allow one probe.
+    assert!(b.allow(), "half-open allows the probe");
+    // While half-open, further allow() calls should reject (only one probe in flight).
+    assert!(!b.allow(), "half-open rejects second probe");
+
+    // Probe succeeded.
+    b.record_success();
+    assert!(b.allow(), "closed after successful probe");
+    assert!(b.allow(), "stays closed");
+}
+
+#[tokio::test(start_paused = true)]
+async fn half_open_failure_reopens() {
+    let b = cb();
+    b.record_failure();
+    b.record_failure();
+    b.record_failure();
+
+    tokio::time::advance(Duration::from_millis(150)).await;
+    assert!(b.allow(), "half-open admit");
+    b.record_failure();
+    // Back to Open.
+    assert!(!b.allow(), "back to open");
+}
+
+#[tokio::test(start_paused = true)]
+async fn would_allow_is_non_mutating_past_cool_down() {
+    // `would_allow()` answers "would `allow()` return true right now?" without
+    // claiming a probe slot. Enumeration / filtering paths (e.g.
+    // `WorkerRegistry::healthy_workers_for`) call it to inspect breakers
+    // without disturbing state.
+    let b = cb();
+    b.record_failure();
+    b.record_failure();
+    b.record_failure();
+    assert!(!b.allow(), "open after threshold");
+
+    tokio::time::advance(Duration::from_millis(150)).await;
+
+    // Repeated would_allow() returns true and leaves state untouched.
+    assert!(b.would_allow());
+    assert!(b.would_allow());
+    assert!(b.would_allow());
+
+    // The first allow() claims the half-open probe.
+    assert!(b.allow(), "allow() admits the probe");
+    // The probe is in flight — subsequent allow() (and would_allow()) reject.
+    assert!(!b.allow(), "only one probe in flight");
+    assert!(!b.would_allow(), "would_allow() agrees: no slot available");
+}
+
+#[tokio::test(start_paused = true)]
+async fn enumeration_then_dispatch_preserves_probe() {
+    // Regression for the bug where `healthy_workers_for` filtered with
+    // mutating `allow()`. Once would_allow() is the filter, an enumeration
+    // pass over many workers must not steal the probe slot from the one
+    // worker that actually gets dispatched to.
+    let b = cb();
+    b.record_failure();
+    b.record_failure();
+    b.record_failure();
+    tokio::time::advance(Duration::from_millis(150)).await;
+
+    // Imagine 3 workers; enumeration filters each with would_allow().
+    for _ in 0..3 {
+        assert!(b.would_allow(), "filter sees the worker as available");
+    }
+
+    // Now the policy picks ONE worker and dispatch claims the probe.
+    assert!(b.allow(), "dispatch on the picked worker succeeds");
+}
+
+#[test]
+fn would_allow_in_closed_state_is_true_and_non_mutating() {
+    let b = cb();
+    for _ in 0..5 {
+        assert!(b.would_allow());
+    }
+    // And allow() should still work afterwards.
+    assert!(b.allow());
+}
+
+#[tokio::test(start_paused = true)]
+async fn open_breaker_recovery_is_not_delayed_by_continued_failures() {
+    // Regression: previously, record_failure on an already-Open breaker
+    // refreshed opened_at, so a failure storm pinned the breaker open
+    // forever. Now the cool_down is measured from first-open.
+    let b = CircuitBreaker::with_config(CircuitBreakerConfig {
+        threshold: std::num::NonZeroU32::new(3).unwrap(),
+        cool_down: Duration::from_millis(100),
+    });
+    // Open it.
+    b.record_failure();
+    b.record_failure();
+    b.record_failure();
+    assert!(!b.allow(), "breaker should be open");
+
+    // Advance halfway through cool_down, then record more failures.
+    tokio::time::advance(Duration::from_millis(50)).await;
+    b.record_failure();
+    b.record_failure();
+    b.record_failure();
+
+    // Advance just past the original cool_down.
+    tokio::time::advance(Duration::from_millis(60)).await;
+
+    // We're past the original cool_down → HalfOpen.
+    assert!(
+        b.allow(),
+        "breaker should be half-open after cool_down from first-open"
+    );
+}

experimental/sgl-router/tests/component/health/mod.rs

@@ -0,0 +1,4 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 The SGLang Authors
+// SPDX-License-Identifier: Apache-2.0
+
+mod circuit_breaker;

experimental/sgl-router/tests/component/main.rs

@@ -0,0 +1,14 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 The SGLang Authors
+// SPDX-License-Identifier: Apache-2.0
+
+//! Component-scope integration tests.
+//!
+//! Each submodule exercises a single library component (policy, registry,
+//! discovery, health, tokenizer) via the crate's public API. None of these
+//! tests spin up the full HTTP router; for those see `tests/proxy/`.
+
+mod discovery;
+mod health;
+mod policies;
+mod tokenizer;
+mod workers;

experimental/sgl-router/tests/component/policies/cache_aware_zmq.rs

@@ -0,0 +1,182 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 The SGLang Authors
+// SPDX-License-Identifier: Apache-2.0
+
+//! E2E test for the cache-aware-zmq policy.
+//!
+//! Drives a real ZMQ PUB socket → `KvEventIndex` subscriber pipeline →
+//! `HashTree` → `CacheAwareZmqPolicy::select`. Verifies that an event
+//! published by one worker's PUB causes subsequent selection to route
+//! to that worker (cache-aware affinity).
+//!
+//! API constraint: the subscriber registry builds endpoints as
+//! `tcp://{host}:{port_base + dp_rank}` where `port_base` is in the
+//! per-worker `EventConfig`. Both mock workers below share
+//! `127.0.0.1` as host, so both subscribe to the same PUB socket and
+//! both end up indexed in the tree. The tiebreak (lowest active_load)
+//! picks the worker we want; same shape as the SMG version of this
+//! test.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use zeromq::SocketSend;
+
+use sgl_router::config::CacheAwareConfig;
+use sgl_router::config::{ActiveLoadConfig, ProxyConfig};
+
+use sgl_router::discovery::{ModelId, WorkerId, WorkerMode, WorkerSpec};
+use sgl_router::policies::cache_aware_zmq::CacheAwareZmqPolicy;
+use sgl_router::policies::kv_events::{compute_block_hashes, discovery::EventConfig, KvEventIndex};
+use sgl_router::policies::{Policy, SelectionContext};
+use sgl_router::tokenizer::TokenizerRegistry;
+use sgl_router::workers::Worker;
+
+use super::zmq_helpers::{
+    build_multipart, encode_block_stored_event, encode_event_batch, make_pub_bound,
+};
+
+fn build_worker(url: &str, model: &str) -> Arc<Worker> {
+    Arc::new(Worker::new(WorkerSpec {
+        id: WorkerId(url.into()),
+        url: url.into(),
+        mode: WorkerMode::Plain,
+        model_ids: vec![ModelId(model.into())],
+        bootstrap_port: None,
+    }))
+}
+
+/// E2E: real PUB socket publishes a `BlockStored` for worker A's
+/// hash chain. The `CacheAwareZmqPolicy`'s shared `KvEventIndex`
+/// receives it, applies it to the tree, and the next `select` call
+/// picks worker A.
+///
+/// Both workers share `127.0.0.1` as host so both subscribers connect
+/// to the same PUB and both get indexed under their KvWorkerIds — the
+/// same shape as the SMG e2e test. We tie-break on min-load: worker B
+/// is bumped above worker A so the matched-worker pick prefers A.
+#[tokio::test]
+async fn zmq_indexer_routes_to_publishing_worker_e2e() {
+    let model_id = ModelId("tiny".into());
+
+    // 1. Tokenizer registry — use the in-tree tiny fixture.
+    let cfg = sgl_router::config::Config {
+        server: sgl_router::config::ServerConfig {
+            host: "0".into(),
+            port: 0,
+        },
+        observability: Default::default(),
+        models: vec![sgl_router::config::ModelConfig {
+            id: "tiny".into(),
+            tokenizer_path: "tests/fixtures/tiny_tokenizer.json".into(),
+            policy: sgl_router::config::PolicyKind::CacheAwareZmq,
+            circuit_breaker: None,
+            cache_aware: None,
+        }],
+        discovery: sgl_router::config::DiscoveryConfig {
+            backend: sgl_router::config::DiscoveryBackend::StaticUrls(
+                sgl_router::config::StaticUrlsDiscoveryConfig {
+                    urls: vec!["http://placeholder:0".into()],
+                },
+            ),
+        },
+        proxy: ProxyConfig::default(),
+        active_load: ActiveLoadConfig::default(),
+    };
+    let tokenizers = Arc::new(TokenizerRegistry::load_from_config(&cfg).unwrap());
+
+    // 2. Bind a real PUB socket on an OS-assigned port.
+    let (mut pub_a, port) = make_pub_bound().await;
+
+    // 3. Compute the hash chain for the routing prompt.
+    let text = "hello world hello world hello world";
+    let tok = tokenizers.get("tiny").unwrap();
+    let token_ids = sgl_router::tokenizer::adapter::encode(&tok, text).unwrap();
+    let block_size = 4u32;
+    let hashes = compute_block_hashes(&token_ids, block_size as usize);
+    assert!(!hashes.is_empty(), "tiny tokenizer must yield ≥1 block");
+
+    // 4. Build the KvEventIndex + policy. The policy holds an
+    //    Arc<HashTree> that the index also owns; events the index
+    //    receives mutate the same tree the policy reads.
+    let kv_index = KvEventIndex::new();
+    // Mirror what `KvEventIndex::add_worker` would do in production: seed
+    // the oracle with the worker-reported page_size before any cache
+    // lookup happens. The integration path calls `add_worker` further
+    // down, but here we want the policy to know `block_size` immediately.
+    let block_size_oracle = kv_index.block_size_oracle();
+    block_size_oracle.try_set(block_size).unwrap();
+    let policy = CacheAwareZmqPolicy::new(
+        CacheAwareConfig {
+            cache_threshold: 0.0,
+            balance_abs_threshold: 32,
+            balance_rel_threshold: 1.1,
+        },
+        kv_index.tree(),
+        Arc::clone(&tokenizers),
+        block_size_oracle,
+    );
+
+    // 5. Register two workers. They share `127.0.0.1` so both
+    //    subscribers connect to the same PUB; preresolved EventConfig
+    //    points at the bound port.
+    let url_a = "http://127.0.0.1:30000";
+    let url_b = "http://127.0.0.1:30001";
+    let preresolved = EventConfig {
+        host: "127.0.0.1".to_string(),
+        port_base: port,
+        topic: String::new(),
+        block_size,
+        dp_size: 1,
+    };
+    kv_index.add_worker(url_a, Some(preresolved.clone())).await;
+    kv_index.add_worker(url_b, Some(preresolved)).await;
+
+    // SUB sockets take a moment to handshake. The polling loop below
+    // soaks up any extra latency; this is just a publish-before-SUB
+    // guard.
+    tokio::time::sleep(Duration::from_millis(150)).await;
+
+    // 6. Publish a BlockStored event for the routing prompt's chain.
+    let event_bytes = encode_block_stored_event(&hashes, None, &token_ids, block_size);
+    let payload = encode_event_batch(0.0, vec![event_bytes], Some(0));
+    pub_a
+        .send(build_multipart(1, payload))
+        .await
+        .expect("send block-stored event");
+
+    // 7. Bump worker B's load so the tie-break picks A among matched
+    //    workers. The bump stays below balance_abs_threshold so the
+    //    imbalance fast-path does not skip cache-aware selection.
+    //    Bind the guards to a Vec held for the rest of the test scope
+    //    so the counter stays > 0 through the polling loop.
+    let w_a = build_worker(url_a, "tiny");
+    let w_b = build_worker(url_b, "tiny");
+    let _b_load: Vec<_> = (0..3).map(|_| w_b.load_guard()).collect();
+    let workers = vec![Arc::clone(&w_a), Arc::clone(&w_b)];
+
+    // 8. Drive select until the event has been applied. The pipeline is
+    //    asynchronous (publish → SUB recv → mpsc → pump → tree); a
+    //    polling loop is less flaky than a fixed sleep.
+    let body = serde_json::to_vec(&serde_json::json!({"prompt": text})).unwrap();
+    let ctx = SelectionContext::new(&model_id, Some(&body));
+
+    let start = std::time::Instant::now();
+    let mut chose_a = false;
+    while start.elapsed() < Duration::from_secs(3) {
+        if let Some(w) = policy.select(&workers, &ctx) {
+            if w.url == url_a {
+                chose_a = true;
+                break;
+            }
+        }
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+    assert!(
+        chose_a,
+        "policy did not route to publishing worker A within timeout",
+    );
+
+    // 9. Shutdown cleanly.
+    let r = tokio::time::timeout(Duration::from_secs(2), kv_index.shutdown()).await;
+    assert!(r.is_ok(), "kv_index shutdown should not hang");
+}

experimental/sgl-router/tests/component/policies/kv_events_hash_parity.rs

@@ -0,0 +1,75 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 The SGLang Authors
+// SPDX-License-Identifier: Apache-2.0
+
+//! Cross-implementation parity test for the KV-event block-hash algorithm.
+//!
+//! The Rust implementation at `src/policies/kv_events/hash.rs` must produce
+//! the same i64 block hashes as SGLang's `radix_cache::RadixKey.hash_page`
+//! followed by `hash_str_to_int64`.  Hard-coded `cross_language_golden_*`
+//! values inside `hash.rs` are correct but brittle: if either side's
+//! algorithm changes, the comments don't get regenerated and the tests
+//! pass with stale expectations.
+//!
+//! This test consumes a fixture produced by
+//! `tests/scripts/generate_kv_events_hash_parity.py`, which replicates the
+//! SGLang algorithm verbatim (see the script's docstring for authority
+//! pointers). CI regenerates the fixture (see
+//! `.github/workflows/pr-test-sgl-router.yml`) and diffs against the
+//! committed file; this test asserts the Rust implementation matches
+//! whatever fixture is checked in.
+
+use serde::Deserialize;
+use sgl_router::policies::kv_events::compute_block_hashes;
+use std::path::PathBuf;
+
+#[derive(Debug, Deserialize)]
+struct ParityCase {
+    name: String,
+    tokens: Vec<u32>,
+    block_size: usize,
+    expected_i64_hashes: Vec<i64>,
+}
+
+fn fixture_path() -> PathBuf {
+    PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+        .join("tests")
+        .join("fixtures")
+        .join("kv_events_hash_parity.json")
+}
+
+fn load_cases() -> Vec<ParityCase> {
+    let path = fixture_path();
+    let bytes = std::fs::read(&path)
+        .unwrap_or_else(|e| panic!("read parity fixture {}: {e}", path.display()));
+    serde_json::from_slice(&bytes)
+        .unwrap_or_else(|e| panic!("decode parity fixture {}: {e}", path.display()))
+}
+
+#[test]
+fn fixture_is_non_empty() {
+    let cases = load_cases();
+    assert!(
+        !cases.is_empty(),
+        "kv_events_hash_parity.json is empty — run \
+         tests/scripts/generate_kv_events_hash_parity.py",
+    );
+}
+
+/// Drives every case in the fixture through `compute_block_hashes` and
+/// asserts equality with the Python-derived expectation.
+#[test]
+fn rust_block_hashes_match_python_radix_cache() {
+    for case in load_cases() {
+        // block_size of 0 is rejected by `compute_block_hashes` with a
+        // panic; the Python generator also rejects it.  The fixture
+        // doesn't include a 0 case, so unwrap is safe.
+        let block_size = std::num::NonZeroUsize::new(case.block_size)
+            .unwrap_or_else(|| panic!("case {} has block_size=0 which is invalid", case.name));
+        let got = compute_block_hashes(&case.tokens, block_size.get());
+        assert_eq!(
+            got, case.expected_i64_hashes,
+            "case {}: tokens={:?} block_size={} — Rust produced {:?}, fixture says {:?}",
+            case.name, case.tokens, case.block_size, got, case.expected_i64_hashes,
+        );
+    }
+}

experimental/sgl-router/tests/component/policies/kv_events_tree_concurrent.rs

@@ -0,0 +1,167 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 The SGLang Authors
+// SPDX-License-Identifier: Apache-2.0
+
+//! Concurrent-mutation stress test for `HashTree`.
+//!
+//! The 19 inline tests in `policies::kv_events::tree` are all
+//! single-threaded.  Under production load, multiple worker subscribers
+//! drive `insert` / `remove` / `clear_worker` against the same tree from
+//! tokio worker threads while the chat handler simultaneously calls
+//! `match_prefix` from many concurrent requests.
+//!
+//! The tree is documented as taking a write-lock for mutations and a
+//! read-lock for `match_prefix`; this test exercises that contract under
+//! heavy contention to catch:
+//!
+//!   * Deadlocks between the reverse index and the arena's RwLock.
+//!   * Logical races where a removed worker still appears in the reverse
+//!     index (or vice versa).
+//!   * Panics from a node arena being mutated mid-read.
+//!
+//! After the storm settles, the tree must be self-consistent: every
+//! worker that was fully cleared must be absent from every node's worker
+//! set, and `node_count()` must converge to zero.
+
+use std::sync::Arc;
+use std::thread;
+
+use sgl_router::policies::kv_events::{HashTree, KvWorkerId};
+
+fn worker(i: usize) -> KvWorkerId {
+    KvWorkerId {
+        url: format!("http://w{i}:30000"),
+        dp_rank: 0,
+    }
+}
+
+/// 8 mutator threads × 200 ops + 4 reader threads × 500 match queries.
+/// Each mutator inserts a chain, queries it, then clears the worker; the
+/// invariant is that after every thread joins, the tree is empty (every
+/// worker was cleared) and no thread panicked.
+#[test]
+fn tree_survives_concurrent_inserts_removes_and_matches() {
+    let tree = Arc::new(HashTree::new());
+
+    let mut handles = Vec::new();
+
+    for tid in 0..8 {
+        let tree = tree.clone();
+        handles.push(thread::spawn(move || {
+            let w = worker(tid);
+            for round in 0..200_u64 {
+                // Each round uses a fresh chain so different mutators
+                // don't trample each other's nodes — we want contention
+                // on the lock, not contention on the keys (those are
+                // covered by the single-threaded reinsert/remove tests).
+                let chain: Vec<i64> = (0..4)
+                    .map(|i| ((tid as i64) << 32) | ((round as i64) << 8) | i as i64)
+                    .collect();
+                tree.insert(&w, None, &chain);
+
+                let m = tree.match_prefix(None, &chain);
+                assert!(
+                    m.matched_blocks <= chain.len(),
+                    "match must never exceed query length",
+                );
+
+                // Half the rounds use remove(&chain); the rest use
+                // clear_worker — both must leave a consistent tree.
+                if round % 2 == 0 {
+                    tree.remove(&w, &chain);
+                } else {
+                    tree.clear_worker(&w);
+                }
+            }
+            // Final blanket clear in case the last iteration used `remove`
+            // on only part of the chain.
+            tree.clear_worker(&w);
+        }));
+    }
+
+    for tid in 0..4 {
+        let tree = tree.clone();
+        handles.push(thread::spawn(move || {
+            for round in 0..500_u64 {
+                let probe: Vec<i64> = (0..3)
+                    .map(|i| ((tid as i64) << 40) | ((round as i64) << 8) | i as i64)
+                    .collect();
+                // Readers must never block-walk and must never panic.
+                let _ = tree.match_prefix(None, &probe);
+            }
+        }));
+    }
+
+    for h in handles {
+        h.join()
+            .expect("worker thread panicked under concurrent load");
+    }
+
+    assert_eq!(
+        tree.node_count(),
+        0,
+        "tree must be empty after every worker was cleared; \
+         residual nodes indicate a missed clear_worker path",
+    );
+
+    // The arena and the reverse index must agree: zero non-root nodes
+    // means zero `by_hash` entries. A bug that prunes the arena but not
+    // the reverse index would leak memory and corrupt future inserts;
+    // this assertion turns that into an immediate test failure.
+    assert_eq!(
+        tree.reverse_index_size(),
+        0,
+        "by_hash reverse index must be empty when no non-root nodes remain",
+    );
+}
+
+/// A mutator races `clear_worker` against a reader that is mid-`match_prefix`
+/// on a deep chain.  The reader must never see a partially-mutated tree
+/// (no panic, no double-counted workers in the result set).
+#[test]
+fn match_prefix_is_consistent_with_concurrent_clear() {
+    let tree = Arc::new(HashTree::new());
+    let w = worker(0);
+    let chain: Vec<i64> = (0..32).map(|i| 1_000 + i).collect();
+
+    // Pre-populate so the reader has something to walk.
+    tree.insert(&w, None, &chain);
+
+    let stop = Arc::new(std::sync::atomic::AtomicBool::new(false));
+
+    let mutator = {
+        let tree = tree.clone();
+        let stop = stop.clone();
+        let w = w.clone();
+        let chain = chain.clone();
+        thread::spawn(move || {
+            let mut round = 0u64;
+            while !stop.load(std::sync::atomic::Ordering::Relaxed) {
+                if round.is_multiple_of(2) {
+                    tree.clear_worker(&w);
+                } else {
+                    tree.insert(&w, None, &chain);
+                }
+                round += 1;
+            }
+        })
+    };
+
+    for _ in 0..2_000 {
+        let m = tree.match_prefix(None, &chain);
+        // Either the worker was present (matched_blocks == chain.len(),
+        // workers set contains w) or it was cleared mid-walk (matched_blocks
+        // == 0 OR matched_blocks > 0 with empty workers if the chain is
+        // partially present).  Whichever — the result must be internally
+        // consistent.
+        if m.matched_blocks == chain.len() {
+            assert!(
+                m.workers.contains(&w),
+                "full match must include worker; got {:?}",
+                m.workers,
+            );
+        }
+    }
+
+    stop.store(true, std::sync::atomic::Ordering::Relaxed);
+    mutator.join().unwrap();
+}

experimental/sgl-router/tests/component/policies/kv_events_two_subscribers.rs

@@ -0,0 +1,305 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 The SGLang Authors
+// SPDX-License-Identifier: Apache-2.0
+
+//! Two independent `KvEventIndex` instances subscribed to the same PUB
+//! socket — the in-process surrogate for "two router replicas watching
+//! the same SGLang worker's KV publisher."
+//!
+//! Why this matters: sgl-router v1 explicitly omits multi-replica state
+//! sync (deferred to v2 in the slim-design spec). Independent ZMQ
+//! subscription is the **only** mechanism by which two routers arrive at
+//! a consistent cache-aware view today. If a future change accidentally
+//! degraded that property — e.g. a worker that only allows one subscriber,
+//! a switch from PUB/SUB to PUSH/PULL, or a teardown bug that drops
+//! events to one of N subscribers — this test fails loudly.
+//!
+//! Property pinned: after publishing N `BlockStored` events, both trees
+//! report the same `match_prefix(matched_blocks, workers)` for the
+//! published key, and an unpublished key remains absent from both.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use zeromq::SocketSend;
+
+use sgl_router::policies::kv_events::discovery::EventConfig;
+use sgl_router::policies::kv_events::{compute_block_hashes, KvEventIndex, KvWorkerId};
+
+use super::zmq_helpers::{
+    build_multipart, encode_block_stored_event, encode_event_batch, make_pub_bound,
+};
+
+#[tokio::test]
+async fn two_independent_subscribers_converge_to_same_tree_state() {
+    // 1. One PUB socket — the worker. Both router surrogates connect to it.
+    let (mut publisher, port) = make_pub_bound().await;
+    let worker_url = "http://127.0.0.1:30000";
+    let block_size = 4u32;
+    let cfg = EventConfig {
+        host: "127.0.0.1".into(),
+        port_base: port,
+        topic: String::new(),
+        block_size,
+        dp_size: 1,
+    };
+
+    // 2. Two independent router-process surrogates, each with its own
+    //    `KvEventIndex` (own tree, own subscriber, own pump task). Both
+    //    call `add_worker` with the same preresolved `EventConfig` — the
+    //    same shape production wires through `WorkerManager`.
+    let router_a = KvEventIndex::new();
+    let router_b = KvEventIndex::new();
+    router_a.add_worker(worker_url, Some(cfg.clone())).await;
+    router_b.add_worker(worker_url, Some(cfg.clone())).await;
+
+    // SUB-side handshake settle. Publishing before the subscribers
+    // finish their initial connect loses messages in PUB/SUB semantics;
+    // the polling loop below would then never converge.
+    tokio::time::sleep(Duration::from_millis(200)).await;
+
+    // 3. Publish a deterministic, multi-block event chain.
+    let tokens: Vec<u32> = (0..16).collect();
+    let hashes = compute_block_hashes(&tokens, block_size as usize);
+    assert!(
+        hashes.len() >= 3,
+        "test needs ≥3 blocks; got {}",
+        hashes.len()
+    );
+    let event_bytes = encode_block_stored_event(&hashes, None, &tokens, block_size);
+    let payload = encode_event_batch(0.0, vec![event_bytes], Some(0));
+    publisher
+        .send(build_multipart(1, payload))
+        .await
+        .expect("publish BlockStored");
+
+    // 4. Poll both trees until both report the FULL chain matched. The
+    //    SUB→mpsc→pump→tree pipeline is async; loopback delivery is
+    //    reliable but not instantaneous.
+    let target = hashes.len();
+    let key = KvWorkerId {
+        url: worker_url.into(),
+        dp_rank: 0,
+    };
+    let start = std::time::Instant::now();
+    loop {
+        let ma = router_a.tree().match_prefix(None, &hashes);
+        let mb = router_b.tree().match_prefix(None, &hashes);
+        let converged = ma.matched_blocks == target
+            && mb.matched_blocks == target
+            && ma.workers.contains(&key)
+            && mb.workers.contains(&key);
+        if converged {
+            // Both trees agree on count AND on the worker that holds the
+            // prefix. This is what the cache-aware-zmq policy reads to
+            // pick a worker; both routers picking the same key here
+            // means they would route the same prompt to the same worker.
+            assert_eq!(
+                ma.matched_blocks, mb.matched_blocks,
+                "subscribers disagreed on matched_blocks",
+            );
+            assert_eq!(
+                ma.workers, mb.workers,
+                "subscribers disagreed on worker set",
+            );
+            break;
+        }
+        if start.elapsed() > Duration::from_secs(3) {
+            panic!(
+                "subscribers did not converge within 3s: \
+                 router_a={{matched={}, workers={:?}}}, \
+                 router_b={{matched={}, workers={:?}}}, target={target}",
+                ma.matched_blocks, ma.workers, mb.matched_blocks, mb.workers,
+            );
+        }
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+
+    // 5. Negative leg: a key that was never published must not appear in
+    //    either tree. Guards against a future bug where one subscriber
+    //    accidentally inherits another's state (shared static, etc.).
+    let unseen: Vec<i64> = vec![999_999_999_001, 999_999_999_002, 999_999_999_003];
+    let na = router_a.tree().match_prefix(None, &unseen);
+    let nb = router_b.tree().match_prefix(None, &unseen);
+    assert_eq!(na.matched_blocks, 0, "router_a leaked unpublished key");
+    assert_eq!(nb.matched_blocks, 0, "router_b leaked unpublished key");
+
+    // 6. Both shutdowns must complete cleanly — no hang from the second
+    //    subscriber holding a reference to a shared resource. The first
+    //    drains under a generous ceiling (worker thread joins, mpsc
+    //    receiver drop); the second has nothing left to wait on and
+    //    must complete promptly. A slow second shutdown indicates the
+    //    two subscribers were sharing a resource that serialized them.
+    let r = tokio::time::timeout(Duration::from_secs(2), Arc::clone(&router_a).shutdown()).await;
+    assert!(r.is_ok(), "router_a shutdown hung");
+
+    let t = std::time::Instant::now();
+    let r = tokio::time::timeout(Duration::from_secs(2), Arc::clone(&router_b).shutdown()).await;
+    assert!(r.is_ok(), "router_b shutdown hung");
+    let elapsed = t.elapsed();
+    assert!(
+        elapsed < Duration::from_millis(100),
+        "router_b shutdown after router_a drained took {elapsed:?}; \
+         expected <100ms (no shared-resource contention)",
+    );
+}
+
+/// Two PUB sockets (two workers) + two `KvEventIndex` instances (two
+/// routers), each subscribed to **both** publishers. This is the real
+/// v1 HA shape: each router replica fans out subscriptions across the
+/// worker pool and merges every publisher's `BlockStored` stream into
+/// its own tree. The companion 1-PUB test above only verifies broadcast
+/// fan-out; this test verifies the per-worker attribution stays correct
+/// when events arrive from multiple sources concurrently.
+///
+/// Property pinned: after publishing prefix `X` on `pub_x` and prefix
+/// `Y` on `pub_y`, both trees report
+///   * `match_prefix(X) = {full, workers={worker_x}}`
+///   * `match_prefix(Y) = {full, workers={worker_y}}`
+/// with no cross-attribution (worker_x must NOT appear in match(Y)).
+/// A regression that wires both subscribers to the same internal
+/// channel — or that mis-keys events by their arrival socket rather
+/// than their announced worker URL — would surface here as cross-
+/// contamination of the worker sets.
+#[tokio::test]
+async fn two_subscribers_merge_events_from_two_publishers() {
+    let (mut pub_x, port_x) = make_pub_bound().await;
+    let (mut pub_y, port_y) = make_pub_bound().await;
+    let worker_x = "http://127.0.0.1:30001";
+    let worker_y = "http://127.0.0.1:30002";
+    let block_size = 4u32;
+    let cfg_x = EventConfig {
+        host: "127.0.0.1".into(),
+        port_base: port_x,
+        topic: String::new(),
+        block_size,
+        dp_size: 1,
+    };
+    let cfg_y = EventConfig {
+        host: "127.0.0.1".into(),
+        port_base: port_y,
+        topic: String::new(),
+        block_size,
+        dp_size: 1,
+    };
+
+    // Both routers subscribe to BOTH workers — the production fan-out.
+    let router_a = KvEventIndex::new();
+    let router_b = KvEventIndex::new();
+    router_a.add_worker(worker_x, Some(cfg_x.clone())).await;
+    router_a.add_worker(worker_y, Some(cfg_y.clone())).await;
+    router_b.add_worker(worker_x, Some(cfg_x.clone())).await;
+    router_b.add_worker(worker_y, Some(cfg_y.clone())).await;
+
+    // Four SUB→PUB handshakes need to settle before publishing; missed
+    // SUBSCRIBE frames lose messages forever in PUB/SUB semantics.
+    tokio::time::sleep(Duration::from_millis(200)).await;
+
+    // Two non-overlapping token streams → two distinct hash chains. The
+    // gap between them (0..16 vs 1000..1016) keeps `compute_block_hashes`
+    // outputs disjoint so a cross-attribution bug can't be masked by
+    // hash collision.
+    let tokens_x: Vec<u32> = (0..16).collect();
+    let tokens_y: Vec<u32> = (1000..1016).collect();
+    let hashes_x = compute_block_hashes(&tokens_x, block_size as usize);
+    let hashes_y = compute_block_hashes(&tokens_y, block_size as usize);
+    assert!(hashes_x.len() >= 3 && hashes_y.len() >= 3);
+
+    let payload_x = encode_event_batch(
+        0.0,
+        vec![encode_block_stored_event(
+            &hashes_x, None, &tokens_x, block_size,
+        )],
+        Some(0),
+    );
+    let payload_y = encode_event_batch(
+        0.0,
+        vec![encode_block_stored_event(
+            &hashes_y, None, &tokens_y, block_size,
+        )],
+        Some(0),
+    );
+    pub_x
+        .send(build_multipart(1, payload_x))
+        .await
+        .expect("publish on pub_x");
+    pub_y
+        .send(build_multipart(1, payload_y))
+        .await
+        .expect("publish on pub_y");
+
+    let key_x = KvWorkerId {
+        url: worker_x.into(),
+        dp_rank: 0,
+    };
+    let key_y = KvWorkerId {
+        url: worker_y.into(),
+        dp_rank: 0,
+    };
+    let target_x = hashes_x.len();
+    let target_y = hashes_y.len();
+
+    let start = std::time::Instant::now();
+    loop {
+        let ax = router_a.tree().match_prefix(None, &hashes_x);
+        let ay = router_a.tree().match_prefix(None, &hashes_y);
+        let bx = router_b.tree().match_prefix(None, &hashes_x);
+        let by = router_b.tree().match_prefix(None, &hashes_y);
+        let converged = ax.matched_blocks == target_x
+            && ay.matched_blocks == target_y
+            && bx.matched_blocks == target_x
+            && by.matched_blocks == target_y
+            && ax.workers.contains(&key_x)
+            && ay.workers.contains(&key_y)
+            && bx.workers.contains(&key_x)
+            && by.workers.contains(&key_y);
+        if converged {
+            // Negative attribution: prefix X must not be attributed to
+            // worker_y in either tree, and vice versa. A regression that
+            // keyed events by arriving socket rather than announced
+            // worker URL would set BOTH worker keys on each prefix.
+            assert!(
+                !ax.workers.contains(&key_y),
+                "router_a cross-attributed worker_y to prefix X: {:?}",
+                ax.workers,
+            );
+            assert!(
+                !ay.workers.contains(&key_x),
+                "router_a cross-attributed worker_x to prefix Y: {:?}",
+                ay.workers,
+            );
+            assert!(
+                !bx.workers.contains(&key_y),
+                "router_b cross-attributed worker_y to prefix X: {:?}",
+                bx.workers,
+            );
+            assert!(
+                !by.workers.contains(&key_x),
+                "router_b cross-attributed worker_x to prefix Y: {:?}",
+                by.workers,
+            );
+            break;
+        }
+        if start.elapsed() > Duration::from_secs(3) {
+            panic!(
+                "trees did not converge within 3s:\n  \
+                 router_a: X={{matched={}, workers={:?}}}, Y={{matched={}, workers={:?}}}\n  \
+                 router_b: X={{matched={}, workers={:?}}}, Y={{matched={}, workers={:?}}}\n  \
+                 targets: X={target_x}, Y={target_y}",
+                ax.matched_blocks,
+                ax.workers,
+                ay.matched_blocks,
+                ay.workers,
+                bx.matched_blocks,
+                bx.workers,
+                by.matched_blocks,
+                by.workers,
+            );
+        }
+        tokio::time::sleep(Duration::from_millis(20)).await;
+    }
+
+    let r = tokio::time::timeout(Duration::from_secs(2), Arc::clone(&router_a).shutdown()).await;
+    assert!(r.is_ok(), "router_a shutdown hung");
+    let r = tokio::time::timeout(Duration::from_secs(2), Arc::clone(&router_b).shutdown()).await;
+    assert!(r.is_ok(), "router_b shutdown hung");
+}

experimental/sgl-router/tests/component/policies/mod.rs

@@ -0,0 +1,11 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 The SGLang Authors
+// SPDX-License-Identifier: Apache-2.0
+
+mod zmq_helpers;
+
+mod cache_aware_zmq;
+mod kv_events_hash_parity;
+mod kv_events_tree_concurrent;
+mod kv_events_two_subscribers;
+mod power_of_two;
+mod round_robin;

experimental/sgl-router/tests/component/policies/power_of_two.rs

@@ -0,0 +1,73 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 The SGLang Authors
+// SPDX-License-Identifier: Apache-2.0
+
+use sgl_router::discovery::{ModelId, WorkerId, WorkerMode, WorkerSpec};
+use sgl_router::policies::power_of_two::PowerOfTwoChoicesPolicy;
+use sgl_router::policies::{Policy, SelectionContext};
+use sgl_router::workers::Worker;
+use std::sync::atomic::Ordering;
+use std::sync::Arc;
+
+fn worker(id: &str) -> Arc<Worker> {
+    Arc::new(Worker::new(WorkerSpec {
+        id: WorkerId(id.into()),
+        url: format!("http://{id}"),
+        mode: WorkerMode::Plain,
+        model_ids: vec![ModelId("m".into())],
+        bootstrap_port: None,
+    }))
+}
+
+#[test]
+fn selects_lower_load() {
+    let a = worker("a");
+    let b = worker("b");
+    a.active_requests.store(10, Ordering::Relaxed);
+    b.active_requests.store(2, Ordering::Relaxed);
+    let p = PowerOfTwoChoicesPolicy::new();
+    let ws = vec![a.clone(), b.clone()];
+    let model_id = ModelId("m".into());
+    let ctx = SelectionContext::new(&model_id, None);
+    let chosen = p.select(&ws, &ctx).unwrap();
+    assert_eq!(chosen.id.0, "b");
+}
+
+#[test]
+fn distribution_skews_to_lower_load() {
+    // With 3 workers and one heavily loaded, the loaded one should win
+    // significantly less than 1/3 of selections.
+    let workers = vec![worker("a"), worker("b"), worker("c")];
+    workers[2].active_requests.store(100, Ordering::Relaxed); // c is loaded
+
+    let p = PowerOfTwoChoicesPolicy::new();
+    let model_id = ModelId("m".into());
+    let ctx = SelectionContext::new(&model_id, None);
+    let mut counts = std::collections::HashMap::new();
+    for _ in 0..1000 {
+        let w = p.select(&workers, &ctx).unwrap();
+        *counts.entry(w.id.0.clone()).or_insert(0) += 1;
+    }
+    let c_picks = *counts.get("c").unwrap_or(&0);
+    assert!(
+        c_picks < 200,
+        "loaded worker should be picked < 20% of the time, got {c_picks}"
+    );
+}
+
+#[test]
+fn empty_returns_none() {
+    let p = PowerOfTwoChoicesPolicy::new();
+    let ws: Vec<Arc<Worker>> = vec![];
+    let model_id = ModelId("m".into());
+    let ctx = SelectionContext::new(&model_id, None);
+    assert!(p.select(&ws, &ctx).is_none());
+}
+
+#[test]
+fn single_worker_returns_it() {
+    let p = PowerOfTwoChoicesPolicy::new();
+    let ws = vec![worker("only")];
+    let model_id = ModelId("m".into());
+    let ctx = SelectionContext::new(&model_id, None);
+    assert_eq!(p.select(&ws, &ctx).unwrap().id.0, "only");
+}

experimental/sgl-router/tests/component/policies/round_robin.rs

@@ -0,0 +1,56 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 The SGLang Authors
+// SPDX-License-Identifier: Apache-2.0
+
+use sgl_router::discovery::{ModelId, WorkerId, WorkerMode, WorkerSpec};
+use sgl_router::policies::round_robin::RoundRobinPolicy;
+use sgl_router::policies::{Policy, SelectionContext};
+use sgl_router::workers::Worker;
+use std::sync::Arc;
+
+fn worker(id: &str) -> Arc<Worker> {
+    Arc::new(Worker::new(WorkerSpec {
+        id: WorkerId(id.into()),
+        url: format!("http://{id}"),
+        mode: WorkerMode::Plain,
+        model_ids: vec![ModelId("m".into())],
+        bootstrap_port: None,
+    }))
+}
+
+#[test]
+fn cycles_through_workers() {
+    let p = RoundRobinPolicy::new();
+    let ws = vec![worker("a"), worker("b"), worker("c")];
+    let model_id = ModelId("m".into());
+    let ctx = SelectionContext::new(&model_id, None);
+    let picks: Vec<_> = (0..6)
+        .filter_map(|_| p.select(&ws, &ctx))
+        .map(|w| w.id.0.clone())
+        .collect();
+    assert_eq!(picks, vec!["a", "b", "c", "a", "b", "c"]);
+}
+
+#[test]
+fn empty_pool_returns_none() {
+    let p = RoundRobinPolicy::new();
+    let ws: Vec<Arc<Worker>> = vec![];
+    let model_id = ModelId("m".into());
+    let ctx = SelectionContext::new(&model_id, None);
+    assert!(p.select(&ws, &ctx).is_none());
+}
+
+#[test]
+fn distribution_across_100_calls() {
+    let p = RoundRobinPolicy::new();
+    let ws = vec![worker("a"), worker("b"), worker("c")];
+    let model_id = ModelId("m".into());
+    let ctx = SelectionContext::new(&model_id, None);
+    let mut counts = std::collections::HashMap::new();
+    for _ in 0..99 {
+        let w = p.select(&ws, &ctx).unwrap();
+        *counts.entry(w.id.0.clone()).or_insert(0) += 1;
+    }
+    assert_eq!(counts["a"], 33);
+    assert_eq!(counts["b"], 33);
+    assert_eq!(counts["c"], 33);
+}

experimental/sgl-router/tests/component/policies/zmq_helpers.rs

@@ -0,0 +1,88 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 The SGLang Authors
+// SPDX-License-Identifier: Apache-2.0
+
+//! Shared ZMQ wire-format helpers for the `policies::kv_events` component
+//! tests. Encodes events in the same msgspec layout SGLang emits, builds
+//! the two-frame `[seq, payload]` ZMQ message a real publisher sends, and
+//! binds a loopback PUB socket on an OS-assigned port.
+
+#![allow(dead_code)]
+
+use bytes::Bytes;
+use rmp::encode as mp;
+use zeromq::{Endpoint, PubSocket, Socket, ZmqMessage};
+
+/// Bind a PUB socket to an OS-assigned 127.0.0.1 port. Returns
+/// `(socket, port)`.
+pub async fn make_pub_bound() -> (PubSocket, u16) {
+    let mut sock = PubSocket::new();
+    let endpoint = sock
+        .bind("tcp://127.0.0.1:0")
+        .await
+        .expect("bind PUB socket");
+    let port = match endpoint {
+        Endpoint::Tcp(_, p) => p,
+        other => panic!("unexpected endpoint: {other:?}"),
+    };
+    (sock, port)
+}
+
+/// Encode a single `BlockStored` event in the wire format msgspec
+/// emits. Layout: `["BlockStored", block_hashes, parent, token_ids,
+/// block_size, lora_id, medium]`.
+pub fn encode_block_stored_event(
+    block_hashes: &[i64],
+    parent: Option<i64>,
+    token_ids: &[u32],
+    block_size: u32,
+) -> Vec<u8> {
+    let mut buf = Vec::new();
+    mp::write_array_len(&mut buf, 7).unwrap();
+    mp::write_str(&mut buf, "BlockStored").unwrap();
+    mp::write_array_len(&mut buf, block_hashes.len() as u32).unwrap();
+    for v in block_hashes {
+        mp::write_sint(&mut buf, *v).unwrap();
+    }
+    match parent {
+        Some(v) => {
+            mp::write_sint(&mut buf, v).unwrap();
+        }
+        None => mp::write_nil(&mut buf).unwrap(),
+    }
+    mp::write_array_len(&mut buf, token_ids.len() as u32).unwrap();
+    for v in token_ids {
+        mp::write_uint(&mut buf, *v as u64).unwrap();
+    }
+    mp::write_uint(&mut buf, block_size as u64).unwrap();
+    mp::write_nil(&mut buf).unwrap(); // lora_id
+    mp::write_str(&mut buf, "GPU").unwrap();
+    buf
+}
+
+/// Wrap one or more pre-encoded events into a KVEventBatch with
+/// timestamp + optional dp-rank.
+pub fn encode_event_batch(ts: f64, events: Vec<Vec<u8>>, attn_dp_rank: Option<u32>) -> Vec<u8> {
+    let mut buf = Vec::new();
+    mp::write_array_len(&mut buf, 3).unwrap();
+    mp::write_f64(&mut buf, ts).unwrap();
+    mp::write_array_len(&mut buf, events.len() as u32).unwrap();
+    for ev in events {
+        buf.extend_from_slice(&ev);
+    }
+    match attn_dp_rank {
+        Some(v) => {
+            mp::write_uint(&mut buf, v as u64).unwrap();
+        }
+        None => mp::write_nil(&mut buf).unwrap(),
+    }
+    buf
+}
+
+/// Build the two-frame ZMQ message a real KV publisher sends:
+/// `[seq (big-endian i64), payload]`.
+pub fn build_multipart(seq: i64, payload: Vec<u8>) -> ZmqMessage {
+    let mut msg = ZmqMessage::from(Bytes::new());
+    msg.push_back(Bytes::copy_from_slice(&seq.to_be_bytes()));
+    msg.push_back(Bytes::from(payload));
+    msg
+}

experimental/sgl-router/tests/component/tokenizer/mod.rs

@@ -0,0 +1,4 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 The SGLang Authors
+// SPDX-License-Identifier: Apache-2.0
+
+mod parity;

experimental/sgl-router/tests/component/tokenizer/parity.rs

@@ -0,0 +1,150 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 The SGLang Authors
+// SPDX-License-Identifier: Apache-2.0
+
+//! Bit-parity check: dynamo-tokenizers must produce the same token_ids as
+//! SGLang's reference (transformers.AutoTokenizer) for every (model, shape)
+//! fixture. Any drift is a regression.
+//!
+//! ## Running
+//!
+//! `cargo test --release --test component tokenizer::parity` runs the test.
+//!
+//! Each fixture cell needs the model's `tokenizer.json` on disk; the test
+//! looks in the local HuggingFace cache (`HF_HOME` or `~/.cache/huggingface`).
+//! Cells whose snapshot isn't cached are skipped (with a warning); cells
+//! whose snapshot IS cached are asserted bit-identical.
+//!
+//! Locally, when no fixtures can be checked (fresh cache) the test emits a
+//! warning and passes — useful for contributors without the model snapshots.
+//! In CI (`SGLANG_IS_IN_CI=true`) the same condition is a hard failure: a
+//! parity matrix that validates nothing is worse than no test at all, since
+//! it gives a false sense of coverage. The e2e HTTP tokenize test remains
+//! the authoritative live-model parity gate, but this matrix must actually
+//! run against cached snapshots when present in CI.
+//!
+//! ## Regenerating fixtures
+//!
+//! Run `tests/scripts/generate_parity_fixtures.py` after changing a prompt
+//! shape or adding a model, then commit the new JSON.
+
+use serde::Deserialize;
+use std::path::PathBuf;
+
+#[derive(Deserialize)]
+struct Fixture {
+    model_id: String,
+    shape: String,
+    prompt_text: String,
+    expected_token_ids: Vec<u32>,
+    #[allow(dead_code)]
+    skip_special_tokens: bool,
+}
+
+fn fixture_root() -> PathBuf {
+    PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("tests/fixtures/tokenizer_parity")
+}
+
+/// Resolve a model's tokenizer.json file from the local HF cache.
+///
+/// Strategy:
+/// 1. Check HF_HOME env var, or default to ~/.cache/huggingface
+/// 2. Look for models--<safe-name>/snapshots/<hash>/tokenizer.json
+/// 3. Return None if not found — the test cell is skipped.
+fn resolve_tokenizer_path(model_id: &str) -> Option<PathBuf> {
+    let hf_home = std::env::var("HF_HOME")
+        .ok()
+        .map(PathBuf::from)
+        .or_else(|| dirs::home_dir().map(|h| h.join(".cache/huggingface")))?;
+    let safe = model_id.replace('/', "--");
+    let candidate = hf_home.join("hub").join(format!("models--{safe}"));
+    if !candidate.exists() {
+        return None;
+    }
+    let snapshots = candidate.join("snapshots");
+    let snap = std::fs::read_dir(&snapshots).ok()?.next()?.ok()?.path();
+    let tj = snap.join("tokenizer.json");
+    tj.exists().then_some(tj)
+}
+
+/// Parity matrix: dynamo-tokenizers vs. transformers.AutoTokenizer.
+///
+/// Skips cells whose tokenizer.json isn't in the local HF cache. See
+/// module-level docs.
+#[test]
+fn parity_matrix() {
+    let mut checked = 0;
+    let mut skipped = vec![];
+    for model_dir in std::fs::read_dir(fixture_root()).unwrap() {
+        let model_dir = model_dir.unwrap().path();
+        if !model_dir.is_dir() {
+            continue;
+        }
+        for shape_file in std::fs::read_dir(&model_dir).unwrap() {
+            let p = shape_file.unwrap().path();
+            if p.extension().and_then(|s| s.to_str()) != Some("json") {
+                continue;
+            }
+            let raw = std::fs::read_to_string(&p).unwrap();
+            let f: Fixture =
+                serde_json::from_str(&raw).unwrap_or_else(|e| panic!("parse {}: {e}", p.display()));
+            let Some(tp) = resolve_tokenizer_path(&f.model_id) else {
+                skipped.push((f.model_id.clone(), f.shape.clone()));
+                continue;
+            };
+            let tok = sgl_router::tokenizer::adapter::load(tp.to_str().unwrap()).unwrap();
+            let ids = sgl_router::tokenizer::adapter::encode(&tok, &f.prompt_text).unwrap();
+            assert_eq!(
+                ids, f.expected_token_ids,
+                "DRIFT on {}/{}",
+                f.model_id, f.shape
+            );
+            checked += 1;
+        }
+    }
+    let expected = std::fs::read_dir(fixture_root())
+        .unwrap()
+        .filter_map(|e| e.ok())
+        .filter(|e| e.path().is_dir())
+        .map(|e| {
+            std::fs::read_dir(e.path())
+                .unwrap()
+                .filter_map(|f| f.ok())
+                .filter(|f| f.path().extension().and_then(|s| s.to_str()) == Some("json"))
+                .count()
+        })
+        .sum::<usize>();
+    assert_eq!(
+        checked + skipped.len(),
+        expected,
+        "expected {expected} fixtures, found {}",
+        checked + skipped.len()
+    );
+    if checked == 0 {
+        let families: Vec<String> = skipped
+            .iter()
+            .map(|(m, _)| m.clone())
+            .collect::<std::collections::BTreeSet<_>>()
+            .into_iter()
+            .collect();
+        let msg = format!(
+            "parity_matrix: no fixtures could be checked — HF cache empty? skipped {} cells \
+             across model families: [{}]. The e2e HTTP tokenize test remains the \
+             authoritative live-model parity gate.",
+            skipped.len(),
+            families.join(", "),
+        );
+        if std::env::var("SGLANG_IS_IN_CI").as_deref() == Ok("true") {
+            panic!(
+                "{msg}\n\nThis is a hard failure in CI: a parity test that validates zero \
+                 cells provides no coverage. Either pre-populate the HF cache for these \
+                 model families on the runner, or remove the parity test."
+            );
+        }
+        eprintln!("{msg}");
+    } else {
+        eprintln!(
+            "parity: {checked} cells passed, {} skipped (no HF snapshot)",
+            skipped.len()
+        );
+    }
+}

experimental/sgl-router/tests/component/workers/concurrent_state.rs

@@ -0,0 +1,148 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 The SGLang Authors
+// SPDX-License-Identifier: Apache-2.0
+
+//! Concurrent-state invariants for the worker/registry/breaker layer.
+//!
+//! These tests stress the lock-free / single-Mutex paths that production
+//! traffic exercises in parallel: many requests calling `breaker.allow()`,
+//! many discovery events racing with workers_for() reads, and LoadGuard
+//! lifecycles under panics.
+
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Arc;
+use std::time::Duration;
+
+use sgl_router::discovery::{ModelId, WorkerId, WorkerMode, WorkerSpec};
+use sgl_router::health::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig};
+use sgl_router::workers::{Worker, WorkerRegistry};
+
+/// HalfOpen state must admit at most one probe at a time even under high
+/// concurrency.  N threads race `allow()` when the breaker is HalfOpen; the
+/// invariant is that exactly one observes `true` (the probe holder); the
+/// rest see `false` because `probe_in_flight` is already set.
+#[tokio::test(start_paused = true)]
+async fn breaker_half_open_admits_only_one_probe_concurrently() {
+    let cb = Arc::new(CircuitBreaker::with_config(CircuitBreakerConfig {
+        threshold: std::num::NonZeroU32::new(1).unwrap(),
+        cool_down: Duration::from_millis(50),
+    }));
+
+    // Trip into Open.
+    cb.record_failure();
+    assert!(!cb.allow(), "must be Open immediately after a failure");
+
+    // Advance the paused clock past cool_down so the next `allow()` will
+    // attempt the Open → HalfOpen transition.
+    tokio::time::advance(Duration::from_millis(60)).await;
+
+    let admitted = Arc::new(AtomicUsize::new(0));
+    let mut handles = Vec::new();
+    for _ in 0..32 {
+        let cb = cb.clone();
+        let admitted = admitted.clone();
+        handles.push(tokio::spawn(async move {
+            if cb.allow() {
+                admitted.fetch_add(1, Ordering::Relaxed);
+            }
+        }));
+    }
+    for h in handles {
+        h.await.unwrap();
+    }
+    assert_eq!(
+        admitted.load(Ordering::Relaxed),
+        1,
+        "exactly one probe must be admitted in HalfOpen",
+    );
+}
+
+/// Concurrent `add_with_cb` (upsert) and `remove` from many threads on the
+/// same WorkerId must not panic, must not deadlock, and must leave a
+/// consistent index — `workers_for(model)` may return 0 or 1 worker, but
+/// must never resolve to a worker that has been removed.
+#[test]
+fn registry_concurrent_add_remove_keeps_indexes_consistent() {
+    let r = Arc::new(WorkerRegistry::default());
+    let model = ModelId("m".into());
+
+    let mut handles = Vec::new();
+    for i in 0..8 {
+        let r = r.clone();
+        let model = model.clone();
+        handles.push(std::thread::spawn(move || {
+            for _ in 0..200 {
+                let _ = r.add(WorkerSpec {
+                    id: WorkerId(format!("w{i}")),
+                    url: format!("http://w{i}:30000"),
+                    mode: WorkerMode::Plain,
+                    model_ids: vec![model.clone()],
+                    bootstrap_port: None,
+                });
+                let snapshot = r.workers_for(&model);
+                for w in &snapshot {
+                    // Cross-index invariant: an entry surfaced via
+                    // `by_model[m]` must come from a Worker whose own
+                    // `model_ids` includes `m`. An earlier version of
+                    // this assertion checked `w.id.0.starts_with('w')`,
+                    // which is a tautology — every id is `w0..w7` by
+                    // construction — and a regression where `by_model`
+                    // pointed at the wrong Worker (e.g., a stale entry
+                    // left after an upsert that should have cleared its
+                    // by_model membership for the dropped model) would
+                    // pass silently. We can't `re-get by_id and ptr_eq`
+                    // because a concurrent remove can drop the by_id
+                    // entry between the two reads — `Arc` keeps the
+                    // Worker alive on our side but the index map is
+                    // gone. The model-membership claim, however, is a
+                    // property of the Arc itself and stays stable.
+                    assert!(
+                        w.model_ids.contains(&model),
+                        "cross-index drift: by_model[{model:?}] surfaced \
+                         {:?} whose own model_ids = {:?}",
+                        w.id,
+                        w.model_ids,
+                    );
+                }
+                r.remove(&WorkerId(format!("w{i}")));
+            }
+        }));
+    }
+    for h in handles {
+        h.join().unwrap();
+    }
+
+    // After every thread finishes, every removed worker must really be gone.
+    assert!(
+        r.workers_for(&model).is_empty(),
+        "registry must be empty after all threads finished their add/remove cycles",
+    );
+}
+
+/// `LoadGuard` must decrement the counter during a panic-unwind, not just
+/// on a normal scope exit.  Rust's RAII contract via `Drop` covers this,
+/// but a future refactor (e.g. adding a manual decrement on a non-panic
+/// path) could silently regress it.  This test pins the invariant.
+#[test]
+fn load_guard_decrements_on_panic_unwind() {
+    let w = Arc::new(Worker::new(WorkerSpec {
+        id: WorkerId("w".into()),
+        url: "http://x:30000".into(),
+        mode: WorkerMode::Plain,
+        model_ids: vec![ModelId("m".into())],
+        bootstrap_port: None,
+    }));
+    assert_eq!(w.active_load(), 0);
+
+    let w_inner = w.clone();
+    let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(move || {
+        let _g = w_inner.load_guard();
+        assert_eq!(w_inner.active_load(), 1);
+        panic!("synthetic panic to exercise Drop on unwind");
+    }));
+    assert!(result.is_err(), "the closure must have panicked");
+    assert_eq!(
+        w.active_load(),
+        0,
+        "LoadGuard's Drop must decrement even when the holder panics",
+    );
+}

experimental/sgl-router/tests/component/workers/manager.rs

@@ -0,0 +1,511 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 The SGLang Authors
+// SPDX-License-Identifier: Apache-2.0
+
+use axum::{routing::get, Json, Router};
+use serde_json::{json, Value};
+use sgl_router::discovery::{DiscoveryEvent, ModelId, WorkerId, WorkerMode, WorkerSpec};
+use sgl_router::workers::{manager, WorkerRegistry};
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+use tokio::net::TcpListener;
+use tokio::sync::{mpsc, oneshot};
+
+/// Spin up a tiny fake worker that returns `body` on `GET /server_info`.
+/// Returns the worker base URL and a shutdown channel.
+async fn spawn_fake_worker(body: Value) -> (String, oneshot::Sender<()>) {
+    let body = Arc::new(body);
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let port = listener.local_addr().unwrap().port();
+    let app = Router::new().route(
+        "/server_info",
+        get(move || {
+            let body = body.clone();
+            async move { Json((*body).clone()) }
+        }),
+    );
+    let (tx, rx) = oneshot::channel::<()>();
+    tokio::spawn(async move {
+        let _ = axum::serve(listener, app)
+            .with_graceful_shutdown(async move {
+                let _ = rx.await;
+            })
+            .await;
+    });
+    (format!("http://127.0.0.1:{port}"), tx)
+}
+
+fn spec_for(id: &str, url: &str, mode: WorkerMode) -> WorkerSpec {
+    // model_ids are intentionally empty: the manager resolves them via
+    // /server_info introspection.  Pre-populating here would lie about
+    // what discovery backends actually emit.
+    WorkerSpec {
+        id: WorkerId(id.into()),
+        url: url.into(),
+        mode,
+        model_ids: Vec::new(),
+        bootstrap_port: None,
+    }
+}
+
+#[tokio::test]
+async fn manager_processes_added_then_removed() {
+    let (url_a, _s_a) = spawn_fake_worker(json!({"served_model_name": "m"})).await;
+    let (url_b, _s_b) = spawn_fake_worker(json!({"served_model_name": "m"})).await;
+
+    let (tx, rx) = mpsc::channel(16);
+    let registry = Arc::new(WorkerRegistry::default());
+    let h = tokio::spawn(manager::run(rx, registry.clone()));
+
+    tx.send(DiscoveryEvent::Added(spec_for(
+        "w1",
+        &url_a,
+        WorkerMode::Plain,
+    )))
+    .await
+    .unwrap();
+    tx.send(DiscoveryEvent::Added(spec_for(
+        "w2",
+        &url_b,
+        WorkerMode::Plain,
+    )))
+    .await
+    .unwrap();
+
+    // Give the manager time to drain.
+    tokio::time::sleep(Duration::from_millis(200)).await;
+    assert_eq!(registry.workers_for(&ModelId("m".into())).len(), 2);
+
+    tx.send(DiscoveryEvent::Removed {
+        id: WorkerId("w1".into()),
+    })
+    .await
+    .unwrap();
+    tokio::time::sleep(Duration::from_millis(50)).await;
+    assert_eq!(registry.workers_for(&ModelId("m".into())).len(), 1);
+
+    drop(tx);
+    h.await.unwrap();
+}
+
+#[tokio::test]
+async fn manager_handles_mode_changed() {
+    let (url, _s) = spawn_fake_worker(json!({"served_model_name": "m"})).await;
+
+    let (tx, rx) = mpsc::channel(16);
+    let registry = Arc::new(WorkerRegistry::default());
+    let h = tokio::spawn(manager::run(rx, registry.clone()));
+
+    tx.send(DiscoveryEvent::Added(spec_for(
+        "w1",
+        &url,
+        WorkerMode::Prefill,
+    )))
+    .await
+    .unwrap();
+    tokio::time::sleep(Duration::from_millis(200)).await;
+    assert_eq!(
+        registry
+            .workers_for_mode(&ModelId("m".into()), WorkerMode::Prefill)
+            .len(),
+        1
+    );
+
+    tx.send(DiscoveryEvent::ModeChanged {
+        id: WorkerId("w1".into()),
+        mode: WorkerMode::Decode,
+    })
+    .await
+    .unwrap();
+    tokio::time::sleep(Duration::from_millis(50)).await;
+    assert_eq!(
+        registry
+            .workers_for_mode(&ModelId("m".into()), WorkerMode::Prefill)
+            .len(),
+        0
+    );
+    assert_eq!(
+        registry
+            .workers_for_mode(&ModelId("m".into()), WorkerMode::Decode)
+            .len(),
+        1
+    );
+
+    drop(tx);
+    h.await.unwrap();
+}
+
+#[tokio::test]
+async fn mode_changed_preserves_active_requests_and_breaker() {
+    let (url, _s) = spawn_fake_worker(json!({"served_model_name": "m"})).await;
+
+    let (tx, rx) = mpsc::channel(16);
+    let registry = Arc::new(WorkerRegistry::default());
+    let h = tokio::spawn(manager::run(rx, registry.clone()));
+
+    tx.send(DiscoveryEvent::Added(spec_for(
+        "w1",
+        &url,
+        WorkerMode::Prefill,
+    )))
+    .await
+    .unwrap();
+    tokio::time::sleep(Duration::from_millis(200)).await;
+
+    // Grab a handle, bump active_requests, and open the breaker.
+    let w = registry.get(&WorkerId("w1".into())).unwrap();
+    w.active_requests.fetch_add(5, Ordering::Relaxed);
+    // Default threshold is 3 — record 10 failures to guarantee Open state.
+    for _ in 0..10 {
+        w.breaker.record_failure();
+    }
+    let breaker_open_before = !w.breaker.allow();
+    assert!(
+        breaker_open_before,
+        "breaker should be open after 10 failures"
+    );
+
+    // Flip mode via ModeChanged.
+    tx.send(DiscoveryEvent::ModeChanged {
+        id: WorkerId("w1".into()),
+        mode: WorkerMode::Decode,
+    })
+    .await
+    .unwrap();
+    tokio::time::sleep(Duration::from_millis(50)).await;
+
+    // Re-fetch the Worker handle from the registry.
+    let w_after = registry.get(&WorkerId("w1".into())).unwrap();
+
+    assert_eq!(
+        w_after.mode(),
+        WorkerMode::Decode,
+        "mode should have flipped to Decode"
+    );
+    assert_eq!(
+        w_after.active_requests.load(Ordering::Relaxed),
+        5,
+        "active_requests should be preserved across mode change"
+    );
+    assert!(
+        !w_after.breaker.allow(),
+        "breaker open state should be preserved across mode change"
+    );
+
+    // Critical: the Arc identity must be the same — mutation in place.
+    assert!(
+        Arc::ptr_eq(&w, &w_after),
+        "Worker handle should be the SAME Arc, not a fresh replacement"
+    );
+
+    drop(tx);
+    h.await.unwrap();
+}
+
+/// An out-of-order `ModeChanged` for a worker the registry does not know
+/// about (e.g. a buggy discovery backend reordered `Removed` and
+/// `ModeChanged`) must not panic, must not silently log INFO claiming the
+/// mode flip happened, and must leave the registry untouched.
+#[tokio::test]
+async fn manager_handles_orphan_mode_changed_without_panic() {
+    let (tx, rx) = mpsc::channel(16);
+    let registry = Arc::new(WorkerRegistry::default());
+    let h = tokio::spawn(manager::run(rx, registry.clone()));
+
+    tx.send(DiscoveryEvent::ModeChanged {
+        id: WorkerId("ghost".into()),
+        mode: WorkerMode::Decode,
+    })
+    .await
+    .unwrap();
+    tokio::time::sleep(Duration::from_millis(50)).await;
+
+    assert!(
+        registry.get(&WorkerId("ghost".into())).is_none(),
+        "an orphan ModeChanged must not create a phantom worker",
+    );
+    assert_eq!(
+        registry.workers_for(&ModelId("m".into())).len(),
+        0,
+        "registry must be empty after an orphan event",
+    );
+
+    drop(tx);
+    h.await.unwrap();
+}
+
+/// A `Removed` for an unknown id is a no-op — registry stays empty, manager
+/// keeps running.
+#[tokio::test]
+async fn manager_handles_orphan_removed_without_panic() {
+    let (tx, rx) = mpsc::channel(16);
+    let registry = Arc::new(WorkerRegistry::default());
+    let h = tokio::spawn(manager::run(rx, registry.clone()));
+
+    tx.send(DiscoveryEvent::Removed {
+        id: WorkerId("ghost".into()),
+    })
+    .await
+    .unwrap();
+    tokio::time::sleep(Duration::from_millis(50)).await;
+    assert!(registry.is_empty());
+
+    drop(tx);
+    h.await.unwrap();
+}
+
+/// Duplicate `Added` for the same id is an upsert — the registry ends up
+/// with exactly one worker. The model resolved by /server_info wins on
+/// re-add (a different worker may advertise a different served model).
+#[tokio::test]
+async fn manager_handles_duplicate_added_as_upsert() {
+    let (url_first, _s_first) = spawn_fake_worker(json!({"served_model_name": "m1"})).await;
+    let (url_second, _s_second) = spawn_fake_worker(json!({"served_model_name": "m1"})).await;
+
+    let (tx, rx) = mpsc::channel(16);
+    let registry = Arc::new(WorkerRegistry::default());
+    let h = tokio::spawn(manager::run(rx, registry.clone()));
+
+    tx.send(DiscoveryEvent::Added(spec_for(
+        "w1",
+        &url_first,
+        WorkerMode::Plain,
+    )))
+    .await
+    .unwrap();
+    tx.send(DiscoveryEvent::Added(spec_for(
+        "w1",
+        &url_second,
+        WorkerMode::Plain,
+    )))
+    .await
+    .unwrap();
+    tokio::time::sleep(Duration::from_millis(300)).await;
+
+    assert_eq!(
+        registry.workers_for(&ModelId("m1".into())).len(),
+        1,
+        "w1 still serves m1 after the second Added",
+    );
+
+    drop(tx);
+    h.await.unwrap();
+}
+
+/// Spawn a fake worker whose `/server_info` returns `body` only after
+/// sleeping for `delay`. Returns the worker URL and a shutdown channel.
+async fn spawn_slow_worker(body: Value, delay: Duration) -> (String, oneshot::Sender<()>) {
+    let body = Arc::new(body);
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let port = listener.local_addr().unwrap().port();
+    let app = Router::new().route(
+        "/server_info",
+        get(move || {
+            let body = body.clone();
+            async move {
+                tokio::time::sleep(delay).await;
+                Json((*body).clone())
+            }
+        }),
+    );
+    let (tx, rx) = oneshot::channel::<()>();
+    tokio::spawn(async move {
+        let _ = axum::serve(listener, app)
+            .with_graceful_shutdown(async move {
+                let _ = rx.await;
+            })
+            .await;
+    });
+    (format!("http://127.0.0.1:{port}"), tx)
+}
+
+/// Spawn a fake worker that counts each `GET /server_info` hit in the
+/// returned `AtomicUsize`.  Used to assert the manager makes exactly
+/// one round-trip per worker.
+async fn spawn_counting_worker(body: Value) -> (String, Arc<AtomicUsize>, oneshot::Sender<()>) {
+    let body = Arc::new(body);
+    let counter = Arc::new(AtomicUsize::new(0));
+    let counter_clone = counter.clone();
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let port = listener.local_addr().unwrap().port();
+    let app = Router::new().route(
+        "/server_info",
+        get(move || {
+            let body = body.clone();
+            let counter = counter_clone.clone();
+            async move {
+                counter.fetch_add(1, Ordering::SeqCst);
+                Json((*body).clone())
+            }
+        }),
+    );
+    let (tx, rx) = oneshot::channel::<()>();
+    tokio::spawn(async move {
+        let _ = axum::serve(listener, app)
+            .with_graceful_shutdown(async move {
+                let _ = rx.await;
+            })
+            .await;
+    });
+    (format!("http://127.0.0.1:{port}"), counter, tx)
+}
+
+/// Registration must run in parallel across multiple `Added` events.
+/// Each fake worker delays its `/server_info` by 200ms; with sequential
+/// processing the manager would take ≥1000ms for 5 workers. We allow
+/// up to 600ms (3x the per-fetch delay) as a generous bound that still
+/// rejects the sequential implementation.
+#[tokio::test]
+async fn added_events_run_in_parallel() {
+    let delay = Duration::from_millis(200);
+    let n = 5;
+    let mut workers = Vec::new();
+    for _ in 0..n {
+        workers.push(spawn_slow_worker(json!({"served_model_name": "m"}), delay).await);
+    }
+
+    let (tx, rx) = mpsc::channel(16);
+    let registry = Arc::new(WorkerRegistry::default());
+    let h = tokio::spawn(manager::run(rx, registry.clone()));
+
+    let start = Instant::now();
+    for (i, (url, _s)) in workers.iter().enumerate() {
+        tx.send(DiscoveryEvent::Added(spec_for(
+            &format!("w{i}"),
+            url,
+            WorkerMode::Plain,
+        )))
+        .await
+        .unwrap();
+    }
+    let registered = tokio::time::timeout(Duration::from_secs(5), async {
+        loop {
+            if registry.workers_for(&ModelId("m".into())).len() == n {
+                return true;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await;
+    let elapsed = start.elapsed();
+    assert!(registered.is_ok(), "manager failed to register {n} workers");
+    assert!(
+        elapsed < Duration::from_millis(600),
+        "registration of {n} workers took {elapsed:?}; sequential per-worker /server_info \
+         fetches would take ≥1000ms — parallel spawn is required"
+    );
+
+    drop(tx);
+    h.await.unwrap();
+}
+
+/// A `Removed` issued while the matching `Added` is still mid-fetch
+/// must await the in-flight registration handle before removing.
+/// Without that ordering the removal runs first (registry has nothing
+/// to remove), then the Added's deferred registry write leaks the
+/// worker.
+#[tokio::test]
+async fn removed_awaits_pending_added() {
+    let (url, _s) = spawn_slow_worker(
+        json!({"served_model_name": "m"}),
+        Duration::from_millis(300),
+    )
+    .await;
+
+    let (tx, rx) = mpsc::channel(16);
+    let registry = Arc::new(WorkerRegistry::default());
+    let h = tokio::spawn(manager::run(rx, registry.clone()));
+
+    tx.send(DiscoveryEvent::Added(spec_for(
+        "w-slow",
+        &url,
+        WorkerMode::Plain,
+    )))
+    .await
+    .unwrap();
+    tx.send(DiscoveryEvent::Removed {
+        id: WorkerId("w-slow".into()),
+    })
+    .await
+    .unwrap();
+
+    // Wait long enough for the Added's /server_info to complete (300ms),
+    // then assert the worker is gone. If Removed ran before Added's
+    // registry write, the post-fetch write would leak the entry.
+    tokio::time::sleep(Duration::from_millis(600)).await;
+    assert!(
+        registry.get(&WorkerId("w-slow".into())).is_none(),
+        "Removed must await the in-flight Added; otherwise the deferred \
+         registry write leaks the worker"
+    );
+
+    drop(tx);
+    h.await.unwrap();
+}
+
+/// The manager must make exactly ONE `/server_info` request per worker.
+/// Before this fix the worker manager fetched `served_model_name` and
+/// `KvEventIndex::add_worker` fetched the `kv_events` block
+/// independently — 2N round-trips for N workers.
+#[tokio::test]
+async fn manager_emits_single_server_info_fetch_per_worker() {
+    use sgl_router::policies::kv_events::KvEventIndex;
+
+    let body = json!({
+        "served_model_name": "m",
+        "kv_events": {
+            "publisher": "zmq",
+            "endpoint_host": "127.0.0.1",
+            "endpoint_port_base": 60100,
+            "topic": "",
+            "block_size": 64,
+            "dp_size": 1,
+        }
+    });
+    let (url, counter, _s) = spawn_counting_worker(body).await;
+
+    let (tx, rx) = mpsc::channel(16);
+    let registry = Arc::new(WorkerRegistry::default());
+    let kv_index = KvEventIndex::new();
+    let h = tokio::spawn(manager::run_with_config(
+        rx,
+        registry.clone(),
+        None,
+        Some(kv_index.clone()),
+        None,
+    ));
+
+    tx.send(DiscoveryEvent::Added(spec_for(
+        "w1",
+        &url,
+        WorkerMode::Plain,
+    )))
+    .await
+    .unwrap();
+
+    // Wait for both the registry and kv-events index to reflect the worker.
+    let ready = tokio::time::timeout(Duration::from_secs(2), async {
+        loop {
+            if registry.get(&WorkerId("w1".into())).is_some() && kv_index.known_worker_count() == 1
+            {
+                return true;
+            }
+            tokio::time::sleep(Duration::from_millis(20)).await;
+        }
+    })
+    .await;
+    assert!(
+        ready.is_ok(),
+        "manager did not finish onboarding the worker"
+    );
+
+    let hits = counter.load(Ordering::SeqCst);
+    assert_eq!(
+        hits, 1,
+        "manager must fetch /server_info exactly once per worker (got {hits})"
+    );
+
+    drop(tx);
+    h.await.unwrap();
+    kv_index.shutdown().await;
+}

experimental/sgl-router/tests/component/workers/mod.rs

@@ -0,0 +1,5 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 The SGLang Authors
+// SPDX-License-Identifier: Apache-2.0
+
+mod concurrent_state;
+mod manager;