refactor(connector): restore pipeline behavior and boundaries

benchmarks/bench_staging_restore.py

@@ -20,7 +20,7 @@
 
 # First Party
 from benchmarks.utils.constants import BLOCK_TOKENS  # noqa: E402
-from daser.connector.staging import copy_staging_to_kv_cache  # noqa: E402
+from daser.connector.worker.staging import copy_staging_to_kv_cache  # noqa: E402
 from daser.ops.rope_apply import clear_rope_apply_cache  # noqa: E402
 
 

benchmarks/utils/vllm_bench.py

@@ -11,7 +11,7 @@
 
 import asyncio
 from collections.abc import Callable
-from dataclasses import asdict
+from dataclasses import asdict, replace
 import json
 import math
 from pathlib import Path
@@ -297,9 +297,19 @@ def run_load(
             asyncio.run(_wait_lmcache_quiescent(manifest, settle_seconds=0.0))
         elif backend_run.backend == "daser":
             _drain_daser(manifest, print_kv=print_kv)
-        warm_metrics, warm_hit_rate = _run_phase(
-            args, manifest, warm_raw, run_command=run_command
-        )
+        if backend_run.backend == "daser" and args.evict:
+            warm_metrics, warm_hit_rate = _run_daser_evict_warm_phase(
+                args,
+                manifest,
+                warm_raw,
+                run_command=run_command,
+            )
+        else:
+            warm_metrics, warm_hit_rate = _run_phase(
+                args, manifest, warm_raw, run_command=run_command
+            )
+        if backend_run.backend == "daser" and args.evict:
+            _require_daser_evict_tier_activity(warm_metrics)
         cold_summary = _normalise_result(cold_raw)
         warm_summary = _normalise_result(warm_raw)
         _apply_phase_metrics(cold_summary, cold_hit_rate)
@@ -373,6 +383,33 @@ def _run_phase(
     return _collect_phase_metrics(manifest, before_metrics)
 
 
+def _run_daser_evict_warm_phase(
+    args: RunBenchArgs,
+    manifest: BenchmarkManifest,
+    raw_path: Path,
+    *,
+    run_command: Callable[[list[str]], Any],
+) -> tuple[dict[str, Any], float | None]:
+    """Perturb LRU order before measuring a complete evict warm phase.
+
+    A same-order scan of a working set larger than L1 can produce 100% L2
+    reads even when most entries were resident. Replaying the first 20% of the
+    deterministic workload first makes the subsequent complete phase exercise
+    both resident L1 entries and evicted L2 entries. Both commands are included
+    in the returned warm metric delta; correctness and latency use only the
+    complete phase result.
+    """
+    before_metrics = asyncio.run(collect_phase_metrics(manifest))
+    prime_args = replace(
+        args,
+        bench_num_prompts=max(1, math.ceil(args.bench_num_prompts * 0.2)),
+    )
+    prime_path = raw_path.with_name("vllm_bench_warm_prime.json")
+    run_command(_bench_command(prime_args, manifest.endpoints["vllm"], prime_path))
+    run_command(_bench_command(args, manifest.endpoints["vllm"], raw_path))
+    return _collect_phase_metrics(manifest, before_metrics)
+
+
 def _collect_phase_metrics(
     manifest: BenchmarkManifest,
     before_metrics: dict[str, Any] | None,
@@ -414,11 +451,21 @@ def _drain_daser(
     endpoint = manifest.endpoints.get("daser")
     if endpoint is None:
         return
-    try:
-        response = httpx.post(f"{endpoint.url}/drain", timeout=30.0)
-        response.raise_for_status()
-    except Exception as exc:  # noqa: BLE001
-        print_kv("daser_drain_status", f"unavailable ({exc})")
+    response = httpx.post(f"{endpoint.url}/drain", timeout=30.0)
+    response.raise_for_status()
+    print_kv("daser_drain_status", "ok")
+
+
+def _require_daser_evict_tier_activity(metrics: dict[str, Any]) -> None:
+    """Require one evict warm phase to exercise both DaseR cache tiers."""
+    counters = metrics.get("backend_prometheus", {})
+    l1_hits = float(counters.get("daser_l1_hits_total", 0.0))
+    l2_reads = float(counters.get("daser_l2_reads_total", 0.0))
+    if l1_hits <= 0 or l2_reads <= 0:
+        raise RuntimeError(
+            "DaseR evict warm phase must exercise both tiers: "
+            f"l1_hits={l1_hits:g} l2_reads={l2_reads:g}"
+        )
 
 
 def _bench_command(