fix: accept pinned Flare derived analysis after sidecar save

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,12 +153,14 @@ 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.
 
 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.

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(

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,21 +180,22 @@ 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"]))
                 + b"\n\n"
                 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,

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",
     [