profile(kvc): rewrite v5+profile report after critic audit + P0/P1 instrument

Hostile audit of the original report flagged three load-bearing errors:

1. held_tokens semantic was inverted. session_held_tokens() at
   session_aware_cache.py:278-282 sums (kv_allocated_len - cache_protected_len)
   per slot, i.e. slot-private (NOT in radix tree). So "other = cap - held -
   avail" actually CONTAINS the radix-tree protected prefix cache (likely the
   single biggest component for shared agentic prefixes), not just running
   batch + in-flight as the original report claimed.

2. Admission-race causal hypothesis for the 415 EXP2+profile errors is
   contradicted by the data: 414/415 errors have kv_transfer_blocks > 0 — they
   passed admission and died downstream ("generate stream ended before
   producing any token", raised by the client when a 200 response had an empty
   stream).

3. Polling deconfound was too quickly dismissed. Mode counts shift ~1:1
   (session-cap-fb -356 / kvcache-centric +406), and /server_info is not a
   passive read — it dispatches into the scheduler main loop and iterates
   every session slot.

Plus: per-D error% confounded by sticky session affinity (only 18 unique
sessions cause 415 errors, decode-3 had 0 errors only because no high-error
session landed there); decile 10 "recovery" was an equal-time binning
artifact (24.5% under equal-count); v5 vs v5+profile time gap was 21h not
6h; p50/p90 latency comparison is N=1.

Rewritten report (docs/V5_PROFILE_INVESTIGATION_ZH.md) marks each correction
with ⚠️ and demotes admission-race to one of four hypotheses (H1-H4).

Action items split into P0 (verify, must do first) and P1 (instrument):

P0 — scripts/sweep_tp1_v5_baseline_rerun_exp2.sh runs 3x v5 baseline EXP2
(no polling, identical config to the original v5 run) to test whether the
9-error baseline result is reproducible. If 3 runs give ~9 errors and
profile gives 415, polling is the leading suspect. Currently running
in background.

P1 — scheduler.py:_compute_pool_breakdown_for_diagnostics adds a read-only
"pool_breakdown" dict to /server_info covering: radix_evictable_tokens,
radix_protected_tokens, slot_private_held_tokens, session_slot_count,
running_batch_{reqs,kv_tokens}, transfer_queue_{reqs,tokens},
prealloc_queue_{reqs,tokens}, retracted_queue_{reqs,tokens}. With these,
"unaccounted = cap - sum(known)" exposes true leakage. replay.py captures
all fields into the per-tick row; analyzer prints the decomposition and
gracefully handles old timeseries (prints "P1 instrument absent").

Mock-tested end-to-end. SGLang patch is read-only and does not affect
admission/scheduling. Old v5+profile data still analyzes correctly.

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

third_party/sglang/python/sglang/srt/managers/scheduler.py

@@ -3181,6 +3181,89 @@ class Scheduler(
             success = False
         return success
 
+    def _compute_pool_breakdown_for_diagnostics(self) -> dict:
+        """Read-only KV pool decomposition for the agentic-pd-hybrid profiler.
+
+        Decomposes capacity into:
+          - radix_evictable_tokens / radix_protected_tokens: tree-managed
+          - slot_private_held_tokens: SessionAwareCache out-of-tree slot holds
+          - running_batch_kv_tokens: kv_allocated_len of currently-decoding reqs
+            (overlaps with radix_protected; not additive)
+          - {transfer,prealloc,retracted}_queue_{reqs,tokens}: disagg queues
+          - available_tokens: free pool
+
+        Caller computes "unaccounted = capacity - sum_of_known" to find leakage.
+        Implementation is best-effort; missing components return omitted keys.
+        """
+        breakdown: dict = {
+            "capacity_tokens": int(self.max_total_num_tokens or 0),
+            "available_tokens": int(self.token_to_kv_pool_allocator.available_size()),
+        }
+
+        # Radix tree (works for SessionAwareCache and most inner caches)
+        try:
+            ev = self.tree_cache.evictable_size()
+            pr = self.tree_cache.protected_size()
+            if isinstance(ev, tuple):
+                ev = ev[0]
+            if isinstance(pr, tuple):
+                pr = pr[0]
+            breakdown["radix_evictable_tokens"] = int(ev or 0)
+            breakdown["radix_protected_tokens"] = int(pr or 0)
+        except Exception:
+            pass
+
+        # SessionAwareCache slot-private holds (already in session_cache.held_tokens
+        # but mirrored here for one-stop decomposition)
+        try:
+            from sglang.srt.mem_cache.session_aware_cache import SessionAwareCache
+            if isinstance(self.tree_cache, SessionAwareCache):
+                breakdown["slot_private_held_tokens"] = int(
+                    self.tree_cache.session_held_tokens()
+                )
+                breakdown["session_slot_count"] = int(
+                    self.tree_cache.session_held_req_count()
+                )
+        except Exception:
+            pass
+
+        # Running batch KV (overlaps with radix_protected for tree-tracked reqs)
+        try:
+            running_reqs = self.running_batch.reqs
+            breakdown["running_batch_reqs"] = len(running_reqs)
+            breakdown["running_batch_kv_tokens"] = sum(
+                int(getattr(req, "kv_allocated_len", 0) or 0)
+                for req in running_reqs
+            )
+        except Exception:
+            pass
+
+        # Disagg decode queues
+        if self.disaggregation_mode == DisaggregationMode.DECODE:
+            try:
+                tq = self.disagg_decode_transfer_queue.queue
+                pq = self.disagg_decode_prealloc_queue.queue
+                rq = self.disagg_decode_prealloc_queue.retracted_queue
+                breakdown["transfer_queue_reqs"] = len(tq)
+                breakdown["transfer_queue_tokens"] = sum(
+                    int(getattr(getattr(dr, "req", None), "kv_allocated_len", 0) or 0)
+                    for dr in tq
+                )
+                breakdown["prealloc_queue_reqs"] = len(pq)
+                breakdown["prealloc_queue_tokens"] = sum(
+                    int(getattr(getattr(dr, "req", None), "kv_allocated_len", 0) or 0)
+                    for dr in pq
+                )
+                breakdown["retracted_queue_reqs"] = len(rq)
+                breakdown["retracted_queue_tokens"] = sum(
+                    int(getattr(req, "kv_allocated_len", 0) or 0)
+                    for req in rq
+                )
+            except Exception:
+                pass
+
+        return breakdown
+
     def get_internal_state(self, recv_req: GetInternalStateReq):
         ret = vars(get_global_server_args())
         ret["last_gen_throughput"] = self.last_gen_throughput
@@ -3196,6 +3279,7 @@ class Scheduler(
         ret["session_cache"] = (
             self.session_controller.get_streaming_session_cache_status()
         )
+        ret["pool_breakdown"] = self._compute_pool_breakdown_for_diagnostics()
 
         if not self.spec_algorithm.is_none() and self.spec_total_num_forward_ct > 0:
             ret["avg_spec_accept_length"] = (