Derive preset decode rate from E2E latency

src/dstack/_internal/cli/models/preset_agent.py

@@ -9,11 +9,16 @@
     ValidationError,
     ValidatorFunctionWrapHandler,
     WrapValidator,
+    model_validator,
 )
 
 from dstack._internal.core.models.common import CoreModel
 from dstack._internal.core.models.configurations import ServiceConfiguration
-from dstack._internal.core.models.presets import PresetBenchmark
+from dstack._internal.core.models.presets import (
+    PresetBenchmark,
+    PresetBenchmarkLatency,
+    PresetBenchmarkMetrics,
+)
 
 
 class PresetAgentInvalidService(CoreModel):
@@ -42,6 +47,33 @@ def _service_from_yaml(value: Any, handler: ValidatorFunctionWrapHandler) -> Any
 ]
 
 
+class PresetAgentBenchmarkMetrics(PresetBenchmarkMetrics):
+    """Fresh reports require the E2E latency missing from legacy stored presets."""
+
+    e2e_ms: PresetBenchmarkLatency
+
+
+class PresetAgentBenchmark(PresetBenchmark):
+    metrics: PresetAgentBenchmarkMetrics
+
+    @model_validator(mode="after")
+    def validate_reported_tpot_is_possible(self) -> "PresetAgentBenchmark":
+        tpot_ms = self.metrics.tpot_ms
+        if tpot_ms is None:
+            return self
+        decode_token_count = self.metrics.total_output_tokens - self.metrics.successful_requests
+        reported_decode_seconds = decode_token_count * tpot_ms.mean / 1000
+        available_slot_seconds = self.metrics.duration_seconds * self.workload.concurrency
+        # Benchmark tools round their displayed latency values. A one-percent
+        # allowance prevents that formatting from rejecting a boundary value.
+        if reported_decode_seconds > available_slot_seconds * 1.01:
+            raise ValueError(
+                "reported tpot_ms exceeds the decode time available from duration_seconds"
+                " and workload.concurrency"
+            )
+        return self
+
+
 class PresetAgentSuccess(CoreModel):
     """A preset the agent created, cross-checked against the run in `verify`."""
 
@@ -53,7 +85,7 @@ class PresetAgentSuccess(CoreModel):
     base: Annotated[str, Field(min_length=1)]
     model: Annotated[str, Field(min_length=1)]
     context_length: PositiveInt
-    benchmark: PresetBenchmark
+    benchmark: PresetAgentBenchmark
 
 
 class PresetAgentFailure(CoreModel):

src/dstack/_internal/cli/services/presets/output.py

@@ -211,9 +211,9 @@ def _add_session(table: Table, session: dict[str, Any], *, verbose: bool = False
         tps = _format_number(best["tok_s"])
         if objective:
             # Lead with per-user tps: comparable across rows regardless of concurrency.
-            tpot_ms = best.get("tpot_ms")
-            if isinstance(tpot_ms, (int, float)) and tpot_ms > 0:
-                parts.append(f"tps/user={_format_number(1000 / tpot_ms)}")
+            per_user_tok_s = best.get("per_user_tok_s")
+            if isinstance(per_user_tok_s, (int, float)) and per_user_tok_s > 0:
+                parts.append(f"tps/user={_format_number(per_user_tok_s)}")
             if verbose:
                 parts.append(f"tps={tps}")
             ttft_ms = best.get("ttft_ms")

src/dstack/_internal/cli/services/presets/resources/system_prompt.md

@@ -367,6 +367,7 @@ trial benchmarks in `trials/<n>/trial.json`, the final benchmark as
     "total_input_tokens": 16384, "total_output_tokens": 2048,
     "output_tok_per_s": 512.0, "per_user_tok_per_s": 64.0,
     "ttft_ms": {"mean": 110.9, "p50": 108.2, "p99": 121.6},
+    "e2e_ms": {"mean": 1063.4, "p50": 1048.0, "p99": 1150.3},
     "tpot_ms": {"mean": 7.5, "p50": 7.4, "p99": 8.1}
   }
 }
@@ -379,7 +380,13 @@ input and output token counts of the benchmark, rounded to whole tokens.
 <!--?end-->
 Compute `output_tok_per_s` as `total_output_tokens / duration_seconds` and
 `per_user_tok_per_s` as `output_tok_per_s / workload.concurrency`. These are
-the numbers used to compare trials (see `## Performance`).
+the numbers used to compare trials (see `## Performance`). Record the benchmark
+tool's per-request end-to-end latency as `e2e_ms`. Record `tpot_ms` only when the
+tool reports time per output token; set it to `null` when the tool only reports
+inter-token latency per streamed chunk. In particular, do not use SGLang's ITL
+row as TPOT: speculative decoding can put several accepted tokens in one chunk.
+The CLI derives its per-user decode rate from `e2e_ms`, `ttft_ms`, and the output
+token totals rather than trusting either self-reported throughput field.
 
 Set `tool` to the command name and subcommands without options or values,
 `tool_version` to the exact version, and `command` to the secret-free

src/dstack/_internal/cli/services/presets/session.py

@@ -628,19 +628,43 @@ def _trial_entry(
     gpu: Optional[str],
 ) -> dict[str, Any]:
     ttft = (metrics.get("ttft_ms") or {}).get("p50")
-    # Mean, not p50: MTP skews TPOT right, and mean is the delivered decode rate.
-    tpot = (metrics.get("tpot_ms") or {}).get("mean")
     context_length = record.get("context_length")
     return {
         "tok_s": tok_s,
-        "tpot_ms": tpot if isinstance(tpot, (int, float)) and tpot > 0 else None,
+        "per_user_tok_s": _effective_per_user_tok_s(metrics),
         "ttft_ms": ttft if isinstance(ttft, (int, float)) else None,
         "context_length": context_length if isinstance(context_length, int) else None,
         "concurrency": workload.get("concurrency"),
         "gpu": gpu,
     }
 
 
+def _effective_per_user_tok_s(metrics: dict[str, Any]) -> Optional[float]:
+    e2e_ms = (metrics.get("e2e_ms") or {}).get("mean")
+    ttft_ms = (metrics.get("ttft_ms") or {}).get("mean")
+    total_output_tokens = metrics.get("total_output_tokens")
+    successful_requests = metrics.get("successful_requests")
+    if (
+        isinstance(e2e_ms, (int, float))
+        and isinstance(ttft_ms, (int, float))
+        and isinstance(total_output_tokens, (int, float))
+        and isinstance(successful_requests, (int, float))
+        and e2e_ms > ttft_ms
+        and successful_requests > 0
+        and total_output_tokens > successful_requests
+    ):
+        mean_output_tokens = total_output_tokens / successful_requests
+        mean_tpot_ms = (e2e_ms - ttft_ms) / (mean_output_tokens - 1)
+        return 1000 / mean_tpot_ms
+
+    # Sessions created by older clients have no E2E latency. Preserve their
+    # display using the reported TPOT, but new reports never depend on it.
+    legacy_tpot_ms = (metrics.get("tpot_ms") or {}).get("mean")
+    if isinstance(legacy_tpot_ms, (int, float)) and legacy_tpot_ms > 0:
+        return 1000 / legacy_tpot_ms
+    return None
+
+
 def _format_trial_gpu(record: dict[str, Any]) -> Optional[str]:
     resources = record.get("resources")
     gpu = resources.get("gpu") if isinstance(resources, dict) else None

src/dstack/_internal/cli/services/presets/verify.py

@@ -192,7 +192,9 @@ def _normalized_benchmark(
             {"dataset": None} if configuration.dataset is None else {"shared_prefix_tokens": 0}
         )
     )
-    return benchmark.model_copy(update={"workload": workload})
+    return PresetBenchmark.model_validate(
+        {**benchmark.model_dump(), "workload": workload.model_dump()}
+    )
 
 
 def _portable_service(

src/dstack/_internal/core/models/presets.py

@@ -83,13 +83,22 @@ class PresetBenchmarkMetrics(CoreModel):
     output_tok_per_s: PositiveFloat
     per_user_tok_per_s: PositiveFloat
     ttft_ms: PresetBenchmarkLatency
-    tpot_ms: PresetBenchmarkLatency
+    # Old stored presets predate `e2e_ms`. New agent reports require it via
+    # `PresetAgentBenchmarkMetrics` and use it for the derived decode rate.
+    e2e_ms: Optional[PresetBenchmarkLatency] = None
+    # Some benchmark tools report inter-token latency per streamed chunk rather
+    # than TPOT per output token. Keep genuine TPOT when available, but do not
+    # force the agent to substitute a different metric.
+    tpot_ms: Optional[PresetBenchmarkLatency] = None
 
 
 class PresetBenchmark(CoreModel):
-    """The agent reports its benchmark in exactly this shape, and is forced to by
-    the schema generated from it. Changing a field here means also changing the
-    `## Benchmark` section of the system prompt, which tells it what to put there."""
+    """The benchmark shape stored in presets.
+
+    Fresh agent reports use a stricter subclass in `cli/models/preset_agent.py`.
+    Changing these fields also means changing the `## Benchmark` section of the
+    system prompt, which tells the agent what to report.
+    """
 
     tool: Annotated[str, Field(min_length=1)]
     tool_version: Annotated[str, Field(min_length=1)]
@@ -103,7 +112,15 @@ def effective_output_tok_per_s(self) -> float:
 
     @property
     def effective_per_user_tok_per_s(self) -> float:
-        return 1000 / self.metrics.tpot_ms.mean
+        e2e_ms = self.metrics.e2e_ms
+        if e2e_ms is None:
+            # Compatibility with presets stored before end-to-end latency was
+            # recorded. New agent reports always provide `e2e_ms`.
+            assert self.metrics.tpot_ms is not None
+            return 1000 / self.metrics.tpot_ms.mean
+        mean_output_tokens = self.metrics.total_output_tokens / self.metrics.successful_requests
+        mean_tpot_ms = (e2e_ms.mean - self.metrics.ttft_ms.mean) / (mean_output_tokens - 1)
+        return 1000 / mean_tpot_ms
 
     @field_validator("command")
     @classmethod
@@ -125,6 +142,14 @@ def validate_metrics(self) -> Self:
             raise ValueError("benchmark must not include failed requests")
         if self.metrics.successful_requests != self.workload.num_requests:
             raise ValueError("benchmark request count must match workload.num_requests")
+        if self.metrics.e2e_ms is None:
+            if self.metrics.tpot_ms is None:
+                raise ValueError("legacy benchmark without e2e_ms must include tpot_ms")
+            return self
+        if self.metrics.total_output_tokens <= self.metrics.successful_requests:
+            raise ValueError("benchmark must average more than one output token per request")
+        if self.metrics.e2e_ms.mean <= self.metrics.ttft_ms.mean:
+            raise ValueError("benchmark mean e2e_ms must be greater than mean ttft_ms")
         return self
 
 

src/tests/_internal/cli/common.py

@@ -114,6 +114,7 @@ def get_preset_benchmark() -> PresetBenchmark:
             "output_tok_per_s": 42.1,
             "per_user_tok_per_s": 42.1,
             "ttft_ms": {"mean": 110.9, "p50": 108.2, "p99": 121.6},
+            "e2e_ms": {"mean": 1063.4, "p50": 1048.0, "p99": 1150.3},
             "tpot_ms": {"mean": 7.5, "p50": 7.4, "p99": 8.1},
         },
     )
@@ -254,5 +255,5 @@ def get_successful_preset_report(run: Run) -> PresetAgentSuccess:
         base="Qwen/Qwen3.5-27B",
         model="community/Qwen3.5-27B-GPTQ-Int4",
         context_length=32768,
-        benchmark=get_preset_benchmark(),
+        benchmark=get_preset_benchmark().model_dump(),
     )

src/tests/_internal/cli/services/presets/test_agent.py

@@ -1096,13 +1096,33 @@ def test_counts_records_even_when_trials_share_a_task(self, tmp_path):
         assert summary["count"] == 4
         assert summary["best"] == {
             "tok_s": 2300.0,
-            "tpot_ms": None,
+            "per_user_tok_s": None,
             "ttft_ms": None,
             "context_length": None,
             "concurrency": 8,
             "gpu": "A40:48GB:1",
         }
 
+    def test_derives_per_user_rate_from_e2e_instead_of_reported_tpot(self, tmp_path):
+        record = {
+            "benchmark": {
+                "metrics": {
+                    "successful_requests": 4,
+                    "duration_seconds": 10.0,
+                    "total_output_tokens": 512,
+                    "ttft_ms": {"mean": 100.0},
+                    "e2e_ms": {"mean": 1370.0},
+                    # Chunk-level ITL from speculative decoding, mislabeled as TPOT.
+                    "tpot_ms": {"mean": 137.18},
+                },
+                "workload": {"concurrency": 4},
+            }
+        }
+
+        summary = _summarize_session_trials(_write_trials(tmp_path, [record]))
+
+        assert summary["best"]["per_user_tok_s"] == pytest.approx(100.0)
+
     def test_trials_are_ordered_numerically_beyond_nine(self, tmp_path):
         # Twelve trials: a string sort would chart 1, 10, 11, 12, 2, ...
         records = [

src/tests/_internal/cli/services/presets/test_create.py

@@ -52,6 +52,7 @@
 from dstack._internal.core.errors import CLIError
 from dstack._internal.core.models.configurations import PresetConfiguration
 from dstack._internal.core.models.envs import EnvSentinel
+from dstack._internal.core.models.presets import PresetBenchmarkLatency
 from dstack._internal.core.models.runs import Run, RunStatus
 from tests._internal.cli.common import (
     get_preset,
@@ -730,11 +731,22 @@ def test_throughput_is_derived_not_read(self):
         benchmark = preset.benchmark
         benchmark.metrics.output_tok_per_s = 999999.0
         benchmark.metrics.per_user_tok_per_s = 999999.0
+        benchmark.metrics.tpot_ms = PresetBenchmarkLatency(
+            mean=999999.0, p50=999999.0, p99=999999.0
+        )
 
         expected = benchmark.metrics.total_output_tokens / benchmark.metrics.duration_seconds
         assert benchmark.effective_output_tok_per_s == expected
-        # Per-user speed is the steady decode rate, not the aggregate over concurrency.
-        assert benchmark.effective_per_user_tok_per_s == 1000 / benchmark.metrics.tpot_ms.mean
+        # Per-user speed is the steady decode rate derived from the E2E and TTFT
+        # means, not a self-reported rate or chunk-level ITL mislabeled as TPOT.
+        assert benchmark.metrics.e2e_ms is not None
+        mean_output_tokens = (
+            benchmark.metrics.total_output_tokens / benchmark.metrics.successful_requests
+        )
+        mean_tpot_ms = (benchmark.metrics.e2e_ms.mean - benchmark.metrics.ttft_ms.mean) / (
+            mean_output_tokens - 1
+        )
+        assert benchmark.effective_per_user_tok_per_s == 1000 / mean_tpot_ms
 
 
 class TestBuildConstraints:

src/tests/_internal/cli/services/presets/test_output.py

@@ -32,10 +32,17 @@ def test_formats_second_scale_ttft_without_scientific_notation(self):
         ttft.mean = 8148.3
         ttft.p50 = 8151.4
         ttft.p99 = 8334.2
+        e2e = preset.benchmark.metrics.e2e_ms
+        tpot = preset.benchmark.metrics.tpot_ms
+        assert e2e is not None and tpot is not None
+        output_token_intervals = preset.benchmark.workload.output_tokens - 1
+        e2e.mean = ttft.mean + output_token_intervals * tpot.mean
+        e2e.p50 = ttft.p50 + output_token_intervals * tpot.p50
+        e2e.p99 = ttft.p99 + output_token_intervals * tpot.p99
 
         output = output_module.format_preset_benchmark(preset, verbose=True)
 
-        # Per-user speed is 1/TPOT (p50 7.4ms), not the aggregate over concurrency.
+        # Per-user speed is the derived mean decode rate, not aggregate throughput.
         assert output.startswith("tps/user=133 ")
         assert output_module.format_preset_objective(preset).startswith("[secondary]io=1K/128 ")
         assert "ctx=32K" in output
@@ -171,7 +178,7 @@ def test_shows_progress_after_status_and_best_benchmark(self):
                     "count": 2,
                     "best": {
                         "tok_s": 2339.0,
-                        "tpot_ms": 3.42,
+                        "per_user_tok_s": 292.0,
                         "concurrency": 8,
                         "gpu": "A40:48GB:1",
                     },
@@ -181,8 +188,8 @@ def test_shows_progress_after_status_and_best_benchmark(self):
 
         # 2 completed, so the 3rd is the one being trialed.
         assert row["STATUS"] == "[bold sea_green3]trialing[/] [secondary](3/3)[/]"
-        # Per-user speed is 1/TPOT, the same definition the preset row uses — not
-        # the aggregate over concurrency, which would read 292 here.
+        # Per-user speed comes from the trial's E2E/TTFT-derived decode rate, not
+        # aggregate throughput.
         assert row["BENCHMARK"].startswith("tps/user=292")
         assert row["RESOURCES"] == "A40:48GB:1"
 
@@ -441,8 +448,8 @@ def test_the_fastest_failed_trial_is_kept_when_nothing_passed(self, tmp_path):
 
         assert summary["best"] is None
         assert summary["best_failed"]["tok_s"] == 300.0
-        # Mean, not p50: MTP skews TPOT right, and mean is the delivered rate.
-        assert summary["best_failed"]["tpot_ms"] == 41.7
+        # Older sessions without E2E preserve their reported mean TPOT display.
+        assert summary["best_failed"]["per_user_tok_s"] == pytest.approx(1000 / 41.7)
 
     def test_a_run_that_met_nothing_still_shows_what_it_measured(self):
         row = _session_row(
@@ -457,7 +464,7 @@ def test_a_run_that_met_nothing_still_shows_what_it_measured(self):
                     "best": None,
                     "best_failed": {
                         "tok_s": 100.0,
-                        "tpot_ms": 34.4,
+                        "per_user_tok_s": 1000 / 34.4,
                         "ttft_ms": 4300.0,
                         "context_length": 1048576,
                         "concurrency": 4,

src/tests/_internal/cli/services/presets/test_verify.py

@@ -349,6 +349,28 @@ def test_redacts_known_secret_in_benchmark_command_instead_of_failing(self, tmp_
         assert report.benchmark.command.endswith("Bearer [redacted]'")
         assert "sk-live" not in report.benchmark.command
 
+    def test_requires_e2e_latency_in_a_fresh_report(self, tmp_path):
+        run = get_running_service_run()
+        data = get_successful_preset_report(run).model_dump()
+        data["run_id"] = str(data["run_id"])
+        data["benchmark"]["metrics"].pop("e2e_ms")
+
+        with pytest.raises(CLIError, match="e2e_ms"):
+            self._load(tmp_path, data, redacted_values=())
+
+    def test_rejects_a_reported_tpot_that_exceeds_available_slot_time(self, tmp_path):
+        run = get_running_service_run()
+        data = get_successful_preset_report(run).model_dump()
+        data["run_id"] = str(data["run_id"])
+        data["benchmark"]["metrics"]["tpot_ms"] = {
+            "mean": 137.18,
+            "p50": 55.34,
+            "p99": 1160.43,
+        }
+
+        with pytest.raises(CLIError, match="exceeds the decode time available"):
+            self._load(tmp_path, data, redacted_values=())
+
     def test_still_rejects_unknown_bearer_token(self, tmp_path):
         run = get_running_service_run()
         data = get_successful_preset_report(run).model_dump()