diff --git a/docs/flare-qualification.md b/docs/flare-qualification.md new file mode 100644 index 00000000..9b0e646e --- /dev/null +++ b/docs/flare-qualification.md @@ -0,0 +1,157 @@ +# Offline Flare delivery evidence + +The adapter reads existing Flare qualification artifacts and writes a separate +receipt. It never launches Flare, downloads a model or executes imported code. +Run it with `python -m inferrsi.model.validators.flare`. The existing +`external_benchmark` scenario, `validation_command` and evaluator own execution +and verdicts. + +## Supported source and metric + +Input contract: Flare +[`bce032c014bae34502ba22d3304fa084d5b40afe`](https://github.com/NVIDIA/flare-local-inference/tree/bce032c014bae34502ba22d3304fa084d5b40afe). +The frozen readset is: + +| Source | Contract used | +| --- | --- | +| [`qualify/performance.py`](https://github.com/NVIDIA/flare-local-inference/blob/bce032c014bae34502ba22d3304fa084d5b40afe/qualify/performance.py) — `validate_corpus`, `stream_request`, `analyze_events`, `idle_metrics`, `run` | Ordered corpus groups, request/token fingerprints, raw SSE and timestamped observations, native completion metadata, preparation capacity and raw counter snapshots | +| [`qualify/provenance.py`](https://github.com/NVIDIA/flare-local-inference/blob/bce032c014bae34502ba22d3304fa084d5b40afe/qualify/provenance.py) — `digest`, `file_record` | File-byte SHA-256 identities; source/package declarations remain distinct from runtime qualification | + +Exactly one scalar is exported: `metrics.delivery_ttft_ns`. It is the median +first nonempty token-event arrival across **all measured `ttft-512` requests** +in the supplied complete corpus, in nanoseconds. Its clock domain is +`client.perf_counter_ns.elapsed`; comparison scope is +`flare_http_delivery_ttft_512`. Connection opening happens before the upstream +send barrier. HTTP delivery TTFT includes request/frontend/scheduling/delivery +work and does not isolate GPU prefill. + +Native `token_timestamps_ns`, `ingress_ns` and `enqueue_ns` are checked within +their own domain. They are never subtracted from client timestamps. One SSE +event containing several token IDs remains one arrival group. Such groups do +not establish exact per-token delivery intervals. +The pinned [native observation contract](https://github.com/NVIDIA/flare-local-inference/blob/bce032c014bae34502ba22d3304fa084d5b40afe/helios/model-contract.md) +allows `usage=null` on preceding chunks; the adapter ignores null optional +metadata and uses the actual terminal usage/timing records. + +## Packet schema v1 + +A packet is a caller-assembled directory with `packet.json` and unchanged source +artifacts beneath the same root. Paths are root-relative; external paths and +symlink escapes are rejected. The output receipt must be outside that directory. +The wrapper does not change the upstream JSON or raw SSE formats. + +| Manifest field | Required value / meaning | +| --- | --- | +| `schema` | `inferrsi-flare-delivery-packet-v1` | +| `source` | Exact `repository`, `revision` and ordered `readset` object from `SOURCE` in the adapter | +| `metric` | Exact `key`, `unit`, `clock_domain`, `grouping`, `comparison_scope` object from `METRIC` in the adapter | +| `synthetic` | Explicit boolean; `true` for constructed fixtures | +| `package` | Path to actual package bytes; synthetic fixtures use an explicitly labelled non-model file | +| `corpus` | Path to the frozen `flare-http-performance-v1-corpus` JSON | +| `memory_report` | Path to the actual preparation memory report, including `prefill_chunk_size` | +| `measurements` | Path to the completed native Flare `measurements.json` | +| `run` | Root-relative directory containing upstream per-group artifacts | +| `snapshots` | Map from every corpus group ID to `[before_path, after_path]`, selecting the raw idle `/metrics` captures used for that group | +| `files` | Map from every referenced asset path to its SHA-256; includes package, corpus, memory report, measurements, selected snapshots and per-request `request.json`, `observation.json`, `response.sse` | + +The complete ordered upstream matrix is required: warmup, both TTFT lengths, +decode concurrency 1/2/4, 65536-token prefill, prefix seed and prefix hit for +each declared repetition. Warmup and prefix seed are checked but excluded from +the exported TTFT samples. Missing/duplicate groups or requests cannot be filled +with zeroes. The adapter checks corpus token geometry and recorded fingerprints; +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. +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. + +The native `prefill_chunk_size` label must match the preparation report. Selected +raw snapshots must contain unlabelled integer samples for the scheduler gauges, +completed/failed/cancelled request counters, reused-token counter, and free/cached +KV-page and snapshot/checkpoint gauges named in the adapter. Duplicate samples, +missing values, nonfinite values, non-idle schedulers, incorrect request/reuse +deltas and resource-conservation failures reject the packet. A cache lookup hit +does not substitute for actual `reused_prompt_tokens`: cold cases require zero, +and the pinned prefix-hit cases require 4096, agreeing with raw counter deltas. +P0 does not reconstruct offload admission decisions or implement a cache policy. + +## CLI and existing runner integration + +Both measurement commands must name their peer. Before exporting either metric, +the adapter validates both packets and requires matching source, metric contract, +synthetic status, package bytes, corpus bytes and prepared capacity. This keeps +an incomplete or incomparable pair out of the evaluator's performance inputs. + +```text +python -m inferrsi.model.validators.flare measure \ + --packet /packet.json --peer /packet.json \ + --output /baseline.json + +python -m inferrsi.model.validators.flare validate-pair \ + --baseline /packet.json --candidate /packet.json \ + --output /pair.json +``` + +For `RunSpec`, use these argv sequences: + +```python +cli = (python, "-m", "inferrsi.model.validators.flare") +baseline_command = cli + ( + "measure", "--packet", baseline, "--peer", candidate, "--output", "{result}", +) +candidate_command = cli + ( + "measure", "--packet", candidate, "--peer", baseline, "--output", "{result}", +) +validation_command = cli + ( + "validate-pair", "--baseline", baseline, "--candidate", candidate, "--output", "{result}", +) +``` + +Set `scenario="external_benchmark"`, `metric_key="metrics.delivery_ttft_ns"`, +`metric_unit="ns"`, `direction="minimize"` and +`comparison_scope="flare_http_delivery_ttft_512"`. The ordinary RunSpec controls +the comparison threshold and trial count. Re-reading one packet several times +does not create independent service measurements; freeze independent captured +inputs when designing a real campaign. + +Exit zero means the offline contract passed. Invalid/incomplete evidence returns +2 and writes a rejection receipt with **no metric**. Through the standard +executor this is an incomplete measurement, so the existing evaluator retains +`INCONCLUSIVE`. A valid pair's receipt is suitable for the fixture integration +test's `validation_command`; a real campaign must also include its separately +frozen task-quality validator. Pair consistency is not a model-quality oracle. +Hardware equality, workload representativeness and actual measurement execution +remain campaign acceptance requirements. + +The receipt retains per-request groups and selected samples, packet/asset hashes, +source/readset, package/corpus identities, actual prepared capacity, and hashes +of the adapter and its artifact helper. These bind the receipt to the supplied +bytes; they do not attest that a package was built from a declared source or +qualify its ABI/model/frontend compatibility. ABI/preparation preflight and +session-state recovery scenarios remain separate future work. + +## Reproducible synthetic sample and negative cases + +From a development checkout with InferRSI installed: + +```text +python tests/fixtures/flare_packet.py +python -m inferrsi.model.validators.flare measure \ + --packet /packet.json --peer /packet.json \ + --output /receipt.json +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 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. diff --git a/docs/operations.md b/docs/operations.md index 89af098b..306e6df3 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -26,6 +26,7 @@ Effect classes used below: | Amend repository-owned knowledge | Scoped repository PR | — | Repository review and integration authority apply; this is not a runtime store Update | | Inspect/prepare experiments | CLI `doctor`; `plan` or `--dry-run` | Read-only inspection / preview | `doctor` may invoke bounded probes and does not certify target compatibility; `plan` previews a specification without persisting a campaign or authorizing execution | | Execute/inspect experiments | CLI `run`; `status`; `report` | Experiment-executing / read-only / report-generating | `run` executes caller-supplied commands and records evidence; `status` reads recorded progress; `report` generates HTML from recorded evidence | +| Adapt existing Flare delivery evidence | [Offline qualification adapter](flare-qualification.md), `python -m inferrsi.model.validators.flare` | Read-only | Existing raw packets and a separate receipt; one HTTP delivery metric, no model/server/device execution | | Evaluate selected skill revisions | [Bounded skill-evaluation driver](skill-evaluation.md) | Experiment-executing | Explicitly authorized real Codex calls, independent task verdicts and private evidence; separate from numeric engine-performance scoring | | Compare skill additions or updates | [Configured skill comparison](skill-eval-comparison.md), `python -m tools.skill_eval_compare` | Read-only preparation / experiment-executing / report-generating | Fixed Git snapshots and one shared exam; explicit plan authorization for execution; JSON + Markdown regeneration makes no model calls | diff --git a/src/inferrsi/model/validators/flare.py b/src/inferrsi/model/validators/flare.py new file mode 100644 index 00000000..9821645c --- /dev/null +++ b/src/inferrsi/model/validators/flare.py @@ -0,0 +1,457 @@ +import argparse +import hashlib +import json +import statistics +import struct +from decimal import Decimal, InvalidOperation +from pathlib import Path + +from ...infra import artifacts + +SOURCE = { + "repository": "https://github.com/NVIDIA/flare-local-inference", + "revision": "bce032c014bae34502ba22d3304fa084d5b40afe", + "readset": ["qualify/performance.py", "qualify/provenance.py"], +} +METRIC = { + "key": "metrics.delivery_ttft_ns", + "unit": "ns", + "clock_domain": "client.perf_counter_ns.elapsed", + "grouping": "SSE event token groups; median across measured ttft-512 requests", + "comparison_scope": "flare_http_delivery_ttft_512", +} +UPSTREAM_CLOCK = ( + "client perf_counter_ns; native token timestamps are a separate process-relative domain" +) +SCHEMA = "inferrsi-flare-delivery-packet-v1" +COUNTERS = ( + "flare_cache_reused_tokens_total", + "flare_requests_completed_total", + "flare_requests_failed_total", + "flare_requests_cancelled_total", +) +RESOURCES = ( + "flare_active_requests", + "flare_free_request_slots", + "flare_retiring_request_slots", + "flare_pending_requests", + "flare_free_kv_pages", + "flare_cached_kv_pages", + "flare_free_snapshots", + "flare_cached_checkpoints", +) + + +def _require(condition, message): + if not condition: + raise ValueError(message) + + +def _integer(value, label, minimum=0): + _require(type(value) is int and value >= minimum, f"invalid {label}") + return value + + +def _encoded(value): + return json.dumps(value, ensure_ascii=False, separators=(",", ":"), allow_nan=False).encode() + + +def _sha(data): + return hashlib.sha256(data).hexdigest() + + +def _load(path): + root = path.resolve().parent + raw_manifest = path.read_bytes() + manifest = json.loads(raw_manifest) + _require(manifest["schema"] == SCHEMA, "unsupported packet schema") + _require(manifest["source"] == SOURCE, "unsupported source/readset identity") + _require(manifest["metric"] == METRIC, "unsupported metric clock/grouping/scope") + _require(type(manifest["synthetic"]) is bool, "synthetic must be explicit") + _require(isinstance(manifest["files"], dict), "files must map paths to digests") + files, identities = {}, {} + for name, expected in manifest["files"].items(): + location = (root / name).resolve() + _require( + not Path(name).is_absolute() and location.is_relative_to(root), + "packet file escapes its root", + ) + _require(location != path.resolve(), "manifest cannot be its own asset") + if name == manifest["package"]: + actual = artifacts.sha256(location) + else: + files[name] = location.read_bytes() + actual = _sha(files[name]) + _require(actual == expected, f"input digest mismatch: {name}") + identities[name] = actual + _require(manifest["package"] in identities, "missing package bytes") + return manifest, files, {"manifest_sha256": _sha(raw_manifest), "files": identities} + + +def _corpus(corpus): + _require(corpus["schema"] == "flare-http-performance-v1-corpus", "unsupported corpus schema") + repetitions = _integer(corpus["repetitions"], "corpus repetitions", 1) + expected = [("warmup", "decode", 4, -1, False)] + for repetition in range(repetitions): + expected.extend( + (f"ttft-{length}-r{repetition}", f"ttft-{length}", 1, repetition, True) + for length in (512, 4096) + ) + expected.extend( + (f"decode-b{batch}-r{repetition}", f"decode-b{batch}", batch, repetition, True) + for batch in (1, 2, 4) + ) + expected.extend( + [ + (f"prefill-65536-r{repetition}", "prefill-65536", 1, repetition, True), + (f"prefix-seed-r{repetition}", "prefix-seed", 1, repetition, False), + (f"prefix-hit-r{repetition}", "prefix-hit", 1, repetition, True), + ] + ) + groups = corpus["groups"] + actual = [(g["id"], g["kind"], g["batch"], g["repetition"], g["measured"]) for g in groups] + _require(actual == expected, "missing or changed corpus group") + cases = {case["id"]: case for case in corpus["cases"]} + _require(len(cases) == len(corpus["cases"]), "duplicate corpus case") + used = [] + for group in groups: + _require(len(group["cases"]) == group["batch"], "group batch/case mismatch") + for name in group["cases"]: + case = cases[name] + used.append(name) + _require(case["group"] == group["id"], "case/group mismatch") + tokens = case["prompt_tokens"] + length = { + "ttft-4096": 4096, + "prefix-seed": 4096, + "prefix-hit": 4107, + "prefill-65536": 65536, + }.get(group["kind"], 512) + output = 256 if group["kind"].startswith("decode") else 1 + _require(len(tokens) == length and case["output_tokens"] == output, "token geometry") + _require( + all(type(t) is int and 0 <= t < 248320 for t in tokens), "invalid prompt tokens" + ) + request = case["request"] + _require( + request["model"] == corpus["model_id"] and request["max_tokens"] == output, + "request/model/output mismatch", + ) + for key, data in ( + ("request_sha256", _encoded(request)), + ("prompt_tokens_sha256", struct.pack(f"<{len(tokens)}I", *tokens)), + ("rendered_prompt_sha256", case["rendered_prompt"].encode()), + ): + _require(case[key] == _sha(data), f"corpus fingerprint mismatch: {key}") + if group["kind"] == "prefix-hit": + seed = cases[case["prefix_seed"]] + _require( + case["expected_cache"] == "prefix-4096" + and seed["group"] == f"prefix-seed-r{group['repetition']}" + and tokens[:4096] == seed["prompt_tokens"], + "prefix token identity mismatch", + ) + else: + _require(case["expected_cache"] == "cold", "unexpected cache condition") + _require(sorted(used) == sorted(cases), "duplicate or missing case coverage") + return groups, cases + + +def _sse(raw): + payloads, data = [], [] + for line in raw.decode("utf-8").splitlines(): + if not line: + if data: + value = "\n".join(data) + payloads.append(value if value == "[DONE]" else json.loads(value)) + data = [] + elif line.startswith("data:"): + data.append(line[5:].removeprefix(" ")) + _require(not data and bool(payloads), "missing or truncated raw SSE") + return payloads + + +def _observation(observation, raw, case): + _require(observation["case"] == case["id"], "observation case mismatch") + _require(observation["status"] == 200 and not observation.get("client_error"), "client failure") + _integer(observation["request_start_perf_counter_ns"], "client request start") + events = observation["events"] + _require([e["data"] for e in events] == _sse(raw), "raw SSE/event grouping mismatch") + delivery, native, ids, prompts = [], [], [], [] + usages, finishes, metadata, cached_counts = [], [], [], [] + done, previous = 0, 0 + for index, event in enumerate(events): + timestamp = _integer(event["elapsed_ns"], "client event timestamp") + _require(timestamp >= previous, "nonmonotonic client clock") + previous = timestamp + payload = event["data"] + if payload == "[DONE]": + done += 1 + _require(index == len(events) - 1, "events after DONE") + continue + _require(isinstance(payload, dict) and "error" not in payload, "stream error") + if payload.get("prompt_token_ids") is not None: + prompts.append(payload["prompt_token_ids"]) + if payload.get("usage") is not None: + usages.append(payload["usage"]) + if payload.get("performance") is not None: + metadata.append(payload["performance"]) + if payload.get("reused_prompt_tokens") is not None: + cached_counts.append(payload["reused_prompt_tokens"]) + for choice in payload.get("choices", []): + _require(choice["index"] == 0, "unexpected choice index") + if choice.get("finish_reason") is not None: + finishes.append(choice["finish_reason"]) + tokens = choice.get("token_ids") + times = choice.get("token_timestamps_ns") + if tokens is None: + _require(not times, "timestamps without tokens") + continue + _require( + isinstance(tokens, list) + and all(type(t) is int and 0 <= t < 248320 for t in tokens), + "invalid output tokens", + ) + _require( + isinstance(times, list) and len(times) == len(tokens), + "missing or unaligned native timestamps", + ) + native.extend(_integer(t, "native timestamp") for t in times) + ids.extend(tokens) + if tokens: + delivery.append({"timestamp_ns": timestamp, "token_ids": tokens}) + _require(done == 1 and delivery and finishes == ["length"], "incomplete stream") + _require(prompts == [case["prompt_tokens"]], "prompt token mismatch") + _require(len(ids) == case["output_tokens"], "incomplete generated tokens") + _require(usages and metadata, "missing usage/native timing") + usage, timing = usages[-1], metadata[-1] + _require( + usage["prompt_tokens"] == len(case["prompt_tokens"]) + and usage["completion_tokens"] == len(ids) + and usage["total_tokens"] == len(case["prompt_tokens"]) + len(ids), + "usage mismatch", + ) + ingress = _integer(timing["ingress_ns"], "native ingress") + enqueue = _integer(timing["enqueue_ns"], "native enqueue") + _require(ingress <= enqueue <= native[0], "native clock order") + _require(all(a <= b for a, b in zip(native, native[1:])), "nonmonotonic native clock") + reused = _integer(timing["reused_prompt_tokens"], "actual reused tokens") + expected_reuse = 4096 if case["expected_cache"] == "prefix-4096" else 0 + _require(reused == expected_reuse, "actual reused tokens mismatch") + cached_counts.append(reused) + if "prompt_tokens_details" in usage: + cached_counts.append(usage["prompt_tokens_details"]["cached_tokens"]) + _require(all(type(n) is int and n == reused for n in cached_counts), "cache counter mismatch") + singleton = all(len(group["token_ids"]) == 1 for group in delivery) + itl = [b["timestamp_ns"] - a["timestamp_ns"] for a, b in zip(delivery, delivery[1:])] + analysis = observation.get("analysis", {}) + if "passed" in analysis: + _require(analysis["passed"] is True, "reported failed observation") + for key, value in ( + ("token_groups", delivery), + ("delivery_itl_exact", singleton), + ("delivery_itl_ns", itl if singleton else []), + ): + if key in analysis: + _require(analysis[key] == value, "expanded or changed coalesced event grouping") + ttft = delivery[0]["timestamp_ns"] + _require(ttft > 0, "delivery TTFT must be positive") + return { + "case": case["id"], + "delivery_ttft_ns": ttft, + "reused_prompt_tokens": reused, + "token_groups": delivery, + "delivery_itl_exact": singleton, + "coalesced_groups": sum(len(group["token_ids"]) > 1 for group in delivery), + } + + +def _counters(raw): + values = {} + required = set(COUNTERS + RESOURCES) + for line in raw.decode("utf-8").splitlines(): + if not line or line.startswith("#"): + continue + parts = line.split() + if parts[0] not in required: + continue + _require(len(parts) == 2 and parts[0] not in values, "ambiguous metrics snapshot") + value = Decimal(parts[1]) + _require(value.is_finite() and value >= 0 and value == int(value), "invalid counter") + values[parts[0]] = int(value) + _require(set(values) == required, "missing raw counter/resource sample") + _require( + all( + values[name] == 0 + for name in ( + "flare_active_requests", + "flare_retiring_request_slots", + "flare_pending_requests", + ) + ) + and values["flare_free_request_slots"] == 4, + "scheduler is not idle", + ) + return values + + +def adapt(packet_path: Path) -> dict: + manifest, files, identities = _load(packet_path) + corpus = json.loads(files[manifest["corpus"]]) + groups, cases = _corpus(corpus) + measurements = json.loads(files[manifest["measurements"]]) + memory = json.loads(files[manifest["memory_report"]]) + _require(measurements["backend"] == "flare", "only native Flare packets are supported") + _require(measurements["measurement"]["clock"] == UPSTREAM_CLOCK, "wrong clock domain") + capacity = _integer(memory["prefill_chunk_size"], "prepared prefill capacity", 1) + _require( + capacity in (512, 4096) and capacity == measurements["prefill_chunk_size"], + "prepared capacity mismatch", + ) + _require( + not measurements["errors"] and measurements["passed"] is True, + "qualification contains failed groups", + ) + _require( + [g["id"] for g in measurements["groups"]] == [g["id"] for g in groups], + "missing or changed measurement group", + ) + _require(all(g["passed"] is True for g in measurements["groups"]), "reported failed group") + observations = measurements["observations"] + _require( + sorted(o["case"] for o in observations) == sorted(cases), + "missing or duplicate raw observations", + ) + recomputed, samples = [], [] + for group in groups: + root = f"{manifest['run']}/{group['id']}" + before_name, after_name = manifest["snapshots"][group["id"]] + before, after = _counters(files[before_name]), _counters(files[after_name]) + results = [] + for name in group["cases"]: + observation = json.loads(files[f"{root}/{name}/observation.json"]) + _require(observation in observations, "observation sidecar differs") + request = json.loads(files[f"{root}/{name}/request.json"]) + expected_request = dict(cases[name]["request"]) + expected_request.update( + stream=True, + return_token_ids=True, + stream_options={"include_usage": True}, + return_token_timestamps=True, + ) + _require(request == expected_request, "sent request differs from frozen corpus") + _require( + observation["request_sha256"] == _sha(_encoded(request)), + "sent request fingerprint mismatch", + ) + result = _observation(observation, files[f"{root}/{name}/response.sse"], cases[name]) + results.append(result) + if group["kind"] == "ttft-512" and group["measured"]: + samples.append(result["delivery_ttft_ns"]) + deltas = {name: after[name] - before[name] for name in COUNTERS} + _require( + deltas[COUNTERS[0]] == sum(r["reused_prompt_tokens"] for r in results) + and deltas[COUNTERS[1]] == len(results) + and deltas[COUNTERS[2]] == deltas[COUNTERS[3]] == 0, + "request/reuse counter mismatch", + ) + for free, cached in ( + ("flare_free_kv_pages", "flare_cached_kv_pages"), + ("flare_free_snapshots", "flare_cached_checkpoints"), + ): + _require( + before[free] + before[cached] == after[free] + after[cached], + "resource conservation mismatch", + ) + recomputed.extend(results) + _require(len(samples) == corpus["repetitions"], "incomplete selected TTFT groups") + return { + "schema": "inferrsi-flare-delivery-receipt-v1", + "status": "validated", + "synthetic": manifest["synthetic"], + "source": SOURCE, + "metric_contract": METRIC, + "metrics": {"delivery_ttft_ns": statistics.median(samples)}, + "samples_ns": samples, + "identities": identities, + "package_sha256": identities["files"][manifest["package"]], + "corpus_sha256": identities["files"][manifest["corpus"]], + "prepared_capacity": capacity, + "observations": recomputed, + "adapter": { + "flare.py": artifacts.sha256(__file__), + "artifacts.py": artifacts.sha256(artifacts.__file__), + }, + "limits": "Offline delivery contract; task quality and service performance unverified.", + } + + +def validate_pair(baseline: Path, candidate: Path) -> dict: + before, after = adapt(baseline), adapt(candidate) + for key in ( + "source", + "metric_contract", + "synthetic", + "package_sha256", + "corpus_sha256", + "prepared_capacity", + ): + _require(before[key] == after[key], f"comparison identity mismatch: {key}") + return { + "schema": "inferrsi-flare-pair-receipt-v1", + "status": "validated", + "baseline": before, + "candidate": after, + "limits": "Pair consistency only; validation_command also needs task-quality gates.", + } + + +def main(argv=None): + parser = argparse.ArgumentParser( + description="Validate existing Flare delivery evidence offline." + ) + commands = parser.add_subparsers(dest="command", required=True) + measure = commands.add_parser("measure") + measure.add_argument("--packet", type=Path, required=True) + measure.add_argument("--peer", type=Path, required=True) + pair = commands.add_parser("validate-pair") + pair.add_argument("--baseline", type=Path, required=True) + pair.add_argument("--candidate", type=Path, required=True) + for command in (measure, pair): + command.add_argument("--output", type=Path, required=True) + args = parser.parse_args(argv) + inputs = ( + [args.packet, args.peer] if args.command == "measure" else [args.baseline, args.candidate] + ) + if any(args.output.resolve().is_relative_to(p.resolve().parent) for p in inputs): + parser.error("output must be outside every input packet directory") + try: + pair_receipt = validate_pair(*inputs) + if args.command == "measure": + receipt = pair_receipt["baseline"] + receipt["comparison_peer"] = pair_receipt["candidate"]["identities"] + else: + receipt = pair_receipt + except ( + OSError, + ValueError, + KeyError, + TypeError, + IndexError, + InvalidOperation, + OverflowError, + ) as exc: + receipt = { + "schema": "inferrsi-flare-rejection-v1", + "status": "rejected", + "error": str(exc), + "metrics": {}, + } + artifacts.write_json(args.output, receipt) + return 2 + artifacts.write_json(args.output, receipt) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/fixtures/flare_packet.py b/tests/fixtures/flare_packet.py new file mode 100644 index 00000000..eee3c8f9 --- /dev/null +++ b/tests/fixtures/flare_packet.py @@ -0,0 +1,251 @@ +import argparse +import hashlib +import json +import struct +from pathlib import Path + +from inferrsi.model.validators.flare import METRIC, SCHEMA, SOURCE, UPSTREAM_CLOCK + + +def encoded(value): + return json.dumps(value, ensure_ascii=False, separators=(",", ":"), allow_nan=False).encode() + + +def write_json(path, value): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_bytes(encoded(value) + b"\n") + + +def build_packet(root: Path, ttft_ns=100_000_000, repetitions=1): + root.mkdir(parents=True) + groups, cases = [], [] + + def add(name, kind, batch, repetition, measured): + names = [] + length = { + "ttft-4096": 4096, + "prefix-seed": 4096, + "prefix-hit": 4107, + "prefill-65536": 65536, + }.get(kind, 512) + output = 256 if kind.startswith("decode") else 1 + for lane in range(batch): + identifier = f"{name}-{lane}" + tokens = [1000 + len(cases)] * length + request = { + "model": "synthetic-fixture-model", + "max_tokens": output, + "temperature": 1, + "top_p": 0.95, + "top_k": 20, + "min_p": 0, + "repetition_penalty": 1, + "presence_penalty": 0, + "frequency_penalty": 0, + "seed": 42, + "messages": [{"role": "user", "content": identifier}], + "chat_template_kwargs": {"enable_thinking": False}, + "add_generation_prompt": True, + } + case = { + "id": identifier, + "group": name, + "request": request, + "prompt_tokens": tokens, + "rendered_prompt": identifier, + "expected_cache": "cold", + "output_tokens": output, + } + if kind == "prefix-hit": + seed = next(c for c in cases if c["group"] == f"prefix-seed-r{repetition}") + case.update(expected_cache="prefix-4096", prefix_seed=seed["id"]) + case["prompt_tokens"] = seed["prompt_tokens"] + [2222] * 11 + case["request"] = json.loads(encoded(seed["request"])) + case["request"]["messages"][-1]["content"] += " apple" * 11 + for key, data in ( + ("request_sha256", encoded(case["request"])), + ("prompt_tokens_sha256", struct.pack(f"<{length}I", *case["prompt_tokens"])), + ("rendered_prompt_sha256", case["rendered_prompt"].encode()), + ): + case[key] = hashlib.sha256(data).hexdigest() + cases.append(case) + names.append(identifier) + groups.append( + { + "id": name, + "kind": kind, + "batch": batch, + "repetition": repetition, + "measured": measured, + "cases": names, + } + ) + + add("warmup", "decode", 4, -1, False) + for repetition in range(repetitions): + for length in (512, 4096): + add(f"ttft-{length}-r{repetition}", f"ttft-{length}", 1, repetition, True) + for batch in (1, 2, 4): + add(f"decode-b{batch}-r{repetition}", f"decode-b{batch}", batch, repetition, True) + for kind in ("prefill-65536", "prefix-seed", "prefix-hit"): + add(f"{kind}-r{repetition}", kind, 1, repetition, kind != "prefix-seed") + corpus = { + "schema": "flare-http-performance-v1-corpus", + "model_id": "synthetic-fixture-model", + "revision": "synthetic-fixture-revision", + "repetitions": repetitions, + "nonce": "synthetic", + "groups": groups, + "cases": cases, + } + write_json(root / "corpus.json", corpus) + (root / "package.fixture").write_bytes(b"SYNTHETIC FIXTURE; NOT A MODEL OR RUNTIME PACKAGE\n") + write_json(root / "memory.json", {"prefill_chunk_size": 512}) + measurements = { + "backend": "flare", + "prefill_chunk_size": 512, + "measurement": {"clock": UPSTREAM_CLOCK}, + "groups": [], + "observations": [], + "errors": [], + "passed": True, + } + snapshots = {} + for group in groups: + folder = root / "run" / group["id"] + folder.mkdir(parents=True) + reused = 0 + for identifier in group["cases"]: + case = next(c for c in cases if c["id"] == identifier) + local = folder / identifier + local.mkdir() + output = list(range(case["output_tokens"])) + reused_tokens = 4096 if case["expected_cache"] == "prefix-4096" else 0 + reused += reused_tokens + # Two tokens arrive together in the first event of each decode request. + chunks = [output] if len(output) == 1 else [output[:2], output[2:]] + events, token_groups = [], [] + native = 1_000_000_000_000 + for index, tokens in enumerate(chunks): + elapsed = ttft_ns + index * 20_000_000 + payload = { + "usage": None, + "prompt_token_ids": None, + "performance": None, + "reused_prompt_tokens": None, + "choices": [ + { + "index": 0, + "token_ids": tokens, + "token_timestamps_ns": [native + t * 100 for t in tokens], + } + ], + } + events.append({"elapsed_ns": elapsed, "data": payload}) + token_groups.append({"timestamp_ns": elapsed, "token_ids": tokens}) + events.append( + { + "elapsed_ns": ttft_ns + 30_000_000, + "data": { + "prompt_token_ids": case["prompt_tokens"], + "choices": [{"index": 0, "finish_reason": "length"}], + "usage": { + "prompt_tokens": len(case["prompt_tokens"]), + "completion_tokens": len(output), + "total_tokens": len(case["prompt_tokens"]) + len(output), + "prompt_tokens_details": {"cached_tokens": reused_tokens}, + }, + "performance": { + "ingress_ns": native - 1000, + "enqueue_ns": native - 500, + "reused_prompt_tokens": reused_tokens, + }, + }, + } + ) + events.append({"elapsed_ns": ttft_ns + 40_000_000, "data": "[DONE]"}) + request = dict(case["request"]) + request.update( + stream=True, + return_token_ids=True, + return_token_timestamps=True, + stream_options={"include_usage": True}, + ) + observation = { + "case": identifier, + "request_sha256": hashlib.sha256(encoded(request)).hexdigest(), + "status": 200, + "request_start_perf_counter_ns": 99_000_000_000, + "events": events, + "analysis": { + "passed": True, + "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) + values = { + "flare_active_requests": 0, + "flare_free_request_slots": 4, + "flare_retiring_request_slots": 0, + "flare_pending_requests": 0, + "flare_free_kv_pages": 100, + "flare_cached_kv_pages": 10, + "flare_free_snapshots": 10, + "flare_cached_checkpoints": 2, + "flare_cache_reused_tokens_total": 0, + "flare_requests_completed_total": 0, + "flare_requests_failed_total": 0, + "flare_requests_cancelled_total": 0, + } + names = [] + for side in ("before", "after"): + if side == "after": + values.update( + flare_cache_reused_tokens_total=reused, + flare_requests_completed_total=group["batch"], + ) + name = f"run/{group['id']}/metrics-{side}-000.txt" + (root / name).write_bytes("".join(f"{k} {v}\n" for k, v in values.items()).encode()) + names.append(name) + snapshots[group["id"]] = names + measurements["groups"].append({"id": group["id"], "passed": True}) + write_json(root / "measurements.json", measurements) + manifest = { + "schema": SCHEMA, + "source": SOURCE, + "metric": METRIC, + "synthetic": True, + "package": "package.fixture", + "corpus": "corpus.json", + "memory_report": "memory.json", + "measurements": "measurements.json", + "run": "run", + "snapshots": snapshots, + "files": { + p.relative_to(root).as_posix(): hashlib.sha256(p.read_bytes()).hexdigest() + for p in sorted(root.rglob("*")) + if p.is_file() + }, + } + write_json(root / "packet.json", manifest) + return root / "packet.json" + + +if __name__ == "__main__": + parser = argparse.ArgumentParser( + description="Write a synthetic offline Flare contract fixture." + ) + parser.add_argument("output", type=Path) + args = parser.parse_args() + print(build_packet(args.output)) diff --git a/tests/test_flare_adapter.py b/tests/test_flare_adapter.py new file mode 100644 index 00000000..fd8372ae --- /dev/null +++ b/tests/test_flare_adapter.py @@ -0,0 +1,219 @@ +import hashlib +import json +import subprocess +import sys +from pathlib import Path + +import pytest +from fixtures.flare_packet import build_packet, write_json + +from inferrsi.harness.runner import run_experiment +from inferrsi.model.validators.flare import adapt, main, validate_pair +from inferrsi.runspec import RunSpec + + +@pytest.fixture +def packet(tmp_path): + return build_packet(tmp_path / "packet") + + +def edit(packet, name, change): + path = packet.parent / name + value = json.loads(path.read_bytes()) + change(value) + write_json(path, value) + manifest = json.loads(packet.read_bytes()) + manifest["files"][name] = hashlib.sha256(path.read_bytes()).hexdigest() + write_json(packet, manifest) + + +def edit_observation(packet, change): + name = "run/warmup/warmup-0/observation.json" + edit(packet, name, change) + observation = json.loads((packet.parent / name).read_bytes()) + edit(packet, "measurements.json", lambda m: m["observations"].__setitem__(0, observation)) + + +def test_complete_raw_packet_recomputes_metric_and_preserves_originals(packet, tmp_path): + 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 + assert len(receipt["observations"]) == 16 + warmup = receipt["observations"][0] + assert warmup["coalesced_groups"] == 2 + assert warmup["delivery_itl_exact"] is False + assert warmup["token_groups"][0]["token_ids"] == [0, 1] + assert all(p.read_bytes() == content for p, content in before.items()) + output = tmp_path / "receipt.json" + assert ( + main(["measure", "--packet", str(packet), "--peer", str(packet), "--output", str(output)]) + == 0 + ) + assert json.loads(output.read_bytes())["identities"]["manifest_sha256"] + + +@pytest.mark.parametrize( + "mutation", + [ + "missing_raw", + "missing_group", + "missing_observation", + "clock", + "capacity", + "counter", + "resource", + "coalesced", + "raw_grouping", + "client_failure", + "nonfinite", + "changed_corpus", + "failed_group", + "digest", + "source", + "manifest_clock", + ], +) +def test_semantic_and_incomplete_packets_reject_without_metric(packet, tmp_path, mutation): + if mutation == "missing_raw": + (packet.parent / "run/warmup/warmup-0/response.sse").unlink() + elif mutation == "missing_group": + edit(packet, "measurements.json", lambda m: m["groups"].pop()) + elif mutation == "missing_observation": + edit(packet, "measurements.json", lambda m: m["observations"].pop()) + elif mutation == "clock": + edit(packet, "measurements.json", lambda m: m["measurement"].update(clock="native")) + elif mutation == "capacity": + edit(packet, "memory.json", lambda m: m.update(prefill_chunk_size=4096)) + elif mutation in ("counter", "resource"): + name = "run/warmup/metrics-after-000.txt" + path = packet.parent / name + old, new = ( + ("flare_requests_completed_total 4", "flare_requests_completed_total 3") + if mutation == "counter" + else ("flare_free_kv_pages 100", "flare_free_kv_pages 99") + ) + path.write_bytes(path.read_bytes().replace(old.encode(), new.encode())) + manifest = json.loads(packet.read_bytes()) + manifest["files"][name] = hashlib.sha256(path.read_bytes()).hexdigest() + write_json(packet, manifest) + elif mutation == "coalesced": + edit_observation(packet, lambda o: o["analysis"].update(delivery_itl_exact=True)) + elif mutation == "raw_grouping": + edit_observation( + packet, lambda o: o["events"][0]["data"]["choices"][0].update(token_ids=[0]) + ) + elif mutation == "client_failure": + edit_observation(packet, lambda o: o.update(client_error="fixture client failure")) + elif mutation == "nonfinite": + edit_observation(packet, lambda o: o["events"][0].update(elapsed_ns=True)) + elif mutation == "changed_corpus": + edit(packet, "corpus.json", lambda c: c["groups"].pop()) + elif mutation == "failed_group": + edit(packet, "measurements.json", lambda m: m["errors"].append({"group": "warmup"})) + elif mutation == "digest": + (packet.parent / "memory.json").write_bytes(b"{}") + else: + manifest = json.loads(packet.read_bytes()) + if mutation == "source": + manifest["source"]["revision"] = "unknown" + else: + manifest["metric"]["clock_domain"] = "native" + write_json(packet, manifest) + 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["status"] == "rejected" + assert rejected["metrics"] == {} + assert rejected["error"] + + +def test_pair_requires_matching_prepared_capacity_and_identities(packet, tmp_path): + other = build_packet(tmp_path / "other", ttft_ns=80_000_000) + pair = validate_pair(packet, other) + assert pair["baseline"]["metrics"]["delivery_ttft_ns"] == 100_000_000 + assert pair["candidate"]["metrics"]["delivery_ttft_ns"] == 80_000_000 + edit(other, "memory.json", lambda m: m.update(prefill_chunk_size=4096)) + edit(other, "measurements.json", lambda m: m.update(prefill_chunk_size=4096)) + with pytest.raises(ValueError, match="comparison identity mismatch: prepared_capacity"): + validate_pair(packet, other) + + +def test_output_cannot_overwrite_original_packet(packet): + before = packet.read_bytes() + with pytest.raises(SystemExit): + main(["measure", "--packet", str(packet), "--peer", str(packet), "--output", str(packet)]) + assert packet.read_bytes() == before + + +def test_real_runner_uses_existing_validation_command_and_rejection_is_inconclusive( + packet, + tmp_path, + monkeypatch, +): + source = Path(__file__).resolve().parents[1] / "src" + monkeypatch.setenv("PYTHONPATH", str(source)) + repo = tmp_path / "repo" + repo.mkdir() + (repo / "source.txt").write_text("fixture source\n", encoding="utf-8") + for command in ( + ["git", "init", "--quiet"], + ["git", "add", "source.txt"], + [ + "git", + "-c", + "user.name=InferRSI fixture", + "-c", + "user.email=fixture@example.invalid", + "-c", + "commit.gpgsign=false", + "commit", + "--quiet", + "-s", + "-m", + "fixture", + ], + ): + subprocess.run(command, cwd=repo, check=True, capture_output=True) + candidate = build_packet(tmp_path / "candidate", ttft_ns=80_000_000) + cli = (sys.executable, "-m", "inferrsi.model.validators.flare") + spec = RunSpec( + repo=str(repo), + repetitions=1, + max_minutes=3, + output_dir=str(tmp_path / "runs"), + baseline_command=cli + + ("measure", "--packet", str(packet), "--peer", str(candidate), "--output", "{result}"), + candidate_command=cli + + ("measure", "--packet", str(candidate), "--peer", str(packet), "--output", "{result}"), + validation_command=cli + + ( + "validate-pair", + "--baseline", + str(packet), + "--candidate", + str(candidate), + "--output", + "{result}", + ), + metric_key="metrics.delivery_ttft_ns", + metric_unit="ns", + comparison_scope="flare_http_delivery_ttft_512", + ) + run = run_experiment(spec, {"backend": "cpu", "synthetic": True}) + evaluation = json.loads((run / "evaluation.json").read_bytes()) + assert evaluation["verdict"] == "PASS" + assert evaluation["improvement_pct"] == 20 + validation = json.loads((run / "validation/result.json").read_bytes()) + assert validation["schema"] == "inferrsi-flare-pair-receipt-v1" + edit(candidate, "measurements.json", lambda m: m["groups"].pop()) + run = run_experiment(spec, {"backend": "cpu", "synthetic": True}) + evaluation = json.loads((run / "evaluation.json").read_bytes()) + assert evaluation["verdict"] == "INCONCLUSIVE" + assert json.loads((run / "baseline/trial-001/result.json").read_bytes())["metrics"] == {}