"""CPU-only behavioral-fidelity comparison for frozen OLMoE arm results.""" from __future__ import annotations import hashlib import json import math import re from collections.abc import Mapping, Sequence from dataclasses import dataclass from itertools import zip_longest from pathlib import Path NUMERIC_TOLERANCE = 0.000001 ARM_SCHEMA = "horizon.olmoe-0924.behavioral-fidelity.arm-result.v1" PAIR_SCHEMA = "horizon.olmoe-0924.behavioral-fidelity.pairs.v1" SUMMARY_SCHEMA = "horizon.olmoe-0924.behavioral-fidelity.summary.v2" MANIFEST_SCHEMA = "horizon.olmoe-0924.behavioral-fidelity.public-manifest.v1" CATEGORIES = ( "exact-token-parity", "semantic-json-parity", "horizon-contract-regression", "horizon-contract-improvement", "semantic-divergence", "non-directional-divergence", ) _PUBLIC_FILE_NAMES = frozenset( { "native-results.json", "horizon-results.json", "paired-results.json", "summary.json", "report.md", "manifest.json", } ) _PUBLIC_ROW_FIELDS = ( "schema_version", "arm_id", "prompt_id", "prompt", "input_token_count", "input_token_sha256", "output_token_ids", "output_text", "generated_tokens", "finish_reason", "status", "error_reason", "observations", ) @dataclass(frozen=True, slots=True) class ParsedJsonObject: valid: bool value: dict[str, object] | None reason: str | None class _DuplicateKey(ValueError): pass class _NonFiniteNumber(ValueError): pass def _object_without_duplicates(pairs: list[tuple[str, object]]) -> dict[str, object]: result: dict[str, object] = {} for key, value in pairs: if key in result: raise _DuplicateKey(key) result[key] = value return result def _has_non_finite_number(value: object) -> bool: if type(value) is float: return not math.isfinite(value) if type(value) is dict: return any(_has_non_finite_number(item) for item in value.values()) if type(value) is list: return any(_has_non_finite_number(item) for item in value) return False def parse_json_object(text: object) -> ParsedJsonObject: """Parse exactly one finite JSON object without duplicate keys.""" if type(text) is not str or not text.strip(): return ParsedJsonObject(False, None, "empty-response") try: value = json.loads( text.strip(), object_pairs_hook=_object_without_duplicates, parse_constant=lambda constant: (_ for _ in ()).throw( _NonFiniteNumber(constant) ), ) except _DuplicateKey: return ParsedJsonObject(False, None, "duplicate-key") except _NonFiniteNumber: return ParsedJsonObject(False, None, "non-finite-number") except (json.JSONDecodeError, UnicodeError): return ParsedJsonObject(False, None, "malformed-json") if type(value) is not dict: return ParsedJsonObject(False, None, "root-not-object") if _has_non_finite_number(value): return ParsedJsonObject(False, None, "non-finite-number") return ParsedJsonObject(True, value, None) def semantic_json_equal( left: object, right: object, tolerance: float = NUMERIC_TOLERANCE, ) -> bool: """Compare parsed JSON values with strict types and bounded numeric drift.""" if isinstance(tolerance, bool) or not isinstance(tolerance, int | float): return False if not math.isfinite(tolerance) or tolerance < 0: return False if type(left) in (int, float) and type(right) in (int, float): return ( math.isfinite(left) and math.isfinite(right) and abs(left - right) <= tolerance ) if type(left) is not type(right): return False if type(left) is dict: if left.keys() != right.keys(): return False return all(semantic_json_equal(left[key], right[key], tolerance) for key in left) if type(left) is list: return len(left) == len(right) and all( semantic_json_equal(left_item, right_item, tolerance) for left_item, right_item in zip(left, right, strict=True) ) return left == right def first_token_divergence(left: Sequence[int], right: Sequence[int]) -> int | None: """Return the zero-based first unequal token offset, including length drift.""" sentinel = object() for index, (left_token, right_token) in enumerate( zip_longest(left, right, fillvalue=sentinel) ): if left_token != right_token: return index return None def response_contract_valid(row: Mapping[str, object]) -> bool: return ( row.get("status") == "completed" and row.get("finish_reason") == "eos" and type(row.get("output_text")) is str and bool(row["output_text"].strip()) and parse_json_object(row["output_text"]).valid ) def _token_ids(row: Mapping[str, object]) -> tuple[int, ...]: value = row.get("output_token_ids") if type(value) is not list or any(type(token_id) is not int for token_id in value): return () return tuple(value) def classify_pair( prompt_id: str, native: Mapping[str, object], horizon: Mapping[str, object], ) -> dict[str, object]: """Classify one prompt pair without accepting any expected-answer input.""" native_tokens = _token_ids(native) horizon_tokens = _token_ids(horizon) native_parsed = parse_json_object(native.get("output_text")) horizon_parsed = parse_json_object(horizon.get("output_text")) native_contract_valid = response_contract_valid(native) horizon_contract_valid = response_contract_valid(horizon) semantic_equal = bool( native_contract_valid and horizon_contract_valid and native_parsed.value is not None and horizon_parsed.value is not None and semantic_json_equal(native_parsed.value, horizon_parsed.value) ) both_completed = ( native.get("status") == "completed" and horizon.get("status") == "completed" ) token_equal = native_tokens == horizon_tokens native_finish = native.get("finish_reason") horizon_finish = horizon.get("finish_reason") if ( both_completed and bool(native_tokens) and token_equal and native_finish == horizon_finish ): category = "exact-token-parity" elif native_contract_valid and horizon_contract_valid and semantic_equal: category = "semantic-json-parity" elif native_contract_valid and not horizon_contract_valid: category = "horizon-contract-regression" elif horizon_contract_valid and not native_contract_valid: category = "horizon-contract-improvement" elif native_contract_valid and horizon_contract_valid: category = "semantic-divergence" else: category = "non-directional-divergence" divergence = first_token_divergence(native_tokens, horizon_tokens) common_prefix_length = ( len(native_tokens) if divergence is None else divergence ) return { "prompt_id": prompt_id, "category": category, "exact_token_equal": token_equal, "semantic_json_equal": semantic_equal, "first_token_divergence": divergence, "common_prefix_length": common_prefix_length, "native_contract_valid": native_contract_valid, "horizon_contract_valid": horizon_contract_valid, "native_parse_valid": native_parsed.valid, "native_parse_reason": native_parsed.reason, "horizon_parse_valid": horizon_parsed.valid, "horizon_parse_reason": horizon_parsed.reason, "native_finish_reason": native_finish, "horizon_finish_reason": horizon_finish, } def validate_arm_records( prompt_ids: Sequence[str], arm_id: str, rows: Sequence[Mapping[str, object]], ) -> tuple[Mapping[str, object], ...]: """Require exactly one ordered row for every supplied prompt identifier.""" expected = tuple(prompt_ids) if ( not expected or any(type(prompt_id) is not str or not prompt_id for prompt_id in expected) or len(set(expected)) != len(expected) or type(arm_id) is not str or not arm_id or len(rows) != len(expected) or any(not isinstance(row, Mapping) for row in rows) or tuple(row.get("prompt_id") for row in rows) != expected or any(row.get("arm_id") != arm_id for row in rows) ): raise ValueError("behavioral_fidelity_arm_records_invalid") return tuple(rows) def compare_arms( prompt_ids: Sequence[str], native_rows: Sequence[Mapping[str, object]], horizon_rows: Sequence[Mapping[str, object]], ) -> list[dict[str, object]]: native = validate_arm_records(prompt_ids, "native-bf16", native_rows) horizon = validate_arm_records(prompt_ids, "horizon-c64", horizon_rows) return [ classify_pair(prompt_id, native_row, horizon_row) for prompt_id, native_row, horizon_row in zip( prompt_ids, native, horizon, strict=True ) ] def summarize_pairs(pairs: Sequence[Mapping[str, object]]) -> dict[str, object]: counts = {category: 0 for category in CATEGORIES} prompt_ids: list[str] = [] native_contract_valid_cases = 0 horizon_contract_valid_cases = 0 both_contract_valid_cases = 0 for pair in pairs: category = pair.get("category") prompt_id = pair.get("prompt_id") native_valid = pair.get("native_contract_valid") horizon_valid = pair.get("horizon_contract_valid") if ( category not in counts or type(prompt_id) is not str or not prompt_id or type(native_valid) is not bool or type(horizon_valid) is not bool ): raise ValueError("behavioral_fidelity_pair_invalid") counts[category] += 1 prompt_ids.append(prompt_id) native_contract_valid_cases += int(native_valid) horizon_contract_valid_cases += int(horizon_valid) both_contract_valid_cases += int(native_valid and horizon_valid) if len(prompt_ids) != len(set(prompt_ids)): raise ValueError("behavioral_fidelity_pair_invalid") return { "schema_version": SUMMARY_SCHEMA, "prompt_count": len(pairs), "category_counts": counts, "exact_token_parity": counts["exact-token-parity"], "semantic_json_parity": counts["semantic-json-parity"], "observable_behavior_parity": ( counts["exact-token-parity"] + counts["semantic-json-parity"] ), "horizon_contract_regressions": counts["horizon-contract-regression"], "native_contract_valid_cases": native_contract_valid_cases, "horizon_contract_valid_cases": horizon_contract_valid_cases, "both_contract_valid_cases": both_contract_valid_cases, "horizon_regression_eligible_cases": native_contract_valid_cases, "horizon_contract_improvements": counts["horizon-contract-improvement"], "directionless_divergences": ( counts["semantic-divergence"] + counts["non-directional-divergence"] ), } def _interesting_prompt_ids( pairs: Sequence[Mapping[str, object]], arm_results: Mapping[str, Mapping[str, object]], ) -> tuple[str, ...]: selected = { str(pair["prompt_id"]) for pair in pairs if pair["category"] in { "horizon-contract-regression", "semantic-divergence", "non-directional-divergence", } } for arm in arm_results.values(): for row in arm["rows"]: if ( row.get("status") in {"error", "not-run"} or row.get("finish_reason") == "length" or not str(row.get("output_text", "")).strip() ): selected.add(str(row["prompt_id"])) return tuple( str(pair["prompt_id"]) for pair in pairs if str(pair["prompt_id"]) in selected ) def render_report( contract: Mapping[str, object], arm_results: Mapping[str, Mapping[str, object]], pairs: Sequence[Mapping[str, object]], summary: Mapping[str, object], ) -> str: """Render the bounded public result without adjudicating model answers.""" prompt_count = summary["prompt_count"] lines = [ "# OLMoE 0924 behavioral fidelity", "", ] pilot = contract.get("design", {}).get("mode") == "pilot10" if pilot: lines.extend( [ "PILOTO OPERACIONAL NÃO PROBATÓRIO — 10 PARES. NÃO USAR PARA " "AFIRMAÇÕES DE QUALIDADE OU PARIDADE.", "", ] ) if pilot: lines.extend( [ "## Pilot scorer exercise — non-probative", "", "The counts below exercise the frozen scorer and publication path only. " "They are not a campaign result and do not support a quality or parity claim.", "", ] ) else: lines.extend( [ "## Bounded result", "", ( f"On OLMoE 0924 under these {prompt_count} predefined structured " "prompts, Horizon matched the BF16 reference's observable response " f"behavior in {summary['observable_behavior_parity']}/{prompt_count} " "pairs and produced " f"{summary['horizon_contract_regressions']}/" f"{summary['horizon_regression_eligible_cases']} objective response-" "contract regressions among BF16 contract-valid reference cases " f"({summary['horizon_contract_regressions']}/{prompt_count} overall). " f"There were {summary['directionless_divergences']}/{prompt_count} " "divergences whose quality direction was not determined." ), "", ] ) lines.extend( [ "## Objective denominators", "", f"- All predefined pairs: {prompt_count}/{prompt_count}", f"- BF16 contract-valid reference cases: " f"{summary['native_contract_valid_cases']}/{prompt_count}", f"- Horizon contract-valid cases: " f"{summary['horizon_contract_valid_cases']}/{prompt_count}", f"- Both arms contract-valid: " f"{summary['both_contract_valid_cases']}/{prompt_count}", "", "## Categories", "", "| Category | Count |", "| --- | ---: |", ] ) counts = summary["category_counts"] lines.extend(f"| `{category}` | {counts[category]} |" for category in CATEGORIES) lines.extend(["", "## Arm configuration", ""]) for arm_id in ("native-bf16", "horizon-c64"): arm = arm_results[arm_id] lines.extend( [ f"### {arm_id}", "", "```json", json.dumps( { "configuration": arm.get("configuration", {}), "identity": arm.get("identity", {}), "cleanup": arm.get("cleanup", {}), "valid": arm.get("valid"), }, indent=2, sort_keys=True, ensure_ascii=False, ), "```", "", ] ) by_prompt = { arm_id: {str(row["prompt_id"]): row for row in arm["rows"]} for arm_id, arm in arm_results.items() } lines.extend(["## Flagged pairs", ""]) interesting = _interesting_prompt_ids(pairs, arm_results) if not interesting: lines.extend(["None.", ""]) pair_by_prompt = {str(pair["prompt_id"]): pair for pair in pairs} for prompt_id in interesting: pair = pair_by_prompt[prompt_id] lines.extend([f"### {prompt_id}", "", f"Category: `{pair['category']}`", ""]) for arm_id in ("native-bf16", "horizon-c64"): row = by_prompt[arm_id][prompt_id] lines.extend( [ f"{arm_id}: status `{row.get('status')}`, finish " f"`{row.get('finish_reason')}`, parse " f"`{pair[arm_id.split('-')[0] + '_parse_reason']}`.", "", "```text", str(row.get("output_text", "")), "```", "", ] ) lines.extend( [ "## Contract and limitations", "", "The scorer did not load expected answers and makes no absolute correctness " "judgment. Semantic and non-directional divergences have no assigned winner.", "", "The result is limited to the exact checkpoint, tokenizer, prompts, generation " "settings, arm configurations, and recorded execution contract.", "", "Contract design:", "", "```json", json.dumps(contract.get("design", {}), indent=2, sort_keys=True, ensure_ascii=False), "```", "", ] ) return "\n".join(lines) def _sha256(path: Path) -> str: with path.open("rb") as stream: return hashlib.file_digest(stream, "sha256").hexdigest() def _write_json_exclusive(path: Path, value: object) -> None: with path.open("x", encoding="utf-8", newline="\n") as stream: json.dump(value, stream, indent=2, sort_keys=True, ensure_ascii=False, allow_nan=False) stream.write("\n") def _load_object(path: Path) -> dict[str, object]: value = json.loads(path.read_bytes()) if type(value) is not dict: raise ValueError("behavioral_fidelity_json_object_required") return value def public_arm_result(value: Mapping[str, object]) -> dict[str, object]: rows = value.get("rows") if type(rows) is not list: raise ValueError("behavioral_fidelity_arm_result_invalid") return { "schema_version": ARM_SCHEMA, "arm_id": value.get("arm_id"), "contract_sha256": value.get("contract_sha256"), "valid": value.get("valid"), "configuration": value.get("configuration", {}), "identity": value.get("identity", {}), "cleanup": value.get("cleanup", {}), "rows": [ {field: row[field] for field in _PUBLIC_ROW_FIELDS if field in row} for row in rows ], } _WINDOWS_PATH = re.compile(r"(?i)(?:^|[\s\"'(])(?:[a-z]:[\\/])") _UNC_PATH = re.compile(r"\\\\[^\\\s]+[\\/]") _URI_CREDENTIAL = re.compile(r"[a-z][a-z0-9+.-]*://[^/\s:@]+:[^/\s@]+@", re.I) def _assert_public_files_safe(public_root: Path) -> None: for path in sorted(public_root.iterdir()): if not path.is_file() or path.is_symlink(): raise ValueError("behavioral_fidelity_public_content_unsafe") text = path.read_text(encoding="utf-8") if ( _WINDOWS_PATH.search(text) or _UNC_PATH.search(text) or _URI_CREDENTIAL.search(text) or "traceback (most recent call last):" in text.casefold() ): raise ValueError("behavioral_fidelity_public_content_unsafe") def _contract_prompt_ids(contract: Mapping[str, object]) -> tuple[str, ...]: design = contract.get("design") inputs = contract.get("prompt_inputs") candidates = ( design.get("prompt_ids") if isinstance(design, Mapping) else None, inputs.get("prompt_ids") if isinstance(inputs, Mapping) else None, ) prompt_ids = next((value for value in candidates if type(value) is list), None) if prompt_ids is None or any(type(value) is not str for value in prompt_ids): raise ValueError("behavioral_fidelity_contract_prompt_ids_invalid") return tuple(prompt_ids) def publish_analysis( run_root: Path, contract_path: Path, contract_sha256: str, *, public_root: Path | None = None, ) -> Path: """Publish a create-only, path-free analysis directory from retained arm rows.""" selected_run_root = run_root.resolve() selected_contract = contract_path.resolve() if _sha256(selected_contract) != contract_sha256: raise ValueError("behavioral_fidelity_contract_digest_mismatch") contract = _load_object(selected_contract) prompt_ids = _contract_prompt_ids(contract) private_arms = { arm_id: _load_object(selected_run_root / arm_id / "arm-result.json") for arm_id in ("native-bf16", "horizon-c64") } if any( arm.get("schema_version") != ARM_SCHEMA or arm.get("arm_id") != arm_id or arm.get("contract_sha256") != contract_sha256 or arm.get("valid") is not True for arm_id, arm in private_arms.items() ): raise ValueError("behavioral_fidelity_claim_publication_requires_valid_arms") arm_results = { arm_id: public_arm_result(arm) for arm_id, arm in private_arms.items() } native_rows = validate_arm_records( prompt_ids, "native-bf16", arm_results["native-bf16"]["rows"] ) horizon_rows = validate_arm_records( prompt_ids, "horizon-c64", arm_results["horizon-c64"]["rows"] ) pairs = compare_arms(prompt_ids, native_rows, horizon_rows) summary = summarize_pairs(pairs) summary["contract_sha256"] = contract_sha256 selected_public_root = ( selected_run_root / "public" if public_root is None else public_root.resolve() ) if selected_public_root.exists(): raise FileExistsError(str(selected_public_root)) candidate_root = selected_public_root.with_name( f".{selected_public_root.name}.candidate" ) candidate_root.mkdir(parents=False, exist_ok=False) _write_json_exclusive( candidate_root / "native-results.json", arm_results["native-bf16"] ) _write_json_exclusive( candidate_root / "horizon-results.json", arm_results["horizon-c64"] ) _write_json_exclusive( candidate_root / "paired-results.json", { "schema_version": PAIR_SCHEMA, "contract_sha256": contract_sha256, "pairs": pairs, }, ) _write_json_exclusive(candidate_root / "summary.json", summary) with (candidate_root / "report.md").open( "x", encoding="utf-8", newline="\n" ) as stream: stream.write(render_report(contract, arm_results, pairs, summary)) _assert_public_files_safe(candidate_root) retained = sorted(path.name for path in candidate_root.iterdir()) manifest = { "schema_version": MANIFEST_SCHEMA, "contract_sha256": contract_sha256, "files": [ {"path": name, "sha256": _sha256(candidate_root / name)} for name in retained ], } _write_json_exclusive(candidate_root / "manifest.json", manifest) verify_public_bundle(candidate_root) candidate_root.rename(selected_public_root) verify_public_bundle(selected_public_root) return selected_public_root def verify_public_bundle(public_root: Path) -> bool: selected_root = public_root.resolve() names = {path.name for path in selected_root.iterdir()} if names != _PUBLIC_FILE_NAMES: raise ValueError("behavioral_fidelity_public_file_set_invalid") _assert_public_files_safe(selected_root) manifest = _load_object(selected_root / "manifest.json") entries = manifest.get("files") if manifest.get("schema_version") != MANIFEST_SCHEMA or type(entries) is not list: raise ValueError("behavioral_fidelity_manifest_invalid") expected_names = _PUBLIC_FILE_NAMES - {"manifest.json"} if ( {entry.get("path") for entry in entries if isinstance(entry, Mapping)} != expected_names or len(entries) != len(expected_names) ): raise ValueError("behavioral_fidelity_manifest_invalid") for entry in entries: if ( not isinstance(entry, Mapping) or type(entry.get("path")) is not str or type(entry.get("sha256")) is not str or _sha256(selected_root / entry["path"]) != entry["sha256"] ): raise ValueError("behavioral_fidelity_manifest_digest_mismatch") contract_sha256 = manifest.get("contract_sha256") native = _load_object(selected_root / "native-results.json") horizon = _load_object(selected_root / "horizon-results.json") paired = _load_object(selected_root / "paired-results.json") summary = _load_object(selected_root / "summary.json") if ( type(contract_sha256) is not str or len(contract_sha256) != 64 or any(character not in "0123456789abcdef" for character in contract_sha256) or native.get("schema_version") != ARM_SCHEMA or native.get("arm_id") != "native-bf16" or native.get("contract_sha256") != contract_sha256 or native.get("valid") is not True or horizon.get("schema_version") != ARM_SCHEMA or horizon.get("arm_id") != "horizon-c64" or horizon.get("contract_sha256") != contract_sha256 or horizon.get("valid") is not True or paired.get("schema_version") != PAIR_SCHEMA or paired.get("contract_sha256") != contract_sha256 or summary.get("schema_version") != SUMMARY_SCHEMA or summary.get("contract_sha256") != contract_sha256 or type(native.get("rows")) is not list or type(horizon.get("rows")) is not list or type(paired.get("pairs")) is not list ): raise ValueError("behavioral_fidelity_public_claim_content_invalid") prompt_ids = tuple(row.get("prompt_id") for row in native["rows"]) try: native_rows = validate_arm_records( prompt_ids, "native-bf16", native["rows"] ) horizon_rows = validate_arm_records( prompt_ids, "horizon-c64", horizon["rows"] ) expected_pairs = compare_arms(prompt_ids, native_rows, horizon_rows) expected_summary = summarize_pairs(expected_pairs) expected_summary["contract_sha256"] = contract_sha256 except ValueError as error: raise ValueError("behavioral_fidelity_public_claim_content_invalid") from error if paired["pairs"] != expected_pairs or summary != expected_summary: raise ValueError("behavioral_fidelity_public_claim_content_invalid") return True __all__ = [ "ARM_SCHEMA", "CATEGORIES", "MANIFEST_SCHEMA", "NUMERIC_TOLERANCE", "PAIR_SCHEMA", "SUMMARY_SCHEMA", "ParsedJsonObject", "classify_pair", "compare_arms", "first_token_divergence", "parse_json_object", "public_arm_result", "publish_analysis", "render_report", "response_contract_valid", "semantic_json_equal", "summarize_pairs", "validate_arm_records", "verify_public_bundle", ]