wisp-coder-110m / evidence /source /e3_rollout_v3 /scripts /test_release_rollout_v3.py
philipjohnbasile's picture
Publish audited Wisp Coder 110M release
818282c verified
Raw
History Blame Contribute Delete
28 kB
"""Release-level adversarial checks for the E3 v3 rollout report."""
import copy
from datetime import datetime, timezone
import hashlib
import json
import os
import sys
import tempfile
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from scripts.release_audit import ( # noqa: E402
validate_rollout_replay_attestation,
validate_rollout_report,
)
from scripts.rollout_metrics import ( # noqa: E402
NEAR_TIE_MAX_ULPS,
V3_ATTESTATION_PROVENANCE_SCOPE,
V3_INSTRUMENT_VERSION,
V3_PAIR_PAYLOAD_DOMAIN,
V3_REPLAY_REFERENCE,
V3_REPLAY_RULE,
V3_REPLAY_VERIFICATION_ARGV,
V3_REPLAY_VERIFIER_METHOD,
V3_REPORT_SCHEMA_VERSION,
canonical_json_sha256,
metric_value,
paired_mean_difference_ci,
select_policy,
summarize_policy_v3,
token_ids_sha256,
validate_cross_policy_trajectories,
)
from scripts.test_rollout_replay import ( # noqa: E402
PROMPT,
VOCAB_SIZE,
make_row,
resign_trace,
)
RUNTIME = {
"python": "3.14.6",
"mlx": "0.32.0",
"numpy": "2.5.1",
"tokenizers": "0.22.2",
}
REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
EXECUTION_ARGV = [
"scripts/eval_rollout.py",
"--ckpt",
"fixture-checkpoint",
"--receipt",
"fixture-receipt.json",
"--out",
"fixture-report.json",
]
def _file_sha256(path):
digest = hashlib.sha256()
with open(path, "rb") as f:
for chunk in iter(lambda: f.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def _identity_manifest_sha256(rows):
encoded = json.dumps(
rows, sort_keys=True, separators=(",", ":")
).encode("utf-8")
return hashlib.sha256(encoded).hexdigest()
def _endpoint(
specification,
adaptive_rows,
fixed_rows,
adaptive_policy,
fixed_policy,
):
adaptive = [
metric_value(row, specification["metric"]) for row in adaptive_rows
]
fixed = [
metric_value(row, specification["metric"]) for row in fixed_rows
]
difference, lo, hi = paired_mean_difference_ci(
adaptive,
fixed,
n_boot=specification["bootstrap_samples"],
seed=specification["bootstrap_seed"],
)
if lo > 0:
verdict = "POSITIVE"
elif hi < 0:
verdict = "NEGATIVE"
else:
verdict = "NULL: the interval includes 0"
return {
"comparison": specification["comparison"],
"metric": specification["metric"],
"adaptive_policy": adaptive_policy,
"fixed_policy": fixed_policy,
"difference": difference,
"ci95": [lo, hi],
"documents": len(adaptive_rows),
"verdict": verdict,
}
def _rows(policy_name, payloads, seed):
rows = []
for local_index, payload in enumerate(payloads):
row = make_row(
[3, 4, 5],
[3, 4, 5],
pair_index=payload["index"],
policy=policy_name,
)
row["seed"] = seed + local_index
if policy_name.startswith("adaptive_h"):
row["rollout"]["policy"] = "adaptive"
row["rollout"]["entropy_threshold"] = float(
policy_name.removeprefix("adaptive_h")
)
rows.append(row)
return rows
def build_v3_rollout_fixture(root):
"""Build a small, fully valid report without invoking MLX."""
fixture_root = os.path.join(root, "release-rollout-v3")
os.makedirs(fixture_root, exist_ok=True)
tokenizer_path = os.path.join(fixture_root, "tokenizer.json")
holdout_path = os.path.join(fixture_root, "holdout.jsonl")
with open(tokenizer_path, "w", encoding="utf-8") as f:
json.dump({"fixture": "tokenizer"}, f)
with open(holdout_path, "w", encoding="utf-8") as f:
f.write('{"fixture":"holdout"}\n')
outputs = [[3, 4, 5] for _ in range(4)]
payloads = []
identities = []
for index, target in enumerate(outputs):
identity = {
"index": index,
"document_id": f"doc-{index}",
"decoy_document_id": f"decoy-{index}",
"token_offset": index,
"prompt_sha256": token_ids_sha256(PROMPT),
"target_sha256": token_ids_sha256(target),
}
identities.append(identity)
payloads.append({
**identity,
"prompt_token_ids": list(PROMPT),
"target_token_ids": list(target),
})
identity_sha256 = _identity_manifest_sha256(identities)
payload_sha256 = canonical_json_sha256(
payloads, V3_PAIR_PAYLOAD_DOMAIN
)
primary_spec = {
"comparison": "selected_adaptive_minus_selected_fixed",
"metric": "accepted_drafts_per_verification",
"bootstrap_unit": "paired_target_document",
"bootstrap_samples": 200,
"bootstrap_seed": 0,
"positive_when": "ci95_lower_gt_0",
}
companion_specs = [
{
"comparison": "selected_adaptive_minus_selected_fixed",
"metric": "output_tokens_per_target_forward",
"bootstrap_unit": "paired_target_document",
"bootstrap_samples": 200,
"bootstrap_seed": 0,
},
{
"comparison": "selected_adaptive_minus_selected_fixed",
"metric": "drafts_issued_per_output_token",
"bootstrap_unit": "paired_target_document",
"bootstrap_samples": 200,
"bootstrap_seed": 0,
},
{
"comparison": "selected_adaptive_minus_selected_fixed",
"metric": "draft_recursions_per_output_token",
"bootstrap_unit": "paired_target_document",
"bootstrap_samples": 200,
"bootstrap_seed": 0,
},
]
checkpoint = {
"path": os.path.join(fixture_root, "fixture-checkpoint"),
"step": 19073,
"meta_sha256": "a" * 64,
"master_sha256": "b" * 64,
}
registered_checkpoint = {
"path": checkpoint["path"],
"meta_sha256": checkpoint["meta_sha256"],
"master_sha256": checkpoint["master_sha256"],
}
receipt = {
"schema_version": 2,
"instrument_version": V3_INSTRUMENT_VERSION,
"registered_at": "2026-07-30T12:00:00Z",
"checkpoint": registered_checkpoint,
"execution_argv": list(EXECUTION_ARGV),
"runtime_requirements": dict(RUNTIME),
"model_vocab_size": VOCAB_SIZE,
"acceptance_receipt": {
"path": os.path.join(fixture_root, "acceptance-receipt.json"),
"sha256": "f" * 64,
},
"holdout": {
"path": holdout_path,
"sha256": _file_sha256(holdout_path),
},
"tokenizer": {
"path": tokenizer_path,
"sha256": _file_sha256(tokenizer_path),
},
"pair_settings": {
"examples": len(payloads),
"pair_manifest_sha256": identity_sha256,
"pair_payload_manifest_sha256": payload_sha256,
},
"split": {
"calibration_documents": 2,
"test_documents": 2,
},
"policy": {
"fixed_candidates": ["fixed_d2"],
"adaptive_candidates": ["adaptive_h0.2"],
"max_depth": 2,
"selection_metric": "accepted_drafts_per_verification",
},
"decoding": {
"max_tokens": 3,
"seed": 2718,
},
"test_endpoint": primary_spec,
"companion_endpoints": companion_specs,
}
receipt_evidence = {
"path": os.path.join(fixture_root, "fixture-receipt.json"),
"sha256": "c" * 64,
"registered_at": "2026-07-30T12:00:00Z",
"registration_git": {
"commit": "e" * 40,
"committed_at": "2026-07-30T12:00:00Z",
"path": "config/eval_rollout_receipt_v3.json",
"blob_sha256": "c" * 64,
"origin_ref": "origin/main",
},
"implementation": {
"git_commit": "d" * 40,
"source_files": {},
},
"checkpoint": registered_checkpoint,
"execution_argv": list(EXECUTION_ARGV),
"runtime_requirements": dict(RUNTIME),
"prior_instruments": [],
}
calibration_payloads = payloads[:2]
test_payloads = payloads[2:]
candidate_order = ["fixed_d2", "adaptive_h0.2"]
calibration_rows = {
name: _rows(name, calibration_payloads, receipt["decoding"]["seed"])
for name in candidate_order
}
calibration_summaries = {
name: summarize_policy_v3(
rows,
calibration_payloads,
max_tokens=3,
vocab_size=VOCAB_SIZE,
)
for name, rows in calibration_rows.items()
}
selected_fixed = select_policy(
calibration_summaries,
receipt["policy"]["fixed_candidates"],
receipt["policy"]["selection_metric"],
)
selected_adaptive = select_policy(
calibration_summaries,
receipt["policy"]["adaptive_candidates"],
receipt["policy"]["selection_metric"],
)
selected_names = [
selected_fixed["policy"],
selected_adaptive["policy"],
]
test_rows = {
name: _rows(name, test_payloads, receipt["decoding"]["seed"])
for name in selected_names
}
test_summaries = {
name: summarize_policy_v3(
rows,
test_payloads,
max_tokens=3,
vocab_size=VOCAB_SIZE,
)
for name, rows in test_rows.items()
}
calibration_trajectory = validate_cross_policy_trajectories(
calibration_rows, candidate_order
)
test_trajectory = validate_cross_policy_trajectories(
test_rows, selected_names
)
all_summaries = [
*calibration_summaries.values(),
*test_summaries.values(),
]
scored_documents = sum(item["documents"] for item in all_summaries)
scored_tokens = sum(item["total_tokens"] for item in all_summaries)
exact_tokens = sum(
item["exact_argmax_tokens"] for item in all_summaries
)
near_tie_tokens = sum(
item["certified_near_tie_tokens"] for item in all_summaries
)
branch_passes = sum(
item["branch_replay_passes"] for item in all_summaries
)
cached_exact = sum(
item["cached_ar_exact_documents"] for item in all_summaries
)
adaptive_test = test_rows[selected_adaptive["policy"]]
fixed_test = test_rows[selected_fixed["policy"]]
primary = _endpoint(
primary_spec,
adaptive_test,
fixed_test,
selected_adaptive["policy"],
selected_fixed["policy"],
)
companions = {
item["metric"]: _endpoint(
item,
adaptive_test,
fixed_test,
selected_adaptive["policy"],
selected_fixed["policy"],
)
for item in companion_specs
}
now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
report = {
"schema_version": V3_REPORT_SCHEMA_VERSION,
"instrument_version": V3_INSTRUMENT_VERSION,
"publication_ready": True,
"execution": {
"started_at": now,
"completed_at": now,
"argv": list(EXECUTION_ARGV),
"runtime": {**RUNTIME, "platform": "fixture-platform"},
},
"receipt": receipt_evidence,
"checkpoint": checkpoint,
"tokenizer": {
"path": tokenizer_path,
"sha256": _file_sha256(tokenizer_path),
},
"holdout": {
"path": holdout_path,
"sha256": _file_sha256(holdout_path),
"pair_count": len(payloads),
"pair_manifest_sha256": identity_sha256,
"pair_manifest": identities,
"pair_payload_manifest_sha256": payload_sha256,
"pair_payload_manifest": payloads,
"decoy_match": {"matched": len(payloads)},
},
"calibration": {
"documents": len(calibration_payloads),
"candidate_order": candidate_order,
"trajectory_identity": calibration_trajectory,
"summaries": calibration_summaries,
"selected_fixed": selected_fixed,
"selected_adaptive": selected_adaptive,
"rows": calibration_rows,
},
"test": {
"documents": len(test_payloads),
"trajectory_identity": test_trajectory,
"summaries": test_summaries,
"rows": test_rows,
},
"quality_gate": {
"reference": V3_REPLAY_REFERENCE,
"rule": V3_REPLAY_RULE,
"near_tie_max_ulps": NEAR_TIE_MAX_ULPS,
"scored_policy_documents": scored_documents,
"scored_tokens": scored_tokens,
"exact_argmax_tokens": exact_tokens,
"certified_near_tie_tokens": near_tie_tokens,
"failed_tokens": 0,
"branch_replay_passes": branch_passes,
"cross_policy_trajectory_matches": (
calibration_trajectory["matching_documents"]
+ test_trajectory["matching_documents"]
),
"passed": True,
},
"cached_ar_diagnostic": {
"scored_policy_documents": scored_documents,
"exact_output_matches": cached_exact,
"different_cached_ar_branches": scored_documents - cached_exact,
"claim_scope": (
"diagnostic_only; release quality is established by "
"branch-local replay, not cached-AR byte identity"
),
},
"primary_endpoint": primary,
"secondary_target_forward_endpoint": companions[
"output_tokens_per_target_forward"
],
"secondary_draft_issued_proxy_endpoint": companions[
"drafts_issued_per_output_token"
],
"secondary_draft_work_endpoint": companions[
"draft_recursions_per_output_token"
],
"wall_clock_note": (
"This reference recomputes full prefixes and has no rollback-capable "
"KV cache. Wall time is recorded for audit, not claimed as deployment "
"latency."
),
"endpoint_scope_note": (
"The primary endpoint measures accepted drafts per verification. "
"It does not establish verification-width cost or deployment "
"latency. Draft recursions per output token is the registered "
"drafter-work companion; issued drafts per output token is retained "
"only as an issuance proxy. Target forwards exclude the added "
"post-hoc branch-replay forward and independent verification pass."
),
}
return {
"report": report,
"receipt": receipt,
"receipt_evidence": receipt_evidence,
"checkpoint": checkpoint,
"tokenizer_path": tokenizer_path,
"holdout_path": holdout_path,
}
def build_replay_attestation_fixture(root, fixture):
source_path = os.path.join(REPO_ROOT, "scripts", "verify_rollout_replay.py")
source = {
"path": "scripts/verify_rollout_replay.py",
"bytes": os.path.getsize(source_path),
"sha256": _file_sha256(source_path),
}
registration = {
"method": V3_REPLAY_VERIFIER_METHOD,
"execution_argv": list(V3_REPLAY_VERIFICATION_ARGV),
"source": source,
}
fixture["receipt"]["independent_replay_verification"] = registration
report = fixture["report"]
rows = []
expected = []
calibration = report["calibration"]
for local_index in range(calibration["documents"]):
for policy in calibration["candidate_order"]:
expected.append(
(
"calibration",
policy,
calibration["rows"][policy][local_index],
)
)
selected = [
calibration["selected_fixed"]["policy"],
calibration["selected_adaptive"]["policy"],
]
for local_index in range(report["test"]["documents"]):
for policy in selected:
expected.append(
(
"test",
policy,
report["test"]["rows"][policy][local_index],
)
)
aggregate = {
"policy_documents": 0,
"output_tokens": 0,
"exact_argmax_tokens": 0,
"certified_near_tie_tokens": 0,
"failed_tokens": 0,
"reproduced_policy_documents": 0,
"failed_reproductions": 0,
}
for index, (split_name, policy, producer) in enumerate(expected):
payload = report["holdout"]["pair_payload_manifest"][
producer["pair_index"]
]
output = producer["output_token_ids"]
rows.append({
"verification_index": index,
"split": split_name,
"policy": policy,
"pair_index": producer["pair_index"],
"document_id": producer["document_id"],
"seed": producer["seed"],
"prompt_sha256": producer["prompt_sha256"],
"output_sha256": producer["output_sha256"],
"reproduction": {
"output_sha256": producer["output_sha256"],
"trace_sha256": producer["trace_sha256"],
"deterministic_stats_sha256": "d" * 64,
"output_tokens_match": True,
"generation_trace_matches": True,
"deterministic_stats_match": True,
},
"branch_quality": {
"input_sha256": token_ids_sha256(
payload["prompt_token_ids"] + output[:-1]
),
"output_sha256": producer["output_sha256"],
"reference_argmax_sha256": producer["output_sha256"],
"aligned_logits_float32_le_sha256": "e" * 64,
"exact_argmax_tokens": len(output),
"certified_near_tie_tokens": 0,
"failed_tokens": 0,
"certified_near_ties": [],
"failures": [],
"passed": True,
},
})
aggregate["policy_documents"] += 1
aggregate["output_tokens"] += len(output)
aggregate["exact_argmax_tokens"] += len(output)
aggregate["reproduced_policy_documents"] += 1
completed = report["execution"]["completed_at"]
rollout_path = os.path.join(root, "rollout.registered.v3.json")
receipt_path = os.path.join(root, "eval_rollout_receipt_v3.json")
rollout_sha256 = "9" * 64
receipt_sha256 = fixture["report"]["receipt"]["sha256"]
attestation = {
"schema_version": 1,
"instrument_version": V3_INSTRUMENT_VERSION,
"publication_ready": True,
"branch_quality_independently_verified": True,
"rollout_execution_reproduced": True,
"passed": True,
"provenance_scope": V3_ATTESTATION_PROVENANCE_SCOPE,
"verification": {
"method": V3_REPLAY_VERIFIER_METHOD,
"started_at": completed,
"completed_at": completed,
"argv": list(V3_REPLAY_VERIFICATION_ARGV),
"runtime": {**RUNTIME, "platform": "fixture-platform"},
"timing_scope": (
"producer elapsed_seconds, tok_per_sec, and ar_tok_per_sec "
"are excluded from deterministic reproduction"
),
},
"time_order": {
"registered_at": fixture["receipt"]["registered_at"],
"registration_committed_at": fixture["report"]["receipt"][
"registration_git"
]["committed_at"],
"report_started_at": report["execution"]["started_at"],
"report_completed_at": completed,
"verification_started_at": completed,
"verification_completed_at": completed,
},
"report": {"path": rollout_path, "sha256": rollout_sha256},
"receipt": {
"path": receipt_path,
"sha256": receipt_sha256,
"registered_at": fixture["receipt"]["registered_at"],
"registration_git": fixture["report"]["receipt"][
"registration_git"
],
},
"checkpoint": fixture["checkpoint"],
"verifier": {
"method": V3_REPLAY_VERIFIER_METHOD,
"execution_argv": list(V3_REPLAY_VERIFICATION_ARGV),
"registered_source": source,
"live_source": source,
},
"frozen_inputs": {
"acceptance_receipt": dict(
fixture["receipt"]["acceptance_receipt"]
),
"holdout": dict(fixture["receipt"]["holdout"]),
"tokenizer": dict(fixture["receipt"]["tokenizer"]),
"pair_count": fixture["receipt"]["pair_settings"]["examples"],
"pair_manifest_sha256": fixture["receipt"]["pair_settings"][
"pair_manifest_sha256"
],
"pair_payload_manifest_sha256": fixture["receipt"][
"pair_settings"
]["pair_payload_manifest_sha256"],
},
"row_manifest_sha256": canonical_json_sha256(
rows, b"WISP_E3_V3_INDEPENDENT_ROW_MANIFEST\0"
),
"rows": rows,
"aggregate": aggregate,
}
attestation_path = os.path.join(root, "rollout.replay-verification.v3.json")
with open(attestation_path, "w", encoding="utf-8") as handle:
json.dump(attestation, handle, sort_keys=True)
return {
"attestation": attestation,
"path": attestation_path,
"rollout_path": rollout_path,
"rollout_sha256": rollout_sha256,
"receipt_path": receipt_path,
"receipt_sha256": receipt_sha256,
}
def _validate(fixture, report):
return validate_rollout_report(
report,
fixture["receipt"],
fixture["receipt_evidence"],
fixture["checkpoint"],
fixture["tokenizer_path"],
fixture["holdout_path"],
)
def _expect_reject(fixture, report, substring):
try:
_validate(fixture, report)
except ValueError as error:
assert substring in str(error), (
f"expected {substring!r} in {error!r}"
)
else:
raise AssertionError(f"mutation unexpectedly passed: {substring}")
def main():
with tempfile.TemporaryDirectory() as root:
fixture = build_v3_rollout_fixture(root)
report = fixture["report"]
result = _validate(fixture, report)
assert result["branch_local_replay_valid"] is True
assert result["cross_policy_trajectory_identical"] is True
assert result["quality_equivalent_to_greedy_ar"] is False
verification = build_replay_attestation_fixture(root, fixture)
def validate_attestation(value):
return validate_rollout_replay_attestation(
value,
verification["path"],
verification["rollout_path"],
verification["rollout_sha256"],
verification["receipt_path"],
verification["receipt_sha256"],
fixture["receipt"],
report,
fixture["checkpoint"],
REPO_ROOT,
)
attestation_result = validate_attestation(
verification["attestation"]
)
assert (
attestation_result["branch_quality_independently_verified"] is True
)
assert attestation_result["rollout_execution_reproduced"] is True
for mutate, substring in (
(
lambda value: value.__setitem__(
"rollout_execution_reproduced", False
),
"did not pass",
),
(
lambda value: value["report"].__setitem__(
"sha256", "0" * 64
),
"different rollout report",
),
(
lambda value: value.__setitem__(
"provenance_scope", "externally trusted"
),
"provenance scope",
),
(
lambda value: value["rows"][0]["reproduction"].__setitem__(
"generation_trace_matches", False
),
"did not reproduce",
),
(
lambda value: value["rows"][0]["branch_quality"].__setitem__(
"failed_tokens", 1
),
"branch counts",
),
(
lambda value: value.__setitem__(
"row_manifest_sha256", "0" * 64
),
"row-manifest hash",
),
):
bad_attestation = copy.deepcopy(verification["attestation"])
mutate(bad_attestation)
try:
validate_attestation(bad_attestation)
except ValueError as error:
assert substring in str(error), (
f"expected {substring!r} in {error!r}"
)
else:
raise AssertionError(
f"mutated independent attestation passed: {substring}"
)
copied = copy.deepcopy(report)
copied["calibration"]["rows"]["fixed_d2"][1] = copy.deepcopy(
copied["calibration"]["rows"]["fixed_d2"][0]
)
_expect_reject(fixture, copied, "differs on pair_index")
bad = copy.deepcopy(report)
bad["holdout"]["pair_payload_manifest"][0]["target_sha256"] = "0" * 64
_expect_reject(fixture, bad, "differs on target_sha256")
bad = copy.deepcopy(report)
bad["calibration"]["rows"]["fixed_d2"][0][
"output_sha256"
] = "0" * 64
_expect_reject(fixture, bad, "output hash")
bad = copy.deepcopy(report)
row = bad["calibration"]["rows"]["fixed_d2"][0]
row["generation_trace"]["cycles"][0]["verification"][
"base_row_index"
] += 1
resign_trace(row)
_expect_reject(fixture, bad, "verification binding")
bad = copy.deepcopy(report)
row = bad["calibration"]["rows"]["fixed_d2"][0]
row["generation_trace"]["cycles"][0]["verification"]["outcomes"][1][
"target_row_index"
] += 1
resign_trace(row)
_expect_reject(fixture, bad, "off by one")
bad = copy.deepcopy(report)
row = bad["calibration"]["rows"]["fixed_d2"][0]
row["generation_trace"]["cycles"][0][
"state_source"
] = "verification_reuse"
resign_trace(row)
_expect_reject(fixture, bad, "stale state")
bad = copy.deepcopy(report)
replay = bad["calibration"]["rows"]["fixed_d2"][0]["branch_replay"]
replay["reference_argmax_token_ids"][-1] = 6
replay["reference_argmax_sha256"] = token_ids_sha256(
replay["reference_argmax_token_ids"]
)
_expect_reject(fixture, bad, "complete ordered mismatch set")
bad = copy.deepcopy(report)
payload = bad["holdout"]["pair_payload_manifest"][2]
alternate = make_row(
[3, 4, 6],
[3, 4, 6],
pair_index=payload["index"],
policy="adaptive_h0.2",
)
alternate["seed"] = 2718
alternate["target_sha256"] = payload["target_sha256"]
alternate["target_position_accuracy"] = 2 / 3
alternate["target_common_prefix_tokens"] = 2
alternate["target_exact_match"] = False
alternate["rollout"]["policy"] = "adaptive"
alternate["rollout"]["entropy_threshold"] = 0.2
bad["test"]["rows"]["adaptive_h0.2"][0] = alternate
_expect_reject(fixture, bad, "cross-policy trajectory differs")
print("release rollout v3 audit: PASS")
if __name__ == "__main__":
main()