diff --git a/runs/frontier-multicase-sufficiency-v0/best_effort/fixed_cohort_rank.py b/runs/frontier-multicase-sufficiency-v0/best_effort/fixed_cohort_rank.py index 33a899f..d81db14 100644 --- a/runs/frontier-multicase-sufficiency-v0/best_effort/fixed_cohort_rank.py +++ b/runs/frontier-multicase-sufficiency-v0/best_effort/fixed_cohort_rank.py @@ -52,6 +52,7 @@ PROTOCOL_SCHEMA = "qwen235b-prefill-fixed-cohort-protocol-v1" FRONTIER_SCHEMA = "frontier-qwen235b-prefill-fixed-cohort-v1" COMMUNITY_SCHEMA = "community-qwen235b-prefill-fixed-cohort-v1" COMPARISON_SCHEMA = "frontier-community-qwen235b-rank-comparison-v1" +TRIAL_STABILITY_SCHEMA = "community-qwen235b-trial-stability-v1" REPEAT_SELECTION_SCHEMA = "community-qwen235b-boundary-repeat-selection-v4" COHORT_SIZE = 64 COHORT_SEED = 2026071501 @@ -1793,6 +1794,304 @@ def compare(args: argparse.Namespace) -> Path: return args.output +def _index_loads( + config_results: list[dict[str, Any]], +) -> dict[tuple[str, float], dict[str, Any]]: + return { + (config["config"]["name"], float(load["offered_request_rate"])): load + for config in config_results + for load in config["loads"] + } + + +def _index_requests( + config_results: list[dict[str, Any]], +) -> dict[tuple[str, float, str], dict[str, Any]]: + return { + ( + config["config"]["name"], + float(load["offered_request_rate"]), + str(request["request_id"]), + ): request + for config in config_results + for load in config["loads"] + for request in load["requests"] + } + + +def _request_passes_slo(request: dict[str, Any], variant: dict[str, Any]) -> bool: + return bool( + request["success"] + and request.get("ttft_ms") is not None + and float(request["ttft_ms"]) + <= slo_threshold_ms(variant, int(request["input_tokens"])) + ) + + +def compare_trials(args: argparse.Namespace) -> Path: + frontier = json.loads(args.frontier_freeze.read_text()) + first = json.loads(args.first_community_freeze.read_text()) + repeat = json.loads(args.repeat_community_freeze.read_text()) + if frontier.get("schema") != FRONTIER_SCHEMA: + raise ValueError("unexpected Frontier freeze") + for label, payload in (("first", first), ("repeat", repeat)): + if payload.get("schema") != COMMUNITY_SCHEMA or payload.get("status") != "completed": + raise ValueError(f"unexpected/incomplete {label} community freeze") + + frontier_loads = _index_loads(frontier["config_results"]) + first_loads = _index_loads(first["config_results"]) + repeat_loads = _index_loads(repeat["config_results"]) + shared_load_keys = sorted(set(frontier_loads) & set(first_loads) & set(repeat_loads)) + if not shared_load_keys: + raise ValueError("trials have no shared config/load points") + for key in shared_load_keys: + trace_hashes = { + index[key].get("trace_sha256") + for index in (frontier_loads, first_loads, repeat_loads) + if index[key].get("trace_sha256") is not None + } + if len(trace_hashes) > 1: + raise ValueError(f"trace mismatch at {key}") + + frontier_requests = _index_requests(frontier["config_results"]) + first_requests = _index_requests(first["config_results"]) + repeat_requests = _index_requests(repeat["config_results"]) + shared_request_keys = sorted( + set(frontier_requests) & set(first_requests) & set(repeat_requests) + ) + if not shared_request_keys: + raise ValueError("trials have no shared requests") + for key in shared_request_keys: + lengths = { + int(index[key]["input_tokens"]) + for index in (frontier_requests, first_requests, repeat_requests) + } + if len(lengths) != 1: + raise ValueError(f"input-token mismatch at {key}") + + trial_ttft_differences = [] + for key in shared_request_keys: + first_request = first_requests[key] + repeat_request = repeat_requests[key] + if ( + first_request["success"] + and repeat_request["success"] + and first_request.get("ttft_ms") is not None + and repeat_request.get("ttft_ms") is not None + ): + trial_ttft_differences.append( + float(repeat_request["ttft_ms"]) - float(first_request["ttft_ms"]) + ) + + variants = {} + for slo_name, variant in SLO_VARIANTS.items(): + request_counts = { + "matching_real_trial_labels": 0, + "frontier_correct_first": 0, + "frontier_correct_repeat": 0, + "frontier_wrong_both_stable_real_label": 0, + "real_trial_label_flips": 0, + } + first_failures = set() + repeat_failures = set() + for key in shared_request_keys: + predicted = _request_passes_slo(frontier_requests[key], variant) + first_observed = _request_passes_slo(first_requests[key], variant) + repeat_observed = _request_passes_slo(repeat_requests[key], variant) + request_counts["matching_real_trial_labels"] += int( + first_observed == repeat_observed + ) + request_counts["frontier_correct_first"] += int(predicted == first_observed) + request_counts["frontier_correct_repeat"] += int(predicted == repeat_observed) + request_counts["frontier_wrong_both_stable_real_label"] += int( + first_observed == repeat_observed and predicted != first_observed + ) + request_counts["real_trial_label_flips"] += int( + first_observed != repeat_observed + ) + if not first_observed: + first_failures.add(key) + if not repeat_observed: + repeat_failures.add(key) + + load_records = [] + matching_load_labels = 0 + frontier_correct_first_loads = 0 + frontier_correct_repeat_loads = 0 + for config_name, rate in shared_load_keys: + predicted_score = frontier_loads[(config_name, rate)]["scores"][slo_name] + first_score = first_loads[(config_name, rate)]["scores"][slo_name] + repeat_score = repeat_loads[(config_name, rate)]["scores"][slo_name] + predicted_label = bool(predicted_score["feasible"]) + first_label = bool(first_score["feasible"]) + repeat_label = bool(repeat_score["feasible"]) + matching_load_labels += int(first_label == repeat_label) + frontier_correct_first_loads += int(predicted_label == first_label) + frontier_correct_repeat_loads += int(predicted_label == repeat_label) + load_records.append( + { + "config": config_name, + "offered_request_rate": rate, + "frontier_passed_request_count": int( + predicted_score["passed_request_count"] + ), + "first_passed_request_count": int(first_score["passed_request_count"]), + "repeat_passed_request_count": int( + repeat_score["passed_request_count"] + ), + "frontier_feasible": predicted_label, + "first_feasible": first_label, + "repeat_feasible": repeat_label, + } + ) + failure_union = first_failures | repeat_failures + variants[slo_name] = { + "shared_requests": len(shared_request_keys), + **request_counts, + "real_trial_request_label_accuracy": ( + request_counts["matching_real_trial_labels"] / len(shared_request_keys) + ), + "frontier_request_label_accuracy_first": ( + request_counts["frontier_correct_first"] / len(shared_request_keys) + ), + "frontier_request_label_accuracy_repeat": ( + request_counts["frontier_correct_repeat"] / len(shared_request_keys) + ), + "real_trial_failure_set_jaccard": ( + len(first_failures & repeat_failures) / len(failure_union) + if failure_union + else 1.0 + ), + "shared_config_load_points": len(shared_load_keys), + "real_trial_load_label_accuracy": matching_load_labels / len(shared_load_keys), + "frontier_load_label_accuracy_first": ( + frontier_correct_first_loads / len(shared_load_keys) + ), + "frontier_load_label_accuracy_repeat": ( + frontier_correct_repeat_loads / len(shared_load_keys) + ), + "config_load_records": load_records, + } + + first_primary_capacity = _capacity_map( + first["rankings"][PRIMARY_SLO], + field="maximum_tested_feasible_request_rate", + ) + first_config = { + item["config"]["name"]: item["config"] for item in first["config_results"] + } + decision_records = [] + repeat_capacity_records = [] + for config_name in sorted(first_primary_capacity): + old_rates = sorted( + rate for name, rate in first_loads if name == config_name + ) + capacity = first_primary_capacity[config_name] + if capacity is None: + continue + higher_rates = [rate for rate in old_rates if rate > capacity] + next_rate = min(higher_rates) if higher_rates else None + boundary_rates = [capacity] + ([next_rate] if next_rate is not None else []) + boundary_repeated = all( + (config_name, rate) in repeat_loads for rate in boundary_rates + ) + capacity_still_feasible = bool( + boundary_repeated + and repeat_loads[(config_name, capacity)]["scores"][PRIMARY_SLO]["feasible"] + ) + next_still_infeasible = bool( + next_rate is None + or ( + boundary_repeated + and not repeat_loads[(config_name, next_rate)]["scores"][PRIMARY_SLO][ + "feasible" + ] + ) + ) + stable = boundary_repeated and capacity_still_feasible and next_still_infeasible + decision_records.append( + { + "config": config_name, + "first_capacity": capacity, + "first_next_tested_rate": next_rate, + "boundary_repeated": boundary_repeated, + "capacity_still_feasible": capacity_still_feasible, + "next_rate_still_infeasible": next_still_infeasible, + "stable_original_grid_capacity": stable, + } + ) + repeat_capacity = capacity if stable else None + repeat_capacity_records.append( + { + "config": first_config[config_name], + "maximum_tested_feasible_request_rate": repeat_capacity, + "maximum_tested_feasible_request_rate_per_gpu": ( + repeat_capacity / int(first_config[config_name]["tp"]) + if repeat_capacity is not None + else None + ), + } + ) + + result = { + "schema": TRIAL_STABILITY_SCHEMA, + "created_unix_s": time.time(), + "inputs": { + "frontier_freeze": { + "path": str(args.frontier_freeze.resolve()), + "sha256": sha256(args.frontier_freeze), + }, + "first_community_freeze": { + "path": str(args.first_community_freeze.resolve()), + "sha256": sha256(args.first_community_freeze), + }, + "repeat_community_freeze": { + "path": str(args.repeat_community_freeze.resolve()), + "sha256": sha256(args.repeat_community_freeze), + }, + }, + "shared_config_load_points": len(shared_load_keys), + "shared_requests": len(shared_request_keys), + "real_trial_ttft_ms": { + "paired_successful_requests": len(trial_ttft_differences), + "repeat_minus_first_mean": ( + statistics.fmean(trial_ttft_differences) + if trial_ttft_differences + else None + ), + "mae": ( + statistics.fmean(abs(value) for value in trial_ttft_differences) + if trial_ttft_differences + else None + ), + "absolute_error_p95": _percentile( + [abs(value) for value in trial_ttft_differences], 0.95 + ), + }, + "variants": variants, + "primary_original_grid_capacity_stability": { + "stable_configs": sum( + item["stable_original_grid_capacity"] for item in decision_records + ), + "total_configs": len(decision_records), + "records": decision_records, + "per_gpu_efficiency": _objective_fidelity( + first["rankings"][PRIMARY_SLO], + repeat_capacity_records, + field="maximum_tested_feasible_request_rate_per_gpu", + ), + "single_replica_raw_capacity": _objective_fidelity( + first["rankings"][PRIMARY_SLO], + repeat_capacity_records, + field="maximum_tested_feasible_request_rate", + ), + }, + } + write_json(args.output, result) + print(args.output) + return args.output + + def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() subparsers = parser.add_subparsers(dest="command", required=True) @@ -1851,6 +2150,12 @@ def parse_args() -> argparse.Namespace: compare_parser.add_argument("--frontier-freeze", type=Path, required=True) compare_parser.add_argument("--community-freeze", type=Path, required=True) compare_parser.add_argument("--output", type=Path, required=True) + + trials = subparsers.add_parser("compare-trials") + trials.add_argument("--frontier-freeze", type=Path, required=True) + trials.add_argument("--first-community-freeze", type=Path, required=True) + trials.add_argument("--repeat-community-freeze", type=Path, required=True) + trials.add_argument("--output", type=Path, required=True) return parser.parse_args() @@ -1874,6 +2179,8 @@ def main() -> None: prepare_community_repeat(args) elif args.command == "compare": compare(args) + elif args.command == "compare-trials": + compare_trials(args) else: raise AssertionError(args.command) diff --git a/runs/frontier-multicase-sufficiency-v0/best_effort/test_fixed_cohort_rank.py b/runs/frontier-multicase-sufficiency-v0/best_effort/test_fixed_cohort_rank.py index f764251..b78c9ad 100644 --- a/runs/frontier-multicase-sufficiency-v0/best_effort/test_fixed_cohort_rank.py +++ b/runs/frontier-multicase-sufficiency-v0/best_effort/test_fixed_cohort_rank.py @@ -338,3 +338,63 @@ def test_frontier_freeze_validation_uses_manifest_rates(tmp_path: Path) -> None: ) validated = MODULE._validate_frontier_freeze(freeze_path, protocol_path) assert validated["protocol_manifest_sha256"] == MODULE.sha256(protocol_path) + + +def test_compare_trials_separates_real_label_flips_from_stable_frontier_errors( + tmp_path: Path, +) -> None: + config = {"name": "a", "tp": 4} + + def config_result(ttfts: list[float]) -> dict[str, object]: + requests = [ + { + "request_id": str(index), + "input_tokens": 0, + "success": True, + "ttft_ms": ttft, + } + for index, ttft in enumerate(ttfts) + ] + load = { + "offered_request_rate": 0.1, + "requests": requests, + **MODULE.score_requests(requests), + } + return {"config": config, "loads": [load]} + + frontier_result = config_result([900.0, 1100.0]) + first_result = config_result([900.0, 900.0]) + repeat_result = config_result([900.0, 1100.0]) + + def freeze(path: Path, schema: str, result: dict[str, object]) -> Path: + payload = { + "schema": schema, + "status": "completed", + "config_results": [result], + "rankings": { + name: MODULE.rank_surface([result], slo_name=name) + for name in MODULE.SLO_VARIANTS + }, + } + path.write_text(json.dumps(payload)) + return path + + frontier_path = freeze(tmp_path / "frontier.json", MODULE.FRONTIER_SCHEMA, frontier_result) + first_path = freeze(tmp_path / "first.json", MODULE.COMMUNITY_SCHEMA, first_result) + repeat_path = freeze(tmp_path / "repeat.json", MODULE.COMMUNITY_SCHEMA, repeat_result) + output = tmp_path / "stability.json" + MODULE.compare_trials( + argparse.Namespace( + frontier_freeze=frontier_path, + first_community_freeze=first_path, + repeat_community_freeze=repeat_path, + output=output, + ) + ) + result = json.loads(output.read_text()) + primary = result["variants"][MODULE.PRIMARY_SLO] + assert primary["shared_requests"] == 2 + assert primary["real_trial_label_flips"] == 1 + assert primary["frontier_wrong_both_stable_real_label"] == 0 + assert primary["real_trial_load_label_accuracy"] == 0.0 + assert result["primary_original_grid_capacity_stability"]["stable_configs"] == 0