feat(preflight): measurement-fidelity gate for the benchmark harness

pyproject.toml

@@ -41,6 +41,7 @@ dependencies = [
     "rich",
     "vidhi",
     "websockets>=14",
+    "aiohttp",
     "jiwer",
     "regex",
     "soundfile",

tests/unit/client/test_vajra_tts_stream_client.py

@@ -210,7 +210,7 @@ def _run_request(
     monkeypatch: pytest.MonkeyPatch,
     events: list[str],
 ):
-    monkeypatch.setattr(client, "_connect", lambda: _FakeConnection(websocket))
+    monkeypatch.setattr(client, "_connect", lambda *a, **k: _FakeConnection(websocket))
     return asyncio.run(
         client.send_request(
             _request(),

tests/unit/evaluator/performance/test_lifecycle_drift.py

@@ -0,0 +1,82 @@
+"""Unit tests for harness request-lifecycle drift metrics + soft WARN."""
+
+import logging
+from types import SimpleNamespace
+
+import pytest
+
+from veeksha.config.evaluator import PerformanceEvaluatorConfig
+from veeksha.evaluator.performance.base import PerformanceEvaluator
+
+
+def _response(ready, dispatched, pickup, sent):
+    return SimpleNamespace(
+        scheduler_ready_at=ready,
+        scheduler_dispatched_at=dispatched,
+        client_picked_up_at=pickup,
+        client_sent_at=sent,
+    )
+
+
+def _evaluator(threshold_ms=5.0):
+    return PerformanceEvaluator(
+        PerformanceEvaluatorConfig(lifecycle_drift_warn_threshold_ms=threshold_ms)
+    )
+
+
+@pytest.mark.unit
+def test_lifecycle_metrics_in_summary():
+    ev = _evaluator()
+    for _ in range(20):
+        # ready->dispatch 1ms, dispatch->pickup 0.5ms, pickup->send 2ms
+        ev._accumulate_lifecycle_drift(_response(100.0, 100.001, 100.0015, 100.0035))
+
+    summary = ev.get_aggregated_summary()
+    assert "Harness Ready-to-Dispatch (ms) (P99)" in summary
+    assert summary["Harness Ready-to-Dispatch (ms) (P99)"] == pytest.approx(
+        1.0, rel=0.05
+    )
+    assert summary["Harness Pickup-to-Send (ms) (P99)"] == pytest.approx(2.0, rel=0.05)
+    assert summary["Harness Ready-to-Send (ms) (P99)"] == pytest.approx(3.5, rel=0.05)
+
+
+@pytest.mark.unit
+def test_missing_timestamps_are_skipped():
+    ev = _evaluator()
+    # client_sent_at missing -> pickup->send and ready->send have no samples
+    ev._accumulate_lifecycle_drift(_response(100.0, 100.001, 100.0015, None))
+    assert ev._lifecycle_sketches["Harness Ready-to-Dispatch (ms)"].sketch.count == 1
+    assert ev._lifecycle_sketches["Harness Pickup-to-Send (ms)"].sketch.count == 0
+    assert ev._lifecycle_sketches["Harness Ready-to-Send (ms)"].sketch.count == 0
+
+
+@pytest.mark.unit
+def test_warn_fires_above_threshold(caplog):
+    ev = _evaluator(threshold_ms=5.0)
+    for _ in range(30):
+        # pickup->send = 20ms, well above the 5ms threshold
+        ev._accumulate_lifecycle_drift(_response(100.0, 100.001, 100.0015, 100.0215))
+    with caplog.at_level(logging.WARNING):
+        ev._warn_on_lifecycle_drift()
+    warnings = [r.getMessage() for r in caplog.records if r.levelno == logging.WARNING]
+    assert any("Pickup-to-Send" in w and "lifecycle drift high" in w for w in warnings)
+
+
+@pytest.mark.unit
+def test_no_warn_within_threshold(caplog):
+    ev = _evaluator(threshold_ms=50.0)
+    for _ in range(30):
+        ev._accumulate_lifecycle_drift(_response(100.0, 100.001, 100.0015, 100.0035))
+    with caplog.at_level(logging.WARNING):
+        ev._warn_on_lifecycle_drift()
+    assert not [r for r in caplog.records if "lifecycle drift high" in r.getMessage()]
+
+
+@pytest.mark.unit
+def test_warn_disabled_when_threshold_nonpositive(caplog):
+    ev = _evaluator(threshold_ms=0.0)
+    for _ in range(30):
+        ev._accumulate_lifecycle_drift(_response(100.0, 100.001, 100.0015, 100.5))
+    with caplog.at_level(logging.WARNING):
+        ev._warn_on_lifecycle_drift()
+    assert not [r for r in caplog.records if "lifecycle drift high" in r.getMessage()]

tests/unit/preflight/test_chat_mock_integration.py

@@ -0,0 +1,103 @@
+"""End-to-end: real chat client -> spawned mock server -> scorer.
+
+Exercises the actual OpenAIChatCompletionsClient streaming path against a
+deterministic mock server running in a separate process, then scores the
+paired client/server timestamps. This is the vertical slice's proof that
+request/response delivery lag and server pacing fidelity are measured correctly.
+"""
+
+import asyncio
+
+import pytest
+
+from veeksha.client import ClientRegistry
+from veeksha.config.client import OpenAIChatCompletionsClientConfig
+from veeksha.core.request import Request
+from veeksha.core.request_content import TextChannelRequestContent
+from veeksha.core.tokenizer import TokenizerHandle, TokenizerProvider
+from veeksha.preflight import scorer
+from veeksha.preflight.spawn import spawn_mock_chat_server
+from veeksha.types import ChannelModality
+
+TTFC_MS = 80.0
+TPOC_MS = 10.0
+NUM_CHUNKS = 24
+
+
+def _word_tokenizer_provider() -> TokenizerProvider:
+    handle = TokenizerHandle(
+        count_tokens=lambda text: len(str(text).split()),
+        decode=lambda token_ids: "",
+        encode=lambda text: [0] * len(str(text).split()),
+    )
+    return TokenizerProvider({ChannelModality.TEXT: handle})
+
+
+def _make_request(rid: int) -> Request:
+    return Request(
+        id=rid,
+        channels={
+            ChannelModality.TEXT: TextChannelRequestContent(input_text=f"hello {rid}")
+        },
+    )
+
+
+@pytest.mark.unit
+def test_chat_client_against_mock_is_measured():
+    with spawn_mock_chat_server(
+        ttfc_ms=TTFC_MS, tpoc_ms=TPOC_MS, num_chunks=NUM_CHUNKS
+    ) as server:
+        config = OpenAIChatCompletionsClientConfig(
+            api_base=server.api_base,
+            api_key="preflight",
+            model="dummy",
+            record_preflight_timing=True,
+        )
+        client = ClientRegistry.get(
+            config.get_type(),
+            config=config,
+            tokenizer_provider=_word_tokenizer_provider(),
+        )
+
+        async def _run():
+            results = []
+            for rid in range(6):
+                results.append(
+                    await client.send_request(request=_make_request(rid), session_id=0)
+                )
+            return results
+
+        results = asyncio.run(_run())
+        server_records = server.fetch_records()
+
+    # every request succeeded and streamed all chunks
+    assert all(r.success for r in results), [r.error_msg for r in results]
+    assert all(len(r.chunk_recv_times) == NUM_CHUNKS for r in results)
+    assert len(server_records) == len(results)
+
+    report = scorer.score(results, server_records, ttfc_ms=TTFC_MS, tpoc_ms=TPOC_MS)
+
+    # all requests paired with a server record
+    assert report.n_paired_requests == len(results)
+    assert report.unpaired_fraction == 0.0
+
+    m = report.metrics
+    # delivery lags exist, are non-negative, and small on loopback
+    rd = m[scorer.M_REQUEST_DELIVERY]
+    resp = m[scorer.M_RESPONSE_DELIVERY]
+    assert rd.count == len(results)
+    assert resp.count == NUM_CHUNKS * len(results)
+    assert rd.minimum >= -0.001  # allow float noise; delivery should be >= 0
+    assert resp.minimum >= -0.001
+    # loopback delivery should be well under a few ms at p99
+    assert rd.p99 < 20.0
+    assert resp.p99 < 20.0
+
+    # the mock keeps its own schedule tightly (its whole job)
+    assert m[scorer.M_SERVER_TTFC_ABS_ERR].p99 < 15.0
+    assert m[scorer.M_SERVER_TPOC_ABS_ERR].p99 < 15.0
+
+    # client-observed ttfc/tpoc are in the right ballpark
+    assert m[scorer.M_CLIENT_TTFC].p50 == pytest.approx(TTFC_MS, abs=30.0)
+    assert m[scorer.M_CLIENT_TPOC].count == (NUM_CHUNKS - 1) * len(results)
+    assert m[scorer.M_CLIENT_TPOC].p50 == pytest.approx(TPOC_MS, abs=TPOC_MS)

tests/unit/preflight/test_driver_integration.py

@@ -0,0 +1,171 @@
+"""End-to-end: the preflight driver runs the REAL benchmark loop vs a mock.
+
+Unlike test_chat_mock_integration (which drives the client directly), this
+exercises the full scheduler -> dispatch -> client_runner -> client path, so the
+dispatch-drift metrics are populated too. Then it gates the result.
+"""
+
+import pytest
+
+from veeksha.config.preflight import (
+    PreflightSttCheckConfig,
+    PreflightTextCheckConfig,
+    PreflightTtsCheckConfig,
+)
+from veeksha.config.traffic import ConcurrentTrafficConfig
+from veeksha.preflight import scorer, validator
+from veeksha.preflight.drivers import (
+    run_completions_preflight,
+    run_realtime_tts_preflight,
+    run_stt_preflight,
+    run_text_preflight,
+    run_tts_preflight,
+    run_vajra_tts_preflight,
+)
+
+TTFC_MS = 60.0
+TPOC_MS = 8.0
+NUM_CHUNKS = 16
+
+
+def _traffic(concurrency):
+    return ConcurrentTrafficConfig(
+        target_concurrent_sessions=concurrency, rampup_seconds=0
+    )
+
+
+def _text_cfg():
+    return PreflightTextCheckConfig(
+        input_tokens=8,
+        num_response_chunks=NUM_CHUNKS,
+        server_ttfc_ms=TTFC_MS,
+        server_tpoc_ms=TPOC_MS,
+    )
+
+
+def _tts_cfg():
+    return PreflightTtsCheckConfig(
+        input_tokens=8,
+        input_chunk_tokens=2,
+        input_pacing_tps=500.0,  # fast so the test isn't slow
+        num_response_chunks=NUM_CHUNKS,
+        server_ttfc_ms=TTFC_MS,
+        server_tpoc_ms=TPOC_MS,
+    )
+
+
+def _stt_cfg():
+    return PreflightSttCheckConfig(
+        input_seconds=0.3,  # short clip keeps the realtime-paced input quick
+        input_chunk_bytes=1024,
+        sample_rate=16000,
+        num_response_chunks=NUM_CHUNKS,
+        server_ttfc_ms=TTFC_MS,
+        server_tpoc_ms=TPOC_MS,
+    )
+
+
+def _assert_measured(report, streaming_input):
+    assert report.n_requests > 0
+    assert report.n_paired_requests > 0
+    assert report.unpaired_fraction < 0.5
+    m = report.metrics
+    assert m[scorer.M_REQUEST_DELIVERY].count > 0
+    assert m[scorer.M_RESPONSE_DELIVERY].count > 0
+    assert m[scorer.M_SERVER_TTFC_ABS_ERR].count > 0
+    # dispatch drift only exists because the real scheduler/dispatch path ran.
+    assert m[scorer.M_LIFECYCLE_READY_TO_SEND].count > 0
+    if streaming_input:
+        assert m[scorer.M_INPUT_DELIVERY].count > 0
+        assert m[scorer.M_INPUT_PACING_ABS_ERR].count > 0
+    else:
+        assert scorer.M_INPUT_DELIVERY not in m
+        assert scorer.M_INPUT_PACING_ABS_ERR not in m
+
+
+@pytest.mark.unit
+def test_text_driver_runs_real_loop_and_scores(tmp_path):
+    report = run_text_preflight(
+        _text_cfg(),
+        traffic_scheduler=_traffic(8),
+        num_sessions=40,
+        output_dir=str(tmp_path / "text"),
+    )
+    _assert_measured(report, streaming_input=False)
+    assert report.metrics[scorer.M_SERVER_TPOC_ABS_ERR].count > 0
+
+    result = validator.run_validation(
+        report,
+        delivery_lag_threshold_ms=25.0,
+        server_pacing_threshold_ms=25.0,
+        dispatch_drift_threshold_ms=50.0,
+        input_pacing_threshold_ms=50.0,
+        max_unpaired_fraction=0.1,
+    )
+    assert result.verdict in (
+        validator.VERDICT_PASS,
+        validator.VERDICT_FAIL,
+        validator.VERDICT_SERVER_AT_CAPACITY,
+    )
+
+
+@pytest.mark.unit
+def test_completions_driver_runs_real_loop_and_scores(tmp_path):
+    report = run_completions_preflight(
+        _text_cfg(),
+        traffic_scheduler=_traffic(8),
+        num_sessions=40,
+        output_dir=str(tmp_path / "completions"),
+    )
+    _assert_measured(report, streaming_input=False)
+    # non-streaming: a single response, so no tpoc samples.
+    assert scorer.M_SERVER_TPOC_ABS_ERR not in report.metrics
+
+
+@pytest.mark.unit
+def test_tts_driver_runs_real_loop_and_scores(tmp_path):
+    report = run_tts_preflight(
+        _tts_cfg(),
+        traffic_scheduler=_traffic(8),
+        num_sessions=40,
+        output_dir=str(tmp_path / "tts"),
+    )
+    # HTTP tts sends the whole text in one POST -> no streaming-input metrics.
+    _assert_measured(report, streaming_input=False)
+    assert report.metrics[scorer.M_SERVER_TPOC_ABS_ERR].count > 0
+
+
+@pytest.mark.unit
+def test_realtime_tts_driver_runs_real_loop_and_scores(tmp_path):
+    report = run_realtime_tts_preflight(
+        _tts_cfg(),
+        traffic_scheduler=_traffic(6),
+        num_sessions=24,
+        output_dir=str(tmp_path / "realtime_tts"),
+    )
+    _assert_measured(report, streaming_input=True)
+    assert report.metrics[scorer.M_SERVER_TPOC_ABS_ERR].count > 0
+
+
+@pytest.mark.unit
+def test_vajra_tts_driver_runs_real_loop_and_scores(tmp_path):
+    report = run_vajra_tts_preflight(
+        _tts_cfg(),
+        traffic_scheduler=_traffic(6),
+        num_sessions=24,
+        output_dir=str(tmp_path / "vajra_tts"),
+    )
+    _assert_measured(report, streaming_input=True)
+    assert report.metrics[scorer.M_SERVER_TPOC_ABS_ERR].count > 0
+
+
+@pytest.mark.unit
+def test_stt_driver_runs_real_loop_and_scores(tmp_path):
+    report = run_stt_preflight(
+        _stt_cfg(),
+        traffic_scheduler=_traffic(4),
+        num_sessions=16,
+        output_dir=str(tmp_path / "stt"),
+    )
+    _assert_measured(report, streaming_input=True)
+    assert report.metrics[scorer.M_SERVER_TPOC_ABS_ERR].count > 0