from __future__ import annotations import csv import math from pathlib import Path from statistics import mean BASE = Path("frame_audit_001_bfcl") INPUT = BASE / "data" / "processed" / "bfcl_v4_baseline.csv" PROCESSED = BASE / "data" / "processed" OUTPUTS = BASE / "outputs" PROFILE_RESULTS = PROCESSED / "bfcl_profile_results.csv" SENSITIVITY_RESULTS = PROCESSED / "bfcl_sensitivity_results.csv" DISQUALIFICATION_LOG = PROCESSED / "bfcl_disqualification_log.csv" PROFILE_REPORT = OUTPUTS / "BFCL_PROFILE_RESULTS_REPORT.md" FLIP_SUMMARY = OUTPUTS / "BFCL_DECISION_FLIP_SUMMARY.md" SCORING_TASK_COUNT = 5088 def f(value: str | None) -> float | None: if value is None: return None text = str(value).strip() if text == "": return None try: return float(text) except ValueError: return None def b(value: str | None) -> bool: return str(value).strip().lower() == "true" def avg(values: list[float | None]) -> float | None: cleaned = [v for v in values if v is not None] if len(cleaned) != len(values) or not cleaned: return None return sum(cleaned) / len(cleaned) def safe_pct(value: float | None) -> float: if value is None: return 0.0 return max(0.0, min(100.0, value)) def cost_per_success(cost: float | None, success_pct: float | None) -> float | None: if cost is None or success_pct is None or success_pct <= 0: return None return cost / (SCORING_TASK_COUNT * (success_pct / 100.0)) def latency_score(p95: float | None, target: float) -> float | None: if p95 is None or p95 <= 0: return None return 100.0 / max(1.0, p95 / target) def lower_better_scores(values: dict[str, float | None]) -> dict[str, float | None]: present = {k: v for k, v in values.items() if v is not None and math.isfinite(v)} if not present: return {k: None for k in values} lo = min(present.values()) hi = max(present.values()) if hi == lo: return {k: 100.0 if k in present else None for k in values} return {k: (100.0 * (hi - v) / (hi - lo)) if k in present else None for k, v in values.items()} def higher_better_rank(rows: list[dict], score_field: str) -> None: ordered = sorted( [r for r in rows if r.get("eligible") and r.get(score_field) is not None], key=lambda r: (-r[score_field], int(r["published_rank"])), ) for rank, row in enumerate(ordered, start=1): row["rank"] = rank row["rank_difference_from_published"] = rank - int(row["published_rank"]) def lower_better_rank(rows: list[dict], score_field: str) -> None: ordered = sorted( [r for r in rows if r.get("eligible") and r.get(score_field) is not None], key=lambda r: (r[score_field], int(r["published_rank"])), ) for rank, row in enumerate(ordered, start=1): row["rank"] = rank row["rank_difference_from_published"] = rank - int(row["published_rank"]) def weighted_score(terms: list[tuple[str, float, float | None, bool]]) -> tuple[float | None, str]: used = [(name, weight, value) for name, weight, value, include in terms if include and value is not None] if not used: return None, "" total_weight = sum(weight for _, weight, _ in used) score = sum(weight * safe_pct(value) for _, weight, value in used) / total_weight return score, "; ".join(f"{name}:{weight:g}" for name, weight, _ in used) def row_metrics(row: dict[str, str]) -> dict[str, float | None]: single_turn = avg( [ f(row["non_live_simple_ast"]), f(row["non_live_multiple_ast"]), f(row["non_live_parallel_ast"]), f(row["non_live_parallel_multiple_ast"]), f(row["live_simple_ast"]), f(row["live_multiple_ast"]), f(row["live_parallel_ast"]), f(row["live_parallel_multiple_ast"]), ] ) relevance_behavior = avg([f(row["relevance_detection"]), f(row["hallucination_irrelevance_detection"])]) single_turn_summary = avg([f(row["live_acc"]), f(row["non_live_ast_acc"])]) task_success_d = avg( [ f(row["published_overall_score"]), f(row["non_live_ast_acc"]), f(row["live_acc"]), f(row["multi_turn_acc"]), ] ) format_risk = None if f(row["format_sensitivity_max_delta"]) is not None and f(row["format_sensitivity_standard_deviation"]) is not None: format_risk = f(row["format_sensitivity_max_delta"]) + f(row["format_sensitivity_standard_deviation"]) return { "single_turn": single_turn, "relevance_behavior": relevance_behavior, "single_turn_summary": single_turn_summary, "agentic": f(row["reconstructed_agentic_component"]), "multi_turn": f(row["multi_turn_acc"]), "memory": f(row["memory_acc"]), "web_search": f(row["web_search_acc"]), "task_success_d": task_success_d, "format_risk": format_risk, "format_present": 1.0 if format_risk is not None else 0.0, } def missing(row: dict[str, str], fields: list[str]) -> list[str]: return [field for field in fields if f(row.get(field)) is None] def disqualify_profile(profile_id: str, row: dict[str, str]) -> list[str]: reasons: list[str] = [] if profile_id == "A": required = [ "non_live_simple_ast", "non_live_multiple_ast", "non_live_parallel_ast", "non_live_parallel_multiple_ast", "live_simple_ast", "live_multiple_ast", "live_parallel_ast", "live_parallel_multiple_ast", "relevance_detection", "hallucination_irrelevance_detection", "total_cost_usd", "latency_mean_s", "latency_standard_deviation_s", "latency_95th_percentile_s", ] miss = missing(row, required) if miss: reasons.append("missing required fields: " + "; ".join(miss)) if f(row["hallucination_irrelevance_detection"]) is not None and f(row["hallucination_irrelevance_detection"]) < 70: reasons.append("Irrelevance Detection below 70%") if f(row["latency_95th_percentile_s"]) is not None and f(row["latency_95th_percentile_s"]) > 20: reasons.append("Latency 95th Percentile above 20 seconds") elif profile_id == "B": required = [ "relevance_detection", "hallucination_irrelevance_detection", "multi_turn_base", "multi_turn_miss_func", "multi_turn_miss_param", "multi_turn_long_context", "live_acc", "non_live_ast_acc", "latency_95th_percentile_s", "total_cost_usd", ] miss = missing(row, required) if miss: reasons.append("missing required safety-related fields: " + "; ".join(miss)) if f(row["hallucination_irrelevance_detection"]) is not None and f(row["hallucination_irrelevance_detection"]) < 90: reasons.append("Irrelevance Detection below 90%") if f(row["relevance_detection"]) is not None and f(row["relevance_detection"]) < 85: reasons.append("Relevance Detection below 85%") if f(row["multi_turn_miss_func"]) is not None and f(row["multi_turn_miss_func"]) < 70: reasons.append("Multi Turn Miss Func below 70%") if f(row["latency_95th_percentile_s"]) is not None and f(row["latency_95th_percentile_s"]) > 60: reasons.append("Latency 95th Percentile above 60 seconds") elif profile_id == "C": required = [ "multi_turn_base", "multi_turn_miss_func", "multi_turn_miss_param", "multi_turn_long_context", "memory_kv", "memory_vector", "memory_recursive_summarization", "web_search_base", "web_search_no_snippet", "relevance_detection", "hallucination_irrelevance_detection", "latency_95th_percentile_s", "total_cost_usd", ] miss = missing(row, required) if miss: reasons.append("missing required multi-turn/agentic fields: " + "; ".join(miss)) if f(row["multi_turn_acc"]) is not None and f(row["multi_turn_acc"]) < 60: reasons.append("Multi Turn Acc below 60%") if f(row["latency_95th_percentile_s"]) is not None and f(row["latency_95th_percentile_s"]) > 60: reasons.append("Latency 95th Percentile above 60 seconds") elif profile_id == "D": required = [ "published_overall_score", "non_live_ast_acc", "live_acc", "multi_turn_acc", "relevance_detection", "hallucination_irrelevance_detection", "total_cost_usd", "latency_mean_s", "latency_standard_deviation_s", "latency_95th_percentile_s", ] miss = missing(row, required) if miss: reasons.append("missing required general tool-user fields: " + "; ".join(miss)) if f(row["hallucination_irrelevance_detection"]) is not None and f(row["hallucination_irrelevance_detection"]) < 70: reasons.append("Irrelevance Detection below 70%") if f(row["latency_95th_percentile_s"]) is not None and f(row["latency_95th_percentile_s"]) > 45: reasons.append("Latency 95th Percentile above 45 seconds") return reasons PROFILE_NAMES = { "A": "High-volume low-risk automation", "B": "Regulated or high-consequence workflow", "C": "Multi-turn enterprise agent", "D": "Cost-constrained general tool user", } def profile_success_metric(profile_id: str, metrics: dict[str, float | None]) -> float | None: if profile_id == "A": return metrics["single_turn"] if profile_id == "B": return avg([metrics["relevance_behavior"], metrics["multi_turn"], metrics["agentic"], metrics["single_turn_summary"]]) if profile_id == "C": return metrics["multi_turn"] if profile_id == "D": return metrics["task_success_d"] return None def build_profile_results(rows: list[dict[str, str]]) -> tuple[list[dict], list[dict]]: all_results: list[dict] = [] disq_log: list[dict] = [] for profile_id in ["A", "B", "C", "D"]: prelim: list[dict] = [] cost_values: dict[str, float | None] = {} format_risks: dict[str, float | None] = {} for row in rows: metrics = row_metrics(row) reasons = disqualify_profile(profile_id, row) eligible = not reasons success_metric = profile_success_metric(profile_id, metrics) cps = cost_per_success(f(row["total_cost_usd"]), success_metric) cost_values[row["model_identifier"]] = cps if eligible else None format_risks[row["model_identifier"]] = metrics["format_risk"] if eligible else None prelim.append( { "profile_id": profile_id, "profile_name": PROFILE_NAMES[profile_id], "model_identifier": row["model_identifier"], "published_rank": int(row["official_rank"]), "published_overall_score": f(row["published_overall_score"]), "eligible": eligible, "disqualification_reason": "; ".join(reasons), "total_cost_usd": f(row["total_cost_usd"]), "latency_mean_s": f(row["latency_mean_s"]), "latency_95th_percentile_s": f(row["latency_95th_percentile_s"]), "success_metric_pct": success_metric, "cost_per_expected_success": cps, "single_turn_score": metrics["single_turn"], "relevance_behavior_score": metrics["relevance_behavior"], "single_turn_summary_score": metrics["single_turn_summary"], "agentic_score": metrics["agentic"], "multi_turn_score": metrics["multi_turn"], "memory_score": metrics["memory"], "web_search_score": metrics["web_search"], "task_success_general_score": metrics["task_success_d"], "format_risk": metrics["format_risk"], "format_present": bool(metrics["format_present"]), } ) if reasons: for reason in reasons: disq_log.append( { "analysis_type": "buyer_profile", "profile_or_test": profile_id, "profile_or_test_name": PROFILE_NAMES[profile_id], "model_identifier": row["model_identifier"], "published_rank": row["official_rank"], "reason": reason, } ) cost_scores = lower_better_scores(cost_values) format_scores = lower_better_scores(format_risks) for result in prelim: if not result["eligible"]: result["profile_weighted_score"] = None result["profile_rank"] = None result["rank_difference_from_published"] = None result["score_terms_used"] = "" all_results.append(result) continue model = result["model_identifier"] p95 = result["latency_95th_percentile_s"] if profile_id == "A": score, terms = weighted_score( [ ("single_turn_tool_call_accuracy", 40, result["single_turn_score"], True), ("relevance_irrelevance_behavior", 20, result["relevance_behavior_score"], True), ("cost_efficiency", 25, cost_scores[model], True), ("latency", 15, latency_score(p95, 10), True), ] ) elif profile_id == "B": cost_latency = avg([cost_scores[model], latency_score(p95, 60)]) score, terms = weighted_score( [ ("hallucination_abstention_behavior", 30, result["relevance_behavior_score"], True), ("multi_turn_correctness", 25, result["multi_turn_score"], True), ("agentic_web_search_memory", 15, result["agentic_score"], True), ("single_turn_tool_call_accuracy", 15, result["single_turn_summary_score"], True), ("format_robustness", 10, format_scores[model], result["format_present"]), ("cost_latency", 5, cost_latency, True), ] ) elif profile_id == "C": score, terms = weighted_score( [ ("multi_turn_performance", 35, result["multi_turn_score"], True), ("agentic_memory_performance", 25, result["memory_score"], True), ("web_search_performance", 15, result["web_search_score"], True), ("relevance_irrelevance_behavior", 10, result["relevance_behavior_score"], True), ("latency", 10, latency_score(p95, 30), True), ("cost", 5, cost_scores[model], True), ] ) elif profile_id == "D": latency_adjusted = None if result["task_success_general_score"] is not None: lat = latency_score(p95, 30) latency_adjusted = result["task_success_general_score"] * (lat / 100.0) if lat is not None else None score, terms = weighted_score( [ ("task_success_general_components", 35, result["task_success_general_score"], True), ("cost_per_expected_success", 35, cost_scores[model], True), ("latency_adjusted_success", 15, latency_adjusted, True), ("relevance_irrelevance_behavior", 10, result["relevance_behavior_score"], True), ("format_robustness", 5, format_scores[model], result["format_present"]), ] ) else: score, terms = None, "" result["profile_weighted_score"] = score result["profile_rank"] = None result["rank_difference_from_published"] = None result["score_terms_used"] = terms all_results.append(result) profile_rows = [r for r in all_results if r["profile_id"] == profile_id and r["eligible"]] higher_better_rank(profile_rows, "profile_weighted_score") rank_by_model = {r["model_identifier"]: r for r in profile_rows} for result in all_results: if result["profile_id"] == profile_id and result["model_identifier"] in rank_by_model: result["profile_rank"] = rank_by_model[result["model_identifier"]]["rank"] result["rank_difference_from_published"] = rank_by_model[result["model_identifier"]]["rank_difference_from_published"] return all_results, disq_log def sensitivity_rows(rows: list[dict[str, str]], profile_results: list[dict]) -> tuple[list[dict], list[dict]]: out: list[dict] = [] disq: list[dict] = [] def base_row(test: str, row: dict[str, str], profile_id: str = "") -> dict: return { "sensitivity_test": test, "profile_id": profile_id, "model_identifier": row["model_identifier"], "published_rank": int(row["official_rank"]), "published_overall_score": f(row["published_overall_score"]), "eligible": True, "sensitivity_score": None, "score_direction": "higher_is_better", "sensitivity_rank": None, "rank_difference_from_published": None, "total_cost_usd": f(row["total_cost_usd"]), "cost_per_expected_success": None, "latency_95th_percentile_s": f(row["latency_95th_percentile_s"]), "latency_adjusted_success": None, "disqualification_reason": "", } equal_rows: list[dict] = [] for row in rows: r = base_row("equal_component_weighting", row) components = [ f(row["reconstructed_agentic_component"]), f(row["multi_turn_acc"]), f(row["live_acc"]), f(row["non_live_ast_acc"]), f(row["hallucination_irrelevance_detection"]), ] if any(v is None for v in components): r["eligible"] = False r["disqualification_reason"] = "missing one or more major BFCL scoring components" else: r["sensitivity_score"] = mean(components) equal_rows.append(r) higher_better_rank(equal_rows, "sensitivity_score") out.extend(equal_rows) for pr in profile_results: r = { "sensitivity_test": "buyer_profile_weighting", "profile_id": pr["profile_id"], "model_identifier": pr["model_identifier"], "published_rank": pr["published_rank"], "published_overall_score": pr["published_overall_score"], "eligible": pr["eligible"], "sensitivity_score": pr["profile_weighted_score"], "score_direction": "higher_is_better", "sensitivity_rank": pr["profile_rank"], "rank_difference_from_published": pr["rank_difference_from_published"], "total_cost_usd": pr["total_cost_usd"], "cost_per_expected_success": pr["cost_per_expected_success"], "latency_95th_percentile_s": pr["latency_95th_percentile_s"], "latency_adjusted_success": None, "disqualification_reason": pr["disqualification_reason"], } out.append(r) for test, predicate, reason_fn in [ ( "hallucination_as_hard_constraint", lambda row: f(row["hallucination_irrelevance_detection"]) is not None and f(row["hallucination_irrelevance_detection"]) >= 90 and f(row["relevance_detection"]) is not None and f(row["relevance_detection"]) >= 85, lambda row: "requires Irrelevance Detection >= 90% and Relevance Detection >= 85%", ), ( "multi_turn_minimum_threshold", lambda row: f(row["multi_turn_acc"]) is not None and f(row["multi_turn_acc"]) >= 60, lambda row: "requires Multi Turn Acc >= 60%", ), ( "exclusion_of_incomplete_model_entries", lambda row: b(row["complete_baseline_components"]) and b(row["complete_cost_latency_fields"]), lambda row: "requires complete baseline components and cost/latency fields", ), ]: rows_for_test: list[dict] = [] for row in rows: r = base_row(test, row) r["sensitivity_score"] = f(row["published_overall_score"]) if not predicate(row): r["eligible"] = False r["sensitivity_score"] = None r["disqualification_reason"] = reason_fn(row) disq.append( { "analysis_type": "sensitivity_test", "profile_or_test": test, "profile_or_test_name": test, "model_identifier": row["model_identifier"], "published_rank": row["official_rank"], "reason": r["disqualification_reason"], } ) rows_for_test.append(r) higher_better_rank(rows_for_test, "sensitivity_score") out.extend(rows_for_test) cost_rows: list[dict] = [] for row in rows: r = base_row("cost_per_successful_call", row) cps = cost_per_success(f(row["total_cost_usd"]), f(row["published_overall_score"])) r["cost_per_expected_success"] = cps r["sensitivity_score"] = cps r["score_direction"] = "lower_is_better" if cps is None: r["eligible"] = False r["disqualification_reason"] = "missing cost or published overall score" cost_rows.append(r) lower_better_rank(cost_rows, "sensitivity_score") out.extend(cost_rows) latency_rows: list[dict] = [] for row in rows: r = base_row("latency_adjusted_success", row) score = f(row["published_overall_score"]) lat = latency_score(f(row["latency_95th_percentile_s"]), 30) if score is None or lat is None: r["eligible"] = False r["disqualification_reason"] = "missing published overall score or latency p95" else: r["latency_adjusted_success"] = score * (lat / 100.0) r["sensitivity_score"] = r["latency_adjusted_success"] latency_rows.append(r) higher_better_rank(latency_rows, "sensitivity_score") out.extend(latency_rows) return out, disq def write_csv(path: Path, rows: list[dict]) -> None: path.parent.mkdir(parents=True, exist_ok=True) if not rows: path.write_text("", encoding="utf-8") return fields: list[str] = [] for row in rows: for key in row: if key not in fields: fields.append(key) with path.open("w", encoding="utf-8", newline="") as f_out: writer = csv.DictWriter(f_out, fieldnames=fields) writer.writeheader() for row in rows: clean = {} for key in fields: value = row.get(key, "") if isinstance(value, float): clean[key] = f"{value:.6f}" elif value is None: clean[key] = "" else: clean[key] = value writer.writerow(clean) def top_eligible(profile_results: list[dict], profile_id: str) -> dict | None: rows = [r for r in profile_results if r["profile_id"] == profile_id and r["eligible"] and r["profile_rank"]] return min(rows, key=lambda r: r["profile_rank"]) if rows else None def top_sensitivity(sens: list[dict], test: str, profile_id: str = "") -> dict | None: rows = [ r for r in sens if r["sensitivity_test"] == test and r.get("profile_id", "") == profile_id and r["eligible"] and r["sensitivity_rank"] ] return min(rows, key=lambda r: r["sensitivity_rank"]) if rows else None def write_reports(rows: list[dict[str, str]], profile_results: list[dict], sens: list[dict]) -> str: published_winner = min(rows, key=lambda r: int(r["official_rank"])) published_winner_name = published_winner["model_identifier"] lines: list[str] = [] lines.append("# BFCL Profile Results Report") lines.append("") lines.append("Status: sensitivity analysis only. No buyer-profile definitions or hard thresholds were changed after loading the results. No paid API calls were made.") lines.append("") lines.append("## Documented BFCL Data") lines.append("") lines.append(f"- Input baseline rows: {len(rows)} model variants.") lines.append("- Used BFCL published overall score, reconstructed agentic component, non-live/live AST scores, multi-turn scores, relevance/irrelevance scores, cost, latency, format-sensitivity fields, organization, license, and reproducibility metadata from `bfcl_v4_baseline.csv`.") lines.append("- Published BFCL rank was used only as the baseline comparator.") lines.append("") lines.append("## Pre-Registered Buyer Assumptions") lines.append("") lines.append("- The four buyer profiles and hard disqualification thresholds were taken from `BFCL_DECISION_PROFILES.md`.") lines.append("- Cost efficiency uses BFCL `Total Cost ($)` divided by expected successful benchmark tasks, then inverse-normalized among eligible models for that profile.") lines.append("- Latency scoring uses the documented profile targets or caps: 10 seconds for Profile A, 60 seconds service cap for Profile B, 30 seconds for Profile C, and 30 seconds for Profile D's moderate latency-adjustment calculation.") lines.append("- Format robustness is used only when BFCL publishes format-sensitivity fields for the variant; when absent, that optional term is omitted and remaining used terms are renormalized. This follows the profile wording `if present for the variant` and the baseline finding that format sensitivity is non-scoring.") lines.append("") lines.append("## Calculated Results") lines.append("") for profile_id in ["A", "B", "C", "D"]: prs = [r for r in profile_results if r["profile_id"] == profile_id] eligible = [r for r in prs if r["eligible"]] disq = [r for r in prs if not r["eligible"]] top = top_eligible(profile_results, profile_id) winner_status = next(r for r in prs if r["model_identifier"] == published_winner_name) lines.append(f"### Profile {profile_id}: {PROFILE_NAMES[profile_id]}") lines.append("") lines.append(f"- Eligible models: {len(eligible)}") lines.append(f"- Disqualified models: {len(disq)}") if top: lines.append(f"- Profile-selected model: {top['model_identifier']}") lines.append(f"- Profile-weighted score: {top['profile_weighted_score']:.2f}") lines.append(f"- Published BFCL rank of selected model: {top['published_rank']}") lines.append(f"- Difference from published rank: {top['rank_difference_from_published']}") lines.append(f"- Cost: ${top['total_cost_usd']:.2f}; latency P95: {top['latency_95th_percentile_s']:.2f}s") else: lines.append("- Profile-selected model: none; no eligible models.") if winner_status["eligible"]: lines.append("- Published aggregate winner status in this profile: eligible.") else: lines.append(f"- Published aggregate winner status in this profile: disqualified ({winner_status['disqualification_reason']}).") lines.append("") top5 = sorted(eligible, key=lambda r: r["profile_rank"])[:5] if top5: lines.append("| Profile rank | Model | Published rank | Profile score | Cost | P95 latency |") lines.append("| ---: | --- | ---: | ---: | ---: | ---: |") for r in top5: lines.append( f"| {r['profile_rank']} | {r['model_identifier']} | {r['published_rank']} | {r['profile_weighted_score']:.2f} | ${r['total_cost_usd']:.2f} | {r['latency_95th_percentile_s']:.2f}s |" ) lines.append("") lines.append("## Sensitivity Tests") lines.append("") for test, label in [ ("equal_component_weighting", "Equal component weighting"), ("hallucination_as_hard_constraint", "Hallucination as a hard constraint"), ("multi_turn_minimum_threshold", "Multi-turn minimum threshold"), ("cost_per_successful_call", "Cost per successful call"), ("latency_adjusted_success", "Latency-adjusted success"), ("exclusion_of_incomplete_model_entries", "Exclusion of incomplete model entries"), ]: top = top_sensitivity(sens, test) eligible_count = sum(1 for r in sens if r["sensitivity_test"] == test and r["eligible"]) disq_count = sum(1 for r in sens if r["sensitivity_test"] == test and not r["eligible"]) if top: lines.append(f"- {label}: selected `{top['model_identifier']}`; eligible {eligible_count}; disqualified {disq_count}; published rank {top['published_rank']}; rank difference {top['rank_difference_from_published']}.") else: lines.append(f"- {label}: no eligible model; eligible {eligible_count}; disqualified {disq_count}.") lines.append("") lines.append("## Interpretation") lines.append("") lines.append("The profile results should be read as decision-specific selectors, not as a replacement public leaderboard. A profile flip counts only where the operational constraint is pre-registered and financially meaningful under `BFCL_PASS_FAIL_RULES.md`.") PROFILE_REPORT.write_text("\n".join(lines) + "\n", encoding="utf-8") flip_lines: list[str] = [] flip_lines.append("# BFCL Decision Flip Summary") flip_lines.append("") flip_lines.append("## Documented BFCL Data") flip_lines.append("") flip_lines.append(f"- Published BFCL aggregate comparator: official rank 1, `{published_winner_name}`.") flip_lines.append(f"- Model variants analyzed: {len(rows)}.") flip_lines.append("") flip_lines.append("## Pre-Registered Buyer Assumptions") flip_lines.append("") flip_lines.append("- Buyer profiles A-D and pass/mixed/fail rules were read from the frozen local protocol files.") flip_lines.append("- Hard constraints were applied exactly where the buyer profiles specified them.") flip_lines.append("") flip_lines.append("## Calculated Results") flip_lines.append("") profile_tops: dict[str, dict | None] = {pid: top_eligible(profile_results, pid) for pid in ["A", "B", "C", "D"]} any_flip = False material_flip = False for pid, top in profile_tops.items(): prs = [r for r in profile_results if r["profile_id"] == pid] winner_status = next(r for r in prs if r["model_identifier"] == published_winner_name) if top is None: flip_lines.append(f"- Profile {pid}: no eligible model.") continue flip = top["model_identifier"] != published_winner_name any_flip = any_flip or flip if flip and not winner_status["eligible"]: material_flip = True flip_text = "different from the published aggregate winner" if flip else "same as the published aggregate winner" if winner_status["eligible"]: cause = "profile weighting among eligible models" else: cause = "published aggregate winner failed pre-registered hard constraints: " + winner_status["disqualification_reason"] flip_lines.append(f"- Profile {pid} ({PROFILE_NAMES[pid]}): selected `{top['model_identifier']}`, {flip_text}; cause: {cause}.") flip_lines.append("") flip_lines.append("## Required Questions") flip_lines.append("") flip_lines.append(f"- Does any defensible buyer profile select a different model than the published BFCL overall winner? {'Yes' if any_flip else 'No'}.") if material_flip: flip_lines.append("- Which operational constraint caused the change? At least one flip is caused by pre-registered hard operational constraints, not just a soft weighted preference.") flip_lines.append("- Is the change material or merely cosmetic? Material: a hard disqualification changes which model can be deployed for that buyer profile.") flip_lines.append("- Was the change caused by arbitrary weighting? No. The decisive case is tied to frozen hard constraints.") verdict = "FRAME AUDIT PASS" elif any_flip: flip_lines.append("- Which operational constraint caused the change? The change is caused by pre-registered cost, latency, or workflow-specific weighting among eligible models.") flip_lines.append("- Is the change material or merely cosmetic? Mixed: ranking changes occur, but no hard operational disqualification drives the decisive change.") flip_lines.append("- Was the change caused by arbitrary weighting? No post-hoc weighting was used, but the materiality is weaker without a hard constraint.") verdict = "FRAME AUDIT MIXED" else: flip_lines.append("- Which operational constraint caused the change? None.") flip_lines.append("- Is the change material or merely cosmetic? No decision-changing flip was observed.") flip_lines.append("- Was the change caused by arbitrary weighting? No rank reversal occurred.") verdict = "FRAME AUDIT FAIL" flip_lines.append("- What real purchasing or deployment decision could change? A buyer using BFCL as a procurement screen could choose a different default model, reject the aggregate winner for a constrained workflow, or route models differently for automation, regulated, multi-turn, or cost-constrained deployments.") flip_lines.append("") flip_lines.append("## Interpretation") flip_lines.append("") flip_lines.append("The result is provisional because it is based on public BFCL summary data and pre-registered local assumptions, not a fresh benchmark rerun. It is sufficient for an internal frame-audit finding, not yet a public-facing dossier.") flip_lines.append("") flip_lines.append(verdict) FLIP_SUMMARY.write_text("\n".join(flip_lines) + "\n", encoding="utf-8") return verdict def main() -> int: with INPUT.open("r", encoding="utf-8", newline="") as f_in: rows = list(csv.DictReader(f_in)) profile_results, disq_log = build_profile_results(rows) sens, sens_disq = sensitivity_rows(rows, profile_results) write_csv(PROFILE_RESULTS, profile_results) write_csv(SENSITIVITY_RESULTS, sens) write_csv(DISQUALIFICATION_LOG, disq_log + sens_disq) verdict = write_reports(rows, profile_results, sens) print(verdict) print(PROFILE_RESULTS) print(SENSITIVITY_RESULTS) print(DISQUALIFICATION_LOG) print(PROFILE_REPORT) print(FLIP_SUMMARY) return 0 if __name__ == "__main__": raise SystemExit(main())