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["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
@@ -404,10 +405,17 @@ 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/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/domain/models.py

@@ -312,6 +312,35 @@ def bounded_coordinate(cls, value: tuple[str | int, ...]) -> tuple[str | int, ..
         return value
 
 
+    @model_validator(mode="after")
+    def status_must_be_coherent_with_evidence(self) -> OracleReceiptV1:
+        """Reject receipts where categorical status contradicts quantitative evidence.
+
+        When both ``absolute_error`` and ``tolerance.absolute`` are present,
+        a ``status='pass'`` receipt with error exceeding tolerance is rejected,
+        and a ``status='fail'`` receipt with error within tolerance is rejected.
+
+        Receipts without quantitative evidence (no error/tolerance) are
+        accepted with any status since there is nothing to contradict.
+        """
+        if self.absolute_error is not None and self.tolerance is not None:
+            abs_tol = self.tolerance.absolute
+            if abs_tol is not None:
+                if self.status == "pass" and self.absolute_error > abs_tol:
+                    raise ValueError(
+                        f"Receipt status='pass' contradicts evidence: "
+                        f"absolute_error={self.absolute_error} exceeds "
+                        f"tolerance.absolute={abs_tol}"
+                    )
+                if self.status == "fail" and self.absolute_error <= abs_tol:
+                    raise ValueError(
+                        f"Receipt status='fail' contradicts evidence: "
+                        f"absolute_error={self.absolute_error} is within "
+                        f"tolerance.absolute={abs_tol}"
+                    )
+        return self
+
+
 class OracleReceiptRecord(ContractModel):
     receipt: OracleReceiptV1
     receipt_artifact_id: Digest

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}"
+    )

tests/application/test_summaries.py

@@ -182,3 +182,26 @@ def test_trial_failure_class_fallback_for_null_column_value() -> None:
 
     fallback = TrialFailureClass(str(None) if None is not None else TrialFailureClass.NONE)
     assert fallback is TrialFailureClass.NONE
+
+
+def test_excerpt_discards_overlong_line_in_bounded_chunks() -> None:
+    """Overlong lines must not be materialized during excerpt extraction.
+
+    Regression for #291: ``_excerpt()`` called unbounded ``stream.readline()``
+    to discard the remainder of a line longer than 200 characters, materializing
+    a multi-gigabyte string in memory. The fix uses bounded ``_DISCARD_CHUNK_SIZE``
+    chunks instead.
+    """
+    import io
+
+    from flameox.application.summaries import _discard_rest_of_line
+
+    long_line = "X" * 10_000
+    content = f"{long_line}\nshort line\n"
+    stream = io.StringIO(content)
+    line = stream.readline(201)
+    assert len(line) == 201
+    assert not line.endswith("\n")
+    _discard_rest_of_line(stream)
+    next_line = stream.readline()
+    assert next_line == "short line\n"

tests/domain/test_models.py

@@ -38,6 +38,7 @@
     ExecutionRunManifest,
     ImportRunManifest,
     Integrity,
+    OracleStatus,
     ScalarOracleReceiptValue,
     SucceededTrial,
     parse_capture_plan,
@@ -574,3 +575,59 @@ def test_managed_runtime_extra_parser_owns_the_persisted_vocabulary() -> None:
         CapabilityExtra.TORCH,
         CapabilityExtra.TRACE,
     )
+
+
+
+
+def test_oracle_receipt_rejects_pass_with_error_exceeding_tolerance() -> None:
+    """Cross-field validation: status='pass' with error > tolerance must be rejected."""
+    from flameox.domain.models import OracleReceiptV1, OracleTolerance
+
+    with pytest.raises(ValidationError, match="contradicts evidence"):
+        OracleReceiptV1(
+            schema_version="flameox.oracle-receipt.v1",
+            status=OracleStatus.PASS,
+            reason="test-ok",
+            absolute_error=100.0,
+            tolerance=OracleTolerance(absolute=0.001),
+        )
+
+
+def test_oracle_receipt_rejects_fail_with_error_within_tolerance() -> None:
+    """Cross-field validation: status='fail' with error within tolerance must be rejected."""
+    from flameox.domain.models import OracleReceiptV1, OracleTolerance
+
+    with pytest.raises(ValidationError, match="contradicts evidence"):
+        OracleReceiptV1(
+            schema_version="flameox.oracle-receipt.v1",
+            status=OracleStatus.FAIL,
+            reason="test-bad",
+            absolute_error=0.0,
+            tolerance=OracleTolerance(absolute=1.0),
+        )
+
+
+def test_oracle_receipt_accepts_pass_with_error_within_tolerance() -> None:
+    """Cross-field validation: status='pass' with error within tolerance is accepted."""
+    from flameox.domain.models import OracleReceiptV1, OracleTolerance
+
+    receipt = OracleReceiptV1(
+        schema_version="flameox.oracle-receipt.v1",
+        status=OracleStatus.PASS,
+        reason="test-ok",
+        absolute_error=0.001,
+        tolerance=OracleTolerance(absolute=0.01),
+    )
+    assert receipt.status == OracleStatus.PASS
+
+
+def test_oracle_receipt_accepts_pass_without_quantitative_evidence() -> None:
+    """Receipts without error/tolerance evidence are accepted with any status."""
+    from flameox.domain.models import OracleReceiptV1
+
+    receipt = OracleReceiptV1(
+        schema_version="flameox.oracle-receipt.v1",
+        status=OracleStatus.PASS,
+        reason="test-ok",
+    )
+    assert receipt.status == OracleStatus.PASS