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

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:

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",
 ]
 

src/flameox/application/experiments.py

@@ -35,6 +35,10 @@
     _FactorExperimentConfig,
     _OutcomeExperimentConfig,
     _ScaledLegacyExperimentConfig,
+    scalar_contains,
+    scalar_equal,
+    scalar_identity_set,
+    scalar_subset,
 )
 from flameox.catalog import Catalog
 from flameox.domain import (
@@ -139,6 +143,7 @@ class ExperimentPlan(ContractModel):
     execution_policy: ExecutionPolicy
     variant_parameter: str
     variants: tuple[str, ...]
+    baseline_variant: str | None = None
     factors: dict[str, tuple[JsonValue, ...]] = Field(default_factory=dict)
     parameter_overrides: dict[str, JsonValue]
     blocks: tuple[ExperimentBlock, ...]
@@ -660,6 +665,11 @@ async def plan(
             execution_policy=execution_policy,
             variant_parameter=variant_parameter,
             variants=variants,
+            baseline_variant=(
+                self._factor_label(config.baseline_value)
+                if isinstance(config, _FactorExperimentConfig) and config.baseline_value is not None
+                else None
+            ),
             factors={
                 name: tuple(cast(JsonValue, value) for value in values)
                 for name, values in factors.items()
@@ -903,12 +913,26 @@ async def report(message: str) -> None:
                 "Automatic experiment comparison currently requires pyperf measurements."
             )
         else:
+            comparison_run_sets: tuple[RunSet, ...] = run_sets
+            if not plan.baseline_variant:
+                limitations.append(
+                    "Baseline was determined by list position, not an explicit "
+                    "baseline_value. Reordering the treatment list reverses the "
+                    "comparison direction."
+                )
+            else:
+                comparison_run_sets = tuple(
+                    sorted(
+                        run_sets,
+                        key=lambda run_set: run_set.selection["variant"] != plan.baseline_variant,
+                    )
+                )
             comparison = await run_atomic_thread(
                 lambda: ComparisonService(self.workspace).record(
                     parse_compare_run_sets_request(
                         {
-                            "baseline_run_set_id": run_sets[0].run_set_id,
-                            "candidate_run_set_id": run_sets[1].run_set_id,
+                            "baseline_run_set_id": comparison_run_sets[0].run_set_id,
+                            "candidate_run_set_id": comparison_run_sets[1].run_set_id,
                             "experiment_id": plan.experiment.experiment_id,
                             "metric": plan.experiment.primary_metric,
                             "unit": (
@@ -988,7 +1012,7 @@ def _materialize_combinations(
             matches = [
                 name
                 for name, choices in workload_parameters.items()
-                if set(config.variants).issubset(set(choices))
+                if scalar_subset(list(config.variants), list(choices))
             ]
             if not matches:
                 raise DomainError(
@@ -1018,7 +1042,9 @@ def _materialize_combinations(
                     ErrorCode.WORKSPACE_INVALID,
                     f"Experiment factor {name!r} is not a workload parameter.",
                 )
-            if len(set(values)) != len(values) or not set(values).issubset(set(allowed)):
+            if len(scalar_identity_set(list(values))) != len(values) or not scalar_subset(
+                list(values), list(allowed)
+            ):
                 raise DomainError(
                     ErrorCode.WORKSPACE_INVALID,
                     f"Experiment factor {name!r} contains duplicate or undeclared values.",
@@ -1040,7 +1066,7 @@ def _materialize_combinations(
                     ErrorCode.WORKSPACE_INVALID,
                     "Explicit combinations must contain every declared factor exactly once.",
                 )
-            if any(combination[name] not in factors[name] for name in factor_names):
+            if any(not scalar_contains(combination[name], factors[name]) for name in factor_names):
                 raise DomainError(
                     ErrorCode.WORKSPACE_INVALID,
                     "Explicit combination contains an undeclared factor value.",
@@ -1060,7 +1086,7 @@ def _materialize_combinations(
                     ErrorCode.WORKSPACE_INVALID,
                     "Every exclusion must name at least one declared factor.",
                 )
-            if any(value not in factors[name] for name, value in rule.items()):
+            if any(not scalar_contains(value, factors[name]) for name, value in rule.items()):
                 raise DomainError(
                     ErrorCode.WORKSPACE_INVALID,
                     "Exclusion contains an undeclared factor value.",
@@ -1069,7 +1095,7 @@ def _materialize_combinations(
             combination
             for combination in combinations
             if not any(
-                all(combination[name] == value for name, value in rule.items())
+                all(scalar_equal(combination[name], value) for name, value in rule.items())
                 for rule in config.exclude
             )
         )
@@ -1102,7 +1128,10 @@ def _outcome_result(
             selected = [
                 trial
                 for trial in trials
-                if trial.factors.get(plan.variant_parameter) == treatment_value
+                if scalar_equal(
+                    cast(Scalar, trial.factors.get(plan.variant_parameter)),
+                    cast(Scalar, treatment_value),
+                )
             ]
             attempted = sum(trial.outcome is not TrialOutcome.UNATTEMPTED for trial in selected)
             eligible = sum(
@@ -1211,9 +1240,13 @@ def _outcome_result(
             disposition = ExperimentOutcomeDisposition.INSUFFICIENT_EVIDENCE
         elif not failures:
             disposition = ExperimentOutcomeDisposition.ALL_CLEAN
-        elif len(plan.variants) == 2 and failed_treatments == {plan.variants[0]}:
+        elif len(plan.variants) == 2 and failed_treatments == {
+            plan.baseline_variant or plan.variants[0]
+        }:
             disposition = ExperimentOutcomeDisposition.BASE_ONLY_FAILURE
-        elif len(plan.variants) == 2 and failed_treatments == {plan.variants[1]}:
+        elif len(plan.variants) == 2 and failed_treatments == {
+            v for v in plan.variants if v != (plan.baseline_variant or plan.variants[0])
+        }:
             disposition = ExperimentOutcomeDisposition.CANDIDATE_ONLY_FAILURE
         else:
             disposition = ExperimentOutcomeDisposition.MIXED

src/flameox/application/faults.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import json
+import random
 import shutil
 import socket
 from collections.abc import Awaitable, Callable
@@ -250,6 +251,9 @@ async def plan(
         )
         blocks: list[ExperimentBlock] = []
         for block_number in range(1, config.blocks * config.repetitions + 1):
+            cell_treatments = list(treatments)
+            generator = random.Random(f"{config.random_seed}:{block_number}")
+            generator.shuffle(cell_treatments)
             cells = tuple(
                 ExperimentCell(
                     trial_id=digest_model(
@@ -260,12 +264,12 @@ async def plan(
                     factors={"scenario": treatment},
                     parameters={},
                 )
-                for treatment in treatments
+                for treatment in cell_treatments
             )
             blocks.append(
                 ExperimentBlock(
                     block_id=f"fault-block-{block_number:04d}",
-                    order=treatments,
+                    order=tuple(cell_treatments),
                     cells=cells,
                 )
             )

src/flameox/application/inference.py

@@ -1556,7 +1556,7 @@ def _protocol_identity(
                 model_revision=plan.model_revision,
                 tokenizer_id=plan.tokenizer or plan.model,
                 tokenizer_revision=plan.tokenizer_revision or plan.model_revision,
-                quantization=plan.quantization or "none",
+                quantization=plan.quantization,
             ),
             server=ServerConfigIdentity(
                 backend=plan.server_provider.value,

src/flameox/application/summaries.py

@@ -4,7 +4,7 @@
 import json
 import re
 from enum import StrEnum
-from typing import Annotated, Literal, cast
+from typing import Annotated, Literal, TextIO, cast
 
 from pydantic import Field, JsonValue, model_validator
 
@@ -28,6 +28,23 @@
 from flameox.models import ContractModel
 from flameox.storage import ArtifactStore, JsonRecordStore, RunStore, Workspace
 
+_DISCARD_CHUNK_SIZE = 4096
+
+
+def _discard_rest_of_line(stream: TextIO) -> None:
+    """Discard the remainder of an overlong line without materializing it.
+
+    Instead of calling unbounded ``stream.readline()`` which loads the
+    entire remaining line into memory, read fixed-size chunks until a
+    newline or EOF is encountered.  Peak memory is bounded by
+    ``_DISCARD_CHUNK_SIZE``, not by the physical line length.
+    """
+    while True:
+        chunk = stream.readline(_DISCARD_CHUNK_SIZE)
+        if not chunk or chunk.endswith("\n"):
+            break
+
+
 _ANSI_ESCAPE = re.compile(r"\x1b(?:\[[0-?]*[ -/]*[@-~]|\][^\x07]*(?:\x07|\x1b\\))")
 _CONTROL = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]")
 
@@ -566,7 +583,7 @@ def _excerpt(self, artifact_id: str) -> tuple[tuple[str, ...], bool]:
                     break
                 if len(line) == 201 and not line.endswith("\n"):
                     truncated = True
-                    stream.readline()
+                    _discard_rest_of_line(stream)
                 selected.append(_CONTROL.sub("", _ANSI_ESCAPE.sub("", line.rstrip("\r\n"))))
             if stream.readline(1):
                 truncated = True