Prepare targeted disputed-boundary trial

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
@@ -915,6 +916,11 @@ def _slo_has_feasible_load(config_result: dict[str, Any], *, slo_name: str) -> b
 
 def prepare_community_repeat(args: argparse.Namespace) -> Path:
     """Prepare a fresh-server reverse-order repeat of decision-boundary loads."""
+    selected_slos = list(getattr(args, "slo", [])) or list(SLO_VARIANTS)
+    unknown_slos = sorted(set(selected_slos) - set(SLO_VARIANTS))
+    if unknown_slos:
+        raise ValueError(f"unknown repeat SLOs: {unknown_slos}")
+    selected_configs = set(getattr(args, "config", []))
     source_manifest = json.loads(args.source_manifest.read_text())
     first_pass = json.loads(args.community_freeze.read_text())
     if source_manifest.get("schema") != COMMUNITY_SCHEMA:
@@ -959,6 +965,15 @@ def prepare_community_repeat(args: argparse.Namespace) -> Path:
     first_execution_order = sorted(
         source_manifest["configs"], key=lambda item: int(item["execution_index"])
     )
+    if selected_configs:
+        unknown_configs = sorted(selected_configs - set(source_records))
+        if unknown_configs:
+            raise ValueError(f"unknown repeat configs: {unknown_configs}")
+        first_execution_order = [
+            item
+            for item in first_execution_order
+            if item["config"]["name"] in selected_configs
+        ]
     records = []
     selections = []
     for execution_index, source_record in enumerate(
@@ -967,7 +982,7 @@ def prepare_community_repeat(args: argparse.Namespace) -> Path:
         name = source_record["config"]["name"]
         excluded_unrankable_slos = [
             slo_name
-            for slo_name in SLO_VARIANTS
+            for slo_name in selected_slos
             if not _slo_has_feasible_load(
                 first_results[name], slo_name=slo_name
             )
@@ -980,7 +995,7 @@ def prepare_community_repeat(args: argparse.Namespace) -> Path:
                     first_results[name], slo_name=slo_name
                 )
             )
-            for slo_name in SLO_VARIANTS
+            for slo_name in selected_slos
         }
         boundary_rates = sorted(
             {
@@ -1063,6 +1078,8 @@ def prepare_community_repeat(args: argparse.Namespace) -> Path:
         "execution_contract": {
             "schema": REPEAT_SELECTION_SCHEMA,
             "selection": "all_adjacent_pre_registered_slo_feasibility_transitions_plus_between_rate_refinement",
+            "selected_slos": selected_slos,
+            "selected_configs": sorted(selected_configs) if selected_configs else "all",
             "unrankable_slo_policy": "exclude_per_config_surfaces_with_no_feasible_load",
             "fallback_when_no_transition": "top_two_if_all_feasible_else_bottom_two",
             "config_order": "reverse_of_first_pass",
@@ -1793,6 +1810,305 @@ 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 = []
+    repeated_config_names = {name for name, _ in repeat_loads}
+    for config_name in sorted(set(first_primary_capacity) & repeated_config_names):
+        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)
@@ -1846,11 +2162,19 @@ def parse_args() -> argparse.Namespace:
     repeat.add_argument("--protocol-manifest", type=Path)
     repeat.add_argument("--frontier-freeze", type=Path)
     repeat.add_argument("--output-root", type=Path, required=True)
+    repeat.add_argument("--slo", action="append", default=[])
+    repeat.add_argument("--config", action="append", default=[])
 
     compare_parser = subparsers.add_parser("compare")
     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 +2198,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)
 

runs/frontier-multicase-sufficiency-v0/best_effort/jobs_fixed_cohort_targeted.toml

@@ -0,0 +1,10 @@
+# Third fresh-server trial for primary-SLO boundaries disputed after refinement.
+version = 1
+
+[[jobs]]
+name = "qwen235b-fixed-cohort-community-v3-primary-disputed-20260715"
+gpus = 8
+gpu_model = "H20"
+hosts = ["dash0"]
+command = "cd ../../.. && /tmp/wjh-frontier-vllm0102-smoke/.venv/bin/python runs/frontier-multicase-sufficiency-v0/best_effort/fixed_cohort_rank.py run-community --manifest /home/admin/cpfs/wjh/frontier-community-qwen235-smoke-20260715/rankable-fixed-cohort-v1/community-v3-primary-disputed/run_manifest.json"
+env = { PYTHONPATH = "/home/admin/cpfs/wjh/aituner/aituner-frontier-best-effort-20260715/src" }

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