diff --git a/benchmarks/inferswarm_97/__init__.py b/benchmarks/inferswarm_97/__init__.py new file mode 100644 index 000000000..172b53339 --- /dev/null +++ b/benchmarks/inferswarm_97/__init__.py @@ -0,0 +1,456 @@ +"""Issue #97 (InferSwarm): Gemma v4 physical calibration producer. + +Executes the accepted issue #95 v4 methodology +(inferswarm @ e12a6e3d5589044bace0c9555c0d364fb57a6229, +docs/qualification/gemma4-12b-it-v4/) on the frozen physical topology: + +- reference: inferswarm04 RTX 3090 24 GiB, matched single-GPU FreeToken + runtime (accepted R6 GemmaDenseStage, replay-prefill greedy); +- candidate: accepted three-stage RTX 3060 chain — inferswarm01 GPU-0 + (stage 1, layers [0,16)), inferswarm01 GPU-1 (stage 2, layers [16,32)), + inferswarm03 (stage 3/last, layers [32,48) via the #76 R4 wire service). + +The execution harness is the accepted #88 producer path, renamed only to bind +the v4 evidence identity. It adds ZERO model/execution math. + +1. the canonical frozen argmax/tie-break rule (ARGMAX_FIRST_MAX / + lowest-token-id-among-exactly-equal-fp32-maxima) applied identically + on both arms, with an executor proof-of-rule recorded per emitted + token (lowest-index-among-equal-maxima check computed on-device + against the FP32 row); +2. the frozen decision domain D(r) construction + (reference-top-1024-with-cutoff-ties/1) computed from reference rows + only, with canonical membership hashes; +3. candidate teacher-forcing against the exact canonical reference + prefix at each of all 8 decisions, with mechanical prefix-identity + proof before each execution; +4. retention of the candidate actual full-vocabulary FP32 winner per + canonical-prefix decision row (diagnostic row hashes, never free-run); +5. evidence-sufficient rows for E_full (full 15-envelope capture set is + unchanged from the #76 harness) and for decision_local_error over the + frozen D(r). + +Free-running post-branch tensors are diagnostic only (never calibration +or holdout evidence). The semantic adjudication itself (evaluate_decision, +threshold derivation, unseal preflight) lives in the accepted InferSwarm +CPU tooling and is NOT reimplemented here. + +Execution-branch discipline (issue #97 Phase A): this package and its +tests freeze as the physical implementation producer BEFORE the first +model execution; after that freeze no execution or model math may change +during the campaign. +""" + +from __future__ import annotations + +import json +import math +import subprocess +from pathlib import Path +from typing import Any, Sequence + +# Accepted v4 methodology identity (inferswarm PR #96 merge). +METHODOLOGY_COMMIT = "e12a6e3d5589044bace0c9555c0d364fb57a6229" +V4_ISSUE = 97 +CONTRACT_ID = "inferswarm.gemma4-prediction-aligned-qualification/1" + +# Frozen subject (issue #86 §"Qualification subject" == issue #88 §"Qualification subject"). +EXPECTED_CHECKPOINT_SHA256 = ( + "5a84cb313260ac447237b890387116dfa8682e49a6b44bc585ae8353abbff18d" +) +MODEL_REVISION = "707f0a3b8a3c7ad586ed01e27eafbad8a27dd0f7" + + +def validate_campaign_case_ids(cases: Sequence[dict[str, Any]]) -> list[dict[str, Any]]: + """Reject holdout, historical, and unversioned cases before CUDA work. + + Issue #97 permits only the public c95 statistical corpus and p95 stress + pool through the physical runners. The h95 holdout has a separate, + explicitly authorized path and is never accepted here. + """ + checked = list(cases) + for case in checked: + case_id = case.get("case_id") if isinstance(case, dict) else None + if not isinstance(case_id, str) or not case_id.startswith(("c95-", "p95-")): + raise ValueError( + f"issue #97 calibration accepts only c95-/p95- case IDs, got {case_id!r}" + ) + return checked + + +def frozen_subject_record(*, checkpoint_sha256: str, model_revision: str) -> dict[str, str]: + """Validate and serialize the immutable v4 physical subject identity.""" + if checkpoint_sha256 != EXPECTED_CHECKPOINT_SHA256: + raise ValueError("issue #97 checkpoint SHA-256 does not match the frozen subject") + if model_revision != MODEL_REVISION: + raise ValueError("issue #97 model revision does not match the frozen subject") + return { + "contract_id": CONTRACT_ID, + "methodology_commit": METHODOLOGY_COMMIT, + "checkpoint_sha256": EXPECTED_CHECKPOINT_SHA256, + "model_revision": MODEL_REVISION, + } + +GENERATED_TOKENS = 8 +CAPTURE_POSITIONS = (0, 1, 3, 7) +RUNTIME_CAPACITY_TOKENS = 64 # frozen single replay chunk bound + +# Frozen v3 semantic identities (byte-identical twins of +# scripts/issue86_v3_methodology.py constants; asserted by unit test +# against the committed inferswarm checkout when available). +ARGMAX_TIE_BREAK_IDENTITY = ( + "ARGMAX_FIRST_MAX/lowest-token-id-among-exactly-equal-fp32-maxima" +) +DECISION_DOMAIN_CONSTRUCTION = "reference-top-1024-with-cutoff-ties/1" +DECISION_DOMAIN_K = 1024 + + +def require_producer_identity(repo: Path, expected_sha: str | None) -> dict[str, Any]: + """Fail closed unless this clean tree is the externally frozen producer. + + ``expected_sha`` belongs in InferSwarm's retained execution authority, not + in the producer source. This check intentionally runs before callers + import torch or construct a runtime. + """ + if not isinstance(expected_sha, str) or len(expected_sha) != 40 or any( + char not in "0123456789abcdef" for char in expected_sha + ): + raise ValueError("issue #97 requires a 40-character lowercase expected producer SHA") + sha = subprocess.check_output( + ["git", "-c", f"safe.directory={repo}", "-C", str(repo), + "rev-parse", "HEAD"], text=True).strip() + status = subprocess.check_output( + ["git", "-c", f"safe.directory={repo}", "-C", str(repo), + "status", "--porcelain"], text=True) + if status: + raise ValueError("issue #97 producer source is dirty") + if sha != expected_sha: + raise ValueError( + f"issue #97 producer HEAD {sha!r} does not equal expected SHA {expected_sha!r}" + ) + return {"commit": sha, "dirty": False, "expected_commit": expected_sha} + + +def frozen_argmax_row(row: Sequence[float]) -> tuple[int, float]: + """Frozen rule twin: lowest token id among exactly equal FP32 maxima. + + Returns (winner_index, best_value). Pure host-side reference + implementation used to PROVE the executor's emitted token follows + the frozen rule; the proof compares the executor token against this + twin computed over the exact same FP32 row bytes. + """ + if not row: + raise ValueError("argmax requires a nonempty row") + best_index = 0 + best_value = float(row[0]) + if not math.isfinite(best_value): + raise ValueError("argmax requires finite logits") + for index in range(1, len(row)): + value = float(row[index]) + if not math.isfinite(value): + raise ValueError("argmax requires finite logits") + if value > best_value: + best_value = value + best_index = index + return best_index, best_value + + +def executor_rule_proof(row: Sequence[float], emitted_token: int) -> dict[str, Any]: + """Proof-of-rule for ONE emitted decision token. + + Proves the executor's emitted token equals the frozen-rule winner of + the exact FP32 row, and that the winner is the LOWEST index among + all exactly-equal FP32 maxima. Any violation is a rule failure, not + a tolerance. + """ + winner, best_value = frozen_argmax_row(row) + equal_max_indices = [ + i for i, v in enumerate(row) + if float(v) == best_value and math.isfinite(float(v)) + ] + ok = int(emitted_token) == winner + return { + "emitted_token": int(emitted_token), + "rule_winner_token": winner, + "rule_ok": ok, + "tie_count": len(equal_max_indices), + "lowest_index_among_equal_maxima": ( + equal_max_indices[0] if equal_max_indices else winner + ), + "tie_break_identity": ARGMAX_TIE_BREAK_IDENTITY, + "rule": "emitted==ARGMAX_FIRST_MAX(winner); winner is the lowest " + "index among exactly-equal FP32 maxima", + } + + +def decision_domain_row(row: Sequence[float], k: int = DECISION_DOMAIN_K) -> dict[str, Any]: + """D(r) twin: every token with reference logit >= k-th-highest cutoff. + + Returns membership (ascending token ids), size, cutoff value, and + the canonical membership sha256 (sorted-id canonical JSON) exactly + as the accepted InferSwarm tooling hashes it. Reference-derived + only; no candidate input. + """ + import hashlib + + if k <= 0: + raise ValueError("decision-domain K must be positive") + values = [float(v) for v in row] + if not values or not all(math.isfinite(v) for v in values): + raise ValueError("reference logits must be nonempty and finite") + cutoff = sorted(values, reverse=True)[min(k, len(values)) - 1] + domain = tuple(i for i, v in enumerate(values) if v >= cutoff) + if not domain: + raise ValueError("decision domain construction produced an empty set") + membership_bytes = ( + json.dumps(list(domain), ensure_ascii=False, sort_keys=True, + separators=(",", ":")) + "\n" + ).encode() + return { + "membership": list(domain), + "domain_size": len(domain), + "cutoff_hex": cutoff.hex(), + "cutoff_rank": min(k, len(values)), + "domain_membership_sha256": hashlib.sha256(membership_bytes).hexdigest(), + "construction": DECISION_DOMAIN_CONSTRUCTION, + "k": k, + } + + +def prefix_sha256(prefix: Sequence[int]) -> str: + """Canonical prefix hash used by EVERY arm and the assembler. + + sha256 over compact canonical JSON of the exact token-id list. The + reference runner, the candidate runner, and the CPU assembler all + call this ONE function, so prefix identity is byte-comparable + across arms. + """ + import hashlib + + payload = ( + json.dumps([int(t) for t in prefix], ensure_ascii=False, + separators=(",", ":")) + "\n" + ).encode() + return hashlib.sha256(payload).hexdigest() + + +def canonical_prefix(token_ids: Sequence[int], reference_generated: Sequence[int], + decision_index: int) -> list[int]: + """The exact canonical prefix for decision ``decision_index`` (0-based). + + prompt tokens + reference-generated tokens [0, decision_index). The + candidate must consume EXACTLY this prefix (teacher forcing); the + identity proof hashes the prefix bytes. + """ + import hashlib + + if not 0 <= decision_index < GENERATED_TOKENS: + raise ValueError(f"decision index out of range: {decision_index}") + if len(reference_generated) != GENERATED_TOKENS: + raise ValueError("canonical prefix requires the complete 8-decision " + "reference trajectory") + prefix = list(token_ids) + [ + int(t) for t in reference_generated[:decision_index] + ] + return prefix + + +def prefix_identity_proof(prefix: Sequence[int]) -> dict[str, Any]: + """Mechanical prefix-identity proof (sha256 over canonical token JSON).""" + return { + "prefix_len": len(prefix), + "prefix_sha256": prefix_sha256(prefix), + } + + +def decision_row_evidence( + *, + decision_index: int, + prefix: Sequence[int], + domain_info: dict[str, Any], + emitted_token: int, + rule_proof: dict[str, Any], +) -> dict[str, Any]: + """One canonical-prefix decision evidence row (harness-side). + + Binds prefix identity, domain membership, the emitted winner and its + rule proof. The FP32 candidate row itself stays in the retained + capture bundle; the row records its sha256 for later + decision_local_error derivation on the CPU assembler. + """ + if not rule_proof["rule_ok"]: + raise ValueError( + f"decision {decision_index}: emitted token violates the frozen " + "argmax/tie-break rule" + ) + proof = prefix_identity_proof(prefix) + return { + "decision_index": decision_index, + "prefix_len": proof["prefix_len"], + "prefix_sha256": proof["prefix_sha256"], + "domain_membership_sha256": domain_info["domain_membership_sha256"], + "domain_size": domain_info["domain_size"], + "domain_cutoff_hex": domain_info["cutoff_hex"], + "emitted_token": int(emitted_token), + "emitted_rule": ARGMAX_TIE_BREAK_IDENTITY, + "rule_proof": rule_proof, + } + + +def assert_teacher_forcing( + *, prefix: Sequence[int], reference_decision: dict[str, Any] +) -> None: + """Prove prefix identity BEFORE candidate execution (fail closed). + + The candidate replay prefix for decision i must be byte-identical + (same canonical sha256 and length) to the reference runner's recorded + prefix for that decision. + """ + expected_len = int(reference_decision["prefix_len"]) + expected_sha = reference_decision["prefix_sha256"] + observed_len = len(prefix) + observed_sha = prefix_sha256(prefix) + if observed_len != expected_len or observed_sha != expected_sha: + raise ValueError( + f"teacher-forcing prefix identity failure: decision " + f"{reference_decision.get('decision_index')}: len " + f"{observed_len}!={expected_len} or sha " + f"{observed_sha}!={expected_sha}" + ) + + +def build_reference_case_summary( + *, + case: dict[str, Any], + generated: Sequence[int], + margins: Sequence[dict[str, Any]], + decision_rows: Sequence[dict[str, Any]], + nan_inf_total: int, + capture_manifest: dict[str, Any], + producer: dict[str, Any], + gpu_uuid: str, + tag: str, + attempt_id: str, + wall_seconds: float, + capture_positions: Sequence[int] = CAPTURE_POSITIONS, +) -> dict[str, Any]: + """Assemble one v3 reference case summary binding exact identity. + + Pure (no torch): unit-tested for identity/provenance binding. + """ + if len(generated) != GENERATED_TOKENS: + raise ValueError("reference case must emit exactly 8 decisions") + if len(decision_rows) != GENERATED_TOKENS: + raise ValueError("reference case must retain exactly 8 decision rows") + indices = sorted(d["decision_index"] for d in decision_rows) + if indices != list(range(GENERATED_TOKENS)): + raise ValueError("decision rows must be emitted exactly once per index") + return { + "schema": "inferswarm.issue97.v4-reference-case/1", + "contract_id": CONTRACT_ID, + "attempt_id": attempt_id, + "case_id": case["case_id"], + "case_sha256": case["case_sha256"], + "prompt_sha256": case["prompt_sha256"], + "token_ids_sha256": case["token_ids_sha256"], + "generated_token_ids": [int(t) for t in generated], + "step_margins": list(margins), + "min_top1_margin_hex": min( + float.fromhex(m["margin_hex"]) for m in margins).hex(), + "nan_inf_count": int(nan_inf_total), + "decision_domain_construction": DECISION_DOMAIN_CONSTRUCTION, + "argmax_tie_break": ARGMAX_TIE_BREAK_IDENTITY, + "decisions": [dict(d) for d in decision_rows], + "producer": producer, + "gpu_uuid": gpu_uuid, + "role": "reference-single", + "capture_positions": list(capture_positions), + "capture_manifest": capture_manifest, + "wall_seconds": wall_seconds, + } + + +def build_chain_case_summary( + *, + case: dict[str, Any], + reference_case: dict[str, Any], + decision_rows: Sequence[dict[str, Any]], + margins: Sequence[dict[str, Any]], + nan_inf_total: int, + capture_manifests: dict[str, Any], + producer: dict[str, Any], + tag: str, + attempt_id: str, + wall_seconds: float, + capture_positions: Sequence[int] = CAPTURE_POSITIONS, +) -> dict[str, Any]: + """Assemble one v3 candidate (chain) case summary. + + ``decision_rows`` carry the ACTUAL candidate full-vocabulary winner + per canonical-prefix decision under the frozen rule (with rule + proofs), plus the reference prefix binding. The candidate generated + trajectory is NOT free-run: this summary records the reference + trajectory it was forced against for audit. + """ + if reference_case["case_id"] != case["case_id"]: + raise ValueError("reference/candidate case identity mismatch") + if reference_case["case_sha256"] != case["case_sha256"]: + raise ValueError("reference/candidate case hash mismatch") + if len(decision_rows) != GENERATED_TOKENS: + raise ValueError("candidate case must retain exactly 8 decision rows") + indices = sorted(d["decision_index"] for d in decision_rows) + if indices != list(range(GENERATED_TOKENS)): + raise ValueError("decision rows must be emitted exactly once per index") + for row, ref_row in zip( + decision_rows, + sorted(reference_case["decisions"], key=lambda d: d["decision_index"]), + ): + if row["prefix_sha256"] != ref_row["prefix_sha256"]: + raise ValueError( + f"decision {row['decision_index']}: candidate prefix does " + "not match the reference canonical prefix" + ) + return { + "schema": "inferswarm.issue97.v4-chain-case/1", + "contract_id": CONTRACT_ID, + "attempt_id": attempt_id, + "case_id": case["case_id"], + "case_sha256": case["case_sha256"], + "prompt_sha256": case["prompt_sha256"], + "token_ids_sha256": case["token_ids_sha256"], + "reference_forced_trajectory": list( + reference_case["generated_token_ids"]), + "step_margins": list(margins), + "nan_inf_count": int(nan_inf_total), + "argmax_tie_break": ARGMAX_TIE_BREAK_IDENTITY, + "decisions": [dict(d) for d in decision_rows], + "producer": producer, + "role": "candidate-chain", + "capture_positions": list(capture_positions), + "capture_manifests": capture_manifests, + "wall_seconds": wall_seconds, + } + + +def canonical_json_bytes(value: Any) -> bytes: + """Byte-identical twin of the frozen inferswarm canonical JSON.""" + return ( + json.dumps(value, ensure_ascii=False, sort_keys=True, + separators=(",", ":")) + "\n" + ).encode() + + +def sha256_bytes(data: bytes) -> str: + import hashlib + + return hashlib.sha256(data).hexdigest() + + +def write_json_with_sha(path: Path, value: Any) -> dict[str, Any]: + """Write canonical JSON + record its sha256 (torch-free sidecar pair).""" + payload = canonical_json_bytes(value) + out = Path(path) + if out.exists(): + raise SystemExit(f"refusing to overwrite {out}") + out.write_bytes(payload) + return {"path": str(out), "sha256": sha256_bytes(payload), "bytes": len(payload)} diff --git a/benchmarks/inferswarm_97/chain_runner.py b/benchmarks/inferswarm_97/chain_runner.py new file mode 100644 index 000000000..c5a6545c6 --- /dev/null +++ b/benchmarks/inferswarm_97/chain_runner.py @@ -0,0 +1,299 @@ +"""#97 v4 chain runner: teacher-forced three-stage RTX 3060 candidate. + +Drives the accepted chain topology (stage 1 = node-01 GPU-0 [0,16), +stage 2 = node-01 GPU-1 [16,32), stage 3 = node-03 last stage via the +#76 R4 wire service) with the v3 canonical-prefix contract: + +At each of all 8 decisions the candidate consumes EXACTLY the frozen +reference prefix for that decision (proved byte-identical via +``assert_teacher_forcing`` BEFORE execution), and the ACTUAL candidate +full-vocabulary FP32 winner for that same canonical-prefix row is +retained (row persisted as decision-.f32 on the last-stage node, +sha256-bound in the case summary, with an executor rule proof under the +frozen argmax/tie-break rule). + +Free-running candidate continuation is never executed: the candidate is +teacher-forced at every decision, so every retained row is a +canonical-prefix row. The 15-envelope checkpoint capture at positions +0/1/3/7 is unchanged from the #76 harness. + +Run from a #97 worktree on inferswarm01 with the #97 last-stage service +already listening on inferswarm03. +""" + +from __future__ import annotations + +import argparse +import json +import time +from pathlib import Path + +from benchmarks.inferswarm_76 import verify_case_identity +from benchmarks.inferswarm_76.reference_runner import resolve_cases +from benchmarks.inferswarm_97 import ( + ARGMAX_TIE_BREAK_IDENTITY, + CONTRACT_ID, + GENERATED_TOKENS, + assert_teacher_forcing, + build_chain_case_summary, + executor_rule_proof, + frozen_subject_record, + prefix_sha256, + require_producer_identity, + validate_campaign_case_ids, +) + + +def _load_reference_case(path: Path, case: dict, subject: dict) -> dict: + reference = json.loads(path.read_text()) + if reference.get("schema") != "inferswarm.issue97.v4-reference-case/1": + raise SystemExit(f"{path}: not a v4 reference case summary") + if reference.get("contract_id") != CONTRACT_ID: + raise SystemExit(f"{path}: reference accepted-contract mismatch") + if reference.get("subject") != subject: + raise SystemExit(f"{path}: reference frozen-subject mismatch") + for field in ("case_id", "case_sha256", "prompt_sha256", + "token_ids_sha256"): + if reference[field] != case[field]: + raise SystemExit( + f"{path}: reference {field} mismatch for {case['case_id']}") + return reference + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--model", required=True) + parser.add_argument("--plan", required=True) + parser.add_argument("--corpus", required=True) + parser.add_argument("--resolve-corpus", default=None) + parser.add_argument("--case-ids", default=None) + parser.add_argument("--reference-dir", required=True, + help="v3 reference run root (per-case dirs with " + "reference-case-.json)") + parser.add_argument("--reference-tag", required=True) + parser.add_argument("--last-stage-host", default="10.0.0.219") + parser.add_argument("--last-stage-port", type=int, default=18485) + parser.add_argument("--out-dir", required=True) + parser.add_argument("--tag", required=True) + parser.add_argument("--attempt-id", required=True) + parser.add_argument("--expected-producer-sha", required=True) + parser.add_argument("--checkpoint-sha256", required=True) + parser.add_argument("--model-revision", required=True) + args = parser.parse_args(argv) + + import multiprocessing + + repo = Path(__file__).resolve().parents[2] + producer = require_producer_identity(repo, args.expected_producer_sha) + + subject = frozen_subject_record( + checkpoint_sha256=args.checkpoint_sha256, model_revision=args.model_revision + ) + cases = validate_campaign_case_ids(resolve_cases(args.corpus, args.resolve_corpus)) + if args.case_ids: + wanted = set(args.case_ids.split(",")) + cases = [c for c in cases if c["case_id"] in wanted] + missing = wanted - {c["case_id"] for c in cases} + if missing: + raise SystemExit(f"unknown case ids: {sorted(missing)}") + + from benchmarks.inferswarm_76.stage_entry import I76StageClient + from benchmarks.inferswarm_76.wire_client import I76LastStageClient + + plan = json.loads(Path(args.plan).read_text()) + shared = plan.get("declared_shared_state") + + context = multiprocessing.get_context("spawn") + stages = [] + try: + for index, block in enumerate(plan["blocks"][:-1]): + stages.append( + I76StageClient( + context, + role="first" if index == 0 else "middle", + adapter_data={ + **block, + "declared_shared_state": shared if index == 0 else None, + "runtime_capacity_tokens": 64, + }, + model_path=args.model, + gpu_index=index, + ) + ) + stages.append( + I76LastStageClient( + host=args.last_stage_host, + port=args.last_stage_port, + experiment_id=plan["digest"], + connect_timeout=600.0, + ) + ) + for stage in stages[:-1]: + ready = stage.recv() + if ready.get("op") == "ERROR": + raise RuntimeError(f"stage failed: {ready}") + + results = [] + out_root = Path(args.out_dir) + out_root.mkdir(parents=True, exist_ok=True) + + for case in cases: + case_dir = out_root / case["case_id"] + case_dir.mkdir(parents=True, exist_ok=True) + ref_path = (Path(args.reference_dir) / case["case_id"] / + f"reference-case-{args.reference_tag}.json") + reference = _load_reference_case(ref_path, case, subject) + ref_decisions = sorted( + reference["decisions"], key=lambda d: d["decision_index"]) + forced = list(reference["generated_token_ids"]) + t0 = time.perf_counter() + + # arm per-case capture on stages 1-2 (fresh sink per case) + import subprocess + + gpu_uuids = [ + subprocess.check_output( + ["nvidia-smi", "-i", str(i), "--query-gpu=uuid", + "--format=csv,noheader"], text=True).strip() + for i in (0, 1) + ] + for stage_index, (stage, uuid) in enumerate( + zip(stages[:-1], gpu_uuids) + ): + stage.request({ + "op": "CASE_ARM", + "out_dir": str(case_dir), + "tag": args.tag, + "gpu_uuid": uuid, + "after_layers": [15] if stage_index == 0 else [31], + }) + stages[-1].request({"op": "CASE_BEGIN", "case_id": case["case_id"]}) + + prompt = list(case["token_ids"]) + margins: list[dict] = [] + decision_rows: list[dict] = [] + nan_inf_total = 0 + for step in range(GENERATED_TOKENS): + # --- v3 teacher forcing: exact reference prefix ---------- + replay = prompt + [int(t) for t in forced[:step]] + assert_teacher_forcing( + prefix=replay, reference_decision=ref_decisions[step]) + + for stage in stages: + stage.request({"op": "RESET"}) + # every decision is evidence-bearing in v3: capture_step is + # sent on ALL 8 decisions (the last-stage service keys its + # retained decision-.f32 rows off this value), while the + # 15-envelope capture positions remain the frozen 0/1/3/7. + hidden = None + response: dict = {} + for index, stage in enumerate(stages): + if index == 0: + response = stage.request({ + "op": "PREFILL", + "token_ids": replay, + "position": 0, + "capture_step": step, + }) + hidden = response.get("hidden") + else: + response = stage.request({ + "op": "PREFILL", + "hidden": hidden, + "position": 0, + "capture_step": step, + }) + hidden = response.get("hidden") + token = response["token_id"] + row_sha = response["row_f32_sha256"] + row_count = response["row_element_count"] + rule = response["rule_proof"] + if response.get("top1_index") is None: + raise RuntimeError( + "last-stage response missing margin diagnostics") + decision_rows.append({ + "decision_index": step, + "prefix_len": len(replay), + "prefix_sha256": prefix_sha256(replay), + "emitted_token": int(token), + "emitted_rule": ARGMAX_TIE_BREAK_IDENTITY, + "row_f32_sha256": row_sha, + "row_element_count": row_count, + "rule_proof": rule, + "row_retained_at": "last-stage-node", + }) + margin = { + "step": step, + "top1_index": response["top1_index"], + "top1_value_hex": response["top1_value_hex"], + "top2_index": response["top2_index"], + "top2_value_hex": response["top2_value_hex"], + "margin_hex": response["margin_hex"], + "nan_inf_count": response["nan_inf_count"], + } + margins.append(margin) + nan_inf_total += int(response["nan_inf_count"]) + + # persist per-case captures from stages 1-2 and the remote stage + manifests = {} + for stage_index, stage in enumerate(stages[:-1]): + ack = stage.request( + {"op": "SAVE_CAPTURE", + "suffix": f"{args.tag}-stage{stage_index + 1}"}) + manifests[f"stage{stage_index + 1}"] = ack["manifest"] + save_ack = stages[-1].request( + {"op": "CASE_SAVE", "tag": args.tag}) + manifests["stage3"] = save_ack["manifest"] + + summary = build_chain_case_summary( + case=case, + reference_case=reference, + decision_rows=decision_rows, + margins=margins, + nan_inf_total=nan_inf_total, + capture_manifests=manifests, + producer=producer, + tag=args.tag, + attempt_id=args.attempt_id, + wall_seconds=time.perf_counter() - t0, + ) + summary["subject"] = subject + path = case_dir / f"chain-case-{args.tag}.json" + if path.exists(): + raise SystemExit(f"refusing to overwrite {path}") + path.write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n") + results.append(summary) + print(json.dumps({ + "case_id": case["case_id"], "status": "CASE_COMMITTED", + "nan_inf": nan_inf_total}), flush=True) + + index = { + "schema": "inferswarm.issue97.v4-chain-run-index/1", + "contract_id": CONTRACT_ID, + "subject": subject, + "attempt_id": args.attempt_id, + "producer": producer, + "tag": args.tag, + "case_count": len(results), + "cases": [ + {"case_id": r["case_id"], "case_sha256": r["case_sha256"], + "nan_inf_count": r["nan_inf_count"]} + for r in results + ], + } + index_path = out_root / f"index-{args.tag}.json" + if index_path.exists(): + raise SystemExit(f"refusing to overwrite {index_path}") + index_path.write_text(json.dumps(index, indent=2, sort_keys=True) + "\n") + print(json.dumps({"status": "RUN_COMPLETE", "cases": len(results)})) + return 0 + finally: + for stage in stages: + try: + stage.shutdown() + except Exception: # noqa: BLE001 + pass + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmarks/inferswarm_97/last_stage_service.py b/benchmarks/inferswarm_97/last_stage_service.py new file mode 100644 index 000000000..3073dba8f --- /dev/null +++ b/benchmarks/inferswarm_97/last_stage_service.py @@ -0,0 +1,328 @@ +"""#97 v4 remote last-stage service (inferswarm03) over the accepted R4 wire. + +Identical framing, boundary contract, and execution semantics to the #76 +last-stage service. Differences (all evidence-side, no execution math): + +- at EVERY prefill the final-row FP32 consumer logits are retained as + ``//decision-.f32`` (the #88 chain + runner sets ``capture_step`` on every decision, so the file index IS + the decision index) with sha256 + element count + a frozen-rule + executor proof returned in the TOKEN_RESULT response; +- the external `--expected-producer-sha` is checked against a clean running + tree before any CUDA initialization; the historical plan producer is retained + only as explicit issue-97 override provenance. + +The service speaks the same request ops as the #76 service: + CASE_BEGIN {case_id} -> swaps in a fresh sink for the case + CASE_SAVE {tag} -> persists the case bundle, returns manifest +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import socket +import subprocess +import time +from pathlib import Path + +from freetoken.research.r4_wire import ( + WireError, + encode_frame, + payload_checksum, + recv_frame, + send_exact, + validate_request, +) + +WIRE_PROTOCOL_ID = "inferswarm.r4.boundary-wire/1" +MAX_TOKEN_COUNT = 64 # matches strategy PREFILL_CHUNK (single-chunk replays) +ROW_WIDTH = 3840 +BOUNDARY_CONTRACT = { + "dtype": "bfloat16", + "layout": "plane-major-contiguous", + "planes": 1, + "row_width": ROW_WIDTH, + "element_bytes": 2, + "max_token_count": MAX_TOKEN_COUNT, +} + + +def serve( + *, + listen_host: str, + listen_port: int, + participant_plan: str, + model_path: str, + gpu_uuid: str, + diagnostic: bool, + ready_file: str | None = None, + out_dir: str, + expected_producer_sha: str, + checkpoint_sha256: str, + model_revision: str, +) -> None: + from benchmarks.inferswarm_97 import frozen_subject_record, require_producer_identity + subject = frozen_subject_record( + checkpoint_sha256=checkpoint_sha256, model_revision=model_revision + ) + repo_root = Path(__file__).resolve().parents[2] + producer = require_producer_identity(repo_root, expected_producer_sha) + plan = json.loads(Path(participant_plan).read_text()) + plan_producer = plan.get("provenance", {}).get("r6", {}).get("producer_sha") + producer_check = { + "mode": "EXPLICIT_OVERRIDE_ISSUE97_EXECUTION" if plan_producer != producer["commit"] else "PLAN_FROZEN", + "plan_frozen_producer": plan_producer, + "running_producer": producer["commit"], + "expected_producer": expected_producer_sha, + "reason": "issue #97 externally pinned producer supersedes the historical R6 plan producer" if plan_producer != producer["commit"] else "plan producer matches the externally pinned issue #97 producer", + } + + os.environ["CUDA_VISIBLE_DEVICES"] = gpu_uuid + import torch + + from freetoken.distributed import set_tp_info, try_get_tp_info + from freetoken.layers.rotary import set_rope_device + + if try_get_tp_info() is None: + set_tp_info(rank=0, size=1) + set_rope_device(torch.device("cuda:0")) + + from benchmarks.inferswarm_76.capture import RowPruningSink, arm_full_capture + from benchmarks.inferswarm_r6.stage_runtime import GemmaDenseStage + + block = plan["blocks"][-1] + runtime = GemmaDenseStage( + role="last", + model_path=model_path, + adapter_data={ + **block, + "runtime_capacity_tokens": int( + plan.get("runtime_capacity_tokens", 256) + ), + "declared_shared_state": plan.get("declared_shared_state"), + }, + ) + + sink: RowPruningSink | None = None + case_dir_root = Path(out_dir) + case_dir_root.mkdir(parents=True, exist_ok=True) + current_case_dir: Path | None = None + current_case_id: str | None = None + + def _arm(case_id: str) -> None: + nonlocal sink, current_case_dir, current_case_id + current_case_id = case_id + current_case_dir = case_dir_root / case_id + current_case_dir.mkdir(parents=True, exist_ok=True) + sink = RowPruningSink(role="last", gpu_uuid=gpu_uuid) + runtime._capture_sink = sink + runtime._capture_after_layers = frozenset() + + buffer_bytes = MAX_TOKEN_COUNT * ROW_WIDTH * 2 + host_u8 = torch.empty(buffer_bytes, dtype=torch.uint8) + stats = {"boundaries_served": 0, "activation_bytes_rx": 0, + "result_bytes_tx": 0, "cases_saved": 0, + "decision_rows_retained": 0} + + server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + server.bind((listen_host, listen_port)) + server.listen(1) + if ready_file: + Path(ready_file).write_text(json.dumps({ + "plan_digest": plan.get("digest"), + "stage": "last", + "gpu_uuid": gpu_uuid, + "listen": [listen_host, listen_port], + "pid": os.getpid(), + "producer_freetoken_sha": producer["commit"], + "subject": subject, + "producer_check": producer_check, + "runtime": runtime.report("P4_ready_for_resident_execution"), + })) + experiment_id = plan.get("digest") or "r6-dense-chain" + try: + conn, _addr = server.accept() + conn.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + session_id = None + while True: + identity = {"protocol": WIRE_PROTOCOL_ID, "experiment_id": experiment_id} + header, payload = recv_frame(conn, identity) + kind = header.get("kind") + if kind == "hello": + session_id = header["session_id"] + runtime.reset_session_state() + response = { + "kind": "response", "protocol": WIRE_PROTOCOL_ID, + "experiment_id": experiment_id, "session_id": session_id, + "op": "HELLO_ACK", "runtime_ready": True, + } + send_exact(conn, encode_frame(response)) + continue + if kind != "request" or session_id is None: + raise WireError(f"unexpected frame kind {kind!r} before session") + if header.get("session_id") != session_id: + raise WireError("session identity mismatch") + op = header["op"] + if op == "OPEN_SESSION": + runtime.reset_session_state() + response = { + "kind": "response", "protocol": WIRE_PROTOCOL_ID, + "experiment_id": experiment_id, "session_id": session_id, + "op": "SESSION_ACK", + } + send_exact(conn, encode_frame(response)) + continue + if op == "CASE_BEGIN": + _arm(header["case_id"]) + runtime.reset_session_state() + response = { + "kind": "response", "protocol": WIRE_PROTOCOL_ID, + "experiment_id": experiment_id, "session_id": session_id, + "op": "CASE_ACK", "case_id": current_case_id, + } + send_exact(conn, encode_frame(response)) + continue + if op == "CASE_SAVE": + if sink is None or current_case_dir is None: + raise WireError("CASE_SAVE before CASE_BEGIN") + manifest = sink.save(str(current_case_dir), header["tag"]) + stats["cases_saved"] += 1 + response = { + "kind": "response", "protocol": WIRE_PROTOCOL_ID, + "experiment_id": experiment_id, "session_id": session_id, + "op": "SAVE_ACK", "case_id": current_case_id, + "manifest": manifest, + } + send_exact(conn, encode_frame(response)) + continue + if op != "BOUNDARY": + raise WireError(f"unsupported request op {op!r}") + token_count = int(header["token_count"]) + validate_request( + header, + contract=BOUNDARY_CONTRACT, + checksum=payload_checksum(payload) if diagnostic else None, + payload=payload, + ) + hidden = ( + torch.frombuffer(bytearray(payload), dtype=torch.uint8) + .view(torch.bfloat16) + .reshape(token_count, ROW_WIDTH) + .to(device="cuda:0", non_blocking=False) + ) + capture_step = header.get("capture_step") + if capture_step is not None: + runtime._capture_step = int(capture_step) + if header["operation"] == "prefill": + token, logits = runtime.prefill(None, hidden, int(header["position"])) + else: + if token_count != 1: + raise WireError("decode boundary requires exactly one token") + token, logits = runtime.decode(hidden, int(header["position"])) + runtime._capture_step = None + row = logits + nan_inf = int(torch.isnan(row).sum().item() + + torch.isinf(row).sum().item()) + top2 = torch.topk(row, 2) + + # --- #88 v3: retain EVERY decision's FP32 row + rule proof --- + rule_proof = None + row_sha = None + row_count = None + if header["operation"] == "prefill" and capture_step is not None: + from benchmarks.inferswarm_97 import executor_rule_proof + + host_row = row.detach().to("cpu", torch.float32).contiguous() + values = host_row.tolist() + rule_proof = executor_rule_proof(values, int(token)) + row_bytes = host_row.view(torch.uint8).numpy().tobytes() + row_sha = hashlib.sha256(row_bytes).hexdigest() + row_count = len(values) + if current_case_dir is None: + raise WireError("decision row before CASE_BEGIN") + row_path = current_case_dir / f"decision-{int(capture_step)}.f32" + if row_path.exists(): + raise WireError(f"refusing to overwrite {row_path}") + row_path.write_bytes(row_bytes) + stats["decision_rows_retained"] += 1 + del host_row, values, row_bytes + + stats["boundaries_served"] += 1 + stats["activation_bytes_rx"] += len(payload) + response = { + "kind": "response", "protocol": WIRE_PROTOCOL_ID, + "experiment_id": experiment_id, "session_id": session_id, + "op": "TOKEN_RESULT", + "token_id": int(token), + "top1_index": int(top2.indices[0].item()), + "top1_value_hex": float(top2.values[0].item()).hex(), + "top2_index": int(top2.indices[1].item()), + "top2_value_hex": float(top2.values[1].item()).hex(), + "margin_hex": float( + top2.values[0].item() - top2.values[1].item()).hex(), + "nan_inf_count": nan_inf, + "row_f32_sha256": row_sha, + "row_element_count": row_count, + "rule_proof": rule_proof, + "consumer_sha256": payload_checksum(payload), + "compute_ns": 0, + } + frame = encode_frame(response) + send_exact(conn, frame) + stats["result_bytes_tx"] += len(frame) + del hidden + finally: + try: + Path(os.environ.get( + "I97_LAST_STAGE_FINAL_REPORT", + "/tmp/i97-last-stage.json")).write_text(json.dumps({ + "schema": "inferswarm.issue97.last-stage-final-report/1", + "plan_digest": plan.get("digest"), + "producer_freetoken_sha": producer["commit"], + "subject": subject, + "producer_check": producer_check, + "stats": stats, + "runtime": runtime.report("P5_post_run"), + })) + except Exception: # noqa: BLE001 + pass + server.close() + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--listen-host", default="0.0.0.0") + parser.add_argument("--listen-port", type=int, default=18485) + parser.add_argument("--participant-plan", required=True) + parser.add_argument("--model", required=True) + parser.add_argument("--gpu-uuid", required=True) + parser.add_argument("--diagnostic", action="store_true") + parser.add_argument("--ready-file") + parser.add_argument("--out-dir", required=True, + help="root dir for per-case capture bundles + rows") + parser.add_argument("--checkpoint-sha256", required=True) + parser.add_argument("--model-revision", required=True) + parser.add_argument("--expected-producer-sha", required=True) + args = parser.parse_args(argv) + serve( + listen_host=args.listen_host, + listen_port=args.listen_port, + participant_plan=args.participant_plan, + model_path=args.model, + gpu_uuid=args.gpu_uuid, + diagnostic=args.diagnostic, + ready_file=args.ready_file, + out_dir=args.out_dir, + expected_producer_sha=args.expected_producer_sha, + checkpoint_sha256=args.checkpoint_sha256, + model_revision=args.model_revision, + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmarks/inferswarm_97/reference_runner.py b/benchmarks/inferswarm_97/reference_runner.py new file mode 100644 index 000000000..b598c8285 --- /dev/null +++ b/benchmarks/inferswarm_97/reference_runner.py @@ -0,0 +1,268 @@ +"""#97 v4 reference runner: RTX 3090 canonical reference execution. + +Runs the frozen c95-*/p95-* case manifests through the accepted execution path +and retains the v4 semantic evidence per case: + +- canonical 8-decision trajectory under the frozen argmax/tie-break rule + with per-decision executor rule proofs; +- FULL decision evidence at ALL 8 decisions: the complete FP32 + consumer-logit row is appended to the case bundle (decision-.f32 + sidecars, sha256-recorded) — sufficient for D(r) construction, + E_full-class full-vocabulary evidence, and later + decision_local_error derivation; +- the frozen decision domain D(r) per decision, computed on this + reference row only, with canonical membership hashes; +- unchanged 15-envelope checkpoint capture at positions 0/1/3/7 and + per-decision margin diagnostics (frozen min-over-8 margin definition). + +One process runs many cases sequentially (model load dominates). Each +case appends its own capture bundle + decision rows; nothing is ever +overwritten. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +import subprocess +import time +from pathlib import Path + +from benchmarks.inferswarm_76 import verify_case_identity +from benchmarks.inferswarm_76.reference_runner import resolve_cases +from benchmarks.inferswarm_97 import ( + ARGMAX_TIE_BREAK_IDENTITY, + CONTRACT_ID, + DECISION_DOMAIN_CONSTRUCTION, + GENERATED_TOKENS, + decision_domain_row, + executor_rule_proof, + frozen_subject_record, + prefix_sha256, + require_producer_identity, + validate_campaign_case_ids, +) + + +def _row_bytes(row) -> bytes: + import torch + + return ( + row.detach().to(torch.float32).contiguous() + .view(torch.uint8).numpy().tobytes() + ) + + +def _sha256_bytes(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def main(argv=None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--model", required=True) + parser.add_argument("--corpus", required=True) + parser.add_argument("--resolve-corpus", default=None) + parser.add_argument("--case-ids", default=None) + parser.add_argument("--out-dir", required=True) + parser.add_argument("--tag", required=True) + parser.add_argument("--gpu", default="0") + parser.add_argument("--attempt-id", required=True) + parser.add_argument("--expected-producer-sha", required=True) + parser.add_argument("--checkpoint-sha256", required=True) + parser.add_argument("--model-revision", required=True) + args = parser.parse_args(argv) + + repo = Path(__file__).resolve().parents[2] + producer = require_producer_identity(repo, args.expected_producer_sha) + subject = frozen_subject_record( + checkpoint_sha256=args.checkpoint_sha256, model_revision=args.model_revision + ) + + import os + + os.environ["CUDA_VISIBLE_DEVICES"] = args.gpu + cases = validate_campaign_case_ids(resolve_cases(args.corpus, args.resolve_corpus)) + if args.case_ids: + wanted = set(args.case_ids.split(",")) + cases = [c for c in cases if c["case_id"] in wanted] + missing = wanted - {c["case_id"] for c in cases} + if missing: + raise SystemExit(f"unknown case ids: {sorted(missing)}") + + import torch + + from benchmarks.inferswarm_76 import ( + CAPTURE_POSITIONS, + RUNTIME_CAPACITY_TOKENS, + ) + from benchmarks.inferswarm_76.capture import RowPruningSink, arm_full_capture + from benchmarks.inferswarm_r6.stage_runtime import GemmaDenseStage + from freetoken.distributed import set_tp_info, try_get_tp_info + from freetoken.layers.rotary import set_rope_device + from freetoken.research.r6_dense_census import ( + DenseBlockSpec, + checkpoint_census, + freeze_dense_block_plan, + ) + + if try_get_tp_info() is None: + set_tp_info(rank=0, size=1) + set_rope_device(torch.device("cuda:0")) + + gpu_uuid = subprocess.check_output( + ["nvidia-smi", "-i", "0", "--query-gpu=uuid", "--format=csv,noheader"], + text=True).strip() + + model = Path(args.model).resolve() + census = checkpoint_census(model, text_prefix="model.language_model") + shared = { + "id": "tied-embedding-lm-head", + "kind": "tied-weight-shared-state", + "tensor_keys": ["model.language_model.embed_tokens.weight"], + "bytes": census["bytes_by_owner_category"]["embedding/input"], + "materialization_policy": "single-cuda-tensor-used-for-input-and-output", + } + plan = freeze_dense_block_plan( + census, [DenseBlockSpec(0, 48, True, True)], declared_shared_state=shared + ) + runtime = GemmaDenseStage( + role="single", + model_path=str(model), + adapter_data={ + **plan["blocks"][0], + "declared_shared_state": shared, + "runtime_capacity_tokens": RUNTIME_CAPACITY_TOKENS, + }, + ) + + # wrappers bind runtime._emit dynamically (which reads _capture_sink), + # so they are installed EXACTLY ONCE (the accepted #76 pattern); per-case + # isolation comes from swapping the sink below. Arming per case would + # CHAIN wrapper layers and duplicate every capture record. + runtime._capture_sink = RowPruningSink(role="single", gpu_uuid=gpu_uuid) + runtime._capture_after_layers = frozenset({15, 31}) + arm_full_capture(runtime, runtime._capture_sink) + + results = [] + out_root = Path(args.out_dir) + out_root.mkdir(parents=True, exist_ok=True) + + for case in cases: + case_dir = out_root / case["case_id"] + case_dir.mkdir(parents=True, exist_ok=True) + runtime._capture_sink = RowPruningSink(role="single", gpu_uuid=gpu_uuid) + sink = runtime._capture_sink + + prompt = list(case["token_ids"]) + generated: list[int] = [] + margins: list[dict] = [] + decision_rows: list[dict] = [] + nan_inf_total = 0 + t0 = time.perf_counter() + for step in range(GENERATED_TOKENS): + runtime.reset_session_state() + replay = prompt + generated + capture_now = step in CAPTURE_POSITIONS + runtime._capture_step = step if capture_now else None + token, logits = runtime.prefill(replay, None, 0) + row = logits # final-row FP32 consumer logits [vocab] + nan_inf_total += int( + torch.isnan(row).sum().item() + torch.isinf(row).sum().item()) + # --- v3 semantic layer on the exact row -------------------- + host_row = row.detach().to("cpu", torch.float32).contiguous() + values = host_row.tolist() + proof = executor_rule_proof(values, int(token)) + domain = decision_domain_row(values) + row_bytes = host_row.view(torch.uint8).numpy().tobytes() + row_path = case_dir / f"decision-{step}.f32" + if row_path.exists(): + raise SystemExit(f"refusing to overwrite {row_path}") + row_path.write_bytes(row_bytes) + decision_rows.append({ + "decision_index": step, + "prefix_len": len(replay), + "prefix_sha256": prefix_sha256(replay), + "domain_membership_sha256": domain["domain_membership_sha256"], + "domain_size": domain["domain_size"], + "domain_cutoff_hex": domain["cutoff_hex"], + "emitted_token": int(token), + "emitted_rule": ARGMAX_TIE_BREAK_IDENTITY, + "row_f32_sha256": _sha256_bytes(row_bytes), + "row_element_count": len(values), + "rule_proof": proof, + }) + top2 = torch.topk(row, 2) + margins.append({ + "step": step, + "top1_index": int(top2.indices[0].item()), + "top1_value_hex": float(top2.values[0].item()).hex(), + "top2_index": int(top2.indices[1].item()), + "top2_value_hex": float(top2.values[1].item()).hex(), + "margin_hex": float( + top2.values[0].item() - top2.values[1].item()).hex(), + }) + generated.append(int(token)) + del logits, row, host_row, values + + manifest = sink.save(str(case_dir), args.tag) + summary = { + "schema": "inferswarm.issue97.v4-reference-case/1", + "contract_id": CONTRACT_ID, + "subject": subject, + "attempt_id": args.attempt_id, + "case_id": case["case_id"], + "case_sha256": case["case_sha256"], + "prompt_sha256": case["prompt_sha256"], + "token_ids_sha256": case["token_ids_sha256"], + "generated_token_ids": generated, + "step_margins": margins, + "min_top1_margin_hex": min( + float.fromhex(m["margin_hex"]) for m in margins).hex(), + "nan_inf_count": nan_inf_total, + "decision_domain_construction": DECISION_DOMAIN_CONSTRUCTION, + "decisions": decision_rows, + "producer": producer, + "gpu_uuid": gpu_uuid, + "role": "reference-single", + "capture_positions": list(CAPTURE_POSITIONS), + "capture_manifest": manifest, + "wall_seconds": time.perf_counter() - t0, + } + path = case_dir / f"reference-case-{args.tag}.json" + if path.exists(): + raise SystemExit(f"refusing to overwrite {path}") + path.write_text(json.dumps(summary, indent=2, sort_keys=True) + "\n") + results.append(summary) + print(json.dumps({ + "case_id": case["case_id"], "status": "CASE_COMMITTED", + "tokens": generated, "nan_inf": nan_inf_total, + "records": manifest["record_count"]}), flush=True) + + index = { + "schema": "inferswarm.issue97.v4-reference-run-index/1", + "contract_id": CONTRACT_ID, + "subject": subject, + "attempt_id": args.attempt_id, + "producer": producer, + "gpu_uuid": gpu_uuid, + "tag": args.tag, + "case_count": len(results), + "cases": [ + {"case_id": r["case_id"], "case_sha256": r["case_sha256"], + "generated_token_ids": r["generated_token_ids"], + "nan_inf_count": r["nan_inf_count"], + "min_top1_margin_hex": r["min_top1_margin_hex"]} + for r in results + ], + } + index_path = out_root / f"index-{args.tag}.json" + if index_path.exists(): + raise SystemExit(f"refusing to overwrite {index_path}") + index_path.write_text(json.dumps(index, indent=2, sort_keys=True) + "\n") + print(json.dumps({"status": "RUN_COMPLETE", "cases": len(results)})) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmarks/inferswarm_97b/__init__.py b/benchmarks/inferswarm_97b/__init__.py new file mode 100644 index 000000000..dfdfe3e0c --- /dev/null +++ b/benchmarks/inferswarm_97b/__init__.py @@ -0,0 +1,74 @@ +"""Issue #97 Phase B holdout-only admission (FreeToken, package init). + +Executes the maintainer instruction issued 2026-09-06 after the +`V4_HOLDOUT_UNSEAL_READY_BLOCKED_BY_MISSING_EXECUTION_AUTHORITY` stop: + +- the accepted Phase A calibration producer stays frozen at + `57dfcb7289efac8f66de5b3abbe8de04f2580f75` (PR #31) and its + c95-/p95- admission discipline is UNCHANGED; +- this package adds the separate, explicitly authorized h95 holdout + path the frozen producer's docstring promised but never shipped; +- the Phase B producer (this commit) is a SEPARATE externally pinned + SHA; nothing in `benchmarks/inferswarm_97/` changes. + +Importing this module performs NO model work and imports no torch. +""" + +from __future__ import annotations + +from typing import Any, Sequence + +# Re-exported frozen identities: the holdout runs under the SAME accepted +# v4 contract and subject as Phase A (no methodology/subject change). +from benchmarks.inferswarm_97 import ( # noqa: F401 (re-export) + CONTRACT_ID, + EXPECTED_CHECKPOINT_SHA256, + MODEL_REVISION, +) + +HOLDOUT_NAMESPACE_PREFIX = "h95-" + +# Phase A calibration producer (immutable historical identity). +PHASE_A_PRODUCER_SHA256 = "57dfcb7289efac8f66de5b3abbe8de04f2580f75" + +# Frozen holdout corpus identity: the 24 committed h95 cell case IDs from +# inferswarm@e1d3a16 docs/qualification/gemma4-12b-it-v4/manifests/ +# sealed-holdout-commitment.json (public commitment, no plaintext). +FROZEN_HOLDOUT_CASE_IDS = ( + "h95-01-01-01", "h95-01-02-01", "h95-01-03-01", "h95-01-04-01", + "h95-01-05-01", "h95-01-06-01", + "h95-02-01-01", "h95-02-02-01", "h95-02-03-01", "h95-02-04-01", + "h95-02-05-01", "h95-02-06-01", + "h95-03-01-01", "h95-03-02-01", "h95-03-03-01", "h95-03-04-01", + "h95-03-05-01", "h95-03-06-01", + "h95-04-01-01", "h95-04-02-01", "h95-04-03-01", "h95-04-04-01", + "h95-04-05-01", "h95-04-06-01", +) + + +def validate_holdout_case_ids(cases: Sequence[dict[str, Any]]) -> list[dict[str, Any]]: + """h95-only admission for the Phase B holdout entrypoints. + + Accepts EXACTLY the 24 frozen committed holdout case IDs. Rejects + every c95-/p95- calibration case, every historical namespace, and + any holdout-shaped ID outside the frozen commitment (unknown, + renumbered, or substituted cells fail closed BEFORE any CUDA work). + """ + checked = list(cases) + seen: list[str] = [] + for case in checked: + case_id = case.get("case_id") if isinstance(case, dict) else None + if not isinstance(case_id, str) or not case_id.startswith(HOLDOUT_NAMESPACE_PREFIX): + raise ValueError( + "issue #97 Phase B accepts only h95- holdout case IDs, " + f"got {case_id!r}" + ) + if case_id not in FROZEN_HOLDOUT_CASE_IDS: + raise ValueError( + f"issue #97 Phase B: {case_id!r} is not one of the 24 " + "committed h95 holdout cells" + ) + if case_id in seen: + raise ValueError(f"duplicate holdout case id: {case_id!r}") + seen.append(case_id) + return checked diff --git a/benchmarks/inferswarm_97b/chain_runner_holdout.py b/benchmarks/inferswarm_97b/chain_runner_holdout.py new file mode 100644 index 000000000..c5dad00a1 --- /dev/null +++ b/benchmarks/inferswarm_97b/chain_runner_holdout.py @@ -0,0 +1,22 @@ +#!/usr/bin/env python3 +"""Issue #97 Phase B holdout chain entrypoint. + +Accepts ONLY h95-* holdout cases and otherwise runs the EXACT frozen +Phase A chain runner (`benchmarks/inferswarm_97/chain_runner.py`, +untouched) teacher-forced against the h95 reference evidence. + +Same mechanism as ``reference_runner_holdout``: swap the frozen runner +module's admission callable for the h95-only one, delegate to the frozen +``main``. No execution/model math is duplicated or altered. +""" + +from __future__ import annotations + +import sys + +import benchmarks.inferswarm_97.chain_runner as frozen_chain +from benchmarks.inferswarm_97b import validate_holdout_case_ids + +if __name__ == "__main__": + frozen_chain.validate_campaign_case_ids = validate_holdout_case_ids + sys.exit(frozen_chain.main()) diff --git a/benchmarks/inferswarm_97b/reference_runner_holdout.py b/benchmarks/inferswarm_97b/reference_runner_holdout.py new file mode 100644 index 000000000..73efacfec --- /dev/null +++ b/benchmarks/inferswarm_97b/reference_runner_holdout.py @@ -0,0 +1,32 @@ +#!/usr/bin/env python3 +"""Issue #97 Phase B holdout reference entrypoint. + +Accepts ONLY h95-* holdout cases and otherwise runs the EXACT frozen +Phase A reference runner (`benchmarks/inferswarm_97/reference_runner.py`, +untouched) on the RTX 3090 reference path. + +Mechanism (no source duplication of execution math): this wrapper swaps +the frozen runner module's admission callable for the h95-only one, then +delegates to the frozen ``main``. The frozen module object carries the +swap, so its internal ``validate_campaign_case_ids`` reference resolves +to h95-only admission; every other function, constant, model call, and +byte of execution math is the frozen Phase A code. + +Equivalence is enforced by tests: the Phase A runner files must be +byte-identical to their Phase A freeze hashes, and this wrapper differs +from the Phase A path ONLY in namespace admission (synthetic fixtures; +no secret material). +""" + +from __future__ import annotations + +import sys + +import benchmarks.inferswarm_97.reference_runner as frozen_reference +from benchmarks.inferswarm_97b import validate_holdout_case_ids + +if __name__ == "__main__": + # Phase A admission replaced by h95-only admission BEFORE main() runs + # (main resolves the admission callable from this module object). + frozen_reference.validate_campaign_case_ids = validate_holdout_case_ids + sys.exit(frozen_reference.main()) diff --git a/tests/research/test_inferswarm_97_harness.py b/tests/research/test_inferswarm_97_harness.py new file mode 100644 index 000000000..8d1763d1e --- /dev/null +++ b/tests/research/test_inferswarm_97_harness.py @@ -0,0 +1,85 @@ +"""Issue #97 v4 physical producer contract tests.""" +from __future__ import annotations + +import inspect +import subprocess +import unittest +from pathlib import Path +from unittest.mock import patch + +from benchmarks.inferswarm_97 import ( + CONTRACT_ID, EXPECTED_CHECKPOINT_SHA256, GENERATED_TOKENS, + METHODOLOGY_COMMIT, MODEL_REVISION, V4_ISSUE, frozen_argmax_row, + frozen_subject_record, require_producer_identity, validate_campaign_case_ids, +) + + +class Issue97ProducerIdentityTests(unittest.TestCase): + def test_binds_accepted_v4_methodology_and_subject(self): + self.assertEqual(V4_ISSUE, 97) + self.assertEqual(METHODOLOGY_COMMIT, "e12a6e3d5589044bace0c9555c0d364fb57a6229") + self.assertEqual(CONTRACT_ID, "inferswarm.gemma4-prediction-aligned-qualification/1") + self.assertEqual(EXPECTED_CHECKPOINT_SHA256, "5a84cb313260ac447237b890387116dfa8682e49a6b44bc585ae8353abbff18d") + self.assertEqual(MODEL_REVISION, "707f0a3b8a3c7ad586ed01e27eafbad8a27dd0f7") + self.assertEqual(GENERATED_TOKENS, 8) + + def test_argmax_rule_retains_lowest_exact_maximum(self): + self.assertEqual(frozen_argmax_row([1.0, 3.0, 3.0, 2.0]), (1, 3.0)) + + def test_campaign_rejects_holdout_and_historical_case_namespaces(self): + accepted = [{"case_id": "c95-00-00-00"}, {"case_id": "p95-05-03-01"}] + self.assertEqual(validate_campaign_case_ids(accepted), accepted) + for forbidden in ("h95-00-00-00", "c86-00-00-00", "p86-00-00-00"): + with self.subTest(forbidden=forbidden): + with self.assertRaises(ValueError): + validate_campaign_case_ids([{"case_id": forbidden}]) + + def test_subject_record_rejects_substitution_and_binds_accepted_contract(self): + record = frozen_subject_record(checkpoint_sha256=EXPECTED_CHECKPOINT_SHA256, model_revision=MODEL_REVISION) + self.assertEqual(record["contract_id"], CONTRACT_ID) + self.assertEqual(record["methodology_commit"], METHODOLOGY_COMMIT) + with self.assertRaises(ValueError): + frozen_subject_record(checkpoint_sha256="0" * 64, model_revision=MODEL_REVISION) + + @patch("benchmarks.inferswarm_97.subprocess.check_output") + def test_exact_clean_expected_producer_passes(self, output): + output.side_effect = ["a" * 40 + "\n", ""] + self.assertEqual(require_producer_identity(Path("/repo"), "a" * 40), {"commit": "a" * 40, "dirty": False, "expected_commit": "a" * 40}) + + @patch("benchmarks.inferswarm_97.subprocess.check_output") + def test_dirty_exact_producer_fails(self, output): + output.side_effect = ["a" * 40 + "\n", " M file.py\n"] + with self.assertRaisesRegex(ValueError, "dirty"): + require_producer_identity(Path("/repo"), "a" * 40) + + @patch("benchmarks.inferswarm_97.subprocess.check_output") + def test_clean_wrong_producer_fails(self, output): + output.side_effect = ["b" * 40 + "\n", ""] + with self.assertRaisesRegex(ValueError, "does not equal"): + require_producer_identity(Path("/repo"), "a" * 40) + + def test_missing_or_malformed_expected_producer_fails(self): + for value in (None, "", "not-a-sha", "a" * 39): + with self.subTest(value=value): + with self.assertRaises(ValueError): + require_producer_identity(Path("/repo"), value) + + def test_all_entrypoints_require_external_expected_producer_sha(self): + root = Path(__file__).resolve().parents[2] / "benchmarks" / "inferswarm_97" + for filename in ("reference_runner.py", "chain_runner.py", "last_stage_service.py"): + with self.subTest(filename=filename): + source = (root / filename).read_text() + self.assertIn('\"--expected-producer-sha\", required=True', source) + self.assertNotIn("--allow-producer", source) + + def test_last_stage_producer_and_subject_checks_precede_cuda_initialization(self): + root = Path(__file__).resolve().parents[2] / "benchmarks" / "inferswarm_97" + source = (root / "last_stage_service.py").read_text() + serve = source[source.index("def serve("):source.index("def main(")] + self.assertLess(serve.index("require_producer_identity"), serve.index("import torch")) + self.assertLess(serve.index("frozen_subject_record"), serve.index("import torch")) + self.assertLess(serve.index("require_producer_identity"), serve.index("set_rope_device")) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/research/test_inferswarm_97b_holdout_authority.py b/tests/research/test_inferswarm_97b_holdout_authority.py new file mode 100644 index 000000000..75a23a24b --- /dev/null +++ b/tests/research/test_inferswarm_97b_holdout_authority.py @@ -0,0 +1,232 @@ +"""Issue #97 Phase B holdout-only execution authority: tests. + +All fixtures are SYNTHETIC and contain no holdout plaintext or secret +material. The h95 case IDs used in admission tests are the PUBLIC +committed cell identities from the sealed-holdout commitment (public +metadata, not plaintext). + +CPU-only: no torch, no CUDA, no model execution anywhere in this suite. + +Covers: +1. Phase A runner/service/__init__ files remain byte-identical to the + Phase A freeze (57dfcb7) hashes — the frozen path is untouched. +2. Holdout admission accepts exactly the 24 committed h95 cells and + rejects everything else (calibration IDs, historical namespaces, + unknown/substituted/renumbered holdout IDs, duplicates). +3. Mechanical equivalence of the holdout entrypoints apart from + namespace admission: each holdout entrypoint, executed as __main__, + delegates to the frozen Phase A ``main`` with ONLY the admission + callable swapped, and exits with the frozen main's return code. +4. The Phase A package itself keeps rejecting h95-* (admission swap is + entrypoint-runtime-only, never installed package-wide). +""" + +from __future__ import annotations + +import hashlib +import importlib +import sys +import types +import unittest +from pathlib import Path + +REPO = Path(__file__).resolve().parents[2] + +# Phase A freeze (57dfcb7) sha256 of every file the campaign froze. +PHASE_A_FREEZE_SHA256 = { + "benchmarks/inferswarm_97/reference_runner.py": + "f04b515c0a9b3dbccc52f8f2b0fe96c705637c0d128421e4c2437a6bbffde46d", + "benchmarks/inferswarm_97/chain_runner.py": + "480f01b6a456f948ec327c0afdbfd84237c6efb1e903b8a3f94a1672c89df921", + "benchmarks/inferswarm_97/last_stage_service.py": + "79bed35acd48873f4513c47fe9a8ab5243b42e9cb041a9d723a47688d2be308b", + "benchmarks/inferswarm_97/__init__.py": + "183d8824838f0ced70284a4c180f1ad83321375c8e15473c8071e44da20e2db3", +} + +for _p in (REPO, REPO / "python", REPO / "benchmarks"): + if str(_p) not in sys.path: + sys.path.insert(0, str(_p)) + + +def _sha256(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +class TestPhaseAFilesUnchanged(unittest.TestCase): + def test_every_frozen_phase_a_file_is_byte_identical(self): + for rel, expected in PHASE_A_FREEZE_SHA256.items(): + with self.subTest(rel=rel): + self.assertEqual(_sha256(REPO / rel), expected) + + +class TestHoldoutAdmission(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.pkg = importlib.import_module("benchmarks.inferswarm_97b") + + def _case(self, case_id: str) -> dict: + return {"case_id": case_id} + + def test_accepts_exactly_the_24_committed_cells(self): + cases = [self._case(cid) for cid in self.pkg.FROZEN_HOLDOUT_CASE_IDS] + out = self.pkg.validate_holdout_case_ids(cases) + self.assertEqual(len(out), 24) + self.assertEqual( + sorted(c["case_id"] for c in out), + sorted(self.pkg.FROZEN_HOLDOUT_CASE_IDS)) + + def test_rejects_calibration_and_historical_namespaces(self): + for forbidden in ("c95-01-01-01", "p95-01-03-01", "c86-00-00-00", + "p86-05-03-01", "c74-00-00-00", "h86-03-05-01", + "p76-02-01-01", "unversioned", ""): + with self.subTest(forbidden=forbidden): + with self.assertRaises(ValueError): + self.pkg.validate_holdout_case_ids([self._case(forbidden)]) + + def test_rejects_unknown_substituted_holdout_ids(self): + for forbidden in ("h95-01-01-02", "h95-99-99-99", "h95-01-01", + "h95-05-01-01", "h95-1-1-1"): + with self.subTest(forbidden=forbidden): + with self.assertRaises(ValueError): + self.pkg.validate_holdout_case_ids([self._case(forbidden)]) + + def test_rejects_duplicates(self): + with self.assertRaises(ValueError): + self.pkg.validate_holdout_case_ids( + [self._case("h95-01-01-01"), self._case("h95-01-01-01")]) + + def test_rejects_non_dict_rows(self): + with self.assertRaises(ValueError): + self.pkg.validate_holdout_case_ids(["h95-01-01-01"]) + + +class TestPhaseAPackageUnaffected(unittest.TestCase): + def test_phase_a_admission_still_rejects_h95(self): + frozen = importlib.import_module("benchmarks.inferswarm_97") + with self.assertRaises(ValueError): + frozen.validate_campaign_case_ids([{"case_id": "h95-01-01-01"}]) + + def test_phase_a_admission_still_accepts_calibration(self): + frozen = importlib.import_module("benchmarks.inferswarm_97") + accepted = [{"case_id": "c95-01-01-01"}, {"case_id": "p95-05-03-01"}] + self.assertEqual(frozen.validate_campaign_case_ids(accepted), accepted) + + def test_importing_97b_does_not_touch_phase_a_admission(self): + frozen = importlib.import_module("benchmarks.inferswarm_97") + before = frozen.validate_campaign_case_ids + importlib.import_module("benchmarks.inferswarm_97b") + self.assertIs(frozen.validate_campaign_case_ids, before) + + +class TestEntrypointEquivalence(unittest.TestCase): + """Each holdout entrypoint = frozen main + ONLY the admission swap. + + Executes the real entrypoint source as __main__ with the frozen + module's ``main`` monkeypatched to record (a) the admission callable + active at delegation time and (b) the argv it would parse, then + asserts the process exit code equals the frozen main's rc. The + frozen module's original admission is restored in finally blocks. + """ + + def _run_entrypoint(self, entry_name: str, frozen_mod_name: str) -> None: + frozen = importlib.import_module(frozen_mod_name) + holdout = importlib.import_module("benchmarks.inferswarm_97b") + + observed = {} + + def fake_main(argv=None): + observed["admission_at_delegation"] = \ + frozen.validate_campaign_case_ids + observed["delegated_main"] = fake_main + return 7 # distinctive sentinel rc + + original_main = frozen.main + original_admission = frozen.validate_campaign_case_ids + setattr(frozen, "main", fake_main) + exit_code = None + try: + source = (REPO / "benchmarks" / "inferswarm_97b" / + entry_name).read_text() + mod = types.ModuleType("holdout_entrypoint_under_test") + mod.__dict__["__name__"] = "__main__" + mod.__dict__["__file__"] = str( + REPO / "benchmarks" / "inferswarm_97b" / entry_name) + try: + exec(compile(source, entry_name, "exec"), mod.__dict__) + except SystemExit as exc: + exit_code = exc.code + finally: + setattr(frozen, "main", original_main) + setattr(frozen, "validate_campaign_case_ids", + original_admission) + + # Delegation reached the frozen main with the holdout admission + # active, and NOTHING else in the frozen module changed. + self.assertIn("admission_at_delegation", observed) + self.assertIs(observed["admission_at_delegation"], + holdout.validate_holdout_case_ids) + self.assertIsNot(observed["admission_at_delegation"], + original_admission) + # sys.exit(frozen.main()) propagated the frozen rc verbatim. + self.assertEqual(exit_code, 7) + # Restoration left the frozen module exactly as it was. + self.assertIs(frozen.validate_campaign_case_ids, original_admission) + self.assertIs(frozen.main, original_main) + + def test_reference_entrypoint(self): + self._run_entrypoint( + "reference_runner_holdout.py", + "benchmarks.inferswarm_97.reference_runner") + + def test_chain_entrypoint(self): + self._run_entrypoint( + "chain_runner_holdout.py", + "benchmarks.inferswarm_97.chain_runner") + + def test_entrypoints_import_no_torch(self): + import ast + + for name in ("reference_runner_holdout.py", "chain_runner_holdout.py"): + with self.subTest(name=name): + tree = ast.parse( + (REPO / "benchmarks" / "inferswarm_97b" / + name).read_text()) + imported = set() + for node in ast.walk(tree): + if isinstance(node, ast.Import): + imported.update(a.name for a in node.names) + elif isinstance(node, ast.ImportFrom) and node.module: + imported.add(node.module) + self.assertFalse( + any(m == "torch" or m.startswith("torch.") or + m.startswith("freetoken") or + m.startswith("benchmarks.inferswarm_76") + for m in imported), + f"{name} must import only admission + frozen runner") + + def test_holdout_package_contains_no_execution_math(self): + """Source audit: 97b has no model/runtime/prefill calls.""" + import ast + + forbidden_calls = { + "prefill", "generate", "forward", "reset_session_state", + "arm_full_capture", "checkpoint_census", "freeze_dense_block_plan", + "GemmaDenseStage", "I76StageClient", "I76LastStageClient", + } + for path in sorted((REPO / "benchmarks" / "inferswarm_97b") + .glob("*.py")): + with self.subTest(path=path.name): + tree = ast.parse(path.read_text()) + for node in ast.walk(tree): + if isinstance(node, ast.Call): + fn = node.func + name = getattr(fn, "id", None) or getattr( + fn, "attr", None) + self.assertNotIn( + name, forbidden_calls, + f"{path.name} must not call {name}") + + +if __name__ == "__main__": + unittest.main()