diff --git a/docs/flare-qualification.md b/docs/flare-qualification.md index 9b0e646e..e8fce79a 100644 --- a/docs/flare-qualification.md +++ b/docs/flare-qualification.md @@ -63,9 +63,16 @@ it does not load a tokenizer or prove that a rendered prompt tokenizes correctly Each observation must match its sidecar and raw SSE payloads, successful HTTP completion, terminal DONE, prompt/output/usage counts and sent request identity. +Pinned Flare saves each sidecar before `run` adds +`analysis.http_prefill_tokens_per_second` to measurements for `ttft-4096` and +`prefill-65536`. The adapter accepts that one addition only with the pinned +formula (`prompt_token_count * 1e9 / analysis.delivery_ttft_ns`); every original +field, including the rest of analysis, must still match. Identical unaugmented +copies remain supported. The adapter does not modify either input copy. Coalesced groups must agree with any recorded grouping analysis. Summary TTFT -numbers are not metric inputs. Explicit failed qualification/group flags and -stream errors remain rejections even if other raw checks pass. +and derived throughput numbers are not metric inputs. Explicit failed +qualification/group flags and stream errors remain rejections even if other raw +checks pass. The native `prefill_chunk_size` label must match the preparation report. Selected raw snapshots must contain unlabelled integer samples for the scheduler gauges, @@ -146,6 +153,7 @@ pytest tests/test_flare_adapter.py The fixture writer saves a complete one-repetition packet, including raw SSE, coalesced token events, a prefix-hit case and before/after resource snapshots. +It follows the pinned producer's sidecar-before-derived-analysis write order. It labels every packet `synthetic=true`; no tokenizer, model, server or device is used. Synthetic delivery timings and model/package placeholders are parser test inputs, not measured service results. @@ -153,5 +161,6 @@ test inputs, not measured service results. Tests cover raw recomputation and unchanged originals; missing raw/group/request records; wrong clock declarations; invented coalesced-token timing; capacity, counter and resource mismatches; stream failures and digest drift; pair scope; -and real local runner execution of both adapter commands. Fixture acceptance -does not establish service quality, a performance gain or agent efficacy. +sidecar/measurements divergence even with refreshed digests; and real local +runner execution of both adapter commands. Fixture acceptance does not establish +service quality, a performance gain or agent efficacy. diff --git a/src/inferrsi/model/validators/flare.py b/src/inferrsi/model/validators/flare.py index 9821645c..5eb58da2 100644 --- a/src/inferrsi/model/validators/flare.py +++ b/src/inferrsi/model/validators/flare.py @@ -330,7 +330,28 @@ def adapt(packet_path: Path) -> dict: results = [] for name in group["cases"]: observation = json.loads(files[f"{root}/{name}/observation.json"]) - _require(observation in observations, "observation sidecar differs") + expected = observation + analysis = observation.get("analysis", {}) + ttft = analysis.get("delivery_ttft_ns") + if ( + group["kind"] in ("ttft-4096", "prefill-65536") + and ttft + and "http_prefill_tokens_per_second" not in analysis + ): + # Pinned Flare adds this derived value only after saving the sidecar. + expected = { + **observation, + "analysis": { + **analysis, + "http_prefill_tokens_per_second": ( + len(cases[name]["prompt_tokens"]) * 1e9 / ttft + ), + }, + } + _require( + observation in observations or expected in observations, + "observation sidecar differs", + ) request = json.loads(files[f"{root}/{name}/request.json"]) expected_request = dict(cases[name]["request"]) expected_request.update( diff --git a/tests/fixtures/flare_packet.py b/tests/fixtures/flare_packet.py index eee3c8f9..bcc261c9 100644 --- a/tests/fixtures/flare_packet.py +++ b/tests/fixtures/flare_packet.py @@ -114,6 +114,7 @@ def add(name, kind, batch, repetition, measured): for group in groups: folder = root / "run" / group["id"] folder.mkdir(parents=True) + observations = [] reused = 0 for identifier in group["cases"]: case = next(c for c in cases if c["id"] == identifier) @@ -179,13 +180,13 @@ def add(name, kind, batch, repetition, measured): "events": events, "analysis": { "passed": True, + "delivery_ttft_ns": ttft_ns, "token_groups": token_groups, "delivery_itl_exact": len(output) == 1, "delivery_itl_ns": [], }, } write_json(local / "request.json", request) - write_json(local / "observation.json", observation) raw = b"".join( b"data: " + (e["data"].encode() if isinstance(e["data"], str) else encoded(e["data"])) @@ -193,7 +194,8 @@ def add(name, kind, batch, repetition, measured): for e in events ) (local / "response.sse").write_bytes(raw) - measurements["observations"].append(observation) + write_json(local / "observation.json", observation) + observations.append(observation) values = { "flare_active_requests": 0, "flare_free_request_slots": 4, @@ -220,6 +222,15 @@ def add(name, kind, batch, repetition, measured): names.append(name) snapshots[group["id"]] = names measurements["groups"].append({"id": group["id"], "passed": True}) + # Pinned Flare augments in-memory analysis after saving observation sidecars. + for observation in observations: + ttft = observation["analysis"]["delivery_ttft_ns"] + if ttft and group["kind"] in ("ttft-4096", "prefill-65536"): + case = next(c for c in cases if c["id"] == observation["case"]) + observation["analysis"]["http_prefill_tokens_per_second"] = ( + len(case["prompt_tokens"]) * 1e9 / ttft + ) + measurements["observations"].extend(observations) write_json(root / "measurements.json", measurements) manifest = { "schema": SCHEMA, diff --git a/tests/test_flare_adapter.py b/tests/test_flare_adapter.py index fd8372ae..6c9e39f2 100644 --- a/tests/test_flare_adapter.py +++ b/tests/test_flare_adapter.py @@ -34,11 +34,50 @@ def edit_observation(packet, change): edit(packet, "measurements.json", lambda m: m["observations"].__setitem__(0, observation)) +def test_pinned_producer_augments_measurements_after_saving_sidecars(packet): + measurements = json.loads((packet.parent / "measurements.json").read_bytes()) + augmented = [] + for observation in measurements["observations"]: + name = observation["case"] + group = name.rsplit("-", 1)[0] + sidecar = json.loads( + (packet.parent / f"run/{group}/{name}/observation.json").read_bytes() + ) + if group in ("ttft-4096-r0", "prefill-65536-r0"): + assert "http_prefill_tokens_per_second" not in sidecar["analysis"] + expected = { + **sidecar, + "analysis": { + **sidecar["analysis"], + "http_prefill_tokens_per_second": ( + (4096 if group == "ttft-4096-r0" else 65536) * 1e9 / 100_000_000 + ), + }, + } + assert observation == expected + augmented.append(name) + else: + assert observation == sidecar + assert len(augmented) == 2 + receipt = adapt(packet) + assert receipt["metrics"] == {"delivery_ttft_ns": 100_000_000} + + def test_complete_raw_packet_recomputes_metric_and_preserves_originals(packet, tmp_path): + def change_summaries(measurements): + measurements.update(summary={"delivery_ttft_ns": -999}) + observation = next( + o for o in measurements["observations"] if o["case"] == "ttft-512-r0-0" + ) + observation["analysis"]["delivery_ttft_ns"] = -999 + + edit(packet, "measurements.json", change_summaries) + edit( + packet, + "run/ttft-512-r0/ttft-512-r0-0/observation.json", + lambda o: o["analysis"].update(delivery_ttft_ns=-999), + ) before = {p: p.read_bytes() for p in packet.parent.rglob("*") if p.is_file()} - edit(packet, "measurements.json", lambda m: m.update(summary={"delivery_ttft_ns": -999})) - before[packet] = packet.read_bytes() - before[packet.parent / "measurements.json"] = (packet.parent / "measurements.json").read_bytes() receipt = adapt(packet) assert receipt["metrics"] == {"delivery_ttft_ns": 100_000_000} assert receipt["synthetic"] is True @@ -56,6 +95,61 @@ def test_complete_raw_packet_recomputes_metric_and_preserves_originals(packet, t assert json.loads(output.read_bytes())["identities"]["manifest_sha256"] +def test_identical_unaugmented_observations_remain_supported(packet): + def remove_derived(measurements): + for observation in measurements["observations"]: + observation["analysis"].pop("http_prefill_tokens_per_second", None) + + edit(packet, "measurements.json", remove_derived) + assert adapt(packet)["metrics"] == {"delivery_ttft_ns": 100_000_000} + + +@pytest.mark.parametrize("side", ["sidecar", "measurements"]) +@pytest.mark.parametrize( + "mutation", ["timestamp", "status", "request", "analysis", "extra_analysis"] +) +def test_divergent_observation_rejects_even_with_refreshed_digest(packet, tmp_path, side, mutation): + def change(observation): + if mutation == "timestamp": + observation["events"][0]["elapsed_ns"] += 1 + elif mutation == "status": + observation["status"] = 201 + elif mutation == "request": + observation["request_sha256"] = "0" * 64 + elif mutation == "analysis": + observation["analysis"]["delivery_ttft_ns"] += 1 + else: + observation["analysis"]["unrecognized_derived_value"] = 123 + + if side == "sidecar": + edit(packet, "run/ttft-4096-r0/ttft-4096-r0-0/observation.json", change) + else: + edit( + packet, + "measurements.json", + lambda m: change(next(o for o in m["observations"] if o["case"] == "ttft-4096-r0-0")), + ) + output = tmp_path / "rejected.json" + assert ( + main(["measure", "--packet", str(packet), "--peer", str(packet), "--output", str(output)]) + == 2 + ) + rejected = json.loads(output.read_bytes()) + assert rejected["error"] == "observation sidecar differs" + assert rejected["metrics"] == {} + + +@pytest.mark.parametrize("case", ["warmup-0", "ttft-4096-r0-0"]) +def test_derived_analysis_exception_rejects_wrong_group_or_value(packet, case): + def change(measurements): + observation = next(o for o in measurements["observations"] if o["case"] == case) + observation["analysis"]["http_prefill_tokens_per_second"] = -1 + + edit(packet, "measurements.json", change) + with pytest.raises(ValueError, match="observation sidecar differs"): + adapt(packet) + + @pytest.mark.parametrize( "mutation", [