fix: structural issue batch — bounded reads, oracle coherence, protocol comparison, vLLM percentiles

.github/workflows/release.yml

@@ -132,7 +132,7 @@ jobs:
         run: |
           mkdir -p dist/python dist/npm
           uv build --out-dir dist/python
-          uvx --from twine twine check dist/python/*
+          uv run --locked twine check dist/python/*
           npm pack ./npm --pack-destination "$PWD/dist/npm" --json > dist/npm-pack.json
           sha256sum dist/python/* dist/npm/*.tgz > dist/SHA256SUMS
 

pyproject.toml

@@ -74,6 +74,7 @@ dev = [
     "pyarrow-stubs>=20.0.0.20260625",
     "ruff>=0.11",
     "scipy-stubs>=1.18.0.1,<1.19",
+    "twine>=6.1",
     "vulture>=2.14",
 ]
 

src/flameox/adapters/inference.py

@@ -821,6 +821,17 @@ class AIPerfCorrelationSummary(ContractModel):
 # vLLM aggregate benchmark-result JSON normalization
 # ---------------------------------------------------------------------------
 #
+def _percentile_label(percentile: int | float) -> str:
+    """Return a canonical label for a percentile rank, preserving fractional precision.
+
+    ``int()`` truncation made p99.1 and p99.9 both produce ``p99``, losing
+    the exact percentile identity in the metric name.
+    """
+    if float(percentile).is_integer():
+        return str(int(percentile))
+    return str(float(percentile))
+
+
 # vLLM's ``benchmark_serving.BenchmarkMetrics`` dataclass is serialized by the
 # Mooncake replayer (and other vLLM benchmark scripts) as a JSON object whose
 # percentile fields are lists of ``[percentile, value_ms]`` pairs. The parser
@@ -907,6 +918,10 @@ def non_negative_latency(self) -> VllmAggregateMetrics:
             "median_itl_ms",
             "mean_e2el_ms",
             "median_e2el_ms",
+            "std_ttft_ms",
+            "std_tpot_ms",
+            "std_itl_ms",
+            "std_e2el_ms",
         ):
             if getattr(self, name) < 0:
                 raise ValueError(f"{name} must be non-negative")
@@ -945,6 +960,13 @@ class VllmResultDocument(ContractModel):
     actual_duration: Annotated[float, Field(ge=0)]
     time_scale: Annotated[float, Field(gt=0)] = 1.0
 
+    @field_validator("actual_duration", "time_scale")
+    @classmethod
+    def finite_duration_and_scale(cls, value: float) -> float:
+        if not math.isfinite(value):
+            raise ValueError("duration and time_scale must be finite")
+        return value
+
     @model_validator(mode="after")
     def totals_match(self) -> VllmResultDocument:
         if self.successful_requests + self.failed_requests != self.total_requests:
@@ -1038,7 +1060,7 @@ def _normalize_document(payload: dict[str, Any]) -> dict[str, Any]:
             "successful_requests": completed,
             "failed_requests": max(0, total_requests - completed),
             "total_requests": total_requests,
-            "actual_duration": payload.get("duration", 0.0),
+            "actual_duration": payload.get("duration"),
             "time_scale": 1.0,
         }
 
@@ -1163,7 +1185,7 @@ def add(
             add(f"vllm.{label}.std_ms", std, unit="ms", aggregation="std", extra={"stat": "std"})
             for percentile, value in getattr(metrics, f"percentiles_{metric}_ms"):
                 add(
-                    f"vllm.{label}.p{int(percentile)}_ms",
+                    f"vllm.{label}.p{_percentile_label(percentile)}_ms",
                     float(value),
                     unit="ms",
                     aggregation="percentile",

src/flameox/analysis/inference_protocol.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 from collections.abc import Callable
 from enum import StrEnum
 from typing import Annotated, Literal
@@ -405,9 +406,14 @@ def _facets() -> tuple[_FacetGetter, ...]:
 
 
 def _normalize(value: object) -> str:
+    """Return a display string for a protocol facet value.
+
+    For dict values, use canonical JSON serialization (sorted keys) so that
+    keys/values containing commas or equals signs cannot produce colliding
+    normal forms. Plain scalar values use ``str()`` as before.
+    """
     if isinstance(value, dict):
-        items = sorted(value.items(), key=lambda item: item[0])
-        return ",".join(f"{k}={v}" for k, v in items)
+        return json.dumps(value, sort_keys=True, ensure_ascii=False)
     if isinstance(value, bool):
         return "true" if value else "false"
     if value is None:
@@ -429,6 +435,8 @@ def _identity_missing_fields(protocol: InferenceProtocolIdentity) -> dict[str, s
         missing["model.tokenizer_id"] = "tokenizer identity is unavailable"
     if protocol.model.tokenizer_revision is None:
         missing["model.tokenizer_revision"] = "tokenizer revision is unavailable"
+    if protocol.model.quantization is None:
+        missing["model.quantization"] = "effective model quantization is unavailable"
     if protocol.server.managed_server_command_digest is None:
         missing["server.managed_server_command_digest"] = (
             "managed server and cache configuration provenance is unavailable"

src/flameox/analysis/recipe_models.py

@@ -443,6 +443,7 @@ class ScalingPoint(ConfidenceIntervalFields):
     variant: str
     block_id: str | None
     input_value: float | None
+    input_kind: Literal["integer", "floating"] | None = None
     value: float
     dispersion: float
     unit: str
@@ -456,6 +457,7 @@ class ScalingTrialSummary(ContractModel):
     variant: str
     block_id: str | None
     input_value: float | None
+    input_kind: Literal["integer", "floating"] | None = None
     median: float
     dispersion: float
     unit: str

src/flameox/analysis/recipe_scaling.py

@@ -2,7 +2,7 @@
 
 import math
 import warnings
-from typing import Any, cast
+from typing import Any, Literal, cast
 
 import numpy as np
 from scipy.stats import bootstrap, spearmanr
@@ -29,6 +29,17 @@
 
 
 class ScalingRecipes(RecipeContext):
+    @staticmethod
+    def _input_identity(
+        integer_value: object,
+        floating_value: object,
+    ) -> tuple[float | None, Literal["integer", "floating"] | None]:
+        if integer_value is not None:
+            return float(cast(Any, integer_value)), "integer"
+        if floating_value is not None:
+            return float(cast(Any, floating_value)), "floating"
+        return None, None
+
     def scaling(
         self,
         experiment_id: str,
@@ -106,22 +117,25 @@ def scaling(
             ).fetchone()
             assert complete_row is not None
         trial_groups: dict[
-            tuple[str, str, str | None, float | None, str, str],
+            tuple[
+                str,
+                str,
+                str | None,
+                float | None,
+                Literal["integer", "floating"] | None,
+                str,
+                str,
+            ],
             list[float],
         ] = {}
         for row in rows:
-            input_value = (
-                float(row[3])
-                if row[3] is not None
-                else float(row[4])
-                if row[4] is not None
-                else None
-            )
+            input_value, input_kind = self._input_identity(row[3], row[4])
             key = (
                 str(row[0]),
                 str(row[1]),
                 str(row[2]) if row[2] is not None else None,
                 input_value,
+                input_kind,
                 str(row[7]),
                 str(row[5]),
             )
@@ -133,6 +147,7 @@ def scaling(
             variant,
             block_id,
             input_value,
+            input_kind,
             unit,
             environment_id,
         ), values in sorted(trial_groups.items()):
@@ -143,6 +158,7 @@ def scaling(
                     variant=variant,
                     block_id=block_id,
                     input_value=input_value,
+                    input_kind=input_kind,
                     median=median,
                     dispersion=float(np.median(np.abs(np.asarray(values) - median))),
                     unit=unit,
@@ -151,18 +167,18 @@ def scaling(
                 )
             )
         point_groups: dict[
-            tuple[str, float | None, str],
+            tuple[str, float | None, Literal["integer", "floating"] | None, str],
             list[ScalingTrialSummary],
         ] = {}
         for trial in trials:
             point_groups.setdefault(
-                (trial.variant, trial.input_value, trial.unit),
+                (trial.variant, trial.input_value, trial.input_kind, trial.unit),
                 [],
             ).append(trial)
         points_list: list[ScalingPoint] = []
-        for (variant, input_value, unit), group in sorted(
+        for (variant, input_value, input_kind, unit), group in sorted(
             point_groups.items(),
-            key=lambda item: (item[0][0], item[0][1] or -math.inf),
+            key=lambda item: (item[0][0], item[0][1] or -math.inf, item[0][2] or ""),
         ):
             trial_medians = np.asarray(
                 [trial.median for trial in group],
@@ -180,6 +196,7 @@ def scaling(
                     variant=variant,
                     block_id=next(iter(block_ids)) if len(block_ids) == 1 else None,
                     input_value=input_value,
+                    input_kind=input_kind,
                     value=median,
                     dispersion=dispersion,
                     confidence_interval=(
@@ -218,6 +235,27 @@ def scaling(
             warnings.append(
                 "Some variants have no numeric input value and were excluded from fits."
             )
+        input_kinds_by_variant: dict[str, set[Literal["integer", "floating"]]] = {}
+        for point in points:
+            if (
+                point.input_value is not None
+                and point.input_value > 0
+                and math.isfinite(point.input_value)
+                and math.isfinite(point.value)
+                and point.input_kind is not None
+            ):
+                input_kinds_by_variant.setdefault(point.variant, set()).add(point.input_kind)
+        mixed_input_kind_variants = {
+            variant
+            for variant, input_kinds in input_kinds_by_variant.items()
+            if input_kinds == {"integer", "floating"}
+        }
+        if mixed_input_kind_variants:
+            warnings.append(
+                "Fits were excluded for variants with mixed integer and floating scaling inputs: "
+                + ", ".join(sorted(mixed_input_kind_variants))
+                + "."
+            )
         environment_stable = all(point.environment_count == 1 for point in points)
         if not environment_stable:
             warnings.append("Environment identity varies within at least one scaling point.")
@@ -294,6 +332,8 @@ def _scaling_fits(
             ]
             if len(numeric) < 3 or len({point.input_value for point in numeric}) < 2:
                 continue
+            if {point.input_kind for point in numeric} == {"integer", "floating"}:
+                continue
             x = np.asarray([point.input_value for point in numeric], dtype=float)
             y = np.asarray([point.value for point in numeric], dtype=float)
             candidates = {
@@ -403,20 +443,16 @@ def _correlated_hotspots(
                 int | None,
                 str,
                 str,
+                Literal["integer", "floating"],
                 float,
             ],
             float,
         ] = {}
         for row in rows:
-            input_value = (
-                float(cast(Any, row[2]))
-                if row[2] is not None
-                else float(cast(Any, row[3]))
-                if row[3] is not None
-                else None
-            )
+            input_value, input_kind = self._input_identity(row[2], row[3])
             if input_value is None or not math.isfinite(input_value):
                 continue
+            assert input_kind is not None
             key = (
                 str(row[0]),
                 str(row[1]),
@@ -426,6 +462,7 @@ def _correlated_hotspots(
                 int(cast(Any, row[7])) if row[7] is not None else None,
                 str(row[8]),
                 str(row[9]),
+                input_kind,
                 input_value,
             )
             per_trial[key] = per_trial.get(key, 0.0) + float(cast(Any, row[10]))
@@ -442,6 +479,7 @@ def _correlated_hotspots(
             line,
             metric,
             unit,
+            _input_kind,
             input_value,
         ), value in per_trial.items():
             groups.setdefault(

src/flameox/application/__init__.py

@@ -363,6 +363,11 @@
     "parse_inference_tool_discovery",
     "probe_existing_vllm_server",
     "render_evidence_summary_markdown",
+    "scalar_contains",
+    "scalar_equal",
+    "scalar_identity",
+    "scalar_identity_set",
+    "scalar_subset",
     "workspace_status",
 ]