From 5a2d16eeaa02c63d2b355665db933d5cea576685 Mon Sep 17 00:00:00 2001 From: GUOGUO Date: Mon, 26 Oct 2026 15:55:25 +0800 Subject: [PATCH] fix: accept Flare post-save derived analysis Match the pinned producer augmentation without relaxing original observation integrity or raw metric recomputation. Exercise its write order and reject divergent fields even with refreshed packet digests. Agent-assisted-by: Codex (author) Signed-off-by: GUOGUO --- docs/flare-qualification.md | 17 ++++- src/inferrsi/model/validators/flare.py | 23 +++++- tests/fixtures/flare_packet.py | 15 +++- tests/test_flare_adapter.py | 100 ++++++++++++++++++++++++- 4 files changed, 145 insertions(+), 10 deletions(-) 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", [