From 02092ab15a8db21e5ccec469ca37799999f5cb52 Mon Sep 17 00:00:00 2001
From: CAICAIIs <3360776475@qq.com>
Date: Sat, 4 Jul 2026 12:56:29 +0800
Subject: [PATCH 1/2] fix(dsv2lite): add http reliability gate

---
 docs/models/deepseek-v2-lite/status.md        |  90 +-
 openinfer-deepseek-v2-lite/src/scheduler.rs   | 211 ++++-
 .../src/scheduler/tests.rs                    | 185 +++-
 .../src/scheduler/trace.rs                    |  55 ++
 openinfer-engine/src/engine.rs                | 151 ++-
 openinfer-vllm-frontend/src/bridge.rs         |  52 +-
 openinfer-vllm-frontend/src/bridge/tests.rs   |  71 +-
 pr-body-dsv2lite-http-reliability-gate.md     | 161 ++++
 scripts/bench_dsv2lite_http_reliability.py    | 864 ++++++++++++++++++
 tests/test_bench_dsv2lite_http_reliability.py | 114 +++
 10 files changed, 1868 insertions(+), 86 deletions(-)
 create mode 100644 pr-body-dsv2lite-http-reliability-gate.md
 create mode 100755 scripts/bench_dsv2lite_http_reliability.py
 create mode 100644 tests/test_bench_dsv2lite_http_reliability.py

diff --git a/docs/models/deepseek-v2-lite/status.md b/docs/models/deepseek-v2-lite/status.md
index d0fb2ef9..dd4942a5 100644
--- a/docs/models/deepseek-v2-lite/status.md
+++ b/docs/models/deepseek-v2-lite/status.md
@@ -1,8 +1,8 @@
 # DeepSeek-V2-Lite Status And Benchmark Ledger
 
-> **TL;DR:** DeepSeek-V2-Lite has an EP2 correctness contract across HF, host-staged, and NCCL. The retained vLLM TP2/EP2 matrix adds reproducible HTTP pressure evidence and preserves stock vLLM setup failures on this SM120/CUDA 12.8 stack. It is a benchmark snapshot, not a vLLM parity or production serving claim.
+> **TL;DR:** DeepSeek-V2-Lite has an EP2 correctness contract across HF, host-staged, and NCCL. The retained HTTP gates now cover mixed-request serving, traceable scheduler lifecycle, and a replayable reliability runner for cancel/disconnect/reject/overload recovery. This is reliability evidence only, not a vLLM parity, production serving, sparse dispatch, multi-node, or throughput-optimization claim.
 
-Last touched: 2026-06
+Last touched: 2026-07
 
 ## Capability Contract
 
@@ -22,6 +22,7 @@ Last touched: 2026-06
 | First mixed-request serving gate | Available | Issue #281 adds greedy-only request admission, FCFS deferral, explicit request-local rejection/error/finish events, and one owned `DecodeCache` per active request. The 2026-06-23 2x RTX 5090 run passed HF / host-staged / NCCL exactness and the mixed-serving E2E for host-staged and NCCL. |
 | Long-shape NCCL collectives | Available | Issue #280 chunks large bf16 dense-exchange and f32 combine all-reduces. The 2026-06-24 2x RTX 5090 NCCL checks preserve HF / host-staged / NCCL exactness and complete 24/64/128-word direct long-shape probes. |
 | HTTP trace and position-subgroup decode batching | HTTP evidence | Issue #280 logs DeepSeek-V2-Lite `openinfer_http_trace` records and batches same-position decode subgroups while letting singleton or lagging positions decode independently. The 2026-06-24 2x RTX 5090 NCCL HTTP sweeps below complete short same-shape, 128-word smoke, and mixed 16/128-word cells with full trace coverage. |
+| HTTP reliability lifecycle gate | Available | Issue #453 adds `scripts/bench_dsv2lite_http_reliability.py`, which drives real streaming `/v1/completions` scenarios for client cancel/disconnect, unsupported params, active-cap overload, mixed short/long prompts with adjacent failures, and clean follow-up recovery. The 2026-07-04 2x RTX 5090 host-staged and NCCL runs both passed with terminal trace coverage, stable output hashes, active/pending/decode maxima, and healthy final scheduler baselines. |
 | Retained vLLM comparison matrix | Snapshot complete with clean failed setup rows and supplemental validation rows | The retained matrix for tracking issue #279 keeps HF/host/NCCL correctness, OpenInfer direct diagnostic batch, `vllm bench serve` HTTP pressure, OpenInfer trace rows, and failed setup rows separate. The 2026-06-28 clean full matrix passed HF / host-staged / NCCL correctness plus OpenInfer host-staged/NCCL direct, HTTP pressure, and trace rows; stock vLLM TP2 and TP2+EP2 failed during setup on the target FlashInfer SM120 path. A separate FlashInfer #3633-equivalent validation completed vLLM TP2 and TP2+EP2 under the same HTTP client/workload contract. |
 | vLLM production parity | Not claimed | The vLLM TP2 / TP2+EP2 rows are gap-finding evidence from a documented contract. The supplemental validation run is not serving parity or a stock-install claim. |
 
@@ -39,6 +40,8 @@ The comparison gate must be run on the same model snapshot for HF, host-staged,
 
 The mixed-request serving E2E computes sequential greedy token-id oracles with `DeepSeekV2LiteEp2Generator::generate_greedy`, then submits concurrent requests through `start_engine`. The retained 2026-06-23 run covers same-length mixed prompts for same-position batch decode, different-length mixed prompts for single-row decode fallback, and a valid request submitted beside an invalid `logprobs` request to prove explicit rejection does not poison the valid stream. Host-staged and NCCL both passed the mixed-serving E2E.
 
+The HTTP reliability gate is intentionally separate from the mixed-serving E2E. It proves that the serving bridge and scheduler surface terminal states in a machine-readable way, then uses a clean follow-up request after every failure scenario to show state recovery. Rejected requests that fail in the HTTP/frontend guard may have no scheduler trace; rejected requests admitted to the DSV2-Lite scheduler must have `openinfer_http_trace` terminal evidence. Cancelled and disconnected streams are classified separately by the request-local `TokenSink` cancellation flag versus a closed shared channel.
+
 The Rust E2E accepts the known HF-confirmed RTX 5090 and A800 hash pairs for this narrow shape, because the same model snapshot has produced different exact greedy text on those hosts while still matching HF on each host. Do not use the static hash pair list as a substitute for the same-host HF comparison when changing accuracy-sensitive code.
 
 ## Benchmark Ledger
@@ -134,11 +137,86 @@ Mixed-shape proof: `prompt_words=16,128`, `num_requests=8` per repeat, four shor
 
 Interpretation: the old long-prompt prefill failure is fixed for this HTTP contract, and the post-fastpath rerun lifts mixed 16/128 throughput a bit, but the row is still dominated by long-prompt prefill and admission queueing. The c4/c8 rows prove subgroup batching can happen with mixed prompt lengths (`decode_batch_size_max=2` here), yet the latency profile is not a production serving claim.
 
+### Issue #453 No-Regression HTTP Benchmark
+
+Retained 2026-07-04 no-regression benchmark for #453 used real `/v1/completions` traffic after the reliability gate. Both host-staged and NCCL completed every cell with `failed=0`, `timeouts=0`, and full trace coverage (`missing_traces=[]`). The short same-shape rows use `prompt_words=64`, `max_tokens=64`, and concurrency `1/4/8`; the mixed rows use `prompt_words=16,128`, `max_tokens=16`, and concurrency `4/8`.
+
+| Backend | Shape | Completed | Failed/timeouts | Output tok/s | TTFT avg ms | TPOT/ITL avg ms | active max | decode batch max | traces | output hash |
+| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | --- | --- |
+| host-staged | short `64/64`, c1 | 8/8 | 0/0 | 22.886 | 1226.8 | 24.9 | 1 | 1 | 8/8 | `fed09e819c83762a` |
+| host-staged | short `64/64`, c4 | 8/8 | 0/0 | 21.698 | 2612.8 | 145.1 | 4 | 3 | 8/8 | `fed09e819c83762a` |
+| host-staged | short `64/64`, c8 | 8/8 | 0/0 | 21.830 | 5439.8 | 284.2 | 8 | 5 | 8/8 | `fed09e819c83762a` |
+| host-staged | mixed `16/128`, c4 | 8/8 | 0/0 | 8.281 | 4543.3 | 210.0 | 4 | 2 | 8/8 | `88d8d31234d66978` |
+| host-staged | mixed `16/128`, c8 | 8/8 | 0/0 | 8.288 | 7897.1 | 496.9 | 8 | 3 | 8/8 | `88d8d31234d66978` |
+| NCCL | short `64/64`, c1 | 8/8 | 0/0 | 24.873 | 1034.2 | 24.4 | 1 | 1 | 8/8 | `fed09e819c83762a` |
+| NCCL | short `64/64`, c4 | 8/8 | 0/0 | 23.678 | 2210.0 | 135.8 | 4 | 3 | 8/8 | `fed09e819c83762a` |
+| NCCL | short `64/64`, c8 | 8/8 | 0/0 | 23.872 | 4594.4 | 266.1 | 8 | 6 | 8/8 | `fed09e819c83762a` |
+| NCCL | mixed `16/128`, c4 | 8/8 | 0/0 | 9.186 | 2216.6 | 314.9 | 4 | 1 | 8/8 | `88d8d31234d66978` |
+| NCCL | mixed `16/128`, c8 | 8/8 | 0/0 | 9.269 | 6920.1 | 453.0 | 8 | 3 | 8/8 | `88d8d31234d66978` |
+
+This table is a no-regression and trace-coverage record. It is not a throughput optimization claim.
+
+### Issue #453 HTTP Reliability Gate
+
+The #453 runner is `scripts/bench_dsv2lite_http_reliability.py`. It uses standard-library HTTP streaming against `/v1/completions`, parses `openinfer_http_trace` from the server log, and writes one JSON artifact with per-scenario counts, output hashes, trace coverage, active/pending/decode maxima, terminal reasons, final healthy-baseline status, and clean follow-up results.
+
+Scenarios covered by the runner:
+
+- `cancel_disconnect`: closes a streaming request after the first token, closes another connection after early bytes, keeps a neighboring request completing, then sends a clean follow-up.
+- `invalid_requests`: sends non-greedy, `logprobs`, empty-prompt, and over-context requests beside a valid neighbor, then sends a clean follow-up.
+- `overload_active_cap`: sends more concurrent requests than the DSV2-Lite active cap (`DEFAULT_MAX_ACTIVE_REQUESTS=8`) and requires pending-queue trace evidence plus clean recovery.
+- `mixed_short_long_with_failures`: mixes 16-word and 128-word prompts with a cancelled stream and a rejected request, then sends a clean follow-up.
+
+Strict failure rules:
+
+- missing terminal trace fails, except for HTTP/frontend guard rejections that never reach the DSV2-Lite scheduler;
+- a clean follow-up failure fails;
+- success-hash drift within the same request kind fails;
+- unexplained timeout fails;
+- missing expected terminal reasons (`completed`, `rejected`, `cancelled`, `disconnected`) fail;
+- missing active-set pressure, pending-queue pressure, or decode-batch evidence fails in scenarios that require it.
+
+Validation completed for the runner schema, runner false-positive guards, scheduler accounting, the shared token-sink cancel/disconnect contract, host-staged/NCCL E2E exactness, and live host-staged/NCCL HTTP reliability:
+
+```bash
+python3 scripts/bench_dsv2lite_http_reliability.py --dry-run --out <artifact>.json
+python3 -m py_compile scripts/bench_dsv2lite_http_reliability.py tests/test_bench_dsv2lite_http_reliability.py
+python3 -m unittest tests.test_bench_dsv2lite_http_reliability
+cargo test --release -p openinfer-engine --lib token_sink -- --nocapture
+cargo test --release -p openinfer-vllm-frontend --lib abort -- --nocapture
+cargo test --release -p openinfer-deepseek-v2-lite --features deepseek-v2-lite --lib scheduler -- --nocapture
+OPENINFER_TEST_MODEL_PATH=models/DeepSeek-V2-Lite OPENINFER_DSV2_LITE_EP_BACKEND=host-staged cargo test --release -p openinfer-deepseek-v2-lite --features deepseek-v2-lite --test e2e_ep2 -- --nocapture
+OPENINFER_TEST_MODEL_PATH=models/DeepSeek-V2-Lite OPENINFER_DSV2_LITE_EP_BACKEND=nccl cargo test --release -p openinfer-deepseek-v2-lite --features deepseek-v2-lite --test e2e_ep2 -- --nocapture
+```
+
+The local dry-run passed all four scenarios and emits deterministic JSON. `tests/test_bench_dsv2lite_http_reliability.py` passed and verifies that the runner fails when a client-observed disconnect lacks a matching scheduler terminal reason or when trace fields are missing. `cargo test --release -p openinfer-engine --lib token_sink -- --nocapture` passed and verifies explicit cancel versus closed receiver behavior. `cargo test --release -p openinfer-vllm-frontend --lib abort -- --nocapture` passed and verifies that frontend aborts drop late tokens and classify disconnect before the first client-visible token separately from cancel after token output. The DSV2-Lite scheduler lifecycle subset passed `23 passed; 0 failed`; host-staged and NCCL `e2e_ep2` each passed `1 passed; 0 failed`. The SM120 NCCL validation used an NCCL 2.30.7 runtime library.
+
+Retained 2026-07-04 live HTTP reliability artifacts from real `/v1/completions` traffic:
+
+| Backend | Artifact | SHA-256 | Result | Scenario coverage |
+| --- | --- | --- | --- | --- |
+| host-staged | `http_reliability_host_staged.json` | `832d65a8e8b2b3a6ad0100c4a35f38475f040d6ffc192ec38a3b7384167187a5` | passed | cancel/disconnect, invalid requests, overload active-cap, mixed short/long with failures, clean follow-up after every scenario |
+| NCCL | `http_reliability_nccl.json` | `53bedd98f19c5241df588a1ade8756e84a5e8c99225a589c4ed303e90fba38fa` | passed | cancel/disconnect, invalid requests, overload active-cap, mixed short/long with failures, clean follow-up after every scenario |
+
+Scenario summaries:
+
+| Backend | Scenario | Counts | Terminal reasons | Trace maxima | Final baseline |
+| --- | --- | --- | --- | --- | --- |
+| host-staged | `cancel_disconnect` | completed `2`, cancelled `1`, disconnected `1`, failed/rejected/timeout `0` | cancelled `1`, disconnected `1`, completed_length `2` | active `2`, pending `0`, decode `1` | healthy |
+| host-staged | `invalid_requests` | completed `2`, rejected `4`, failed/timeout `0` | rejected and completed_length observed | active `1`, pending `0`, decode `1` | healthy |
+| host-staged | `overload_active_cap` | completed `13`, failed/rejected/timeout `0` | completed_length `13` | active `8`, pending `4`, decode `7` | healthy |
+| host-staged | `mixed_short_long_with_failures` | completed `5`, cancelled `1`, rejected `1`, failed/timeout `0` | cancelled, rejected, completed_length observed | active `4`, pending `0`, decode `2` | healthy |
+| NCCL | `cancel_disconnect` | completed `2`, cancelled `1`, disconnected `1`, failed/rejected/timeout `0` | cancelled `1`, disconnected `1`, completed_length `2` | active `2`, pending `0`, decode `1` | healthy |
+| NCCL | `invalid_requests` | completed `2`, rejected `4`, failed/timeout `0` | rejected and completed_length observed | active `1`, pending `0`, decode `1` | healthy |
+| NCCL | `overload_active_cap` | completed `13`, failed/rejected/timeout `0` | completed_length `13` | active `8`, pending `4`, decode `7` | healthy |
+| NCCL | `mixed_short_long_with_failures` | completed `5`, cancelled `1`, rejected `1`, failed/timeout `0` | cancelled, rejected, completed_length observed | active `5`, pending `0`, decode `2` | healthy |
+
 ### Interpretation
 
 - direct same-prompt diagnostics show NCCL is still much slower than host-staged, although aggregate decode throughput improves with larger diagnostic batch size;
 - NCCL remains a correctness-first backend and is still significantly slower than host-staged;
 - the #280 HTTP trace proves active request sets and subgroup decode batches, but throughput still scales only weakly on NCCL EP2 and long prompts have high TTFT;
+- the #453 runner and trace fields make failure isolation auditable; host-staged and NCCL live HTTP artifacts now prove cancel/disconnect/reject/overload/mixed-failure cleanup and clean follow-up recovery on the retained 2-GPU validation contract;
 - the 2026-06-28 clean matrix keeps stock vLLM startup failures visible because they are part of the reproducibility record;
 - the supplemental vLLM validation shows the HTTP contract can run after the FlashInfer SM120/CUDA 12.8 path is fixed, but it should stay separate from stock-package rows;
 - future performance claims should use the retained matrix contract, not older short-shape vLLM experiments.
@@ -153,6 +231,7 @@ Use these labels consistently:
 | `direct same-prompt diagnostic batch` | Fixed same-prompt direct batch sizes `1/4/8`. | Production continuous batching or mixed-request scheduling. |
 | `first mixed-request serving gate` | Greedy-only EP2 scheduler path with explicit admission/rejection/deferral, per-request host-side decode `DecodeCache`, active cap `8`, and exact sequential-oracle E2E. | vLLM parity, sparse dispatch, production EP readiness, HTTP throughput scaling, non-greedy sampling, or logprobs support. |
 | `HTTP trace/subgroup evidence` | `/v1/completions` requests have per-request `openinfer_http_trace` rows, and HTTP sweeps show non-1 `active_set_size` and `decode_batch_size_max`. | Fair vLLM parity, long-prompt latency readiness, or a before/after percentage unless a paired baseline run is recorded. |
+| `HTTP reliability lifecycle gate` | `/v1/completions` cancel/disconnect/reject/overload/mixed-failure scenarios have terminal reason counts, trace coverage, active/pending/decode maxima, output hashes, and clean follow-up recovery evidence. | Production EP readiness, soak stability, SLO latency, vLLM parity, throughput improvement, sparse dispatch, or multi-node EP support. |
 | `covered NCCL decode graph probe` | Probe-only batch-1 `Hello` decode step captured, instantiated, replayed, and token-verified under CUDA Graph. | Default serving graph coverage, multi-step graph replay, batch `4/8` graph coverage, or performance improvement. |
 | `HTTP concurrency pressure` | `vllm bench serve --max-concurrency N` against an HTTP endpoint. | True OpenInfer batch size unless the engine path proves it. |
 | `vLLM comparison from documented environment` | vLLM TP2 / TP2+EP2 from the retained matrix or the separate FlashInfer-fixed validation. | Stock vLLM install support, OpenInfer serving parity, or production readiness. |
@@ -200,6 +279,11 @@ The next implementation should be chosen from measured evidence:
    - reduce long-prompt prefill and admission-queue TTFT before claiming long-prompt serving readiness;
    - add paired baseline runs before claiming a percentage speedup from subgroup batching.
 
-5. Keep MoE internals readable.
+5. Keep the #453 HTTP reliability evidence retained.
+   - rerun the reliability runner for host-staged and NCCL when scheduler or HTTP lifecycle code changes;
+   - keep the JSON artifact hashes and server-log trace coverage in the PR evidence;
+   - keep #452 long/mixed-prompt latency, #465 soak, #466 SLO report, and #467 benchmark manifest separate.
+
+6. Keep MoE internals readable.
    - routing, dispatch, expert execution, and combine should remain distinguishable in code and attribution;
    - avoid introducing a generic EP framework before the DeepSeek-V2-Lite EP2 path has a measured reason to need it.
diff --git a/openinfer-deepseek-v2-lite/src/scheduler.rs b/openinfer-deepseek-v2-lite/src/scheduler.rs
index 07171d8e..132ddb6d 100644
--- a/openinfer-deepseek-v2-lite/src/scheduler.rs
+++ b/openinfer-deepseek-v2-lite/src/scheduler.rs
@@ -144,6 +144,8 @@ impl MixedRequestScheduler {
                         FinishReason::Error,
                         0,
                         Some(&message),
+                        self.active.len(),
+                        self.pending.len(),
                     );
                     let _ = pending.token_tx.send(TokenEvent::Rejected {
                         message,
@@ -151,7 +153,12 @@ impl MixedRequestScheduler {
                         completion_tokens: 0,
                     });
                 }
-                Err(scheduled) => log_schedule_disconnect_trace(&pending, &scheduled),
+                Err(scheduled) => log_schedule_disconnect_trace(
+                    &pending,
+                    &scheduled,
+                    self.active.len(),
+                    self.pending.len(),
+                ),
             }
         }
 
@@ -159,14 +166,27 @@ impl MixedRequestScheduler {
             match send_scheduled(&pending) {
                 Ok(scheduled) => {
                     let _ = send_prompt_echo(&pending);
-                    log_pending_terminal_trace(&pending, &scheduled, FinishReason::Length, 0, None);
+                    log_pending_terminal_trace(
+                        &pending,
+                        &scheduled,
+                        FinishReason::Length,
+                        0,
+                        None,
+                        self.active.len(),
+                        self.pending.len(),
+                    );
                     let _ = pending.token_tx.send(TokenEvent::Finished {
                         finish_reason: FinishReason::Length,
                         prompt_tokens: pending.prompt_tokens.len(),
                         completion_tokens: 0,
                     });
                 }
-                Err(scheduled) => log_schedule_disconnect_trace(&pending, &scheduled),
+                Err(scheduled) => log_schedule_disconnect_trace(
+                    &pending,
+                    &scheduled,
+                    self.active.len(),
+                    self.pending.len(),
+                ),
             }
         }
 
@@ -186,18 +206,26 @@ impl MixedRequestScheduler {
         let scheduled = match send_scheduled(&pending) {
             Ok(scheduled) => scheduled,
             Err(scheduled) => {
-                log_schedule_disconnect_trace(&pending, &scheduled);
+                log_schedule_disconnect_trace(
+                    &pending,
+                    &scheduled,
+                    self.active.len(),
+                    self.pending.len(),
+                );
                 return None;
             }
         };
 
         if !send_prompt_echo(&pending) {
+            let terminal_message = terminal_send_failure_message(&pending.token_tx, "prompt echo");
             log_pending_terminal_trace(
                 &pending,
                 &scheduled,
                 FinishReason::Error,
                 0,
-                Some("client disconnected before prompt echo"),
+                Some(&terminal_message),
+                self.active.len(),
+                self.pending.len(),
             );
             return None;
         }
@@ -221,6 +249,8 @@ impl MixedRequestScheduler {
                     prompt_len,
                     prefill_start.elapsed().as_secs_f64() * 1000.0,
                     &message,
+                    self.active.len(),
+                    self.pending.len(),
                 );
                 let _ = pending.token_tx.send(TokenEvent::Error {
                     message,
@@ -253,9 +283,11 @@ impl MixedRequestScheduler {
                 prefill_ms,
             ),
         };
-        active.trace.note_active_set(self.active.len() + 1);
+        active
+            .trace
+            .note_scheduler_state(self.active.len() + 1, self.pending.len());
 
-        if active.emit_token_or_finish(next) {
+        if active.emit_token_or_finish(next, self.active.len(), self.pending.len()) {
             return None;
         }
         Some(active)
@@ -268,11 +300,14 @@ impl MixedRequestScheduler {
             return;
         }
         for state in &mut self.active {
-            state.trace.note_active_set(active_set_size);
+            state
+                .trace
+                .note_scheduler_state(active_set_size, self.pending.len());
         }
         if active_set_size == 1 {
             let row = self.active.pop().expect("single active row present");
-            if let Some(survivor) = self.decode_single_row((0, row)) {
+            let mut active_remaining = active_set_size;
+            if let Some(survivor) = self.decode_single_row((0, row), &mut active_remaining) {
                 self.active.push(survivor.1);
             }
             return;
@@ -280,19 +315,29 @@ impl MixedRequestScheduler {
 
         if let Some(position) = common_decode_position(&self.active) {
             let rows: Vec<_> = self.active.drain(..).enumerate().collect();
-            self.active = restore_surviving_rows(self.decode_batch_group(position, rows));
+            let mut active_remaining = active_set_size;
+            self.active = restore_surviving_rows(self.decode_batch_group(
+                position,
+                rows,
+                &mut active_remaining,
+            ));
             return;
         }
 
         let groups = take_decode_position_groups(&mut self.active);
         let mut survivors = Vec::new();
+        let mut active_remaining = active_set_size;
         for group in groups {
             if group.rows.len() > 1 {
-                survivors.extend(self.decode_batch_group(group.position, group.rows));
+                survivors.extend(self.decode_batch_group(
+                    group.position,
+                    group.rows,
+                    &mut active_remaining,
+                ));
             } else {
                 let mut rows = group.rows;
                 if let Some(row) = rows.pop() {
-                    if let Some(survivor) = self.decode_single_row(row) {
+                    if let Some(survivor) = self.decode_single_row(row, &mut active_remaining) {
                         survivors.push(survivor);
                     }
                 }
@@ -303,11 +348,22 @@ impl MixedRequestScheduler {
 
     fn retire_bad_cache_positions(&mut self) {
         let config = self.generator.config();
+        let active_set_size = self.active.len();
+        let pending_queue_size = self.pending.len();
+        let mut active_set_size_after_terminal = active_set_size;
         let mut survivors = Vec::with_capacity(self.active.len());
         for state in self.active.drain(..) {
             match state.cache_position(config) {
                 Ok(()) => survivors.push(state),
-                Err(message) => state.emit_error(message.to_string()),
+                Err(message) => {
+                    active_set_size_after_terminal =
+                        active_set_size_after_terminal.saturating_sub(1);
+                    state.emit_error(
+                        message.to_string(),
+                        active_set_size_after_terminal,
+                        pending_queue_size,
+                    );
+                }
             }
         }
         self.active = survivors;
@@ -317,6 +373,7 @@ impl MixedRequestScheduler {
         &mut self,
         position: usize,
         rows: Vec<(usize, ActiveRequestState)>,
+        active_remaining: &mut usize,
     ) -> Vec<(usize, ActiveRequestState)> {
         let group_size = rows.len();
         let (indices, mut states): (Vec<_>, Vec<_>) = rows.into_iter().unzip();
@@ -344,13 +401,20 @@ impl MixedRequestScheduler {
         );
         let decode_ms = decode_start.elapsed().as_secs_f64() * 1000.0;
 
+        let pending_queue_size = self.pending.len();
         match result {
             Ok(next_tokens) if next_tokens.len() == group_size => {
                 for (state, cache) in states.iter_mut().zip(caches) {
                     state.cache = cache;
                     state.trace.note_decode_step(group_size, decode_ms);
                 }
-                apply_decoded_tokens_to_rows(indices, states, next_tokens)
+                apply_decoded_tokens_to_rows(
+                    indices,
+                    states,
+                    next_tokens,
+                    active_remaining,
+                    pending_queue_size,
+                )
             }
             // The batched path mutates per-row caches as it advances through the
             // model. This gate avoids full-cache rollback clones; a batch decode
@@ -363,6 +427,8 @@ impl MixedRequestScheduler {
                         next_tokens.len(),
                         group_size
                     ),
+                    active_remaining,
+                    pending_queue_size,
                 );
                 Vec::new()
             }
@@ -373,6 +439,8 @@ impl MixedRequestScheduler {
                         "DeepSeek-V2-Lite batched decode failed for {} active requests: {err}",
                         group_size
                     ),
+                    active_remaining,
+                    pending_queue_size,
                 );
                 Vec::new()
             }
@@ -382,6 +450,7 @@ impl MixedRequestScheduler {
     fn decode_single_row(
         &mut self,
         (idx, mut state): (usize, ActiveRequestState),
+        active_remaining: &mut usize,
     ) -> Option<(usize, ActiveRequestState)> {
         let token = state.last_token;
         let position = state.next_decode_position();
@@ -399,10 +468,17 @@ impl MixedRequestScheduler {
         match result {
             Ok(next) => {
                 state.trace.note_decode_step(1, decode_ms);
-                (!state.emit_token_or_finish(next)).then_some((idx, state))
+                let terminal_active_set_size = active_remaining.saturating_sub(1);
+                if state.emit_token_or_finish(next, terminal_active_set_size, self.pending.len()) {
+                    *active_remaining = terminal_active_set_size;
+                    None
+                } else {
+                    Some((idx, state))
+                }
             }
             Err(err) => {
-                state.emit_error(err.to_string());
+                *active_remaining = active_remaining.saturating_sub(1);
+                state.emit_error(err.to_string(), *active_remaining, self.pending.len());
                 None
             }
         }
@@ -413,19 +489,33 @@ fn apply_decoded_tokens_to_rows(
     indices: Vec<usize>,
     states: Vec<ActiveRequestState>,
     next_tokens: Vec<u32>,
+    active_remaining: &mut usize,
+    pending_queue_size: usize,
 ) -> Vec<(usize, ActiveRequestState)> {
     indices
         .into_iter()
         .zip(states.into_iter().zip(next_tokens))
         .filter_map(|(idx, (mut state, token))| {
-            (!state.emit_token_or_finish(token)).then_some((idx, state))
+            let terminal_active_set_size = active_remaining.saturating_sub(1);
+            if state.emit_token_or_finish(token, terminal_active_set_size, pending_queue_size) {
+                *active_remaining = terminal_active_set_size;
+                None
+            } else {
+                Some((idx, state))
+            }
         })
         .collect()
 }
 
-fn retire_rows_with_error(states: Vec<ActiveRequestState>, message: String) {
+fn retire_rows_with_error(
+    states: Vec<ActiveRequestState>,
+    message: String,
+    active_remaining: &mut usize,
+    pending_queue_size: usize,
+) {
     for state in states {
-        state.emit_error(message.clone());
+        *active_remaining = active_remaining.saturating_sub(1);
+        state.emit_error(message.clone(), *active_remaining, pending_queue_size);
     }
 }
 
@@ -462,10 +552,20 @@ impl ActiveRequestState {
         Ok(())
     }
 
-    fn emit_token_or_finish(&mut self, token: u32) -> bool {
+    fn emit_token_or_finish(
+        &mut self,
+        token: u32,
+        active_set_size_at_terminal: usize,
+        pending_queue_size_at_terminal: usize,
+    ) -> bool {
         self.last_token = token;
         if !self.finish_policy.ignore_eos && token == self.finish_policy.eos_token_id {
-            self.log_http_trace(FinishReason::Stop, None);
+            self.log_http_trace(
+                FinishReason::Stop,
+                None,
+                active_set_size_at_terminal,
+                pending_queue_size_at_terminal,
+            );
             let _ = self.token_tx.send(TokenEvent::Finished {
                 finish_reason: FinishReason::Stop,
                 prompt_tokens: self.prompt_len,
@@ -487,9 +587,12 @@ impl ActiveRequestState {
             })
             .is_err()
         {
+            let terminal_message = terminal_send_failure_message(&self.token_tx, "token emit");
             self.log_http_trace(
                 FinishReason::Error,
-                Some("client disconnected before token emit"),
+                Some(&terminal_message),
+                active_set_size_at_terminal,
+                pending_queue_size_at_terminal,
             );
             return true;
         }
@@ -499,7 +602,12 @@ impl ActiveRequestState {
         self.generated += 1;
 
         if self.generated == self.max_tokens {
-            self.log_http_trace(FinishReason::Length, None);
+            self.log_http_trace(
+                FinishReason::Length,
+                None,
+                active_set_size_at_terminal,
+                pending_queue_size_at_terminal,
+            );
             let _ = self.token_tx.send(TokenEvent::Finished {
                 finish_reason: FinishReason::Length,
                 prompt_tokens: self.prompt_len,
@@ -510,8 +618,18 @@ impl ActiveRequestState {
         false
     }
 
-    fn emit_error(self, message: String) {
-        self.log_http_trace(FinishReason::Error, Some(&message));
+    fn emit_error(
+        mut self,
+        message: String,
+        active_set_size_at_terminal: usize,
+        pending_queue_size_at_terminal: usize,
+    ) {
+        self.log_http_trace(
+            FinishReason::Error,
+            Some(&message),
+            active_set_size_at_terminal,
+            pending_queue_size_at_terminal,
+        );
         let _ = self.token_tx.send(TokenEvent::Error {
             message,
             prompt_tokens: self.prompt_len,
@@ -519,7 +637,15 @@ impl ActiveRequestState {
         });
     }
 
-    fn log_http_trace(&self, finish_reason: FinishReason, error: Option<&str>) {
+    fn log_http_trace(
+        &mut self,
+        finish_reason: FinishReason,
+        error: Option<&str>,
+        active_set_size_at_terminal: usize,
+        pending_queue_size_at_terminal: usize,
+    ) {
+        self.trace
+            .note_terminal_state(active_set_size_at_terminal, pending_queue_size_at_terminal);
         log_http_trace(
             trace_request_id(self.request_id.as_deref(), &self.token_tx),
             &self.trace,
@@ -567,6 +693,16 @@ fn send_prompt_echo(pending: &PendingRequest) -> bool {
         .is_ok()
 }
 
+fn terminal_send_failure_message(token_tx: &TokenSink, stage: &str) -> String {
+    if token_tx.is_disconnected() {
+        format!("client disconnected before {stage}")
+    } else if token_tx.is_cancelled() {
+        format!("client cancelled before {stage}")
+    } else {
+        format!("client disconnected before {stage}")
+    }
+}
+
 fn trace_request_id<'a>(request_id: Option<&'a str>, token_tx: &'a TokenSink) -> &'a str {
     request_id.unwrap_or_else(|| token_tx.tag().as_ref())
 }
@@ -577,8 +713,12 @@ fn log_pending_terminal_trace(
     finish_reason: FinishReason,
     completion_tokens: usize,
     error: Option<&str>,
+    active_set_size_at_terminal: usize,
+    pending_queue_size_at_terminal: usize,
 ) {
-    let trace = RequestTrace::terminal(scheduled.queued_at_unix_s, scheduled.scheduled_at_unix_s);
+    let mut trace =
+        RequestTrace::terminal(scheduled.queued_at_unix_s, scheduled.scheduled_at_unix_s);
+    trace.note_terminal_state(active_set_size_at_terminal, pending_queue_size_at_terminal);
     log_http_trace(
         trace_request_id(pending.request_id.as_deref(), &pending.token_tx),
         &trace,
@@ -589,13 +729,23 @@ fn log_pending_terminal_trace(
     );
 }
 
-fn log_schedule_disconnect_trace(pending: &PendingRequest, scheduled: &ScheduledTrace) {
+fn log_schedule_disconnect_trace(
+    pending: &PendingRequest,
+    scheduled: &ScheduledTrace,
+    active_set_size_at_terminal: usize,
+    pending_queue_size_at_terminal: usize,
+) {
     log_pending_terminal_trace(
         pending,
         scheduled,
         FinishReason::Error,
         0,
-        Some("client disconnected before scheduled event"),
+        Some(&terminal_send_failure_message(
+            &pending.token_tx,
+            "scheduled event",
+        )),
+        active_set_size_at_terminal,
+        pending_queue_size_at_terminal,
     );
 }
 
@@ -605,11 +755,14 @@ fn log_prefill_error_trace(
     prompt_tokens: usize,
     prefill_ms: f64,
     message: &str,
+    active_set_size_at_terminal: usize,
+    pending_queue_size_at_terminal: usize,
 ) {
     let mut trace =
         RequestTrace::terminal(scheduled.queued_at_unix_s, scheduled.scheduled_at_unix_s);
     trace.prefill_done_unix_s = Some(unix_now_s());
     trace.prefill_ms = Some(prefill_ms);
+    trace.note_terminal_state(active_set_size_at_terminal, pending_queue_size_at_terminal);
     log_http_trace(
         trace_request_id(pending.request_id.as_deref(), &pending.token_tx),
         &trace,
diff --git a/openinfer-deepseek-v2-lite/src/scheduler/tests.rs b/openinfer-deepseek-v2-lite/src/scheduler/tests.rs
index ab9bb9fc..e1f9ab40 100644
--- a/openinfer-deepseek-v2-lite/src/scheduler/tests.rs
+++ b/openinfer-deepseek-v2-lite/src/scheduler/tests.rs
@@ -1,6 +1,6 @@
-use std::sync::{Arc, atomic::AtomicBool};
+use std::sync::{Arc, atomic::AtomicU8};
 
-use openinfer_engine::engine::RequestTag;
+use openinfer_engine::engine::{RequestAbortReason, RequestTag};
 use openinfer_engine::sampler::SamplingParams;
 use tokio::sync::mpsc;
 
@@ -227,6 +227,44 @@ fn terminal_admission_events_keep_scheduler_contract() {
     ));
 }
 
+#[test]
+fn immediate_prefill_finish_reports_existing_active_count() {
+    let config = test_lite_config();
+    let (token_tx, mut token_rx) = TokenSink::standalone();
+    let mut state = active_state("one-token", token_tx, 3, 0, 11, &config);
+    state.max_tokens = 1;
+
+    assert!(state.emit_token_or_finish(12, 2, 1));
+
+    match recv_event(&mut token_rx) {
+        TokenEvent::Token { id, .. } => assert_eq!(id, 12),
+        _ => panic!("expected first generated token"),
+    }
+    match recv_event(&mut token_rx) {
+        TokenEvent::Finished {
+            finish_reason,
+            completion_tokens,
+            ..
+        } => {
+            assert_eq!(finish_reason, FinishReason::Length);
+            assert_eq!(completion_tokens, 1);
+        }
+        _ => panic!("expected terminal length event"),
+    }
+
+    let payload = http_trace_payload(
+        "one-token",
+        &state.trace,
+        state.prompt_len,
+        state.generated,
+        FinishReason::Length,
+        None,
+    );
+    assert_eq!(payload["active_set_size_at_terminal"], 2);
+    assert_eq!(payload["pending_queue_size_at_terminal"], 1);
+    assert_eq!(payload["healthy_baseline_after_terminal"], false);
+}
+
 #[test]
 fn send_scheduled_returns_trace_when_client_is_closed() {
     let (pending, rx) = request("closed-before-schedule", 2, 1);
@@ -259,8 +297,8 @@ fn eos_retirement_is_independent_per_request() {
     let mut stop_state = active_state("stop", tx_stop, 3, 1, 10, &config);
     let mut live_state = active_state("live", tx_live, 2, 1, 11, &config);
 
-    assert!(stop_state.emit_token_or_finish(config.eos_token_id));
-    assert!(!live_state.emit_token_or_finish(12));
+    assert!(stop_state.emit_token_or_finish(config.eos_token_id, 2, 0));
+    assert!(!live_state.emit_token_or_finish(12, 2, 0));
 
     match recv_event(&mut rx_stop) {
         TokenEvent::Finished {
@@ -288,12 +326,16 @@ fn batch_decoded_tokens_retire_eos_independently() {
     let stop_state = active_state("stop", tx_stop, 3, 1, 10, &config);
     let live_state = active_state("live", tx_live, 3, 1, 11, &config);
 
+    let mut active_remaining = 2;
     let survivors = apply_decoded_tokens_to_rows(
         vec![0, 1],
         vec![stop_state, live_state],
         vec![config.eos_token_id, 12],
+        &mut active_remaining,
+        0,
     );
 
+    assert_eq!(active_remaining, 1);
     assert_eq!(survivors.len(), 1);
     assert_eq!(survivors[0].0, 1);
     assert_eq!(survivors[0].1.request_id.as_deref(), Some("live"));
@@ -319,7 +361,7 @@ fn batch_decoded_tokens_retire_eos_independently() {
 fn cancelled_token_sink_retires_request() {
     let config = test_lite_config();
     let (stream_tx, mut stream_rx) = mpsc::unbounded_channel();
-    let cancelled = Arc::new(AtomicBool::new(true));
+    let cancelled = Arc::new(AtomicU8::new(RequestAbortReason::Cancelled as u8));
     let sink = TokenSink::new(
         RequestTag::from("cancelled"),
         stream_tx,
@@ -327,7 +369,7 @@ fn cancelled_token_sink_retires_request() {
     );
     let mut state = active_state("cancelled", sink, 2, 1, 11, &config);
 
-    assert!(state.emit_token_or_finish(12));
+    assert!(state.emit_token_or_finish(12, 1, 0));
     assert!(stream_rx.try_recv().is_err());
 }
 
@@ -338,7 +380,7 @@ fn closed_token_sink_retires_request() {
     drop(rx);
     let mut state = active_state("closed", sink, 2, 1, 11, &config);
 
-    assert!(state.emit_token_or_finish(12));
+    assert!(state.emit_token_or_finish(12, 1, 0));
 }
 
 #[test]
@@ -351,7 +393,9 @@ fn batch_decode_error_retires_all_active_requests() {
         active_state("second", second_tx, 4, 1, 12, &config),
     ];
 
-    retire_rows_with_error(active, "batch failed".to_string());
+    let mut active_remaining = 2;
+    retire_rows_with_error(active, "batch failed".to_string(), &mut active_remaining, 0);
+    assert_eq!(active_remaining, 0);
 
     match recv_event(&mut first_rx) {
         TokenEvent::Error {
@@ -389,7 +433,14 @@ fn subgroup_decode_error_retires_only_group_rows() {
     let _untouched = active_state("second", second_tx, 4, 1, 12, &config);
     let third = active_state("third", third_tx, 3, 2, 13, &config);
 
-    retire_rows_with_error(vec![first, third], "group failed".to_string());
+    let mut active_remaining = 3;
+    retire_rows_with_error(
+        vec![first, third],
+        "group failed".to_string(),
+        &mut active_remaining,
+        0,
+    );
+    assert_eq!(active_remaining, 1);
 
     match recv_event(&mut first_rx) {
         TokenEvent::Error { message, .. } => assert_eq!(message, "group failed"),
@@ -402,25 +453,139 @@ fn subgroup_decode_error_retires_only_group_rows() {
     }
 }
 
+#[test]
+fn cross_group_terminal_accounting_uses_shared_active_remaining() {
+    let config = test_lite_config();
+    let (first_tx, mut first_rx) = TokenSink::standalone();
+    let (second_tx, mut second_rx) = TokenSink::standalone();
+    let (third_tx, mut third_rx) = TokenSink::standalone();
+    let first = active_state("first", first_tx, 3, 2, 11, &config);
+    let second = active_state("second", second_tx, 3, 2, 12, &config);
+    let mut third = active_state("third", third_tx, 3, 1, 13, &config);
+
+    let mut active_remaining = 3;
+    retire_rows_with_error(
+        vec![first, second],
+        "group failed".to_string(),
+        &mut active_remaining,
+        0,
+    );
+    assert_eq!(active_remaining, 1);
+
+    assert!(third.emit_token_or_finish(config.eos_token_id, active_remaining.saturating_sub(1), 0));
+    active_remaining = active_remaining.saturating_sub(1);
+    assert_eq!(active_remaining, 0);
+
+    let first_payload = match recv_event(&mut first_rx) {
+        TokenEvent::Error { .. } => http_trace_payload(
+            "first",
+            &trace(),
+            3,
+            2,
+            FinishReason::Error,
+            Some("group failed"),
+        ),
+        _ => panic!("first subgroup row should receive decode error"),
+    };
+    assert_eq!(first_payload["terminal_reason"], "error");
+    match recv_event(&mut second_rx) {
+        TokenEvent::Error { message, .. } => assert_eq!(message, "group failed"),
+        _ => panic!("second subgroup row should receive decode error"),
+    }
+    match recv_event(&mut third_rx) {
+        TokenEvent::Finished { finish_reason, .. } => {
+            assert_eq!(finish_reason, FinishReason::Stop);
+        }
+        _ => panic!("third row should finish independently"),
+    }
+}
+
 #[test]
 fn http_trace_payload_includes_error_and_batch_fields() {
     let mut trace = trace();
-    trace.note_active_set(4);
+    trace.note_scheduler_state(4, 2);
     trace.note_decode_step(2, 7.5);
+    trace.note_terminal_state(1, 0);
 
     let payload = http_trace_payload("req-a", &trace, 3, 2, FinishReason::Error, Some("boom"));
 
     assert_eq!(payload["request_id"], "req-a");
     assert_eq!(payload["finish_reason"], "error");
+    assert_eq!(payload["terminal_reason"], "error");
     assert_eq!(payload["prompt_tokens"], 3);
     assert_eq!(payload["completion_tokens"], 2);
     assert_eq!(payload["active_set_size"], 4);
+    assert_eq!(payload["active_set_size_max"], 4);
+    assert_eq!(payload["pending_queue_size_max"], 2);
+    assert_eq!(payload["active_set_size_at_terminal"], 1);
+    assert_eq!(payload["pending_queue_size_at_terminal"], 0);
+    assert_eq!(payload["healthy_baseline_after_terminal"], false);
     assert_eq!(payload["decode_batch_size_max"], 2);
     assert_eq!(payload["batch_decode_steps"], 1);
     assert_eq!(payload["first_decode_ms"], 7.5);
     assert_eq!(payload["error"], "boom");
 }
 
+#[test]
+fn terminal_reason_labels_are_machine_readable() {
+    let trace = trace();
+
+    let cancelled = http_trace_payload(
+        "cancelled",
+        &trace,
+        2,
+        1,
+        FinishReason::Error,
+        Some("client cancelled before token emit"),
+    );
+    assert_eq!(cancelled["terminal_reason"], "cancelled");
+
+    let disconnected = http_trace_payload(
+        "disconnected",
+        &trace,
+        2,
+        1,
+        FinishReason::Error,
+        Some("client disconnected before token emit"),
+    );
+    assert_eq!(disconnected["terminal_reason"], "disconnected");
+
+    let rejected = http_trace_payload(
+        "rejected",
+        &trace,
+        2,
+        0,
+        FinishReason::Error,
+        Some("DeepSeek-V2-Lite EP=2 mixed serving gate does not return logprobs yet"),
+    );
+    assert_eq!(rejected["terminal_reason"], "rejected");
+
+    let length = http_trace_payload("length", &trace, 2, 1, FinishReason::Length, None);
+    assert_eq!(length["terminal_reason"], "completed_length");
+}
+
+#[test]
+fn terminal_send_failure_message_distinguishes_cancel_from_disconnect() {
+    let (stream_tx, _stream_rx) = mpsc::unbounded_channel();
+    let cancelled_flag = Arc::new(AtomicU8::new(RequestAbortReason::Cancelled as u8));
+    let cancelled = TokenSink::new(
+        RequestTag::from("cancelled"),
+        stream_tx,
+        Arc::clone(&cancelled_flag),
+    );
+    assert_eq!(
+        terminal_send_failure_message(&cancelled, "token emit"),
+        "client cancelled before token emit"
+    );
+
+    let (closed, closed_rx) = TokenSink::standalone();
+    drop(closed_rx);
+    assert_eq!(
+        terminal_send_failure_message(&closed, "token emit"),
+        "client disconnected before token emit"
+    );
+}
+
 #[test]
 fn decode_grouping_batches_same_position_subgroups() {
     assert!(decode_position_groups_for_positions(&[]).is_empty());
diff --git a/openinfer-deepseek-v2-lite/src/scheduler/trace.rs b/openinfer-deepseek-v2-lite/src/scheduler/trace.rs
index 94b90de0..10b6756e 100644
--- a/openinfer-deepseek-v2-lite/src/scheduler/trace.rs
+++ b/openinfer-deepseek-v2-lite/src/scheduler/trace.rs
@@ -9,6 +9,9 @@ pub(super) struct RequestTrace {
     pub(super) prefill_ms: Option<f64>,
     pub(super) first_decode_ms: Option<f64>,
     pub(super) active_set_size_max: usize,
+    pub(super) pending_queue_size_max: usize,
+    pub(super) active_set_size_at_terminal: usize,
+    pub(super) pending_queue_size_at_terminal: usize,
     pub(super) decode_batch_size_max: usize,
     pub(super) batch_decode_steps: usize,
 }
@@ -34,6 +37,9 @@ impl RequestTrace {
             prefill_ms: Some(prefill_ms),
             first_decode_ms: None,
             active_set_size_max: 1,
+            pending_queue_size_max: 0,
+            active_set_size_at_terminal: 0,
+            pending_queue_size_at_terminal: 0,
             decode_batch_size_max: 0,
             batch_decode_steps: 0,
         }
@@ -48,6 +54,9 @@ impl RequestTrace {
             prefill_ms: None,
             first_decode_ms: None,
             active_set_size_max: 0,
+            pending_queue_size_max: 0,
+            active_set_size_at_terminal: 0,
+            pending_queue_size_at_terminal: 0,
             decode_batch_size_max: 0,
             batch_decode_steps: 0,
         }
@@ -57,6 +66,15 @@ impl RequestTrace {
         self.active_set_size_max = self.active_set_size_max.max(active_set_size);
     }
 
+    pub(super) fn note_scheduler_state(
+        &mut self,
+        active_set_size: usize,
+        pending_queue_size: usize,
+    ) {
+        self.note_active_set(active_set_size);
+        self.pending_queue_size_max = self.pending_queue_size_max.max(pending_queue_size);
+    }
+
     pub(super) fn note_decode_step(&mut self, batch_size: usize, decode_ms: f64) {
         self.first_decode_ms.get_or_insert(decode_ms);
         self.decode_batch_size_max = self.decode_batch_size_max.max(batch_size);
@@ -64,6 +82,16 @@ impl RequestTrace {
             self.batch_decode_steps += 1;
         }
     }
+
+    pub(super) fn note_terminal_state(
+        &mut self,
+        active_set_size_at_terminal: usize,
+        pending_queue_size_at_terminal: usize,
+    ) {
+        self.active_set_size_at_terminal = active_set_size_at_terminal;
+        self.pending_queue_size_at_terminal = pending_queue_size_at_terminal;
+        self.note_scheduler_state(active_set_size_at_terminal, pending_queue_size_at_terminal);
+    }
 }
 
 pub(super) fn http_trace_payload(
@@ -85,7 +113,14 @@ pub(super) fn http_trace_payload(
         "prompt_tokens": prompt_tokens,
         "completion_tokens": completion_tokens,
         "finish_reason": finish_reason_label(finish_reason),
+        "terminal_reason": terminal_reason_label(finish_reason, error),
         "active_set_size": trace.active_set_size_max,
+        "active_set_size_max": trace.active_set_size_max,
+        "pending_queue_size_max": trace.pending_queue_size_max,
+        "active_set_size_at_terminal": trace.active_set_size_at_terminal,
+        "pending_queue_size_at_terminal": trace.pending_queue_size_at_terminal,
+        "healthy_baseline_after_terminal": trace.active_set_size_at_terminal == 0
+            && trace.pending_queue_size_at_terminal == 0,
         "decode_batch_size_max": trace.decode_batch_size_max,
         "batch_decode_steps": trace.batch_decode_steps,
     });
@@ -107,3 +142,23 @@ fn finish_reason_label(finish_reason: FinishReason) -> &'static str {
         FinishReason::Error => "error",
     }
 }
+
+fn terminal_reason_label(finish_reason: FinishReason, error: Option<&str>) -> &'static str {
+    match (finish_reason, error) {
+        (FinishReason::Length, _) => "completed_length",
+        (FinishReason::Stop, _) => "completed_stop",
+        (FinishReason::Error, Some(message)) if message.contains("disconnected") => "disconnected",
+        (FinishReason::Error, Some(message)) if message.contains("cancelled") => "cancelled",
+        (FinishReason::Error, Some(message))
+            if message.contains("supports greedy")
+                || message.contains("logprobs")
+                || message.contains("LoRA")
+                || message.contains("non-empty prompt")
+                || message.contains("context")
+                || message.contains("unsupported") =>
+        {
+            "rejected"
+        }
+        (FinishReason::Error, _) => "error",
+    }
+}
diff --git a/openinfer-engine/src/engine.rs b/openinfer-engine/src/engine.rs
index 6751313a..96b82886 100644
--- a/openinfer-engine/src/engine.rs
+++ b/openinfer-engine/src/engine.rs
@@ -5,7 +5,7 @@ use std::{
     path::PathBuf,
     sync::{
         Arc, Mutex,
-        atomic::{AtomicBool, Ordering},
+        atomic::{AtomicU8, Ordering},
     },
     thread::{self, JoinHandle},
 };
@@ -191,34 +191,39 @@ pub type TokenStreamReceiver = mpsc::UnboundedReceiver<(RequestTag, TokenEvent)>
 /// unchanged. Internally each event is tagged with the request's
 /// [`RequestTag`] and pushed onto one shared [`TokenStreamSender`].
 ///
-/// Cancellation moved from "drop the per-request receiver" to a shared
-/// `cancelled` flag: the frontend aborts a *single* request by flipping its
-/// flag without closing the channel the other requests still use. `send` and
-/// `is_closed` then report that request as gone, so the scheduler retires it on
-/// its next emit — the same *reactive* retirement the old consumer-drop gave,
-/// reached through the flag rather than channel closure. `tx.is_closed()` is
-/// the engine-wide signal (the whole demux is gone); the per-request signal is
-/// the flag. The flag is set with `Release` and read with `Acquire` so the
-/// abort is ordered against the frontend dropping the request's stream state.
+/// Cancellation moved from "drop the per-request receiver" to a shared abort
+/// reason: the frontend aborts a *single* request by setting its reason without
+/// closing the channel the other requests still use. `send` and `is_closed`
+/// then report that request as gone, so the scheduler retires it on its next
+/// emit — the same *reactive* retirement the old consumer-drop gave, reached
+/// through the reason rather than channel closure. `tx.is_closed()` is the
+/// engine-wide signal (the whole demux is gone); the per-request signal is the
+/// abort reason. The reason is set with `Release` and read with `Acquire` so
+/// the abort is ordered against the frontend dropping the request's stream
+/// state.
 #[derive(Clone)]
 pub struct TokenSink {
     tag: RequestTag,
     tx: TokenStreamSender,
-    cancelled: Arc<AtomicBool>,
+    abort_reason: Arc<AtomicU8>,
 }
 
 impl TokenSink {
-    pub fn new(tag: RequestTag, tx: TokenStreamSender, cancelled: Arc<AtomicBool>) -> Self {
-        Self { tag, tx, cancelled }
+    pub fn new(tag: RequestTag, tx: TokenStreamSender, abort_reason: Arc<AtomicU8>) -> Self {
+        Self {
+            tag,
+            tx,
+            abort_reason,
+        }
     }
 
     /// Emit one event for this request. Returns `Err` (handing the event back)
-    /// when the request was cancelled or the shared receiver is gone — both of
+    /// when the request was aborted or the shared receiver is gone — both of
     /// which the scheduler reads as "consumer dropped, retire the request",
     /// the same contract as the old per-request channel.
     #[allow(clippy::result_large_err)]
     pub fn send(&self, event: TokenEvent) -> Result<(), mpsc::error::SendError<TokenEvent>> {
-        if self.cancelled.load(Ordering::Acquire) {
+        if self.abort_reason() != RequestAbortReason::None {
             return Err(mpsc::error::SendError(event));
         }
         self.tx.send((self.tag.clone(), event)).map_err(|err| {
@@ -227,9 +232,26 @@ impl TokenSink {
         })
     }
 
-    /// `true` once the request is cancelled or the shared receiver is gone.
+    /// `true` once the request is aborted or the shared receiver is gone.
     pub fn is_closed(&self) -> bool {
-        self.cancelled.load(Ordering::Acquire) || self.tx.is_closed()
+        self.abort_reason() != RequestAbortReason::None || self.tx.is_closed()
+    }
+
+    /// `true` once the frontend explicitly cancelled this request after the
+    /// stream had already started.
+    pub fn is_cancelled(&self) -> bool {
+        self.abort_reason() == RequestAbortReason::Cancelled
+    }
+
+    /// `true` once the frontend observed a client disconnect before the first
+    /// response chunk for this request reached the client.
+    pub fn is_disconnected(&self) -> bool {
+        self.abort_reason() == RequestAbortReason::Disconnected
+    }
+
+    /// Current per-request abort reason.
+    pub fn abort_reason(&self) -> RequestAbortReason {
+        RequestAbortReason::from_raw(self.abort_reason.load(Ordering::Acquire))
     }
 
     /// The request id this sink tags its events with.
@@ -243,11 +265,37 @@ impl TokenSink {
     /// receiver yields the tagged events; the cancel flag is never tripped.
     pub fn standalone() -> (Self, TokenStreamReceiver) {
         let (tx, rx) = mpsc::unbounded_channel();
-        let sink = Self::new(Arc::from("local"), tx, Arc::new(AtomicBool::new(false)));
+        let sink = Self::new(
+            Arc::from("local"),
+            tx,
+            Arc::new(AtomicU8::new(RequestAbortReason::None as u8)),
+        );
         (sink, rx)
     }
 }
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+#[repr(u8)]
+pub enum RequestAbortReason {
+    None = 0,
+    Cancelled = 1,
+    Disconnected = 2,
+}
+
+impl RequestAbortReason {
+    pub fn from_raw(raw: u8) -> Self {
+        match raw {
+            1 => Self::Cancelled,
+            2 => Self::Disconnected,
+            _ => Self::None,
+        }
+    }
+
+    pub fn store(self, abort_reason: &AtomicU8) {
+        abort_reason.store(self as u8, Ordering::Release);
+    }
+}
+
 /// Seconds since `UNIX_EPOCH` as `f64` — the clock base for `TokenEvent`
 /// timestamps.
 pub fn unix_now_s() -> f64 {
@@ -618,6 +666,73 @@ mod tests {
         assert!(handle.supports_lora_control());
     }
 
+    #[test]
+    fn token_sink_distinguishes_cancelled_from_closed_receiver() {
+        let abort_reason = Arc::new(AtomicU8::new(RequestAbortReason::None as u8));
+        let (tx, mut rx) = mpsc::unbounded_channel();
+        let sink = TokenSink::new(Arc::from("request-a"), tx, Arc::clone(&abort_reason));
+
+        assert!(!sink.is_cancelled());
+        assert!(!sink.is_disconnected());
+        assert!(!sink.is_closed());
+        sink.send(TokenEvent::Token {
+            id: 7,
+            logprob: None,
+        })
+        .expect("uncancelled sink should send");
+        assert_eq!(rx.try_recv().expect("tagged event").0.as_ref(), "request-a");
+
+        RequestAbortReason::Cancelled.store(&abort_reason);
+        assert!(sink.is_cancelled());
+        assert!(!sink.is_disconnected());
+        assert!(sink.is_closed());
+        assert!(
+            sink.send(TokenEvent::Token {
+                id: 8,
+                logprob: None,
+            })
+            .is_err()
+        );
+    }
+
+    #[test]
+    fn token_sink_closed_receiver_is_not_explicit_cancel() {
+        let (sink, rx) = TokenSink::standalone();
+
+        drop(rx);
+
+        assert!(!sink.is_cancelled());
+        assert!(!sink.is_disconnected());
+        assert!(sink.is_closed());
+        assert!(
+            sink.send(TokenEvent::Token {
+                id: 7,
+                logprob: None,
+            })
+            .is_err()
+        );
+    }
+
+    #[test]
+    fn token_sink_distinguishes_disconnected_from_cancelled() {
+        let abort_reason = Arc::new(AtomicU8::new(RequestAbortReason::None as u8));
+        let (tx, _rx) = mpsc::unbounded_channel();
+        let sink = TokenSink::new(Arc::from("request-a"), tx, Arc::clone(&abort_reason));
+
+        RequestAbortReason::Disconnected.store(&abort_reason);
+
+        assert!(!sink.is_cancelled());
+        assert!(sink.is_disconnected());
+        assert!(sink.is_closed());
+        assert!(
+            sink.send(TokenEvent::Token {
+                id: 7,
+                logprob: None,
+            })
+            .is_err()
+        );
+    }
+
     #[tokio::test]
     async fn load_lora_adapter_sends_control_command() {
         let (command_tx, mut command_rx) = mpsc::unbounded_channel::<EngineCommand>();
diff --git a/openinfer-vllm-frontend/src/bridge.rs b/openinfer-vllm-frontend/src/bridge.rs
index 45abae2f..4212d527 100644
--- a/openinfer-vllm-frontend/src/bridge.rs
+++ b/openinfer-vllm-frontend/src/bridge.rs
@@ -1,7 +1,6 @@
 use std::collections::{BTreeSet, HashMap};
 use std::path::{Path, PathBuf};
-use std::sync::Arc;
-use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::{Arc, atomic::AtomicU8};
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 use anyhow::{Context, Result, bail};
@@ -24,7 +23,8 @@ use zeromq::util::PeerIdentity;
 use zeromq::{DealerSocket, PushSocket, SocketOptions, ZmqMessage};
 
 use openinfer_engine::engine::{
-    EngineHandle, GenerateRequest, RequestTag, TokenEvent, TokenSink, TokenStreamReceiver,
+    EngineHandle, GenerateRequest, RequestAbortReason, RequestTag, TokenEvent, TokenSink,
+    TokenStreamReceiver,
 };
 
 use crate::wire::{
@@ -112,7 +112,7 @@ impl LocalEngineBridge {
 
         // One shared channel carries every request's token events, tagged by
         // request id; this loop is the sole consumer. Per-request state lives
-        // in `streams`, keyed by the same tag, and holds the cancel flag the
+        // in `streams`, keyed by the same tag, and holds the abort reason the
         // scheduler observes (via `TokenSink`) when an abort flips it.
         let (event_tx, mut event_rx) = mpsc::unbounded_channel();
         let mut streams: HashMap<RequestTag, RequestStreamState> = HashMap::new();
@@ -149,7 +149,7 @@ impl LocalEngineBridge {
         // Cancel every in-flight request so the scheduler retires them on its
         // next emit instead of pushing into a channel no one drains.
         for state in streams.values() {
-            state.cancelled.store(true, Ordering::Release);
+            state.abort(RequestAbortReason::Cancelled);
         }
         drop(output_tx);
         output_task.abort();
@@ -182,14 +182,23 @@ impl LocalEngineBridge {
                 let request_ids: Vec<String> =
                     vllm_engine_core_client::protocol::decode_msgpack(&frames[1])?;
                 for request_id in request_ids {
-                    // Drop our state first, then flip the cancel flag (so the
+                    // Drop our state first, then set the abort reason (so the
                     // scheduler's next emit fails and retires the request). The
                     // `Release` store orders the `streams.remove` before the
-                    // flag the scheduler reads with `Acquire`; any token already
-                    // in flight for this id is discarded by the demux when it
-                    // finds no stream entry.
+                    // reason the scheduler reads with `Acquire`; any token
+                    // already in flight for this id is discarded by the demux
+                    // when it finds no stream entry. An abort before the first
+                    // token output reached the frontend is a disconnect; after
+                    // that first token it is stream cancellation. Scheduled
+                    // metadata can flush with the first engine output but is
+                    // not enough to prove a client-visible token.
                     if let Some(state) = streams.remove(request_id.as_str()) {
-                        state.cancelled.store(true, Ordering::Release);
+                        let reason = if state.has_emitted_tokens {
+                            RequestAbortReason::Cancelled
+                        } else {
+                            RequestAbortReason::Disconnected
+                        };
+                        state.abort(reason);
                     }
                 }
                 Ok(())
@@ -262,8 +271,8 @@ impl LocalEngineBridge {
         }
 
         let tag: RequestTag = Arc::from(request_id.as_str());
-        let cancelled = Arc::new(AtomicBool::new(false));
-        let token_tx = TokenSink::new(tag.clone(), event_tx.clone(), Arc::clone(&cancelled));
+        let abort_reason = Arc::new(AtomicU8::new(RequestAbortReason::None as u8));
+        let token_tx = TokenSink::new(tag.clone(), event_tx.clone(), Arc::clone(&abort_reason));
         self.handle
             .submit(GenerateRequest {
                 request_id: Some(request_id),
@@ -278,30 +287,36 @@ impl LocalEngineBridge {
             })
             .context("failed to submit request to scheduler")?;
 
-        streams.insert(tag, RequestStreamState::new(cancelled));
+        streams.insert(tag, RequestStreamState::new(abort_reason));
         Ok(())
     }
 }
 
 /// Per-request demux state held by the bridge loop, keyed by [`RequestTag`].
 /// Replaces the former per-request task's locals; `first_token_*` flush onto
-/// the request's first output, `cancelled` is the flag the scheduler's
+/// the request's first output, `abort_reason` is the state the scheduler's
 /// [`TokenSink`] checks so an abort retires the request without closing the
 /// shared channel.
 struct RequestStreamState {
     first_token_events: Option<Vec<EngineCoreEvent>>,
     first_token_prefill_stats: Option<PrefillStats>,
-    cancelled: Arc<AtomicBool>,
+    abort_reason: Arc<AtomicU8>,
+    has_emitted_tokens: bool,
 }
 
 impl RequestStreamState {
-    fn new(cancelled: Arc<AtomicBool>) -> Self {
+    fn new(abort_reason: Arc<AtomicU8>) -> Self {
         Self {
             first_token_events: None,
             first_token_prefill_stats: None,
-            cancelled,
+            abort_reason,
+            has_emitted_tokens: false,
         }
     }
+
+    fn abort(&self, reason: RequestAbortReason) {
+        reason.store(&self.abort_reason);
+    }
 }
 
 /// Drain the ready burst from the shared token channel (the `first` event plus
@@ -343,6 +358,9 @@ fn dispatch_burst(
         };
         let (output, terminated) = reduce_request(&tag, state, events);
         if let Some(output) = output {
+            if !output.new_token_ids.is_empty() {
+                state.has_emitted_tokens = true;
+            }
             outputs.push(output);
         }
         if terminated {
diff --git a/openinfer-vllm-frontend/src/bridge/tests.rs b/openinfer-vllm-frontend/src/bridge/tests.rs
index ccfcbed6..9f517f8b 100644
--- a/openinfer-vllm-frontend/src/bridge/tests.rs
+++ b/openinfer-vllm-frontend/src/bridge/tests.rs
@@ -7,7 +7,9 @@
 //! abort dropping late tokens. The full HTTP→ZMQ→bridge happy path is covered
 //! end to end by `openinfer-sim`'s `frontend_e2e` integration test (the CI gate).
 
-use openinfer_engine::engine::{FinishReason, TokenLogprob};
+use std::sync::atomic::Ordering;
+
+use openinfer_engine::engine::{FinishReason, RequestAbortReason, TokenLogprob};
 
 use super::*;
 
@@ -36,15 +38,15 @@ impl Demux {
         }
     }
 
-    /// Register a request as `start_request` does and return its cancel flag.
-    fn add(&mut self, id: &str) -> Arc<AtomicBool> {
+    /// Register a request as `start_request` does and return its abort reason.
+    fn add(&mut self, id: &str) -> Arc<AtomicU8> {
         let tag: RequestTag = Arc::from(id);
-        let cancelled = Arc::new(AtomicBool::new(false));
+        let abort_reason = Arc::new(AtomicU8::new(RequestAbortReason::None as u8));
         self.streams.insert(
             Arc::clone(&tag),
-            RequestStreamState::new(Arc::clone(&cancelled)),
+            RequestStreamState::new(Arc::clone(&abort_reason)),
         );
-        cancelled
+        abort_reason
     }
 
     fn emit(&self, id: &str, event: TokenEvent) {
@@ -333,10 +335,11 @@ fn burst_batches_multiple_requests_into_one_message() {
 #[test]
 fn aborted_request_drops_late_tokens() {
     let mut d = Demux::new();
-    let cancelled = d.add("req-abort");
+    let abort_reason = d.add("req-abort");
 
-    // Replicate the Abort handler: flip the cancel flag, drop the stream.
-    cancelled.store(true, Ordering::Relaxed);
+    // Replicate the Abort handler for a request that already emitted output:
+    // drop the stream and mark it as cancelled.
+    RequestAbortReason::Cancelled.store(&abort_reason);
     d.streams.remove("req-abort");
 
     d.emit(
@@ -353,6 +356,56 @@ fn aborted_request_drops_late_tokens() {
     );
 }
 
+#[test]
+fn abort_reason_tracks_first_output_boundary() {
+    let mut d = Demux::new();
+    let disconnected = d.add("req-before-output");
+    let cancelled = d.add("req-after-output");
+
+    d.emit(
+        "req-after-output",
+        TokenEvent::Token {
+            id: 7,
+            logprob: None,
+        },
+    );
+    assert!(d.drain());
+    assert_eq!(d.next_output().expect("first output").outputs.len(), 1);
+    assert!(
+        d.streams
+            .get("req-after-output")
+            .is_some_and(|state| state.has_emitted_tokens)
+    );
+
+    let state = d
+        .streams
+        .remove("req-before-output")
+        .expect("disconnect stream");
+    let reason = if state.has_emitted_tokens {
+        RequestAbortReason::Cancelled
+    } else {
+        RequestAbortReason::Disconnected
+    };
+    state.abort(reason);
+
+    let state = d.streams.remove("req-after-output").expect("cancel stream");
+    let reason = if state.has_emitted_tokens {
+        RequestAbortReason::Cancelled
+    } else {
+        RequestAbortReason::Disconnected
+    };
+    state.abort(reason);
+
+    assert_eq!(
+        RequestAbortReason::from_raw(disconnected.load(Ordering::Acquire)),
+        RequestAbortReason::Disconnected
+    );
+    assert_eq!(
+        RequestAbortReason::from_raw(cancelled.load(Ordering::Acquire)),
+        RequestAbortReason::Cancelled
+    );
+}
+
 /// A rejected request (could not be admitted, e.g. too large for the KV cache)
 /// surfaces to the client as an error with the rejection message, and its stream
 /// is retired.
diff --git a/pr-body-dsv2lite-http-reliability-gate.md b/pr-body-dsv2lite-http-reliability-gate.md
new file mode 100644
index 00000000..f25b36e6
--- /dev/null
+++ b/pr-body-dsv2lite-http-reliability-gate.md
@@ -0,0 +1,161 @@
+## Summary
+
+- Adds a replayable DeepSeek-V2-Lite EP2 HTTP reliability gate for real streaming `/v1/completions` requests.
+- Extends the request lifecycle contract across the vLLM bridge, `TokenSink`, and DSV2-Lite scheduler so cancel, disconnect, rejection, error, and completion are machine-distinguishable.
+- Adds strict runner JSON, focused tests, live host-staged/NCCL evidence, no-regression benchmark rows, and status-doc claim boundaries.
+
+Fixes #453.
+
+## Extra Risks Found And Fixed
+
+### Cancel and disconnect were not machine-distinguishable
+
+- Risk: a dropped per-request stream and an explicit frontend abort both looked like a send failure, so the HTTP gate could not prove whether it exercised cancel versus disconnect.
+- Why it affects #453: the issue asks for cancel and disconnect coverage as separate failure modes.
+- Minimal fix: add request-local abort state in `TokenSink` / bridge handling, expose `is_cancelled()`, `is_disconnected()`, and `abort_reason()`, and map send failures to `cancelled` versus `disconnected` terminal reasons in the DSV2-Lite scheduler.
+- Evidence: `cargo test --release -p openinfer-engine --lib token_sink -- --nocapture`, `cargo test --release -p openinfer-vllm-frontend --lib abort -- --nocapture`, and the live `cancel_disconnect` scenario on both backends.
+
+### Early disconnect could be misclassified after non-token bridge output
+
+- Risk: the bridge could mark a request as having emitted output before a client-visible token, making an early connection close look like cancel instead of disconnect.
+- Why it affects #453: disconnect-before-token and cancel-after-token are different reliability cases.
+- Minimal fix: track `has_emitted_tokens` at the bridge boundary and classify aborts using client-visible token output, not internal metadata progress.
+- Evidence: `abort_reason_tracks_first_output_boundary` and live `cancel_disconnect` artifacts showing one `cancelled` and one `disconnected` terminal reason per backend.
+
+### Terminal traces could not prove state retirement
+
+- Risk: prior traces had active/decode maxima but no terminal reason or terminal active/pending state, so the gate could not fail on leaked active or pending state.
+- Why it affects #453: clean follow-up success alone is weaker than direct scheduler-state evidence.
+- Minimal fix: add `terminal_reason`, `pending_queue_size_max`, `active_set_size_at_terminal`, `pending_queue_size_at_terminal`, and `healthy_baseline_after_terminal` to `openinfer_http_trace`.
+- Evidence: scheduler lifecycle tests plus the reliability runner's strict trace checks and live artifacts with healthy final baselines after every scenario.
+
+### Overload pressure needed pending-queue recovery evidence
+
+- Risk: an overload run could complete some requests while still hiding bad pending/active cleanup.
+- Why it affects #453: active-cap pressure must prove queued work does not poison active requests and the queue can recover.
+- Minimal fix: thread active/pending state through batch decode, single-row fallback, cache-position errors, and terminal trace logging; require pending pressure in the runner.
+- Evidence: `overload_active_cap` passed on host-staged and NCCL with `active_set_size_max=8`, `pending_queue_size_max=4`, `decode_batch_size_max=7`, `completed=13`, and healthy final baseline.
+
+### Cross-group terminal accounting could overstate active state
+
+- Risk: when different decode-position groups retired rows in the same scheduler round, terminal traces could use a per-group active count instead of a shared round-level active count.
+- Why it affects #453: the reliability gate must prove active state retires after failures, so stale terminal active counts weaken the cleanup claim.
+- Minimal fix: carry one shared active-remaining counter across same-position batch groups and single-row fallback paths.
+- Evidence: `cross_group_terminal_accounting_uses_shared_active_remaining` and live terminal traces with healthy baselines after mixed failure scenarios.
+
+### Benchmark/gate could pass without terminal trace coverage
+
+- Risk: a completed HTTP benchmark can miss failed, cancelled, rejected, or disconnected terminal traces.
+- Why it affects #453: the acceptance criteria require counts, terminal reasons, trace coverage, active/pending/decode maxima, and clean follow-up recovery.
+- Minimal fix: add `scripts/bench_dsv2lite_http_reliability.py` with strict per-scenario failure rules. HTTP/frontend guard rejections may lack scheduler traces; scheduler-level terminal requests must have traces.
+- Evidence: `python3 -m unittest tests.test_bench_dsv2lite_http_reliability`, host-staged reliability artifact `832d65a8e8b2b3a6ad0100c4a35f38475f040d6ffc192ec38a3b7384167187a5`, and NCCL reliability artifact `53bedd98f19c5241df588a1ade8756e84a5e8c99225a589c4ed303e90fba38fa`.
+
+## Evidence
+
+Local checks completed:
+
+```bash
+cargo fmt --all --check
+git diff --check
+cargo test --release -p openinfer-engine --lib token_sink -- --nocapture
+cargo test --release -p openinfer-vllm-frontend --lib abort -- --nocapture
+python3 -m unittest tests.test_bench_dsv2lite_http_reliability
+python3 -m py_compile scripts/bench_dsv2lite_http_reliability.py tests/test_bench_dsv2lite_http_reliability.py
+```
+
+Remote 2-GPU checks completed:
+
+```bash
+cargo build --release -p openinfer-server --no-default-features --features deepseek-v2-lite
+cargo test --release -p openinfer-vllm-frontend --lib abort -- --nocapture
+cargo test --release -p openinfer-deepseek-v2-lite --features deepseek-v2-lite --lib scheduler -- --nocapture
+OPENINFER_TEST_MODEL_PATH=models/DeepSeek-V2-Lite OPENINFER_DSV2_LITE_EP_BACKEND=host-staged cargo test --release -p openinfer-deepseek-v2-lite --features deepseek-v2-lite --test e2e_ep2 -- --nocapture
+OPENINFER_TEST_MODEL_PATH=models/DeepSeek-V2-Lite OPENINFER_DSV2_LITE_EP_BACKEND=nccl cargo test --release -p openinfer-deepseek-v2-lite --features deepseek-v2-lite --test e2e_ep2 -- --nocapture
+```
+
+Results:
+
+- `openinfer-engine` token sink tests passed: 3 passed, 0 failed.
+- `openinfer-vllm-frontend` abort tests passed: 2 passed, 0 failed.
+- DSV2-Lite scheduler lifecycle tests passed: 23 passed, 0 failed.
+- Host-staged `e2e_ep2` passed: 1 passed, 0 failed.
+- NCCL `e2e_ep2` passed: 1 passed, 0 failed. The SM120 validation used an NCCL 2.30.7 runtime library.
+- Python reliability runner tests passed: 3 passed, 0 failed.
+- Python syntax check passed.
+- Server release build passed.
+
+Live HTTP reliability artifacts:
+
+| Backend | Artifact | SHA-256 | Result |
+| --- | --- | --- | --- |
+| host-staged | `http_reliability_host_staged.json` | `832d65a8e8b2b3a6ad0100c4a35f38475f040d6ffc192ec38a3b7384167187a5` | passed |
+| NCCL | `http_reliability_nccl.json` | `53bedd98f19c5241df588a1ade8756e84a5e8c99225a589c4ed303e90fba38fa` | passed |
+
+Scenario summary:
+
+| Backend | Scenario | Counts | Trace maxima | Final baseline |
+| --- | --- | --- | --- | --- |
+| host-staged | `cancel_disconnect` | completed `2`, cancelled `1`, disconnected `1`, failed/rejected/timeout `0` | active `2`, pending `0`, decode `1` | healthy |
+| host-staged | `invalid_requests` | completed `2`, rejected `4`, failed/timeout `0` | active `1`, pending `0`, decode `1` | healthy |
+| host-staged | `overload_active_cap` | completed `13`, failed/rejected/timeout `0` | active `8`, pending `4`, decode `7` | healthy |
+| host-staged | `mixed_short_long_with_failures` | completed `5`, cancelled `1`, rejected `1`, failed/timeout `0` | active `4`, pending `0`, decode `2` | healthy |
+| NCCL | `cancel_disconnect` | completed `2`, cancelled `1`, disconnected `1`, failed/rejected/timeout `0` | active `2`, pending `0`, decode `1` | healthy |
+| NCCL | `invalid_requests` | completed `2`, rejected `4`, failed/timeout `0` | active `1`, pending `0`, decode `1` | healthy |
+| NCCL | `overload_active_cap` | completed `13`, failed/rejected/timeout `0` | active `8`, pending `4`, decode `7` | healthy |
+| NCCL | `mixed_short_long_with_failures` | completed `5`, cancelled `1`, rejected `1`, failed/timeout `0` | active `5`, pending `0`, decode `2` | healthy |
+
+## No-Regression Benchmark
+
+The no-regression benchmark used real `/v1/completions` traffic after the reliability gate. Every row passed with `failed=0`, `timeouts=0`, and `missing_traces=[]`.
+
+| Backend | Shape | Completed | Failed/timeouts | Output tok/s | TTFT avg ms | TPOT/ITL avg ms | active max | decode batch max |
+| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: |
+| host-staged | short `64/64`, c1 | 8/8 | 0/0 | 22.886 | 1226.8 | 24.9 | 1 | 1 |
+| host-staged | short `64/64`, c4 | 8/8 | 0/0 | 21.698 | 2612.8 | 145.1 | 4 | 3 |
+| host-staged | short `64/64`, c8 | 8/8 | 0/0 | 21.830 | 5439.8 | 284.2 | 8 | 5 |
+| host-staged | mixed `16/128`, c4 | 8/8 | 0/0 | 8.281 | 4543.3 | 210.0 | 4 | 2 |
+| host-staged | mixed `16/128`, c8 | 8/8 | 0/0 | 8.288 | 7897.1 | 496.9 | 8 | 3 |
+| NCCL | short `64/64`, c1 | 8/8 | 0/0 | 24.873 | 1034.2 | 24.4 | 1 | 1 |
+| NCCL | short `64/64`, c4 | 8/8 | 0/0 | 23.678 | 2210.0 | 135.8 | 4 | 3 |
+| NCCL | short `64/64`, c8 | 8/8 | 0/0 | 23.872 | 4594.4 | 266.1 | 8 | 6 |
+| NCCL | mixed `16/128`, c4 | 8/8 | 0/0 | 9.186 | 2216.6 | 314.9 | 4 | 1 |
+| NCCL | mixed `16/128`, c8 | 8/8 | 0/0 | 9.269 | 6920.1 | 453.0 | 8 | 3 |
+
+This is a no-regression and trace-coverage record. It is not a throughput optimization claim.
+
+## Claim Boundary
+
+- This PR claims a stronger HTTP reliability gate and auditable failure isolation for DeepSeek-V2-Lite EP2 serving.
+- It does not claim production EP readiness.
+- It does not claim vLLM parity.
+- It does not claim sparse dispatch, multi-node EP, CUDA Graph productization, or host-staged deprecation.
+- It does not claim throughput improvement from the reliability runner or no-regression benchmark.
+
+## Remaining Risks
+
+No #453 reliability blocker remains after the gates above.
+
+Separate follow-ups remain outside this PR's claim boundary:
+
+- #452 remains the mixed/long-prompt latency follow-up.
+- #465 remains the sustained soak benchmark follow-up.
+- #466 remains the retained SLO report follow-up.
+- #467 remains the benchmark artifact manifest follow-up.
+
+## Commands Run And Exact Result
+
+| Command | Result |
+| --- | --- |
+| `cargo fmt --all --check` | Passed locally. |
+| `git diff --check` | Passed locally. |
+| `python3 -m py_compile scripts/bench_dsv2lite_http_reliability.py tests/test_bench_dsv2lite_http_reliability.py` | Passed locally. |
+| `python3 -m unittest tests.test_bench_dsv2lite_http_reliability` | Passed locally; 3 tests verify dry-run success plus false-positive failures for bad terminal traces and missing trace fields. |
+| `cargo test --release -p openinfer-engine --lib token_sink -- --nocapture` | Passed locally; 3 tests verify explicit cancel, disconnect, and closed receiver behavior. |
+| `cargo test --release -p openinfer-vllm-frontend --lib abort -- --nocapture` | Passed locally and remotely; 2 tests verify abort handling. |
+| `cargo build --release -p openinfer-server --no-default-features --features deepseek-v2-lite` | Passed remotely. |
+| `cargo test --release -p openinfer-deepseek-v2-lite --features deepseek-v2-lite --lib scheduler -- --nocapture` | Passed remotely; 23 passed, 0 failed. |
+| `OPENINFER_TEST_MODEL_PATH=models/DeepSeek-V2-Lite OPENINFER_DSV2_LITE_EP_BACKEND=host-staged cargo test --release -p openinfer-deepseek-v2-lite --features deepseek-v2-lite --test e2e_ep2 -- --nocapture` | Passed remotely; 1 passed, 0 failed. |
+| `OPENINFER_TEST_MODEL_PATH=models/DeepSeek-V2-Lite OPENINFER_DSV2_LITE_EP_BACKEND=nccl cargo test --release -p openinfer-deepseek-v2-lite --features deepseek-v2-lite --test e2e_ep2 -- --nocapture` | Passed remotely; 1 passed, 0 failed. |
+| host-staged live HTTP reliability runner | Passed; artifact `http_reliability_host_staged.json`, SHA-256 `832d65a8e8b2b3a6ad0100c4a35f38475f040d6ffc192ec38a3b7384167187a5`. |
+| NCCL live HTTP reliability runner | Passed; artifact `http_reliability_nccl.json`, SHA-256 `53bedd98f19c5241df588a1ade8756e84a5e8c99225a589c4ed303e90fba38fa`. |
+| host-staged/NCCL no-regression HTTP benchmark | Passed; all 10 rows completed 8/8 requests with failed/timeouts `0/0` and `missing_traces=[]`. |
diff --git a/scripts/bench_dsv2lite_http_reliability.py b/scripts/bench_dsv2lite_http_reliability.py
new file mode 100755
index 00000000..a6ccb92c
--- /dev/null
+++ b/scripts/bench_dsv2lite_http_reliability.py
@@ -0,0 +1,864 @@
+#!/usr/bin/env python3
+"""DeepSeek-V2-Lite HTTP reliability gate for /v1/completions.
+
+The gate intentionally uses the real OpenAI-compatible HTTP path and consumes
+`openinfer_http_trace` server logs. It fails when terminal traces are missing,
+state does not retire back to a healthy baseline, clean follow-up requests fail,
+success hashes drift, or an expected cancel/disconnect/reject outcome is not
+visible in traces.
+"""
+
+from __future__ import annotations
+
+import argparse
+import concurrent.futures
+import hashlib
+import http.client
+import json
+import re
+import socket
+import time
+import urllib.parse
+from dataclasses import asdict, dataclass, field
+from pathlib import Path
+from typing import Any
+
+
+TRACE_RE = re.compile(r"openinfer_http_trace\s+(\{.*\})")
+DSV2_LITE_ACTIVE_CAP = 8
+PROMPT_WORDS = (
+    "alpha beta gamma delta epsilon zeta eta theta iota kappa lambda mu "
+    "nu xi omicron pi rho sigma tau upsilon phi chi psi omega"
+).split()
+
+
+@dataclass
+class RequestResult:
+    request_id: str
+    kind: str
+    ok: bool
+    status: int | None
+    terminal_reason: str
+    client_observed_outcome: str
+    error: str | None = None
+    timed_out: bool = False
+    start_wall_s: float = 0.0
+    end_wall_s: float = 0.0
+    latency_ms: float = 0.0
+    first_token_seen: bool = False
+    output_chunks: int = 0
+    output_chars: int = 0
+    output_hash: str = ""
+    trace: dict[str, Any] | None = None
+
+
+@dataclass
+class ScenarioReport:
+    name: str
+    passed: bool
+    failures: list[str]
+    counts: dict[str, int]
+    output_hashes: dict[str, str]
+    trace_coverage: dict[str, Any]
+    trace_maxima: dict[str, int | None]
+    terminal_reasons: dict[str, int]
+    final_healthy_baseline: bool
+    clean_follow_up: RequestResult
+    requests: list[RequestResult] = field(default_factory=list)
+
+
+def make_prompt(index: int, words: int) -> str:
+    return " ".join(PROMPT_WORDS[(index + offset) % len(PROMPT_WORDS)] for offset in range(words))
+
+
+def parse_sse_text(payload: dict[str, Any]) -> str:
+    choices = payload.get("choices") or []
+    if not choices:
+        return ""
+    choice = choices[0]
+    if "text" in choice:
+        return choice.get("text") or ""
+    delta = choice.get("delta") or {}
+    return delta.get("content") or ""
+
+
+def parse_sse_error(payload: dict[str, Any]) -> str | None:
+    error = payload.get("error")
+    if isinstance(error, str):
+        return error
+    if isinstance(error, dict):
+        message = error.get("message")
+        if isinstance(message, str):
+            return message
+        return json.dumps(error, sort_keys=True)
+    return None
+
+
+def completion_path(base: urllib.parse.ParseResult) -> str:
+    prefix = base.path.rstrip("/")
+    return f"{prefix}/v1/completions" if prefix else "/v1/completions"
+
+
+def connect(base: urllib.parse.ParseResult, timeout: float) -> http.client.HTTPConnection:
+    conn_cls = http.client.HTTPSConnection if base.scheme == "https" else http.client.HTTPConnection
+    return conn_cls(base.hostname, port=base.port, timeout=timeout)
+
+
+def request_body(
+    model: str,
+    prompt: str,
+    max_tokens: int,
+    request_id: str,
+    **overrides: Any,
+) -> dict[str, Any]:
+    body: dict[str, Any] = {
+        "model": model,
+        "prompt": prompt,
+        "max_tokens": max_tokens,
+        "temperature": 0.0,
+        "top_k": 0,
+        "top_p": 1.0,
+        "stream": True,
+        "ignore_eos": True,
+        "request_id": request_id,
+    }
+    body.update(overrides)
+    return body
+
+
+def run_stream_request(
+    base: urllib.parse.ParseResult,
+    model: str,
+    request_id: str,
+    prompt: str,
+    max_tokens: int,
+    timeout: float,
+    kind: str,
+    close_after_first_token: bool = False,
+    close_after_headers: bool = False,
+    **overrides: Any,
+) -> RequestResult:
+    started = time.time()
+    chunks: list[str] = []
+    first_token_seen = False
+    status: int | None = None
+    conn: http.client.HTTPConnection | None = None
+    try:
+        conn = connect(base, timeout)
+        body = request_body(model, prompt, max_tokens, request_id, **overrides)
+        conn.request(
+            "POST",
+            completion_path(base),
+            body=json.dumps(body).encode("utf-8"),
+            headers={"Content-Type": "application/json"},
+        )
+        response = conn.getresponse()
+        status = response.status
+        if status != 200:
+            error_body = response.read(4096).decode("utf-8", errors="replace")
+            return finish_result(
+                request_id,
+                kind,
+                False,
+                status,
+                started,
+                "rejected",
+                error=f"HTTP {status}: {error_body}",
+            )
+        if close_after_headers:
+            conn.close()
+            return finish_result(request_id, kind, False, status, started, "disconnected")
+        while True:
+            raw = response.readline()
+            if not raw:
+                break
+            line = raw.decode("utf-8", errors="replace").strip()
+            if not line or not line.startswith("data:"):
+                continue
+            data = line.removeprefix("data:").strip()
+            if data == "[DONE]":
+                break
+            payload = json.loads(data)
+            stream_error = parse_sse_error(payload)
+            if stream_error is not None:
+                return finish_result(
+                    request_id,
+                    kind,
+                    False,
+                    status,
+                    started,
+                    "failed",
+                    error=f"SSE error: {stream_error}",
+                )
+            text = parse_sse_text(payload)
+            if not text:
+                continue
+            first_token_seen = True
+            chunks.append(text)
+            if close_after_first_token:
+                conn.close()
+                return finish_result(
+                    request_id,
+                    kind,
+                    False,
+                    status,
+                    started,
+                    "cancelled",
+                    first_token_seen=True,
+                    output_chunks=len(chunks),
+                    output_text="".join(chunks),
+                )
+        text = "".join(chunks)
+        if max_tokens > 0 and not chunks:
+            return finish_result(
+                request_id,
+                kind,
+                False,
+                status,
+                started,
+                "failed",
+                error="stream completed without text chunks",
+            )
+        return finish_result(
+            request_id,
+            kind,
+            True,
+            status,
+            started,
+            "completed",
+            first_token_seen=first_token_seen,
+            output_chunks=len(chunks),
+            output_text=text,
+        )
+    except (TimeoutError, socket.timeout) as exc:
+        return finish_result(
+            request_id,
+            kind,
+            False,
+            status,
+            started,
+            "timeout",
+            timed_out=True,
+            error=str(exc),
+        )
+    except Exception as exc:  # noqa: BLE001 - reliability report keeps raw failure.
+        return finish_result(
+            request_id,
+            kind,
+            False,
+            status,
+            started,
+            "failed",
+            error=str(exc),
+        )
+    finally:
+        if conn is not None:
+            try:
+                conn.close()
+            except Exception:
+                pass
+
+
+def finish_result(
+    request_id: str,
+    kind: str,
+    ok: bool,
+    status: int | None,
+    started: float,
+    terminal_reason: str,
+    *,
+    error: str | None = None,
+    timed_out: bool = False,
+    first_token_seen: bool = False,
+    output_chunks: int = 0,
+    output_text: str = "",
+) -> RequestResult:
+    ended = time.time()
+    return RequestResult(
+        request_id=request_id,
+        kind=kind,
+        ok=ok,
+        status=status,
+        terminal_reason=terminal_reason,
+        client_observed_outcome=terminal_reason,
+        error=error,
+        timed_out=timed_out,
+        start_wall_s=started,
+        end_wall_s=ended,
+        latency_ms=(ended - started) * 1000.0,
+        first_token_seen=first_token_seen,
+        output_chunks=output_chunks,
+        output_chars=len(output_text),
+        output_hash=hashlib.sha256(output_text.encode("utf-8")).hexdigest()[:16]
+        if output_text
+        else "",
+    )
+
+
+def load_server_traces(path: Path | None) -> dict[str, dict[str, Any]]:
+    if path is None or not path.exists():
+        return {}
+    traces: dict[str, dict[str, Any]] = {}
+    for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
+        match = TRACE_RE.search(line)
+        if not match:
+            continue
+        try:
+            trace = json.loads(match.group(1))
+        except json.JSONDecodeError:
+            continue
+        request_id = trace.get("request_id")
+        if isinstance(request_id, str):
+            traces[request_id] = trace
+    return traces
+
+
+def find_trace(request_id: str, traces: dict[str, dict[str, Any]]) -> dict[str, Any] | None:
+    prefix = f"cmpl-{request_id}-"
+    for trace_id, trace in traces.items():
+        if trace_id == request_id or trace_id == f"cmpl-{request_id}" or trace_id.startswith(prefix):
+            return trace
+    return None
+
+
+def attach_traces(results: list[RequestResult], traces: dict[str, dict[str, Any]]) -> None:
+    for result in results:
+        trace = find_trace(result.request_id, traces)
+        if trace is None:
+            continue
+        result.trace = trace
+        terminal_reason = trace.get("terminal_reason")
+        if isinstance(terminal_reason, str):
+            result.terminal_reason = normalize_trace_terminal_reason(terminal_reason)
+
+
+def normalize_trace_terminal_reason(reason: str) -> str:
+    if reason in {"completed", "completed_length", "completed_stop"}:
+        return "completed"
+    if reason in {"cancelled", "disconnected", "rejected"}:
+        return reason
+    return "failed"
+
+
+def count_results(results: list[RequestResult]) -> dict[str, int]:
+    counts = {
+        "completed": 0,
+        "failed": 0,
+        "rejected": 0,
+        "cancelled": 0,
+        "disconnected": 0,
+        "timeout": 0,
+    }
+    for result in results:
+        if result.timed_out or result.client_observed_outcome == "timeout":
+            counts["timeout"] += 1
+        elif result.terminal_reason in counts:
+            counts[result.terminal_reason] += 1
+        elif result.ok:
+            counts["completed"] += 1
+        else:
+            counts["failed"] += 1
+    return counts
+
+
+def trace_max(results: list[RequestResult], field: str) -> int | None:
+    values = [
+        int(result.trace[field])
+        for result in results
+        if result.trace is not None and isinstance(result.trace.get(field), int)
+    ]
+    return max(values) if values else None
+
+
+def terminal_reason_counts(results: list[RequestResult]) -> dict[str, int]:
+    counts: dict[str, int] = {}
+    for result in results:
+        reason = result.trace.get("terminal_reason") if result.trace is not None else result.terminal_reason
+        if not isinstance(reason, str):
+            reason = "missing_trace"
+        counts[reason] = counts.get(reason, 0) + 1
+    return counts
+
+
+def stable_success_hashes(results: list[RequestResult]) -> tuple[bool, dict[str, str]]:
+    hashes: dict[str, str] = {}
+    stable = True
+    for result in results:
+        if not result.ok:
+            continue
+        key = result.kind
+        previous = hashes.setdefault(key, result.output_hash)
+        if previous != result.output_hash:
+            stable = False
+    return stable, hashes
+
+
+def clean_follow_up(base: urllib.parse.ParseResult, args: argparse.Namespace, name: str) -> RequestResult:
+    return run_stream_request(
+        base,
+        args.model,
+        f"dsv2-rel-{name}-follow-up",
+        make_prompt(97, 16),
+        args.follow_up_tokens,
+        args.timeout,
+        "clean_follow_up",
+    )
+
+
+def build_scenario(
+    name: str,
+    results: list[RequestResult],
+    follow_up: RequestResult,
+    *,
+    required_reasons: set[str],
+    require_pending_pressure: bool = False,
+    require_active_pressure: bool = False,
+    allow_http_guard_rejections: bool = False,
+) -> ScenarioReport:
+    all_results = [*results, follow_up]
+    counts = count_results(all_results)
+    stable_hashes, hashes = stable_success_hashes(all_results)
+    traced = [result for result in all_results if result.trace is not None]
+    missing = [
+        result.request_id
+        for result in all_results
+        if result.trace is None and not trace_optional(result, allow_http_guard_rejections)
+    ]
+    reasons = terminal_reason_counts(all_results)
+    failures: list[str] = []
+
+    if missing:
+        failures.append(f"missing terminal traces: {missing}")
+    if not follow_up.ok:
+        failures.append(f"clean follow-up failed: {follow_up.error or follow_up.terminal_reason}")
+    if not stable_hashes:
+        failures.append("successful output hashes are not stable by request kind")
+    for reason in sorted(required_reasons):
+        if counts.get(reason, 0) <= 0:
+            failures.append(f"expected terminal reason {reason!r} was not observed")
+    for result in all_results:
+        if result.timed_out or result.client_observed_outcome == "timeout":
+            failures.append(f"request timed out without terminal recovery evidence: {result.request_id}")
+        elif result.terminal_reason == "failed":
+            detail = result.error or result.trace.get("error") if result.trace is not None else result.error
+            failures.append(f"request failed unexpectedly: {result.request_id}: {detail or 'unknown error'}")
+        if result.trace is None or trace_optional(result, allow_http_guard_rejections):
+            continue
+        missing_fields = [
+            field
+            for field in (
+                "terminal_reason",
+                "active_set_size_max",
+                "pending_queue_size_max",
+                "decode_batch_size_max",
+                "active_set_size_at_terminal",
+                "pending_queue_size_at_terminal",
+                "healthy_baseline_after_terminal",
+            )
+            if field not in result.trace
+        ]
+        if missing_fields:
+            failures.append(f"trace for {result.request_id} is missing fields: {missing_fields}")
+        elif result.trace.get("healthy_baseline_after_terminal") is not True and result.kind == "clean_follow_up":
+            failures.append(f"clean follow-up did not retire to baseline: {result.request_id}")
+
+    final_healthy = bool(
+        follow_up.trace is not None and follow_up.trace.get("healthy_baseline_after_terminal") is True
+    )
+    if not final_healthy:
+        failures.append("clean follow-up trace did not return to healthy baseline")
+
+    active_max = trace_max(all_results, "active_set_size_max")
+    pending_max = trace_max(all_results, "pending_queue_size_max")
+    if require_active_pressure and (active_max is None or active_max < 2):
+        failures.append("active-set pressure was not visible in trace")
+    if active_max is not None and active_max > DSV2_LITE_ACTIVE_CAP:
+        failures.append(
+            f"active_set_size_max exceeded DSV2-Lite active cap {DSV2_LITE_ACTIVE_CAP}: {active_max}"
+        )
+    if require_pending_pressure and (pending_max is None or pending_max < 1):
+        failures.append("pending-queue pressure was not visible in trace")
+
+    return ScenarioReport(
+        name=name,
+        passed=not failures,
+        failures=failures,
+        counts=counts,
+        output_hashes=hashes,
+        trace_coverage={
+            "traced": len(traced),
+            "total": len(all_results),
+            "missing_request_ids": missing,
+            "http_guard_rejections_without_trace": [
+                result.request_id
+                for result in all_results
+                if result.trace is None and trace_optional(result, allow_http_guard_rejections)
+            ],
+        },
+        trace_maxima={
+            "active_set_size_max": active_max,
+            "pending_queue_size_max": pending_max,
+            "decode_batch_size_max": trace_max(all_results, "decode_batch_size_max"),
+        },
+        terminal_reasons=reasons,
+        final_healthy_baseline=final_healthy,
+        clean_follow_up=follow_up,
+        requests=all_results,
+    )
+
+
+def trace_optional(result: RequestResult, allow_http_guard_rejections: bool) -> bool:
+    return (
+        allow_http_guard_rejections
+        and result.terminal_reason == "rejected"
+        and result.status is not None
+        and result.status != 200
+    )
+
+
+def run_cancel_disconnect(base: urllib.parse.ParseResult, args: argparse.Namespace) -> ScenarioReport:
+    specs = [
+        ("dsv2-rel-cancel-after-token", "cancel_after_first_token", make_prompt(0, 16), {"close_after_first_token": True}),
+        ("dsv2-rel-disconnect-before-token", "disconnect_before_token", make_prompt(1, 16), {"close_after_headers": True}),
+        ("dsv2-rel-neighbor-ok", "neighbor_success", make_prompt(2, 16), {}),
+    ]
+    with concurrent.futures.ThreadPoolExecutor(max_workers=len(specs)) as pool:
+        futures = []
+        for request_id, kind, prompt, options in specs:
+            futures.append(
+                pool.submit(
+                    run_stream_request,
+                    base,
+                    args.model,
+                    request_id,
+                    prompt,
+                    args.max_tokens,
+                    args.timeout,
+                    kind,
+                    **options,
+                )
+            )
+        results = [future.result() for future in concurrent.futures.as_completed(futures)]
+    results.sort(key=lambda item: item.request_id)
+    return build_scenario(
+        "cancel_disconnect",
+        results,
+        clean_follow_up(base, args, "cancel-disconnect"),
+        required_reasons={"cancelled", "disconnected", "completed"},
+    )
+
+
+def run_invalid_requests(base: urllib.parse.ParseResult, args: argparse.Namespace) -> ScenarioReport:
+    over_context_words = args.over_context_words
+    specs = [
+        ("dsv2-rel-invalid-sampling", "invalid_non_greedy", make_prompt(3, 16), {"temperature": 0.8}),
+        ("dsv2-rel-invalid-logprobs", "invalid_logprobs", make_prompt(4, 16), {"logprobs": 1}),
+        ("dsv2-rel-invalid-empty", "invalid_empty_prompt", "", {}),
+        ("dsv2-rel-invalid-context", "invalid_over_context", make_prompt(5, over_context_words), {}),
+        ("dsv2-rel-invalid-neighbor-ok", "neighbor_success", make_prompt(6, 16), {}),
+    ]
+    with concurrent.futures.ThreadPoolExecutor(max_workers=len(specs)) as pool:
+        futures = [
+            pool.submit(
+                run_stream_request,
+                base,
+                args.model,
+                request_id,
+                prompt,
+                args.max_tokens,
+                args.timeout,
+                kind,
+                **options,
+            )
+            for request_id, kind, prompt, options in specs
+        ]
+        results = [future.result() for future in concurrent.futures.as_completed(futures)]
+    results.sort(key=lambda item: item.request_id)
+    return build_scenario(
+        "invalid_requests",
+        results,
+        clean_follow_up(base, args, "invalid"),
+        required_reasons={"rejected", "completed"},
+        allow_http_guard_rejections=True,
+    )
+
+
+def run_overload(base: urllib.parse.ParseResult, args: argparse.Namespace) -> ScenarioReport:
+    count = args.overload_concurrency
+    with concurrent.futures.ThreadPoolExecutor(max_workers=count) as pool:
+        futures = [
+            pool.submit(
+                run_stream_request,
+                base,
+                args.model,
+                f"dsv2-rel-overload-{idx}",
+                make_prompt(20, 16),
+                args.max_tokens,
+                args.timeout,
+                "overload_success",
+            )
+            for idx in range(count)
+        ]
+        results = [future.result() for future in concurrent.futures.as_completed(futures)]
+    results.sort(key=lambda item: item.request_id)
+    return build_scenario(
+        "overload_active_cap",
+        results,
+        clean_follow_up(base, args, "overload"),
+        required_reasons={"completed"},
+        require_pending_pressure=True,
+        require_active_pressure=True,
+    )
+
+
+def run_mixed_faults(base: urllib.parse.ParseResult, args: argparse.Namespace) -> ScenarioReport:
+    specs = [
+        ("dsv2-rel-mixed-short-ok-0", "mixed_short_success", make_prompt(10, 16), {}),
+        ("dsv2-rel-mixed-long-ok-0", "mixed_long_success", make_prompt(11, 128), {}),
+        ("dsv2-rel-mixed-cancel", "mixed_cancel", make_prompt(12, 16), {"close_after_first_token": True}),
+        ("dsv2-rel-mixed-logprobs", "mixed_rejected", make_prompt(13, 16), {"logprobs": 1}),
+        ("dsv2-rel-mixed-short-ok-1", "mixed_short_success", make_prompt(10, 16), {}),
+        ("dsv2-rel-mixed-long-ok-1", "mixed_long_success", make_prompt(11, 128), {}),
+    ]
+    with concurrent.futures.ThreadPoolExecutor(max_workers=args.mixed_concurrency) as pool:
+        futures = []
+        for request_id, kind, prompt, options in specs:
+            options = dict(options)
+            close_after_first_token = bool(options.pop("close_after_first_token", False))
+            futures.append(
+                pool.submit(
+                    run_stream_request,
+                    base,
+                    args.model,
+                    request_id,
+                    prompt,
+                    args.mixed_max_tokens,
+                    args.timeout,
+                    kind,
+                    close_after_first_token=close_after_first_token,
+                    **options,
+                )
+            )
+        results = [future.result() for future in concurrent.futures.as_completed(futures)]
+    results.sort(key=lambda item: item.request_id)
+    return build_scenario(
+        "mixed_short_long_with_failures",
+        results,
+        clean_follow_up(base, args, "mixed"),
+        required_reasons={"cancelled", "rejected", "completed"},
+        require_active_pressure=True,
+    )
+
+
+def attach_scenario_traces(scenarios: list[ScenarioReport], traces: dict[str, dict[str, Any]]) -> None:
+    for scenario in scenarios:
+        attach_traces(scenario.requests, traces)
+        refreshed = build_scenario(
+            scenario.name,
+            scenario.requests[:-1],
+            scenario.requests[-1],
+            required_reasons=required_reasons_for(scenario.name),
+            require_pending_pressure=scenario.name == "overload_active_cap",
+            require_active_pressure=scenario.name in {"overload_active_cap", "mixed_short_long_with_failures"},
+            allow_http_guard_rejections=scenario.name == "invalid_requests",
+        )
+        scenario.passed = refreshed.passed
+        scenario.failures = refreshed.failures
+        scenario.counts = refreshed.counts
+        scenario.output_hashes = refreshed.output_hashes
+        scenario.trace_coverage = refreshed.trace_coverage
+        scenario.trace_maxima = refreshed.trace_maxima
+        scenario.terminal_reasons = refreshed.terminal_reasons
+        scenario.final_healthy_baseline = refreshed.final_healthy_baseline
+        scenario.clean_follow_up = refreshed.clean_follow_up
+
+
+def required_reasons_for(name: str) -> set[str]:
+    return {
+        "cancel_disconnect": {"cancelled", "disconnected", "completed"},
+        "invalid_requests": {"rejected", "completed"},
+        "overload_active_cap": {"completed"},
+        "mixed_short_long_with_failures": {"cancelled", "rejected", "completed"},
+    }[name]
+
+
+def dry_run_report(args: argparse.Namespace) -> dict[str, Any]:
+    synthetic_wall_s = 0.0
+
+    def dry_finish(
+        request_id: str,
+        kind: str,
+        ok: bool,
+        terminal_reason: str,
+        output_text: str = "",
+        output_chunks: int = 0,
+    ) -> RequestResult:
+        result = finish_result(
+            request_id,
+            kind,
+            ok,
+            200,
+            synthetic_wall_s,
+            terminal_reason,
+            output_text=output_text,
+            output_chunks=output_chunks,
+        )
+        result.end_wall_s = synthetic_wall_s
+        result.latency_ms = 0.0
+        return result
+
+    def trace(request_id: str, reason: str, active: int, pending: int, decode: int) -> dict[str, Any]:
+        return {
+            "request_id": request_id,
+            "terminal_reason": reason,
+            "active_set_size_max": active,
+            "pending_queue_size_max": pending,
+            "decode_batch_size_max": decode,
+            "active_set_size_at_terminal": 0,
+            "pending_queue_size_at_terminal": 0,
+            "healthy_baseline_after_terminal": True,
+            "prompt_tokens": 16,
+            "completion_tokens": 4,
+        }
+
+    scenarios: list[ScenarioReport] = []
+    samples = {
+        "cancel_disconnect": [("a", "cancelled"), ("b", "disconnected"), ("c", "completed")],
+        "invalid_requests": [("a", "rejected"), ("b", "rejected"), ("c", "completed")],
+        "overload_active_cap": [("a", "completed"), ("b", "completed"), ("c", "completed")],
+        "mixed_short_long_with_failures": [
+            ("a", "completed"),
+            ("b", "cancelled"),
+            ("c", "rejected"),
+            ("d", "completed"),
+        ],
+    }
+    for scenario_name, rows in samples.items():
+        results = []
+        for suffix, reason in rows:
+            request_id = f"dry-{scenario_name}-{suffix}"
+            ok = reason == "completed"
+            result = dry_finish(
+                request_id,
+                "dry",
+                ok,
+                normalize_trace_terminal_reason(reason),
+                output_text="stable" if ok else "",
+                output_chunks=1 if ok else 0,
+            )
+            result.trace = trace(
+                request_id,
+                "completed_length" if reason == "completed" else reason,
+                DSV2_LITE_ACTIVE_CAP if scenario_name == "overload_active_cap" else 3,
+                2 if scenario_name == "overload_active_cap" else 0,
+                2,
+            )
+            results.append(result)
+        follow = dry_finish(
+            f"dry-{scenario_name}-follow",
+            "clean_follow_up",
+            True,
+            "completed",
+            output_text="follow",
+            output_chunks=1,
+        )
+        follow.trace = trace(follow.request_id, "completed_length", 1, 0, 1)
+        scenarios.append(
+            build_scenario(
+                scenario_name,
+                results,
+                follow,
+                required_reasons=required_reasons_for(scenario_name),
+                require_pending_pressure=scenario_name == "overload_active_cap",
+                require_active_pressure=scenario_name
+                in {"overload_active_cap", "mixed_short_long_with_failures"},
+            )
+        )
+    return render_report(args, scenarios, dry_run=True)
+
+
+def render_report(args: argparse.Namespace, scenarios: list[ScenarioReport], *, dry_run: bool) -> dict[str, Any]:
+    passed = all(scenario.passed for scenario in scenarios)
+    report = {
+        "schema_version": 1,
+        "kind": "deepseek_v2_lite_http_reliability_gate",
+        "passed": passed,
+        "dry_run": dry_run,
+        "base_url": args.base_url,
+        "model": args.model,
+        "workload": {
+            "max_tokens": args.max_tokens,
+            "follow_up_tokens": args.follow_up_tokens,
+            "overload_concurrency": args.overload_concurrency,
+            "mixed_concurrency": args.mixed_concurrency,
+            "mixed_max_tokens": args.mixed_max_tokens,
+            "over_context_words": args.over_context_words,
+            "timeout_s": args.timeout,
+        },
+        "summary": {
+            "passed": passed,
+            "scenario_count": len(scenarios),
+            "passed_scenarios": sum(1 for scenario in scenarios if scenario.passed),
+            "failed_scenarios": [scenario.name for scenario in scenarios if not scenario.passed],
+        },
+        "scenarios": [scenario_to_json(scenario) for scenario in scenarios],
+    }
+    return report
+
+
+def scenario_to_json(scenario: ScenarioReport) -> dict[str, Any]:
+    data = asdict(scenario)
+    data["requests"] = [asdict(result) for result in scenario.requests]
+    data["clean_follow_up"] = asdict(scenario.clean_follow_up)
+    return data
+
+
+def run_live(args: argparse.Namespace) -> dict[str, Any]:
+    base = urllib.parse.urlparse(args.base_url)
+    if base.scheme not in {"http", "https"} or not base.hostname:
+        raise SystemExit(f"invalid --base-url: {args.base_url}")
+    scenarios = [
+        run_cancel_disconnect(base, args),
+        run_invalid_requests(base, args),
+        run_overload(base, args),
+        run_mixed_faults(base, args),
+    ]
+    attach_scenario_traces(scenarios, load_server_traces(args.server_log))
+    return render_report(args, scenarios, dry_run=False)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--base-url", default="http://127.0.0.1:8000")
+    parser.add_argument("--model", default="deepseek-v2-lite")
+    parser.add_argument("--server-log", type=Path, help="server log with openinfer_http_trace lines")
+    parser.add_argument("--out", type=Path)
+    parser.add_argument("--timeout", type=float, default=180.0)
+    parser.add_argument("--max-tokens", type=int, default=16)
+    parser.add_argument("--follow-up-tokens", type=int, default=8)
+    parser.add_argument("--overload-concurrency", type=int, default=12)
+    parser.add_argument("--mixed-concurrency", type=int, default=6)
+    parser.add_argument("--mixed-max-tokens", type=int, default=16)
+    parser.add_argument("--over-context-words", type=int, default=9000)
+    parser.add_argument("--dry-run", action="store_true", help="validate schema and pass/fail logic without HTTP")
+    args = parser.parse_args()
+
+    if args.max_tokens <= 0 or args.follow_up_tokens <= 0 or args.mixed_max_tokens <= 0:
+        raise SystemExit("token counts must be positive")
+    if args.overload_concurrency <= 8:
+        raise SystemExit("--overload-concurrency must exceed the DSV2-Lite active cap of 8")
+    if args.mixed_concurrency <= 1:
+        raise SystemExit("--mixed-concurrency must be > 1")
+
+    report = dry_run_report(args) if args.dry_run else run_live(args)
+    rendered = json.dumps(report, indent=2, sort_keys=True)
+    if args.out is not None:
+        args.out.parent.mkdir(parents=True, exist_ok=True)
+        args.out.write_text(rendered + "\n", encoding="utf-8")
+    print(rendered)
+    if not report["summary"]["passed"]:
+        raise SystemExit(1)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/tests/test_bench_dsv2lite_http_reliability.py b/tests/test_bench_dsv2lite_http_reliability.py
new file mode 100644
index 00000000..d0041dcd
--- /dev/null
+++ b/tests/test_bench_dsv2lite_http_reliability.py
@@ -0,0 +1,114 @@
+"""Regression tests for scripts/bench_dsv2lite_http_reliability.py."""
+
+from __future__ import annotations
+
+import argparse
+import importlib.util
+import sys
+import time
+import unittest
+from pathlib import Path
+
+
+SCRIPT_PATH = Path(__file__).resolve().parents[1] / "scripts" / "bench_dsv2lite_http_reliability.py"
+SPEC = importlib.util.spec_from_file_location("bench_dsv2lite_http_reliability", SCRIPT_PATH)
+assert SPEC is not None and SPEC.loader is not None
+bench_dsv2lite_http_reliability = importlib.util.module_from_spec(SPEC)
+sys.modules[SPEC.name] = bench_dsv2lite_http_reliability
+SPEC.loader.exec_module(bench_dsv2lite_http_reliability)
+
+
+def _result(request_id: str, kind: str, ok: bool, reason: str):
+    return bench_dsv2lite_http_reliability.finish_result(
+        request_id,
+        kind,
+        ok,
+        200,
+        time.time(),
+        reason,
+        output_text="stable" if ok else "",
+        output_chunks=1 if ok else 0,
+    )
+
+
+def _trace(request_id: str, reason: str, active: int = 2, pending: int = 0, decode: int = 1):
+    return {
+        "request_id": request_id,
+        "terminal_reason": reason,
+        "active_set_size_max": active,
+        "pending_queue_size_max": pending,
+        "decode_batch_size_max": decode,
+        "active_set_size_at_terminal": 0,
+        "pending_queue_size_at_terminal": 0,
+        "healthy_baseline_after_terminal": True,
+        "prompt_tokens": 16,
+        "completion_tokens": 4,
+    }
+
+
+class ReliabilityGateTests(unittest.TestCase):
+    def test_trace_terminal_reason_overrides_client_disconnect_label(self):
+        disconnected = _result("disconnect", "disconnect_early_bytes", False, "disconnected")
+        disconnected.trace = _trace("disconnect", "error")
+        disconnected.terminal_reason = bench_dsv2lite_http_reliability.normalize_trace_terminal_reason(
+            "error"
+        )
+        completed = _result("ok", "neighbor_success", True, "completed")
+        completed.trace = _trace("ok", "completed_length")
+        follow = _result("follow", "clean_follow_up", True, "completed")
+        follow.trace = _trace("follow", "completed_length", active=1)
+
+        scenario = bench_dsv2lite_http_reliability.build_scenario(
+            "cancel_disconnect",
+            [disconnected, completed],
+            follow,
+            required_reasons={"disconnected", "completed"},
+        )
+
+        self.assertFalse(scenario.passed)
+        self.assertEqual(scenario.counts["disconnected"], 0)
+        self.assertIn("expected terminal reason 'disconnected' was not observed", scenario.failures)
+        self.assertTrue(any("request failed unexpectedly: disconnect" in item for item in scenario.failures))
+
+    def test_missing_trace_fields_fail_gate(self):
+        completed = _result("ok", "neighbor_success", True, "completed")
+        completed.trace = {
+            "request_id": "ok",
+            "terminal_reason": "completed_length",
+            "active_set_size_max": 1,
+        }
+        follow = _result("follow", "clean_follow_up", True, "completed")
+        follow.trace = _trace("follow", "completed_length", active=1)
+
+        scenario = bench_dsv2lite_http_reliability.build_scenario(
+            "invalid_requests",
+            [completed],
+            follow,
+            required_reasons={"completed"},
+        )
+
+        self.assertFalse(scenario.passed)
+        self.assertTrue(any("trace for ok is missing fields" in item for item in scenario.failures))
+
+    def test_dry_run_report_stays_green(self):
+        args = argparse.Namespace(
+            base_url="http://127.0.0.1:8000",
+            model="deepseek-v2-lite",
+            max_tokens=16,
+            follow_up_tokens=8,
+            overload_concurrency=12,
+            mixed_concurrency=6,
+            mixed_max_tokens=16,
+            over_context_words=9000,
+            timeout=180.0,
+        )
+
+        report = bench_dsv2lite_http_reliability.dry_run_report(args)
+
+        self.assertTrue(report["passed"])
+        self.assertTrue(report["summary"]["passed"])
+        self.assertEqual(report["summary"]["passed_scenarios"], 4)
+
+
+if __name__ == "__main__":
+    unittest.main()

From cb4e25b6d45409472c5c9f39fadb7c8332c5439a Mon Sep 17 00:00:00 2001
From: CAICAIIs <3360776475@qq.com>
Date: Sat, 4 Jul 2026 22:02:27 +0800
Subject: [PATCH 2/2] fix(dsv2lite): update dynamo token sink abort state

---
 openinfer-dynamo-backend/src/engine.rs    |  32 ++--
 pr-body-dsv2lite-http-reliability-gate.md | 189 +++++++++-------------
 2 files changed, 96 insertions(+), 125 deletions(-)

diff --git a/openinfer-dynamo-backend/src/engine.rs b/openinfer-dynamo-backend/src/engine.rs
index 89449510..08da4112 100644
--- a/openinfer-dynamo-backend/src/engine.rs
+++ b/openinfer-dynamo-backend/src/engine.rs
@@ -15,7 +15,7 @@
 use std::path::PathBuf;
 use std::sync::Arc;
 use std::sync::Mutex;
-use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::atomic::{AtomicU8, Ordering};
 use std::time::Duration;
 
 use async_trait::async_trait;
@@ -27,7 +27,8 @@ use dynamo_backend_common::{
 };
 use futures::stream::BoxStream;
 use openinfer_engine::engine::{
-    EngineHandle, GenerateRequest, KvBlockEvent, LoadSnapshot, TokenSink, TokenStreamReceiver,
+    EngineHandle, GenerateRequest, KvBlockEvent, LoadSnapshot, RequestAbortReason, TokenSink,
+    TokenStreamReceiver,
 };
 use openinfer_qwen3::{
     DEFAULT_GPU_MEMORY_UTILIZATION, DEFAULT_KV_CACHE_MEMORY_MARGIN_BYTES,
@@ -263,18 +264,17 @@ impl LLMEngine for OpeninferBackend {
         let params = convert::to_sampling_params(&request);
         let max_tokens = convert::resolve_max_tokens(&request);
 
-        // Per-request private channel + cancel flag. openinfer's scheduler
-        // learns to retire this request by observing the flag (its next emit
-        // sees the sink closed) — the reactive abort the engine is built
-        // around. `TokenSink::standalone()` hard-codes a never-tripped flag,
-        // so we build the sink by hand to own the flag. The channel is
-        // unbounded, but each request emits at most `max_tokens` items before
-        // its terminal, so growth is bounded by the token cap, not by consumer
-        // backpressure.
+        // Per-request private channel + abort reason. openinfer's scheduler
+        // learns to retire this request by observing the reason (its next emit
+        // sees the sink aborted) — the reactive abort the engine is built
+        // around. `TokenSink::standalone()` hard-codes a never-tripped reason,
+        // so we build the sink by hand to own it. The channel is unbounded, but
+        // each request emits at most `max_tokens` items before its terminal, so
+        // growth is bounded by the token cap, not by consumer backpressure.
         let (tx, rx): (_, TokenStreamReceiver) = mpsc::unbounded_channel();
-        let cancelled = Arc::new(AtomicBool::new(false));
+        let abort_reason = Arc::new(AtomicU8::new(RequestAbortReason::None as u8));
         let tag: Arc<str> = Arc::from(ctx.id());
-        let sink = TokenSink::new(tag, tx, cancelled.clone());
+        let sink = TokenSink::new(tag, tx, Arc::clone(&abort_reason));
 
         let req = GenerateRequest {
             request_id: Some(ctx.id().to_string()),
@@ -299,7 +299,7 @@ impl LLMEngine for OpeninferBackend {
 
         Ok(Box::pin(token_stream(
             rx,
-            cancelled,
+            abort_reason,
             ctx.inner_arc(),
             self.cancel_token(),
             prompt_tokens,
@@ -389,7 +389,7 @@ impl LLMEngine for OpeninferBackend {
 /// — or on cancellation, whichever comes first.
 fn token_stream(
     mut rx: TokenStreamReceiver,
-    cancelled: Arc<AtomicBool>,
+    abort_reason: Arc<AtomicU8>,
     ctx: Arc<dyn AsyncEngineContext>,
     cancel: CancellationToken,
     prompt_tokens: u32,
@@ -418,13 +418,13 @@ fn token_stream(
             tokio::select! {
                 biased;
                 _ = cancel.cancelled() => {
-                    cancelled.store(true, Ordering::Release);
+                    RequestAbortReason::Cancelled.store(&abort_reason);
                     yield Ok(LLMEngineOutput::cancelled()
                         .with_usage(cancel_usage(completion_tokens, cached_tokens)));
                     break;
                 }
                 _ = ctx.stopped() => {
-                    cancelled.store(true, Ordering::Release);
+                    RequestAbortReason::Cancelled.store(&abort_reason);
                     yield Ok(LLMEngineOutput::cancelled()
                         .with_usage(cancel_usage(completion_tokens, cached_tokens)));
                     break;
diff --git a/pr-body-dsv2lite-http-reliability-gate.md b/pr-body-dsv2lite-http-reliability-gate.md
index f25b36e6..27840d84 100644
--- a/pr-body-dsv2lite-http-reliability-gate.md
+++ b/pr-body-dsv2lite-http-reliability-gate.md
@@ -1,58 +1,44 @@
 ## Summary
 
-- Adds a replayable DeepSeek-V2-Lite EP2 HTTP reliability gate for real streaming `/v1/completions` requests.
-- Extends the request lifecycle contract across the vLLM bridge, `TokenSink`, and DSV2-Lite scheduler so cancel, disconnect, rejection, error, and completion are machine-distinguishable.
-- Adds strict runner JSON, focused tests, live host-staged/NCCL evidence, no-regression benchmark rows, and status-doc claim boundaries.
+Adds a replayable DeepSeek-V2-Lite EP2 HTTP reliability gate for real streaming `/v1/completions` traffic.
 
-Fixes #453.
-
-## Extra Risks Found And Fixed
-
-### Cancel and disconnect were not machine-distinguishable
-
-- Risk: a dropped per-request stream and an explicit frontend abort both looked like a send failure, so the HTTP gate could not prove whether it exercised cancel versus disconnect.
-- Why it affects #453: the issue asks for cancel and disconnect coverage as separate failure modes.
-- Minimal fix: add request-local abort state in `TokenSink` / bridge handling, expose `is_cancelled()`, `is_disconnected()`, and `abort_reason()`, and map send failures to `cancelled` versus `disconnected` terminal reasons in the DSV2-Lite scheduler.
-- Evidence: `cargo test --release -p openinfer-engine --lib token_sink -- --nocapture`, `cargo test --release -p openinfer-vllm-frontend --lib abort -- --nocapture`, and the live `cancel_disconnect` scenario on both backends.
+This PR also tightens the request lifecycle contract across the vLLM bridge, `TokenSink`, and the DSV2-Lite scheduler so completion, rejection, cancellation, disconnect, and error retirement are machine-distinguishable in traces.
 
-### Early disconnect could be misclassified after non-token bridge output
-
-- Risk: the bridge could mark a request as having emitted output before a client-visible token, making an early connection close look like cancel instead of disconnect.
-- Why it affects #453: disconnect-before-token and cancel-after-token are different reliability cases.
-- Minimal fix: track `has_emitted_tokens` at the bridge boundary and classify aborts using client-visible token output, not internal metadata progress.
-- Evidence: `abort_reason_tracks_first_output_boundary` and live `cancel_disconnect` artifacts showing one `cancelled` and one `disconnected` terminal reason per backend.
-
-### Terminal traces could not prove state retirement
-
-- Risk: prior traces had active/decode maxima but no terminal reason or terminal active/pending state, so the gate could not fail on leaked active or pending state.
-- Why it affects #453: clean follow-up success alone is weaker than direct scheduler-state evidence.
-- Minimal fix: add `terminal_reason`, `pending_queue_size_max`, `active_set_size_at_terminal`, `pending_queue_size_at_terminal`, and `healthy_baseline_after_terminal` to `openinfer_http_trace`.
-- Evidence: scheduler lifecycle tests plus the reliability runner's strict trace checks and live artifacts with healthy final baselines after every scenario.
+Fixes #453.
 
-### Overload pressure needed pending-queue recovery evidence
+## What Changed
 
-- Risk: an overload run could complete some requests while still hiding bad pending/active cleanup.
-- Why it affects #453: active-cap pressure must prove queued work does not poison active requests and the queue can recover.
-- Minimal fix: thread active/pending state through batch decode, single-row fallback, cache-position errors, and terminal trace logging; require pending pressure in the runner.
-- Evidence: `overload_active_cap` passed on host-staged and NCCL with `active_set_size_max=8`, `pending_queue_size_max=4`, `decode_batch_size_max=7`, `completed=13`, and healthy final baseline.
+- Added `scripts/bench_dsv2lite_http_reliability.py` for live HTTP reliability scenarios:
+  - client cancel / disconnect during streaming;
+  - invalid or unsupported request parameters;
+  - active-cap overload pressure;
+  - mixed short/long prompts with adjacent failures;
+  - clean follow-up request after every scenario.
+- Added DSV2-Lite scheduler lifecycle instrumentation:
+  - terminal reason;
+  - active/pending/decode maxima;
+  - terminal active/pending state;
+  - healthy final baseline evidence.
+- Added request abort reason tracking so cancel and disconnect are distinguishable.
+- Added focused unit coverage for token sink abort state, vLLM bridge abort behavior, scheduler retirement, active-cap recovery, and terminal accounting.
+- Updated the DeepSeek-V2-Lite status ledger with the reliability evidence and claim boundary.
 
-### Cross-group terminal accounting could overstate active state
+## Extra Risks Fixed
 
-- Risk: when different decode-position groups retired rows in the same scheduler round, terminal traces could use a per-group active count instead of a shared round-level active count.
-- Why it affects #453: the reliability gate must prove active state retires after failures, so stale terminal active counts weaken the cleanup claim.
-- Minimal fix: carry one shared active-remaining counter across same-position batch groups and single-row fallback paths.
-- Evidence: `cross_group_terminal_accounting_uses_shared_active_remaining` and live terminal traces with healthy baselines after mixed failure scenarios.
+While implementing #453, the gate exposed a few reliability risks that would have weakened the result:
 
-### Benchmark/gate could pass without terminal trace coverage
+- Cancel and disconnect previously collapsed into similar send-failure behavior, so the gate could not prove both cases independently.
+- Early disconnect could be misclassified after non-token bridge output.
+- Terminal traces lacked enough state to prove active/pending retirement.
+- Overload runs could complete requests while hiding pending-queue cleanup issues.
+- Cross-group terminal accounting could overstate active state within one scheduler round.
+- A benchmark could pass while missing terminal trace coverage for failed/cancelled/disconnected requests.
 
-- Risk: a completed HTTP benchmark can miss failed, cancelled, rejected, or disconnected terminal traces.
-- Why it affects #453: the acceptance criteria require counts, terminal reasons, trace coverage, active/pending/decode maxima, and clean follow-up recovery.
-- Minimal fix: add `scripts/bench_dsv2lite_http_reliability.py` with strict per-scenario failure rules. HTTP/frontend guard rejections may lack scheduler traces; scheduler-level terminal requests must have traces.
-- Evidence: `python3 -m unittest tests.test_bench_dsv2lite_http_reliability`, host-staged reliability artifact `832d65a8e8b2b3a6ad0100c4a35f38475f040d6ffc192ec38a3b7384167187a5`, and NCCL reliability artifact `53bedd98f19c5241df588a1ade8756e84a5e8c99225a589c4ed303e90fba38fa`.
+Each fix is scoped to the DSV2-Lite HTTP/scheduler reliability path.
 
 ## Evidence
 
-Local checks completed:
+Local checks:
 
 ```bash
 cargo fmt --all --check
@@ -63,7 +49,7 @@ python3 -m unittest tests.test_bench_dsv2lite_http_reliability
 python3 -m py_compile scripts/bench_dsv2lite_http_reliability.py tests/test_bench_dsv2lite_http_reliability.py
 ```
 
-Remote 2-GPU checks completed:
+Remote 2-GPU checks:
 
 ```bash
 cargo build --release -p openinfer-server --no-default-features --features deepseek-v2-lite
@@ -75,87 +61,72 @@ OPENINFER_TEST_MODEL_PATH=models/DeepSeek-V2-Lite OPENINFER_DSV2_LITE_EP_BACKEND
 
 Results:
 
-- `openinfer-engine` token sink tests passed: 3 passed, 0 failed.
-- `openinfer-vllm-frontend` abort tests passed: 2 passed, 0 failed.
-- DSV2-Lite scheduler lifecycle tests passed: 23 passed, 0 failed.
-- Host-staged `e2e_ep2` passed: 1 passed, 0 failed.
-- NCCL `e2e_ep2` passed: 1 passed, 0 failed. The SM120 validation used an NCCL 2.30.7 runtime library.
-- Python reliability runner tests passed: 3 passed, 0 failed.
-- Python syntax check passed.
+- `openinfer-engine` token sink tests: 3 passed.
+- `openinfer-vllm-frontend` abort tests: 2 passed.
+- DSV2-Lite scheduler lifecycle tests: 23 passed.
+- Host-staged `e2e_ep2`: 1 passed.
+- NCCL `e2e_ep2`: 1 passed.
+- Python reliability runner tests: 3 passed.
 - Server release build passed.
 
-Live HTTP reliability artifacts:
+## HTTP Reliability Gate
 
-| Backend | Artifact | SHA-256 | Result |
-| --- | --- | --- | --- |
-| host-staged | `http_reliability_host_staged.json` | `832d65a8e8b2b3a6ad0100c4a35f38475f040d6ffc192ec38a3b7384167187a5` | passed |
-| NCCL | `http_reliability_nccl.json` | `53bedd98f19c5241df588a1ade8756e84a5e8c99225a589c4ed303e90fba38fa` | passed |
+Live reliability artifacts passed on both backends:
 
-Scenario summary:
+| Backend | Artifact | SHA-256 |
+| --- | --- | --- |
+| host-staged | `http_reliability_host_staged.json` | `832d65a8e8b2b3a6ad0100c4a35f38475f040d6ffc192ec38a3b7384167187a5` |
+| NCCL | `http_reliability_nccl.json` | `53bedd98f19c5241df588a1ade8756e84a5e8c99225a589c4ed303e90fba38fa` |
 
-| Backend | Scenario | Counts | Trace maxima | Final baseline |
-| --- | --- | --- | --- | --- |
-| host-staged | `cancel_disconnect` | completed `2`, cancelled `1`, disconnected `1`, failed/rejected/timeout `0` | active `2`, pending `0`, decode `1` | healthy |
-| host-staged | `invalid_requests` | completed `2`, rejected `4`, failed/timeout `0` | active `1`, pending `0`, decode `1` | healthy |
-| host-staged | `overload_active_cap` | completed `13`, failed/rejected/timeout `0` | active `8`, pending `4`, decode `7` | healthy |
-| host-staged | `mixed_short_long_with_failures` | completed `5`, cancelled `1`, rejected `1`, failed/timeout `0` | active `4`, pending `0`, decode `2` | healthy |
-| NCCL | `cancel_disconnect` | completed `2`, cancelled `1`, disconnected `1`, failed/rejected/timeout `0` | active `2`, pending `0`, decode `1` | healthy |
-| NCCL | `invalid_requests` | completed `2`, rejected `4`, failed/timeout `0` | active `1`, pending `0`, decode `1` | healthy |
-| NCCL | `overload_active_cap` | completed `13`, failed/rejected/timeout `0` | active `8`, pending `4`, decode `7` | healthy |
-| NCCL | `mixed_short_long_with_failures` | completed `5`, cancelled `1`, rejected `1`, failed/timeout `0` | active `5`, pending `0`, decode `2` | healthy |
+Both backends passed:
+
+- `cancel_disconnect`
+- `invalid_requests`
+- `overload_active_cap`
+- `mixed_short_long_with_failures`
+
+Every scenario ended with a healthy scheduler baseline and a successful clean follow-up request.
 
 ## No-Regression Benchmark
 
-The no-regression benchmark used real `/v1/completions` traffic after the reliability gate. Every row passed with `failed=0`, `timeouts=0`, and `missing_traces=[]`.
+The no-regression HTTP benchmark ran after the reliability gate.
+
+Coverage:
+
+- short same-shape: `input_len=64`, `output_len=64`, concurrency `1/4/8`;
+- mixed short/long: `prompt_words=16,128`, `max_tokens=16`, concurrency `4/8`;
+- host-staged and NCCL.
 
-| Backend | Shape | Completed | Failed/timeouts | Output tok/s | TTFT avg ms | TPOT/ITL avg ms | active max | decode batch max |
-| --- | --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: |
-| host-staged | short `64/64`, c1 | 8/8 | 0/0 | 22.886 | 1226.8 | 24.9 | 1 | 1 |
-| host-staged | short `64/64`, c4 | 8/8 | 0/0 | 21.698 | 2612.8 | 145.1 | 4 | 3 |
-| host-staged | short `64/64`, c8 | 8/8 | 0/0 | 21.830 | 5439.8 | 284.2 | 8 | 5 |
-| host-staged | mixed `16/128`, c4 | 8/8 | 0/0 | 8.281 | 4543.3 | 210.0 | 4 | 2 |
-| host-staged | mixed `16/128`, c8 | 8/8 | 0/0 | 8.288 | 7897.1 | 496.9 | 8 | 3 |
-| NCCL | short `64/64`, c1 | 8/8 | 0/0 | 24.873 | 1034.2 | 24.4 | 1 | 1 |
-| NCCL | short `64/64`, c4 | 8/8 | 0/0 | 23.678 | 2210.0 | 135.8 | 4 | 3 |
-| NCCL | short `64/64`, c8 | 8/8 | 0/0 | 23.872 | 4594.4 | 266.1 | 8 | 6 |
-| NCCL | mixed `16/128`, c4 | 8/8 | 0/0 | 9.186 | 2216.6 | 314.9 | 4 | 1 |
-| NCCL | mixed `16/128`, c8 | 8/8 | 0/0 | 9.269 | 6920.1 | 453.0 | 8 | 3 |
+Result:
 
-This is a no-regression and trace-coverage record. It is not a throughput optimization claim.
+- all 10 rows completed `8/8` requests;
+- `failed=0`;
+- `timeouts=0`;
+- `missing_traces=[]`;
+- trace maxima captured active request pressure and decode batch activity.
+
+This benchmark shows that the reliability changes did not regress normal HTTP serving behavior under the covered concurrent workloads. It is not a throughput optimization claim.
 
 ## Claim Boundary
 
-- This PR claims a stronger HTTP reliability gate and auditable failure isolation for DeepSeek-V2-Lite EP2 serving.
-- It does not claim production EP readiness.
-- It does not claim vLLM parity.
-- It does not claim sparse dispatch, multi-node EP, CUDA Graph productization, or host-staged deprecation.
-- It does not claim throughput improvement from the reliability runner or no-regression benchmark.
+This PR claims stronger DeepSeek-V2-Lite EP2 HTTP reliability evidence and auditable failure isolation.
+
+It does not claim:
+
+- production EP readiness;
+- vLLM parity;
+- sparse dispatch readiness;
+- multi-node EP support;
+- CUDA Graph productization;
+- throughput improvement.
 
 ## Remaining Risks
 
 No #453 reliability blocker remains after the gates above.
 
-Separate follow-ups remain outside this PR's claim boundary:
-
-- #452 remains the mixed/long-prompt latency follow-up.
-- #465 remains the sustained soak benchmark follow-up.
-- #466 remains the retained SLO report follow-up.
-- #467 remains the benchmark artifact manifest follow-up.
-
-## Commands Run And Exact Result
-
-| Command | Result |
-| --- | --- |
-| `cargo fmt --all --check` | Passed locally. |
-| `git diff --check` | Passed locally. |
-| `python3 -m py_compile scripts/bench_dsv2lite_http_reliability.py tests/test_bench_dsv2lite_http_reliability.py` | Passed locally. |
-| `python3 -m unittest tests.test_bench_dsv2lite_http_reliability` | Passed locally; 3 tests verify dry-run success plus false-positive failures for bad terminal traces and missing trace fields. |
-| `cargo test --release -p openinfer-engine --lib token_sink -- --nocapture` | Passed locally; 3 tests verify explicit cancel, disconnect, and closed receiver behavior. |
-| `cargo test --release -p openinfer-vllm-frontend --lib abort -- --nocapture` | Passed locally and remotely; 2 tests verify abort handling. |
-| `cargo build --release -p openinfer-server --no-default-features --features deepseek-v2-lite` | Passed remotely. |
-| `cargo test --release -p openinfer-deepseek-v2-lite --features deepseek-v2-lite --lib scheduler -- --nocapture` | Passed remotely; 23 passed, 0 failed. |
-| `OPENINFER_TEST_MODEL_PATH=models/DeepSeek-V2-Lite OPENINFER_DSV2_LITE_EP_BACKEND=host-staged cargo test --release -p openinfer-deepseek-v2-lite --features deepseek-v2-lite --test e2e_ep2 -- --nocapture` | Passed remotely; 1 passed, 0 failed. |
-| `OPENINFER_TEST_MODEL_PATH=models/DeepSeek-V2-Lite OPENINFER_DSV2_LITE_EP_BACKEND=nccl cargo test --release -p openinfer-deepseek-v2-lite --features deepseek-v2-lite --test e2e_ep2 -- --nocapture` | Passed remotely; 1 passed, 0 failed. |
-| host-staged live HTTP reliability runner | Passed; artifact `http_reliability_host_staged.json`, SHA-256 `832d65a8e8b2b3a6ad0100c4a35f38475f040d6ffc192ec38a3b7384167187a5`. |
-| NCCL live HTTP reliability runner | Passed; artifact `http_reliability_nccl.json`, SHA-256 `53bedd98f19c5241df588a1ade8756e84a5e8c99225a589c4ed303e90fba38fa`. |
-| host-staged/NCCL no-regression HTTP benchmark | Passed; all 10 rows completed 8/8 requests with failed/timeouts `0/0` and `missing_traces=[]`. |
+Separate follow-ups remain outside this PR:
+
+- #452: mixed/long-prompt latency;
+- #465: sustained soak benchmark;
+- #466: retained SLO report;
+- #467: benchmark artifact manifest.