Use critical path totals for TP8 breakdown

runs/frontier-fidelity-envelope-v1/analyze_qwen235_fixed_pd_state.py

@@ -126,7 +126,6 @@ OP_CATEGORIES = {
         "expert_parallel_alltoall_dispatch",
         "expert_parallel_alltoall_combine",
         "expert_parallel_allreduce",
-        "expert_parallel_allreduce_wait",
     },
     "tp_dp_communication": {
         "attn_tensor_parallel_allreduce",
@@ -153,6 +152,7 @@ OP_CATEGORIES = {
         "ray_comm_time",
     },
 }
+DIAGNOSTIC_OPS = {"expert_parallel_allreduce_wait"}
 
 
 def atomic_write(path: Path, text: str) -> None:
@@ -328,7 +328,9 @@ def categorized_components(components: dict[str, Any]) -> dict[str, float]:
     }
 
 
-def op_category(name: str) -> str:
+def op_category(name: str) -> str | None:
+    if name in DIAGNOSTIC_OPS:
+        return None
     matches = [category for category, names in OP_CATEGORIES.items() if name in names]
     if len(matches) != 1:
         raise ValueError(f"unclassified or multiply classified op trace event: {name}")
@@ -395,6 +397,7 @@ def load_op_trace_rows(
                 {
                     "tokens": tokens,
                     "categories_ms": {key: 0.0 for key in CATEGORIES},
+                    "diagnostic_wait_ms": 0.0,
                     "events": 0,
                 },
             )
@@ -402,36 +405,42 @@ def load_op_trace_rows(
                 raise ValueError(
                     f"op trace token metadata changed within batch {batch_id}"
                 )
+            if category is None:
+                entry["diagnostic_wait_ms"] += float(event["duration_ms"])
+            else:
                 entry["categories_ms"][category] += float(event["duration_ms"])
             entry["events"] += 1
     if not grouped:
         raise ValueError(f"no op trace events: {trace_path}")
 
-    rows = []
-    for batch_id, entry in sorted(grouped.items()):
-        tokens = entry["tokens"]
-        total = sum(entry["categories_ms"].values())
-        rows.append(
-            {
-                "batch_id": batch_id,
-                "batch_size": len(tokens),
-                "decode_requests": sum(value == 1 for value in tokens),
-                "prefill_requests": sum(value > 1 for value in tokens),
-                "prefill_tokens": sum(value for value in tokens if value > 1),
-                "total_time_ms": total,
-                "categories_ms": entry["categories_ms"],
-                "per_expert_tokens": None,
-            }
-        )
-
     ledger_path = Path(result["state_artifacts"]["ledger"]["path"])
     stage_span_by_batch = {
         int(row["batch_id"]):
         (float(row["stage_end_ts"]) - float(row["stage_start_ts"])) * 1000
         for row in load_jsonl(ledger_path)
     }
+    rows = []
+    for batch_id, entry in sorted(grouped.items()):
+        tokens = entry["tokens"]
+        if batch_id not in stage_span_by_batch:
+            raise ValueError(f"op trace batch missing from ledger: {batch_id}")
+        rows.append(
+            {
+                "batch_id": batch_id,
+                "batch_size": len(tokens),
+                "decode_requests": sum(value == 1 for value in tokens),
+                "prefill_requests": sum(value > 1 for value in tokens),
+                "prefill_tokens": sum(value for value in tokens if value > 1),
+                "total_time_ms": stage_span_by_batch[batch_id],
+                "categories_ms": entry["categories_ms"],
+                "diagnostic_wait_ms": entry["diagnostic_wait_ms"],
+                "per_expert_tokens": None,
+            }
+        )
+
     ratios = [
-        row["total_time_ms"] / stage_span_by_batch[row["batch_id"]]
+        sum(row["categories_ms"].values())
+        / stage_span_by_batch[row["batch_id"]]
         for row in rows
         if row["batch_id"] in stage_span_by_batch
         and stage_span_by_batch[row["batch_id"]] > 0
@@ -440,7 +449,12 @@ def load_op_trace_rows(
         "trace_path": str(trace_path),
         "trace_batches": len(rows),
         "ledger_batches": len(stage_span_by_batch),
-        "trace_total_over_stage_span": numeric(ratios),
+        "serialized_component_sum_over_stage_span": numeric(ratios),
+        "diagnostic_wait_ms": numeric(
+            row["diagnostic_wait_ms"]
+            for row in rows
+            if row["diagnostic_wait_ms"] > 0
+        ),
         "event_names": dict(sorted(event_names.items())),
     }
     return rows, result, validation
@@ -464,7 +478,10 @@ def weighted_component_mean(
         / denominator
         for category in CATEGORIES
     }
-    values["total"] = sum(values.values())
+    values["total"] = sum(
+        row["total_time_ms"] * weight
+        for row, weight in zip(selected, weights, strict=True)
+    ) / denominator
     return values
 
 
@@ -697,8 +714,13 @@ def analyze(
                 "context lengths and graph buckets are unavailable in frozen logs"
             ),
             "graph_observability": (
-                "Frontier ledger exposes additive execution components but no direct graph "
+                "Frontier state outputs have no direct graph bucket/padding/launch-overhead "
-                "bucket/padding/launch-overhead field; graph effects remain folded into predictors"
+                "field; graph effects remain folded into predictors"
+            ),
+            "op_trace_accounting": (
+                "For TP8 shared-domain sync, total is ledger stage_end-start (critical path); "
+                "op categories are serialized work estimates and are not additive because "
+                "overlap and lane-summed wait diagnostics are represented separately"
             ),
         },
         "real": real,
@@ -798,6 +820,7 @@ def markdown(result: dict[str, Any]) -> str:
             f"- Real observed TPOT contrast: {result['reference']['observed_real_tp8_minus_tp4_ms']:+.4f} ms/token.",
             f"- {result['scope']['real_proxy_limitation']}.",
             f"- {result['scope']['graph_observability']}.",
+            f"- {result['scope']['op_trace_accounting']}.",
             "- Component deltas identify where Frontier creates its own TP8 advantage; without real per-stage measurements they are not yet root-cause proof.",
             "",
         ]

runs/frontier-fidelity-envelope-v1/test_fidelity_envelope.py

@@ -248,6 +248,7 @@ class FidelityEnvelopeTest(unittest.TestCase):
         for category, names in module.OP_CATEGORIES.items():
             for name in names:
                 self.assertEqual(module.op_category(name), category)
+        self.assertIsNone(module.op_category("expert_parallel_allreduce_wait"))
         with self.assertRaisesRegex(ValueError, "unclassified"):
             module.op_category("unknown_graph_overhead")
         component_row = {name: 1.0 for name in (*module.CATEGORIES, "total")}