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 (
@@ -988,7 +992,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 +1022,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 +1046,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 +1066,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 +1075,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 +1108,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(

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

src/flameox/application/workloads.py

@@ -62,6 +62,50 @@
 
 Scalar = str | int | float | bool
 
+
+def scalar_identity(value: Scalar) -> tuple[str, object]:
+    """Return a typed identity key distinguishing bool/int/float/str by exact JSON type.
+
+    Python's numeric tower treats ``True == 1`` and ``1 == 1.0`` as equal, and
+    ``hash(True) == hash(1)``. Configuration and evidence protocols must not
+    treat those as the same scalar. This helper returns a ``(type_tag, value)``
+    tuple where the type tag distinguishes the four scalar JSON kinds so that
+    ``scalar_identity(True) != scalar_identity(1)`` and
+    ``scalar_identity(1) != scalar_identity(1.0)``.
+    """
+    if type(value) is bool:
+        return ("bool", value)
+    if type(value) is int:
+        return ("int", value)
+    if type(value) is float:
+        return ("float", value)
+    return ("string", value)
+
+
+def scalar_equal(left: Scalar, right: Scalar) -> bool:
+    """Return True only when both scalars share the exact JSON type and value."""
+    return scalar_identity(left) == scalar_identity(right)
+
+
+def scalar_contains(value: Scalar, choices: tuple[Scalar, ...] | list[Scalar]) -> bool:
+    """Return True only when ``value`` is present in ``choices`` by exact scalar identity."""
+    identity = scalar_identity(value)
+    return any(scalar_identity(choice) == identity for choice in choices)
+
+
+def scalar_identity_set(
+    values: tuple[Scalar, ...] | list[Scalar],
+) -> set[tuple[str, object]]:
+    """Return the set of scalar identities for ``values`` without Python-equality collisions."""
+    return {scalar_identity(value) for value in values}
+
+
+def scalar_subset(subset_values: list[Scalar], superset_values: list[Scalar]) -> bool:
+    """Return True only when every identity in ``subset_values`` is in ``superset_values``."""
+    superset = scalar_identity_set(superset_values)
+    return all(identity in superset for identity in (scalar_identity(v) for v in subset_values))
+
+
 RUNTIME_RESOURCE_METRICS = frozenset(
     {
         "runtime_resource.peak_rss_bytes",
@@ -276,7 +320,7 @@ class _ScaledLegacyExperimentConfig(_LegacyExperimentConfig):
     @field_validator("scaling_values")
     @classmethod
     def scaling_values_are_unique(cls, value: tuple[Scalar, ...]) -> tuple[Scalar, ...]:
-        if len(set(value)) != len(value):
+        if len(scalar_identity_set(list(value))) != len(value):
             raise ValueError("experiment scaling values must be unique")
         return value
 
@@ -1887,7 +1931,7 @@ def _parameters(
                     f"Dynamic workload parameter {name!r} must be supplied.",
                 )
             value = overrides.get(name, choices[0])
-            if name not in dynamic_parameters and value not in choices:
+            if name not in dynamic_parameters and not scalar_contains(value, choices):
                 raise DomainError(
                     ErrorCode.INVALID_CAPTURE_PLAN,
                     f"Value for {name!r} is outside the declared choices.",

tests/adapters/test_inference_artifacts.py

@@ -1307,6 +1307,16 @@ def test_vllm_parse_rejects_boolean_native_request_counts() -> None:
     assert error.value.code is ErrorCode.ARTIFACT_PARSE_FAILED
 
 
+def test_vllm_parse_rejects_native_result_without_duration() -> None:
+    metrics = _vllm_metrics()
+    metrics["num_prompts"] = 313
+
+    with pytest.raises(DomainError) as error:
+        VllmResultParser().parse_payload(metrics)
+
+    assert error.value.code is ErrorCode.ARTIFACT_PARSE_FAILED
+
+
 def test_mooncake_extractor_publishes_bounded_request_evidence(tmp_path: Path) -> None:
     workspace = Workspace.initialize(tmp_path)
     trace = tmp_path / "trace.jsonl"

tests/analysis/test_inference_protocol.py

@@ -366,3 +366,23 @@ def test_server_rejects_out_of_range_gpu_utilization() -> None:
 def test_oracle_result_rejects_empty_reason() -> None:
     with pytest.raises(ValueError):
         OracleResult(status=OracleStatus.PASS, reason="")
+
+
+def test_kv_transfer_config_with_delimiter_chars_does_not_collide() -> None:
+    """Distinct KV-transfer configs with commas/equals in keys/values must not collide.
+
+    Regression for #288: the old ``_normalize()`` used unescaped
+    ``",".join(f"{k}={v}")`` which could make distinct dicts produce the
+    same normalized string. The fix uses canonical JSON serialization.
+    """
+    from flameox.analysis.inference_protocol import _normalize
+
+    config_a = {"a=b,c": "d"}
+    config_b = {"a": "b=c,d"}
+
+    norm_a = _normalize(config_a)
+    norm_b = _normalize(config_b)
+    assert norm_a != norm_b, (
+        f"Distinct configs must not collide: {config_a!r} vs {config_b!r} "
+        f"both normalized to {norm_a!r}"
+    )