diff --git a/runs/frontier-fidelity-envelope-v1/run_frontier_qwen30_exact_trace_surface.py b/runs/frontier-fidelity-envelope-v1/run_frontier_qwen30_exact_trace_surface.py index 02e30a9..fc102a7 100644 --- a/runs/frontier-fidelity-envelope-v1/run_frontier_qwen30_exact_trace_surface.py +++ b/runs/frontier-fidelity-envelope-v1/run_frontier_qwen30_exact_trace_surface.py @@ -68,6 +68,7 @@ def parse_args() -> argparse.Namespace: parser.add_argument("--allreduce-csv", type=Path) parser.add_argument("--timeout-seconds", type=float, default=1800.0) parser.add_argument("--resume", action="store_true") + parser.add_argument("--continue-on-failure", action="store_true") return parser.parse_args() @@ -156,6 +157,12 @@ def trace_manifest_entry(trace: dict[str, Any]) -> dict[str, Any]: } +def classify_frontier_failure(stderr: str) -> str: + if "Sequential simulation ended with non-empty scheduler state" in stderr: + return "scheduler_stall" + return "frontier_error" + + def score(path: Path, expected_shapes: list[tuple[int, int]]) -> dict[str, Any]: with path.open(newline="") as source: rows = list(csv.DictReader(source)) @@ -312,9 +319,48 @@ def main() -> None: check=False, ) if completed.returncode != 0: - raise RuntimeError( - f"Frontier failed for {config.name}/{trace['label']}: {completed.returncode}" + stderr_path = run_dir / "stderr.log" + result = { + "status": "frontier_failed", + "failure_kind": classify_frontier_failure( + stderr_path.read_text(errors="replace") + ), + "returncode": completed.returncode, + "config": { + "tp": config.tp, + "mns": config.mns, + "name": config.name, + }, + "trace_label": trace["label"], + "offered_request_rate": trace["offered_request_rate"], + "offered_request_rate_per_gpu": ( + trace["offered_request_rate"] / config.tp + ), + "request_count": trace["requests"], + "elapsed_seconds": time.time() - started, + "trace_sha256": trace["sha256"], + "stderr_sha256": BASE.sha256(stderr_path), + } + BASE.write_json(result_path, result) + loads.append(result) + print( + json.dumps( + { + "config": config.name, + "trace": trace["label"], + "status": result["status"], + "failure_kind": result["failure_kind"], + }, + sort_keys=True, + ), + flush=True, ) + if not args.continue_on_failure: + raise RuntimeError( + f"Frontier failed for {config.name}/{trace['label']}: " + f"{completed.returncode}" + ) + continue metrics = find_request_metrics(run_dir) result = { "status": "completed", @@ -353,21 +399,38 @@ def main() -> None: for slo in (f"tpot_{int(value)}ms" for value in TPOT_SLOS_MS): records = [] for item in config_results: + completed_loads = [ + load for load in item["loads"] if load["status"] == "completed" + ] + invalid_loads = [ + load for load in item["loads"] if load["status"] != "completed" + ] feasible = [ load["offered_request_rate"] - for load in item["loads"] + for load in completed_loads if load["score"]["slos"][slo]["feasible"] ] capacity = max(feasible, default=None) records.append( { "config": item["config"], + "ranking_valid": not invalid_loads, + "invalid_loads": [ + { + "trace_label": load["trace_label"], + "failure_kind": load.get("failure_kind", "unknown"), + } + for load in invalid_loads + ], "maximum_tested_feasible_request_rate": capacity, "maximum_tested_feasible_request_rate_per_gpu": ( capacity / item["config"]["tp"] if capacity is not None else None ), - "lower_censored": capacity is None, - "upper_censored": capacity == traces[-1]["offered_request_rate"], + "lower_censored": capacity is None and not invalid_loads, + "upper_censored": ( + capacity == traces[-1]["offered_request_rate"] + and not invalid_loads + ), } ) records.sort( @@ -382,9 +445,20 @@ def main() -> None: ) rankings[slo] = records + has_invalid_cells = any( + load["status"] != "completed" + for item in config_results + for load in item["loads"] + ) + if selected != list(BASE.GRID): + manifest_status = "partial_not_decision_bearing" + elif has_invalid_cells: + manifest_status = "frozen_with_invalid_cells" + else: + manifest_status = "frozen_before_real" manifest = { "schema": "frontier-qwen30-exact-trace-surface-v1", - "status": "frozen_before_real" if selected == list(BASE.GRID) else "partial_not_decision_bearing", + "status": manifest_status, "contract": { "window_seconds": ( WINDOW_SECONDS if args.rate_contract == "trace-window" else None diff --git a/runs/frontier-fidelity-envelope-v1/test_fidelity_envelope.py b/runs/frontier-fidelity-envelope-v1/test_fidelity_envelope.py index 5ac9a6b..ac0f3e0 100644 --- a/runs/frontier-fidelity-envelope-v1/test_fidelity_envelope.py +++ b/runs/frontier-fidelity-envelope-v1/test_fidelity_envelope.py @@ -106,6 +106,16 @@ class FidelityEnvelopeTest(unittest.TestCase): def test_exact_trace_parser_and_joint_slo_score(self) -> None: module = load("run_frontier_qwen30_exact_trace_surface.py") + self.assertEqual( + module.classify_frontier_failure( + "Sequential simulation ended with non-empty scheduler state" + ), + "scheduler_stall", + ) + self.assertEqual( + module.classify_frontier_failure("unexpected failure"), + "frontier_error", + ) with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) trace = root / "trace.csv"