llm-torch-profiler-analysis

1"""Internal overlap helpers for triage-only torch-profiler analysis."""2 3from __future__ import annotations4 5import re6from bisect import bisect_left, bisect_right7from collections import Counter, defaultdict8from dataclasses import dataclass, field9from pathlib import Path10from typing import Dict, Iterable, List, Optional, Sequence, Tuple11 12import triage_kernel_helpers as kernel_helpers13from profile_common import (14    coerce_optional_int,15    contains_any_keyword,16    extract_trace_events,17    has_stream_marker,18    is_annotation_event,19    is_complete_duration_event,20    is_non_kernel_trace_category,21    is_trace_metadata_name,22    looks_like_python_scope_name,23    normalize_repo_relative_path,24    normalize_text,25    select_heaviest_pid,26)27 28SOURCE_MAP_SAMPLE_LIMIT_PER_NAME = 1629 30COMMUNICATION_STRONG_KEYWORDS = (31    "allreduce",32    "all_reduce",33    "reduce_scatter",34    "allgather",35    "all_gather",36    "nccl",37    "cross_device_reduce",38    "deepep",39    "a2a",40    "alltoall",41    "allreduce_fusion",42    "mooncake",43)44 45COMMUNICATION_WEAK_KEYWORDS = (46    "broadcast",47    "dispatch",48    "combine",49)50 51MEMORY_STRONG_KEYWORDS = (52    "memcpy",53    "memset",54    "dma",55    "prefetch",56)57 58MEMORY_WEAK_KEYWORDS = (59    "fill",60    "copy",61)62 63ELEMENTWISE_KEYWORDS = (64    "sigmoid",65    "silu",66    "gelu",67    "relu",68    "softmax",69    "layernorm",70    "rmsnorm",71    "norm",72    "rotary",73    "rope",74    "topk",75    "gate",76    "bias",77    "_cast",78    "index",79    "gather",80    "scatter",81    "masked",82    "elementwise",83    "activation",84)85 86COMPUTE_KEYWORDS = (87    "cublas",88    "cudnn",89    "cutlass",90    "triton",91    "gemm",92    "gemv",93    "matmul",94    "grouped_mm",95    "flash",96    "attention",97    "fmha",98    "marlin",99    "fused_moe",100    "moe_kernel",101    "groupgemm",102    "mma",103    "wgmma",104    "conv",105    "bmm",106    "mm_kernel",107)108 109LOW_SIGNAL_FUNCTION_TOKENS = (110    "__torch_function__",111    "__torch_dispatch__",112    "__call__",113    "_call_impl",114    "_wrapped_call_impl",115)116 117LOW_SIGNAL_PATH_TOKENS = (118    "model_executor/parameter.py(",119    "model_executor/parameter.py:",120    "model_executor/cuda_graph_runner.py(",121    "model_executor/cuda_graph_runner.py:",122    "compilation/cuda_graph.py(",123    "compilation/cuda_graph.py:",124    "pyexecutor/cuda_graph_runner.py(",125    "pyexecutor/cuda_graph_runner.py:",126    "pyexecutor/py_executor.py(",127    "pyexecutor/py_executor.py:",128    "_torch/utils.py(",129    "_torch/utils.py:",130    "torch/fx/graph_module.py(",131    "torch/fx/graph_module.py:",132)133 134CATEGORY_PRIORITY = {135    "compute": 4,136    "communication": 3,137    "memory": 2,138    "elementwise": 1,139    "other": 0,140}141 142PYTHON_SCOPE_IGNORE_PREFIXES = (143    "threading.py(",144    "selectors.py(",145    "contextlib.py(",146    "queue.py(",147    "logging/",148    "logging/__init__.py(",149    "socket.py(",150    "asyncio/",151    "concurrent/futures/",152    "tqdm/",153    "uvicorn/",154    "fastapi/",155    "starlette/",156    "http/",157    "torch/_ops.py(",158    "torch/nn/modules/module.py(",159    "torch/utils/_contextlib.py(",160    "torch/autograd/",161    "torch/_tensor.py(",162    "torch/distributed/",163    "torch/_dynamo/",164    "torch/_inductor/",165)166KERNEL_NAME_HINTS = (167    COMMUNICATION_STRONG_KEYWORDS168    + COMMUNICATION_WEAK_KEYWORDS169    + MEMORY_STRONG_KEYWORDS170    + MEMORY_WEAK_KEYWORDS171    + COMPUTE_KEYWORDS172)173 174 175@dataclass176class KernelEvent:177    idx: int178    name: str179    canonical_name: str180    category: str181    pid: str182    tid: str183    stream: str184    ts: float185    dur: float186    end: float187    stage: str = "all"188    external_id: Optional[int] = None189    correlation: Optional[int] = None190    hidden_us: float = 0.0191    exclusive_us: float = 0.0192    hidden_by_compute_us: float = 0.0193    overlap_with: Counter = field(default_factory=Counter)194 195 196@dataclass197class AggregateStats:198    name: str199    category: str200    count: int = 0201    total_us: float = 0.0202    hidden_us: float = 0.0203    exclusive_us: float = 0.0204    hidden_by_compute_us: float = 0.0205    overlap_with: Counter = field(default_factory=Counter)206    representative_idx: Optional[int] = None207    representative_score: float = -1.0208 209    @property210    def hidden_ratio(self) -> float:211        return self.hidden_us / self.total_us if self.total_us else 0.0212 213    @property214    def exclusive_ratio(self) -> float:215        return self.exclusive_us / self.total_us if self.total_us else 0.0216 217 218@dataclass219class PythonScope:220    name: str221    normalized_name: str222    pid: str223    tid: str224    ts: float225    dur: float226    end: float227    is_meaningful: bool = False228    is_fallback: bool = False229 230 231@dataclass232class CPUOpContext:233    external_id: int234    cpu_op_name: str235    pid: str236    tid: str237    ts: float238    dur: float239    end: float240    scope_chain: Tuple[str, ...]241 242 243@dataclass244class KernelSourceStats:245    name: str246    total_count: int = 0247    mapped_count: int = 0248    scope_counter: Counter = field(default_factory=Counter)249    chain_counter: Counter = field(default_factory=Counter)250    launch_op_counter: Counter = field(default_factory=Counter)251    site_share_counter: Counter = field(default_factory=Counter)252 253    @property254    def mapping_ratio(self) -> float:255        return self.mapped_count / self.total_count if self.total_count else 0.0256 257    @property258    def best_scope(self) -> Optional[str]:259        return self.scope_counter.most_common(1)[0][0] if self.scope_counter else None260 261    @property262    def best_chain(self) -> Optional[str]:263        return self.chain_counter.most_common(1)[0][0] if self.chain_counter else None264 265    @property266    def best_launch_op(self) -> Optional[str]:267        return (268            self.launch_op_counter.most_common(1)[0][0]269            if self.launch_op_counter270            else None271        )272 273 274@dataclass275class TraceBundle:276    label: str277    trace_path: Path278    server_args: Optional[dict]279    raw_events: Sequence[dict]280    events: List[KernelEvent]281    pid: Optional[str]282    overlap_stats: Optional[Dict[str, float]] = None283 284 285@dataclass286class ActionRow:287    priority: str288    verdict: str289    kernel: str290    category: str291    total_us: float292    share_pct: float293    exclusive_ratio: float294    hidden_ratio: float295    python_scope: str296    launch_op: str297    mapping_ratio: float298    dependency_signal: str299    prev_neighbor: str300    next_neighbor: str301    recommendation: str302    suggestion: str303    representative_idx: Optional[int]304 305 306def short_name(name: str, max_len: int = 80) -> str:307    name = normalize_text(name)308    if len(name) <= max_len:309        return name310    return name[: max_len - 3] + "..."311 312 313def canonicalize_name(name: str) -> str:314    name = normalize_text(name)315    name = re.sub(r"0x[0-9a-fA-F]+", "0xADDR", name)316    if name.startswith("void ") and name.endswith(")"):317        depth = 0318        split_idx: Optional[int] = None319        for idx in range(len(name) - 1, -1, -1):320            char = name[idx]321            if char == ")":322                depth += 1323            elif char == "(":324                depth -= 1325                if depth == 0:326                    split_idx = idx327                    break328        if split_idx is not None:329            name = name[:split_idx]330    return name331 332 333def canonicalize_python_scope_name(name: str) -> str:334    name = normalize_text(name)335    name = re.sub(r"0x[0-9a-fA-F]+", "0xADDR", name)336    match = re.match(r"(?P<path>.+?)\((?P<line>\d+)\): (?P<func>.+)$", name)337    if match:338        path = normalize_repo_relative_path(match.group("path"))339        name = f"{path}({match.group('line')}): {match.group('func')}"340    return name341 342 343def canonicalize_cpu_op_name(name: str) -> str:344    return short_name(normalize_text(name), max_len=100)345 346 347def classify_kernel(name: str) -> str:348    # This script only needs broad overlap buckets, so keep the precedence small349    # and deterministic: memory/communication first, then compute/elementwise.350    lowered = name.lower()351    looks_compute_like = contains_any_keyword(lowered, COMPUTE_KEYWORDS)352    if contains_any_keyword(lowered, MEMORY_STRONG_KEYWORDS):353        return "memory"354    if contains_any_keyword(lowered, COMMUNICATION_STRONG_KEYWORDS):355        return "communication"356    if contains_any_keyword(lowered, COMPUTE_KEYWORDS):357        return "compute"358    if contains_any_keyword(lowered, ELEMENTWISE_KEYWORDS):359        return "elementwise"360    if contains_any_keyword(lowered, MEMORY_WEAK_KEYWORDS) and not looks_compute_like:361        return "memory"362    if (363        contains_any_keyword(lowered, COMMUNICATION_WEAK_KEYWORDS)364        and not looks_compute_like365    ):366        return "communication"367    if lowered.startswith("void "):368        return "other"369    return "other"370 371 372def is_kernel_event(event: dict) -> bool:373    # The overlap helpers prefer a slightly broader kernel detector than the374    # kernel-attribution helpers, but still reject annotations and Python375    # frames up front so the later overlap math only sees real GPU work.376    if not is_complete_duration_event(event):377        return False378    name = normalize_text(event.get("name", ""))379    if is_trace_metadata_name(name):380        return False381    cat = normalize_text(event.get("cat", "")).lower()382    args = event.get("args", {}) or {}383    if is_non_kernel_trace_category(cat):384        return False385    if is_annotation_event(name, cat):386        return False387    if "kernel" in cat or cat.startswith("gpu_"):388        return True389    lowered = name.lower()390    if looks_like_python_scope_name(name):391        return False392    if has_stream_marker(args) and (393        lowered.startswith("void ")394        or lowered.startswith("ampere_")395        or lowered.startswith("sm80_")396        or lowered.startswith("sm90_")397        or contains_any_keyword(lowered, KERNEL_NAME_HINTS)398    ):399        return True400    return False401 402 403def is_meaningful_python_scope(name: str) -> bool:404    normalized = canonicalize_python_scope_name(name)405    if not normalized:406        return False407    if normalized.startswith("<built-in method"):408        return False409    if normalized.startswith("nn.Module:"):410        return False411    if any(normalized.startswith(prefix) for prefix in PYTHON_SCOPE_IGNORE_PREFIXES):412        return False413    if normalized.startswith("python/sglang/"):414        return True415    if normalized.startswith("sglang/"):416        return True417    if normalized.startswith("vllm/"):418        return True419    if normalized.startswith("python/tokenspeed/") or normalized.startswith(420        "tokenspeed/"421    ):422        return True423    if normalized.startswith("tensorrt_llm/"):424        return True425    if normalized.startswith("sgl_kernel/"):426        return True427    return ".py(" in normalized428 429 430def is_fallback_python_scope(name: str) -> bool:431    normalized = canonicalize_python_scope_name(name)432    if (433        not normalized434        or normalized.startswith("<built-in method")435        or normalized.startswith("nn.Module:")436    ):437        return False438    if normalized.startswith("threading.py("):439        return False440    return ".py(" in normalized or normalized.startswith("python/")441 442 443def extract_thread_names(events: Sequence[dict]) -> Dict[Tuple[str, str], str]:444    mapping: Dict[Tuple[str, str], str] = {}445    for event in events:446        if event.get("ph") != "M" or event.get("name") != "thread_name":447            continue448        pid = str(event.get("pid"))449        tid = str(event.get("tid"))450        thread_name = str((event.get("args") or {}).get("name", ""))451        if thread_name:452            mapping[(pid, tid)] = thread_name453    return mapping454 455 456def build_correlation_external_lookup(raw_events: Sequence[dict]) -> Dict[int, int]:457    lookup: Dict[int, int] = {}458    for event in raw_events:459        args = event.get("args", {}) or {}460        correlation = coerce_optional_int(args.get("correlation"))461        external_id = coerce_optional_int(args.get("External id"))462        if correlation is not None and external_id is not None:463            lookup[correlation] = external_id464    return lookup465 466 467def extract_kernel_events(468    trace: dict, pid_substring: Optional[str]469) -> Tuple[List[KernelEvent], Optional[str]]:470    # We first build a clean kernel list from the chosen TP rank, then later471    # overlap analysis can stay focused on stream timing instead of trace noise.472    raw_events = extract_trace_events(trace)473    thread_names = extract_thread_names(raw_events)474    correlation_external = build_correlation_external_lookup(raw_events)475    (476        annotations_by_external_id,477        gpu_stage_annotations,478        cpu_stage_annotations,479    ) = kernel_helpers.build_stage_annotations(raw_events)480    chosen_pid = select_heaviest_pid(481        raw_events,482        is_kernel_event,483        pid_substring=pid_substring,484        preferred_substrings=(() if pid_substring else ("TP00",)),485    )486    kernel_events: List[KernelEvent] = []487    if chosen_pid is None:488        return kernel_events, None489 490    idx = 0491    for event in raw_events:492        if not is_kernel_event(event):493            continue494        pid = str(event.get("pid"))495        if pid != chosen_pid:496            continue497        tid = str(event.get("tid"))498        args = event.get("args", {}) or {}499        stream = (500            args.get("stream")501            or args.get("cuda_stream")502            or thread_names.get((pid, tid))503            or f"tid={tid}"504        )505        correlation = coerce_optional_int(args.get("correlation"))506        external_id = coerce_optional_int(args.get("External id"))507        if external_id is None and correlation is not None:508            external_id = correlation_external.get(correlation)509        name = str(event["name"])510        dur = float(event["dur"])511        ts = float(event["ts"])512        kernel_events.append(513            KernelEvent(514                idx=idx,515                name=name,516                canonical_name=canonicalize_name(name),517                category=classify_kernel(name),518                stage=kernel_helpers.resolve_kernel_stage(519                    kernel_ts=ts,520                    external_id=external_id,521                    annotations_by_external_id=annotations_by_external_id,522                    gpu_annotations=gpu_stage_annotations,523                    cpu_annotations=cpu_stage_annotations,524                ),525                pid=pid,526                tid=tid,527                stream=str(stream),528                ts=ts,529                dur=dur,530                end=ts + dur,531                external_id=external_id,532                correlation=correlation,533            )534        )535        idx += 1536    return kernel_events, chosen_pid537 538 539def group_events_by_stage(540    events: Sequence[KernelEvent], default_stage: str541) -> Dict[str, List[KernelEvent]]:542    grouped: Dict[str, List[KernelEvent]] = defaultdict(list)543    for event in events:544        stage = default_stage if default_stage != "all" else (event.stage or "all")545        grouped[stage].append(event)546    return dict(grouped)547 548 549def dominant_overlap_name(550    event: KernelEvent, active_events: Iterable[KernelEvent]551) -> Optional[str]:552    candidates = [553        other554        for other in active_events555        if other.idx != event.idx and other.stream != event.stream556    ]557    if not candidates:558        return None559    candidates.sort(560        key=lambda other: (CATEGORY_PRIORITY.get(other.category, 0), other.dur),561        reverse=True,562    )563    return candidates[0].canonical_name564 565 566def analyze_overlap(events: Sequence[KernelEvent]) -> Dict[str, float]:567    # Sweep line over kernel start/end points. For each active time slice we568    # decide whether a kernel was exposed on the critical path or hidden by work569    # on other streams.570    points: List[Tuple[float, int, int]] = []571    event_map = {event.idx: event for event in events}572    for event in events:573        points.append((event.ts, 1, event.idx))574        points.append((event.end, 0, event.idx))575    points.sort(key=lambda item: (item[0], item[1]))576 577    total_busy = 0.0578    total_overlap = 0.0579    max_concurrent = 0580    active: Dict[int, KernelEvent] = {}581    prev_time: Optional[float] = None582 583    for time_point, is_start, event_idx in points:584        if prev_time is not None and time_point > prev_time and active:585            segment = time_point - prev_time586            active_events = list(active.values())587            distinct_streams = {event.stream for event in active_events}588            total_busy += segment589            max_concurrent = max(max_concurrent, len(distinct_streams))590            if len(distinct_streams) >= 2:591                total_overlap += segment592            for event in active_events:593                overlapping_events = [594                    other595                    for other in active_events596                    if other.idx != event.idx and other.stream != event.stream597                ]598                if overlapping_events:599                    event.hidden_us += segment600                    if any(other.category == "compute" for other in overlapping_events):601                        event.hidden_by_compute_us += segment602                    overlap_name = dominant_overlap_name(event, active_events)603                    if overlap_name:604                        event.overlap_with[overlap_name] += segment605                else:606                    event.exclusive_us += segment607 608        if is_start == 0:609            active.pop(event_idx, None)610        else:611            active[event_idx] = event_map[event_idx]612        prev_time = time_point613 614    return {615        "total_busy_us": total_busy,616        "total_overlap_us": total_overlap,617        "max_concurrent_streams": float(max_concurrent),618    }619 620 621def aggregate_events(622    events: Sequence[KernelEvent],623) -> Dict[Tuple[str, str], AggregateStats]:624    aggregates: Dict[Tuple[str, str], AggregateStats] = {}625    for event in events:626        key = (event.canonical_name, event.category)627        if key not in aggregates:628            aggregates[key] = AggregateStats(629                name=event.canonical_name, category=event.category630            )631        stats = aggregates[key]632        stats.count += 1633        stats.total_us += event.dur634        stats.hidden_us += event.hidden_us635        stats.exclusive_us += event.exclusive_us636        stats.hidden_by_compute_us += event.hidden_by_compute_us637        stats.overlap_with.update(event.overlap_with)638        score = event.hidden_us + event.exclusive_us639        if score > stats.representative_score:640            stats.representative_score = score641            stats.representative_idx = event.idx642    return aggregates643 644 645def top_hidden_low_roi(646    aggregates: Dict[Tuple[str, str], AggregateStats],647) -> List[AggregateStats]:648    candidates = [649        stats650        for stats in aggregates.values()651        if stats.category in {"elementwise", "memory"}652        and stats.total_us >= 5.0653        and stats.hidden_ratio >= 0.65654    ]655    candidates.sort(656        key=lambda stats: (657            stats.hidden_us658            * (1.0 + stats.hidden_by_compute_us / max(stats.hidden_us, 1.0)),659            stats.hidden_ratio,660        ),661        reverse=True,662    )663    return candidates[:5]664 665 666def top_overlap_opportunities(667    aggregates: Dict[Tuple[str, str], AggregateStats],668) -> List[AggregateStats]:669    category_weight = {670        "communication": 1.3,671        "memory": 1.15,672        "elementwise": 1.0,673        "compute": 0.35,674        "other": 0.8,675    }676    candidates = [677        stats678        for stats in aggregates.values()679        if stats.total_us >= 5.0 and stats.exclusive_ratio >= 0.45680    ]681    primary = [stats for stats in candidates if stats.category != "compute"]682    fallback = [stats for stats in candidates if stats.category == "compute"]683    primary.sort(684        key=lambda stats: stats.exclusive_us * category_weight.get(stats.category, 1.0),685        reverse=True,686    )687    fallback.sort(688        key=lambda stats: stats.exclusive_us * category_weight.get(stats.category, 1.0),689        reverse=True,690    )691    return (primary + fallback)[:5]692 693 694def choose_best_scope(scope_chain: Sequence[str]) -> Optional[str]:695    ranked: List[Tuple[float, str]] = []696    for index, scope in enumerate(scope_chain):697        score = float(index)698        if scope.startswith("python/sglang/"):699            score += 50.0700        elif scope.startswith("sglang/"):701            score += 48.0702        elif scope.startswith("vllm/"):703            score += 46.0704        elif scope.startswith("python/tokenspeed/") or scope.startswith("tokenspeed/"):705            score += 45.0706        elif scope.startswith("tensorrt_llm/"):707            score += 44.0708        elif scope.startswith("sgl_kernel/"):709            score += 30.0710        elif ".py(" in scope:711            score += 10.0712        if "utils.py" in scope and "__call__" in scope:713            score -= 15.0714        if "scheduler_profiler_mixin.py" in scope:715            score -= 20.0716        if is_low_signal_scope(scope):717            score -= 25.0718        ranked.append((score, scope))719    return max(ranked, key=lambda item: item[0])[1] if ranked else None720 721 722def is_low_signal_scope(scope: str) -> bool:723    lowered = canonicalize_python_scope_name(scope).lower()724    if not lowered:725        return False726    return any(token in lowered for token in LOW_SIGNAL_FUNCTION_TOKENS) or any(727        token in lowered for token in LOW_SIGNAL_PATH_TOKENS728    )729 730 731def scope_chain_key(scope_chain: Sequence[str]) -> Optional[str]:732    if not scope_chain:733        return None734    trimmed = list(scope_chain[-4:])735    return " -> ".join(trimmed)736 737 738def normalize_match_text(text: object) -> str:739    return re.sub(r"[^0-9A-Za-z]+", "", normalize_text(text)).lower()740 741 742def source_scope_priority(scope: Optional[str]) -> int:743    normalized = canonicalize_python_scope_name(scope or "")744    if not normalized or normalized == "unmapped":745        return 0746    penalty = 80 if is_low_signal_scope(normalized) else 0747    if normalized.startswith("python/sglang/"):748        return 300 - penalty749    if normalized.startswith("sglang/"):750        return 290 - penalty751    if normalized.startswith("vllm/"):752        return 285 - penalty753    if normalized.startswith("python/tokenspeed/") or normalized.startswith(754        "tokenspeed/"755    ):756        return 283 - penalty757    if normalized.startswith("tensorrt_llm/"):758        return 280 - penalty759    if normalized.startswith("sgl_kernel/"):760        return 260 - penalty761    if ".py(" in normalized:762        return 120 - penalty763    return 0764 765 766def kernel_alias_token_groups(kernel_name: str) -> List[Tuple[str, ...]]:767    lowered = normalize_match_text(kernel_name)768    groups: List[Tuple[str, ...]] = []769    if "flashattnfwdcombine" in lowered:770        groups.append(771            (772                "flashattnfwdsm90",773                "flashattnvarlenfunc",774                "vllmflashattnflashattninterface",775                "vllmfa3cfwd",776            )777        )778    if "kernelmha" in lowered:779        groups.append(780            (781                "maskedmultiheadattentionkernel",782                "attentioninplace",783                "attentionbackendtrtllm",784            )785        )786    if "applybiasropeupdatekvcachev2" in lowered:787        groups.append(788            (789                "fusedqknormropekernel",790                "applyqknormrope",791                "modelingqwen3py98applyqknormrope",792            )793        )794    if lowered.startswith("memset"):795        groups.append(("memset",))796    return groups797 798 799def source_stats_lookup_text(800    kernel_name: str, stats: Optional[KernelSourceStats]801) -> str:802    parts = [kernel_name]803    if stats:804        parts.append(str(stats.best_scope or ""))805        parts.append(str(stats.best_chain or ""))806        parts.append(str(stats.best_launch_op or ""))807    return normalize_match_text(" ".join(parts))808 809 810def relaxed_source_stats_lookup(811    source_map: Dict[str, KernelSourceStats], kernel_name: str812) -> Optional[KernelSourceStats]:813    if kernel_name in source_map:814        return source_map[kernel_name]815 816    lowered = kernel_name.lower()817    best_key = None818    best_score = -1819    for candidate_key in source_map:820        candidate_lowered = candidate_key.lower()821        if candidate_lowered.startswith(lowered) or lowered.startswith(822            candidate_lowered823        ):824            score = min(len(candidate_lowered), len(lowered))825        elif candidate_lowered in lowered or lowered in candidate_lowered:826            score = min(len(candidate_lowered), len(lowered)) // 2827        else:828            continue829        if score > best_score:830            best_key = candidate_key831            best_score = score832    if best_key:833        return source_map.get(best_key)834 835    lowered_compact = normalize_match_text(kernel_name)836    if len(lowered_compact) >= 96:837 838        def common_prefix_len(left: str, right: str) -> int:839            count = 0840            for left_ch, right_ch in zip(left, right):841                if left_ch != right_ch:842                    break843                count += 1844            return count845 846        best_key = None847        best_score = -1848        for candidate_key in source_map:849            candidate_compact = normalize_match_text(candidate_key)850            if len(candidate_compact) < 96:851                continue852            prefix_len = common_prefix_len(lowered_compact, candidate_compact)853            shorter_len = min(len(lowered_compact), len(candidate_compact))854            if prefix_len < 64 or prefix_len < int(shorter_len * 0.4):855                continue856            score = prefix_len857            if lowered_compact.startswith(858                "voidcutlassdevicekernelflash"859            ) and candidate_compact.startswith("voidcutlassdevicekernelflash"):860                score += 32861            if score > best_score:862                best_key = candidate_key863                best_score = score864        if best_key:865            return source_map.get(best_key)866 867    alias_groups = kernel_alias_token_groups(kernel_name)868    if not alias_groups:869        return None870    best_key = None871    best_score = -1872    for candidate_key, stats in source_map.items():873        candidate_text = source_stats_lookup_text(candidate_key, stats)874        score = 0875        for group_index, group in enumerate(alias_groups):876            group_score = max(877                (len(token) for token in group if token in candidate_text),878                default=0,879            )880            if group_score:881                score += 1000 * (group_index + 1) + group_score882        if score <= 0:883            continue884        score += source_scope_priority(stats.best_scope)885        score += int(stats.mapping_ratio * 100)886        if score > best_score:887            best_key = candidate_key888            best_score = score889    return source_map.get(best_key) if best_key else None890 891 892def extract_cpu_launch_contexts(893    raw_events: Sequence[dict],894    target_external_ids: Optional[set[int]] = None,895) -> Dict[int, List[CPUOpContext]]:896    # Rebuild `External id -> CPU op -> active Python scopes` only for the897    # small set of launch ids that the source-map step will actually consume.898    # vLLM eager traces can have millions of Python frames on one thread, so899    # avoid global timeline reconstruction across unrelated threads and ids.900    cpu_ops_by_thread: Dict[Tuple[str, str], List[CPUOpContext]] = defaultdict(list)901 902    for event in raw_events:903        if not is_complete_duration_event(event):904            continue905        if str(event.get("cat", "")) != "cpu_op":906            continue907        args = event.get("args", {}) or {}908        external_id = coerce_optional_int(args.get("External id"))909        if external_id is None:910            continue911        if target_external_ids is not None and external_id not in target_external_ids:912            continue913        pid = str(event.get("pid"))914        tid = str(event.get("tid"))915        ts = float(event.get("ts", 0.0))916        dur = float(event.get("dur", 0.0))917        cpu_ops_by_thread[(pid, tid)].append(918            CPUOpContext(919                external_id=external_id,920                cpu_op_name=str(event.get("name", "")),921                pid=pid,922                tid=tid,923                ts=ts,924                dur=dur,925                end=ts + dur,926                scope_chain=(),927            )928        )929 930    if not cpu_ops_by_thread:931        return {}932 933    scopes_by_thread: Dict[Tuple[str, str], List[PythonScope]] = defaultdict(list)934    relevant_threads = set(cpu_ops_by_thread)935    for event in raw_events:936        if not is_complete_duration_event(event):937            continue938        if str(event.get("cat", "")) != "python_function":939            continue940        pid = str(event.get("pid"))941        tid = str(event.get("tid"))942        thread_key = (pid, tid)943        if thread_key not in relevant_threads:944            continue945        normalized_name = canonicalize_python_scope_name(event.get("name", ""))946        is_meaningful = is_meaningful_python_scope(normalized_name)947        is_fallback = is_fallback_python_scope(normalized_name)948        if not is_meaningful and not is_fallback:949            continue950        ts = float(event.get("ts", 0.0))951        dur = float(event.get("dur", 0.0))952        scopes_by_thread[thread_key].append(953            PythonScope(954                name=str(event.get("name", "")),955                normalized_name=normalized_name,956                pid=pid,957                tid=tid,958                ts=ts,959                dur=dur,960                end=ts + dur,961                is_meaningful=is_meaningful,962                is_fallback=is_fallback,963            )964        )965 966    contexts_by_external_id: Dict[int, List[CPUOpContext]] = defaultdict(list)967    for thread_key in relevant_threads:968        scopes = scopes_by_thread.get(thread_key, [])969        cpu_ops = cpu_ops_by_thread.get(thread_key, [])970        timeline = []971        for scope_idx, scope in enumerate(scopes):972            timeline.append((scope.ts, 0, scope_idx))973            timeline.append((scope.end, 2, scope_idx))974        for cpu_op_idx, cpu_op in enumerate(cpu_ops):975            timeline.append((cpu_op.ts, 1, cpu_op_idx))976        timeline.sort(key=lambda item: (item[0], item[1]))977 978        active_scopes: Dict[int, PythonScope] = {}979        for _, kind, payload in timeline:980            if kind == 0:981                active_scopes[payload] = scopes[payload]982            elif kind == 1:983                meaningful = [984                    scope.normalized_name985                    for scope in active_scopes.values()986                    if scope.is_meaningful987                ]988                fallback = (989                    []990                    if meaningful991                    else [992                        scope.normalized_name993                        for scope in active_scopes.values()994                        if scope.is_fallback995                    ]996                )997                chosen_chain = tuple((meaningful or fallback)[-6:])998                cpu_op = cpu_ops[payload]999                contexts_by_external_id[cpu_op.external_id].append(1000                    CPUOpContext(1001                        external_id=cpu_op.external_id,1002                        cpu_op_name=cpu_op.cpu_op_name,1003                        pid=cpu_op.pid,1004                        tid=cpu_op.tid,1005                        ts=cpu_op.ts,1006                        dur=cpu_op.dur,1007                        end=cpu_op.end,1008                        scope_chain=chosen_chain,1009                    )1010                )1011            else:1012                active_scopes.pop(payload, None)1013    return contexts_by_external_id1014 1015 1016def is_cuda_launch_event(name: str, cat: str) -> bool:1017    lowered_name = normalize_text(name).lower()1018    lowered_cat = normalize_text(cat).lower()1019    if lowered_cat == "cuda_runtime":1020        return lowered_name in {1021            "cudaLaunchKernel",1022            "cudaLaunchKernelExC",1023        }1024    return lowered_name in {1025        "cuLaunchKernel",1026        "cuLaunchKernelEx",1027        "cudaLaunchKernel",1028        "cudaLaunchKernelExC",1029    }1030 1031 1032@dataclass1033class LaunchContext:1034    correlation: int1035    pid: str1036    tid: str1037    ts: float1038    dur: float1039    end: float1040    launch_name: str1041 1042 1043def build_launch_contexts(1044    raw_events: Sequence[dict],1045) -> Dict[int, List[LaunchContext]]:1046    output: Dict[int, List[LaunchContext]] = defaultdict(list)1047    for event in raw_events:1048        if not is_complete_duration_event(event):1049            continue1050        cat = str(event.get("cat", ""))1051        name = str(event.get("name", ""))1052        args = event.get("args", {}) or {}1053        correlation = coerce_optional_int(args.get("correlation"))1054        if correlation is None or not is_cuda_launch_event(name, cat):1055            continue1056        ts = float(event.get("ts", 0.0))1057        dur = float(event.get("dur", 0.0))1058        output[correlation].append(1059            LaunchContext(1060                correlation=correlation,1061                pid=str(event.get("pid")),1062                tid=str(event.get("tid")),1063                ts=ts,1064                dur=dur,1065                end=ts + dur,1066                launch_name=name,1067            )1068        )1069    for items in output.values():1070        items.sort(key=lambda item: item.ts)1071    return output1072 1073 1074def choose_launch_context(1075    contexts: Sequence[LaunchContext], kernel_ts: float1076) -> Optional[LaunchContext]:1077    if not contexts:1078        return None1079    return min(contexts, key=lambda context: (abs(context.ts - kernel_ts), context.dur))1080 1081 1082def choose_cpu_context(1083    contexts: Sequence[CPUOpContext], kernel_ts: float1084) -> Optional[CPUOpContext]:1085    if not contexts:1086        return None1087    return min(contexts, key=lambda context: (abs(context.ts - kernel_ts), context.dur))1088 1089 1090def extract_meaningful_python_scopes(raw_events: Sequence[dict]) -> List[PythonScope]:1091    scopes: List[PythonScope] = []1092    for event in raw_events:1093        if not is_complete_duration_event(event):1094            continue1095        if str(event.get("cat", "")) != "python_function":1096            continue1097        ts = float(event.get("ts", 0.0))1098        dur = float(event.get("dur", 0.0))1099        normalized_name = canonicalize_python_scope_name(event.get("name", ""))1100        if not is_meaningful_python_scope(normalized_name):1101            continue1102        scopes.append(1103            PythonScope(1104                name=str(event.get("name", "")),1105                normalized_name=normalized_name,1106                pid=str(event.get("pid")),1107                tid=str(event.get("tid")),1108                ts=ts,1109                dur=dur,1110                end=ts + dur,1111            )1112        )1113    return scopes1114 1115 1116def choose_temporal_scope_chain(1117    scopes: Sequence[PythonScope], kernel_ts: float1118) -> Tuple[str, ...]:1119    matches = [scope for scope in scopes if scope.ts <= kernel_ts <= scope.end]1120    if not matches:1121        return ()1122    matches.sort(key=lambda scope: (scope.ts, -scope.dur, scope.normalized_name))1123    chain = []1124    seen = set()1125    for scope in matches:1126        if scope.normalized_name in seen:1127            continue1128        seen.add(scope.normalized_name)1129        chain.append(scope.normalized_name)1130    return tuple(chain[-6:])1131 1132 1133def build_temporal_scope_lookup(1134    scopes: Sequence[PythonScope],1135    query_points: Sequence[Tuple[int, float]],1136) -> Dict[int, Tuple[str, ...]]:1137    if not scopes or not query_points:1138        return {}1139 1140    timeline: List[Tuple[float, int, object]] = []1141    for scope in scopes:1142        timeline.append((scope.ts, 0, scope))1143        timeline.append((scope.end, 2, scope))1144    for event_idx, probe_ts in query_points:1145        timeline.append((probe_ts, 1, event_idx))1146    timeline.sort(key=lambda item: (item[0], item[1]))1147 1148    active_scopes: List[PythonScope] = []1149    resolved: Dict[int, Tuple[str, ...]] = {}1150    for _, kind, payload in timeline:1151        if kind == 0:1152            active_scopes.append(payload)1153            continue1154        if kind == 2:1155            if payload in active_scopes:1156                active_scopes.remove(payload)1157            continue1158 1159        chain: List[str] = []1160        seen: set[str] = set()1161        for scope in sorted(1162            active_scopes,1163            key=lambda scope: (scope.ts, -scope.dur, scope.normalized_name),1164        ):1165            name = scope.normalized_name1166            if name in seen:1167                continue1168            seen.add(name)1169            chain.append(name)1170        resolved[payload] = tuple(chain[-6:])1171    return resolved1172 1173 1174def build_temporal_scope_lookup_from_raw_events(1175    raw_events: Sequence[dict],1176    query_points: Sequence[Tuple[int, float]],1177) -> Dict[int, Tuple[str, ...]]:1178    if not query_points:1179        return {}1180 1181    ordered_queries = sorted(1182        ((float(query_ts), int(query_id)) for query_id, query_ts in query_points),1183        key=lambda item: item[0],1184    )1185    query_times = [query_ts for query_ts, _ in ordered_queries]1186    query_ids = [query_id for _, query_id in ordered_queries]1187    first_query_ts = query_times[0]1188    last_query_ts = query_times[-1]1189 1190    matches_by_query: Dict[int, List[PythonScope]] = defaultdict(list)1191    for event in raw_events:1192        if not is_complete_duration_event(event):1193            continue1194        if str(event.get("cat", "")) != "python_function":1195            continue1196        ts = float(event.get("ts", 0.0))1197        dur = float(event.get("dur", 0.0))1198        end = ts + dur1199        if end < first_query_ts or ts > last_query_ts:1200            continue1201 1202        normalized_name = canonicalize_python_scope_name(event.get("name", ""))1203        if not is_meaningful_python_scope(normalized_name):1204            continue1205 1206        left = bisect_left(query_times, ts - 1e-3)1207        right = bisect_right(query_times, end + 1e-3)1208        if left >= right:1209            continue1210 1211        scope = PythonScope(1212            name=str(event.get("name", "")),1213            normalized_name=normalized_name,1214            pid=str(event.get("pid")),1215            tid=str(event.get("tid")),1216            ts=ts,1217            dur=dur,1218            end=end,1219            is_meaningful=True,1220            is_fallback=False,1221        )1222        for pos in range(left, right):1223            matches_by_query[query_ids[pos]].append(scope)1224 1225    resolved: Dict[int, Tuple[str, ...]] = {}1226    for query_id, scopes in matches_by_query.items():1227        chain: List[str] = []1228        seen: set[str] = set()1229        for scope in sorted(1230            scopes,1231            key=lambda scope: (scope.ts, -scope.dur, scope.normalized_name),1232        ):1233            name = scope.normalized_name1234            if name in seen:1235                continue1236            seen.add(name)1237            chain.append(name)1238        resolved[query_id] = tuple(chain[-6:])1239    return resolved1240 1241 1242def build_kernel_source_map(1243    mapping_bundle: TraceBundle,1244    kernel_map_entry_lookup=None,1245    stage: str = "all",1246) -> Dict[str, KernelSourceStats]:1247    sampled_events = sample_source_map_events(mapping_bundle.events)1248    target_external_ids = {1249        event.external_id for event in sampled_events if event.external_id is not None1250    }1251    contexts_by_external_id = extract_cpu_launch_contexts(1252        mapping_bundle.raw_events,1253        target_external_ids=target_external_ids or None,1254    )1255    correlation_external = build_correlation_external_lookup(mapping_bundle.raw_events)1256    launch_contexts_by_correlation = build_launch_contexts(mapping_bundle.raw_events)1257    fallback_queries = [1258        (event.idx, event.ts)1259        for event in sampled_events1260        if event.external_id is None1261        or not contexts_by_external_id.get(event.external_id)1262    ]1263    temporal_scope_lookup = build_temporal_scope_lookup_from_raw_events(1264        mapping_bundle.raw_events,1265        fallback_queries,1266    )1267    source_map: Dict[str, KernelSourceStats] = {}1268    for event in sampled_events:1269        stats = source_map.setdefault(1270            event.canonical_name, KernelSourceStats(name=event.canonical_name)1271        )1272        stats.total_count += 11273        kernel_entry = (1274            kernel_map_entry_lookup(stage, event.canonical_name)1275            if kernel_map_entry_lookup is not None1276            else None1277        )1278        cpu_context = None1279        effective_external_id = event.external_id1280        if effective_external_id is None and event.correlation is not None:1281            effective_external_id = correlation_external.get(event.correlation)1282        if effective_external_id is not None:1283            cpu_context = choose_cpu_context(1284                contexts_by_external_id.get(effective_external_id, []), event.ts1285            )1286 1287        launch_op = None1288        scope_chain: Tuple[str, ...] = ()1289        if cpu_context is not None:1290            launch_op = canonicalize_cpu_op_name(cpu_context.cpu_op_name)1291            scope_chain = cpu_context.scope_chain1292        else:1293            launch_context = (1294                choose_launch_context(1295                    launch_contexts_by_correlation.get(event.correlation, []), event.ts1296                )1297                if event.correlation is not None1298                else None1299            )1300            if launch_context is not None:1301                scope_chain = build_temporal_scope_lookup_from_raw_events(1302                    mapping_bundle.raw_events,1303                    [(event.idx, launch_context.ts)],1304                ).get(event.idx, ())1305                if scope_chain:1306                    launch_op = canonicalize_cpu_op_name(launch_context.launch_name)1307            if not scope_chain:1308                scope_chain = temporal_scope_lookup.get(event.idx, ())1309            if scope_chain:1310                launch_op = "time-window fallback"1311 1312        if not scope_chain:1313            if kernel_entry:1314                best_location = str(kernel_entry.get("best_location") or "").strip()1315                if best_location and best_location != "unresolved":1316                    stats.mapped_count += 11317                    stats.scope_counter[best_location] += 11318                    stats.site_share_counter[best_location] += 11319                    for site in kernel_entry.get("sites") or []:1320                        display_location = str(1321                            site.get("display_location") or site.get("location") or ""1322                        ).strip()1323                        if display_location and display_location != "unresolved":1324                            launches = int(site.get("launches") or 0)1325                            stats.site_share_counter[display_location] += max(1326                                1, launches1327                            )1328                            if launches > 0:1329                                stats.scope_counter[display_location] += launches1330                        top_cpu_op = site.get("top_cpu_op")1331                        if top_cpu_op:1332                            launches = int(site.get("launches") or 0)1333                            stats.launch_op_counter[str(top_cpu_op)] += max(1, launches)1334            continue1335 1336        stats.mapped_count += 11337        best_scope = choose_best_scope(scope_chain)1338        if best_scope:1339            stats.scope_counter[best_scope] += 11340            stats.site_share_counter[best_scope] += 11341        chain = scope_chain_key(scope_chain)1342        if chain:1343            stats.chain_counter[chain] += 11344        if launch_op:1345            stats.launch_op_counter[launch_op] += 11346    return source_map1347 1348 1349def merge_source_map_from_kernel_payload(1350    source_map: Dict[str, KernelSourceStats],1351    stage_payload: Optional[dict],1352) -> Dict[str, KernelSourceStats]:1353    if not stage_payload:1354        return source_map1355 1356    for kernel_name, entry in (stage_payload.get("kernels") or {}).items():1357        sites = entry.get("sites") or []1358        best_location = str(entry.get("best_location") or "").strip()1359        if not sites and (not best_location or best_location == "unresolved"):1360            continue1361 1362        stats = source_map.setdefault(kernel_name, KernelSourceStats(name=kernel_name))1363        if sites:1364            for site in sites:1365                location = str(site.get("location") or best_location or "").strip()1366                launches = max(1, int(site.get("launches") or 0))1367                stats.total_count += launches1368                if location and location != "unresolved":1369                    stats.mapped_count += launches1370                    stats.scope_counter[location] += launches1371                    stats.site_share_counter[location] += launches1372                top_cpu_op = str(site.get("top_cpu_op") or "").strip()1373                if top_cpu_op:1374                    stats.launch_op_counter[top_cpu_op] += launches1375                stack = str(site.get("stack") or "").strip()1376                if stack:1377                    stats.chain_counter[stack] += launches1378            continue1379 1380        stats.total_count += 11381        stats.mapped_count += 11382        stats.scope_counter[best_location] += 11383        stats.site_share_counter[best_location] += 11384    return source_map1385 1386 1387def sample_source_map_events(1388    events: Sequence[KernelEvent],1389    per_name_limit: int = SOURCE_MAP_SAMPLE_LIMIT_PER_NAME,1390) -> List[KernelEvent]:1391    if per_name_limit <= 0:1392        return list(events)1393 1394    grouped: Dict[str, List[KernelEvent]] = defaultdict(list)1395    for event in events:1396        grouped[event.canonical_name].append(event)1397 1398    sampled: List[KernelEvent] = []1399    for kernel_name in sorted(grouped):1400        items = grouped[kernel_name]1401        if len(items) <= per_name_limit:1402            sampled.extend(items)1403            continue1404        for sample_idx in range(per_name_limit):1405            pos = round(sample_idx * (len(items) - 1) / (per_name_limit - 1))1406            sampled.append(items[pos])1407    sampled.sort(key=lambda event: (event.ts, event.idx))1408    return sampled1409 1410 1411def format_overlap_counter(counter: Counter, limit: int = 2) -> str:1412    if not counter:1413        return "n/a"1414    parts = []1415    for name, duration in counter.most_common(limit):1416        parts.append(f"{short_name(name, 48)} ({duration:.1f} us)")1417    return ", ".join(parts)1418 1419 1420def build_headroom_suggestion(stats: AggregateStats) -> str:1421    if stats.category == "communication":1422        return "Communication is still exposed. Check overlap with nearby compute."1423    if stats.category in {"elementwise", "memory"}:1424        return "This work is still exposed. Check fusion or nearby compute coverage."1425    return (1426        "This work is still exposed. Check stream placement and immediate dependencies."1427    )1428 1429 1430def build_hidden_suggestion(stats: AggregateStats) -> str:1431    overlap = format_overlap_counter(stats.overlap_with, limit=1)1432    if overlap != "n/a":1433        return f"Mostly hidden under {overlap}. Revisit only if schedule or fusion changes."1434    return "Mostly hidden already. Revisit only if schedule or fusion changes."1435 1436 1437def build_other_suggestion(stats: AggregateStats) -> str:1438    if stats.exclusive_ratio >= 0.6:1439        return "Still exposed, but not one of the leading overlap targets."1440    if stats.hidden_ratio >= 0.6:1441        return "Often hidden already. Revisit it if launch count or schedule changes."1442    return "Mixed exposure and overlap. Inspect it after the higher-share rows above."1443 1444 1445def parse_scope_signature(scope: str) -> Tuple[str, str]:1446    if not scope or scope in {"unmapped", "n/a"}:1447        return "", ""1448    match = re.match(r"(.+?)\(\d+\):\s*(.+)$", scope)1449    if match:1450        return match.group(1), match.group(2)1451    return scope, ""1452 1453 1454def same_scope_family(left: str, right: str) -> bool:1455    left_path, left_func = parse_scope_signature(left)1456    right_path, right_func = parse_scope_signature(right)1457    if not left_path or not right_path:1458        return False1459    if left_path == right_path:1460        return True1461    return bool(left_func and right_func and left_func == right_func)1462 1463 1464def is_neighbor_dependency_like(1465    current: KernelEvent, neighbor: Optional[KernelEvent]1466) -> bool:1467    if neighbor is None:1468        return False1469    if current.category == "communication":1470        return neighbor.category in {"compute", "elementwise", "memory", "other"}1471    if current.category in {"elementwise", "memory"}:1472        return neighbor.category in {1473            "compute",1474            "communication",1475            "elementwise",1476            "memory",1477        }1478    return False1479 1480 1481def build_stream_neighbor_index(1482    events: Sequence[KernelEvent],1483) -> Dict[int, Tuple[Optional[KernelEvent], Optional[KernelEvent]]]:1484    by_stream: Dict[str, List[KernelEvent]] = defaultdict(list)1485    for event in events:1486        by_stream[event.stream].append(event)1487 1488    index: Dict[int, Tuple[Optional[KernelEvent], Optional[KernelEvent]]] = {}1489    for stream_events in by_stream.values():1490        stream_events.sort(key=lambda event: (event.ts, event.end, event.idx))1491        for pos, event in enumerate(stream_events):1492            prev_event = stream_events[pos - 1] if pos > 0 else None1493            next_event = (1494                stream_events[pos + 1] if pos + 1 < len(stream_events) else None1495            )1496            index[event.idx] = (prev_event, next_event)1497    return index1498 1499 1500def describe_neighbor(1501    neighbor: Optional[KernelEvent],1502    gap_us: Optional[float],1503    source_map: Dict[str, KernelSourceStats],1504) -> str:1505    if neighbor is None:1506        return "none"1507    source = relaxed_source_stats_lookup(source_map, neighbor.canonical_name)1508    scope = source.best_scope if source and source.best_scope else "unmapped"1509    if gap_us is not None:1510        gap_us = max(gap_us, 0.0)1511        gap_text = f"{gap_us:.1f} us"1512    else:1513        gap_text = "n/a"1514    return (1515        f"{short_name(neighbor.canonical_name, 28)} "1516        f"@ {short_name(scope, 28)} "1517        f"(gap {gap_text})"1518    )1519 1520 1521def classify_dependency_signal(1522    current: KernelEvent,1523    source: Optional[KernelSourceStats],1524    prev_event: Optional[KernelEvent],1525    next_event: Optional[KernelEvent],1526    source_map: Dict[str, KernelSourceStats],1527) -> Tuple[str, str, str]:1528    current_scope = source.best_scope if source and source.best_scope else "unmapped"1529    current_launch = (1530        source.best_launch_op if source and source.best_launch_op else "n/a"1531    )1532 1533    prev_gap = current.ts - prev_event.end if prev_event is not None else None1534    next_gap = next_event.ts - current.end if next_event is not None else None1535    prev_source = (1536        relaxed_source_stats_lookup(source_map, prev_event.canonical_name)1537        if prev_event is not None1538        else None1539    )1540    next_source = (1541        relaxed_source_stats_lookup(source_map, next_event.canonical_name)1542        if next_event is not None1543        else None1544    )1545    prev_scope = (1546        prev_source.best_scope if prev_source and prev_source.best_scope else "unmapped"1547    )1548    next_scope = (1549        next_source.best_scope if next_source and next_source.best_scope else "unmapped"1550    )1551    prev_launch = (1552        prev_source.best_launch_op1553        if prev_source and prev_source.best_launch_op1554        else "n/a"1555    )1556    next_launch = (1557        next_source.best_launch_op1558        if next_source and next_source.best_launch_op1559        else "n/a"1560    )1561 1562    if prev_gap is not None:1563        prev_gap = max(prev_gap, 0.0)1564    if next_gap is not None:1565        next_gap = max(next_gap, 0.0)1566 1567    tight_gap_threshold = max(2.0, min(20.0, current.dur * 0.15))1568    prev_tight = prev_gap is not None and prev_gap <= tight_gap_threshold1569    next_tight = next_gap is not None and next_gap <= tight_gap_threshold1570 1571    prev_risk = prev_tight and (1572        same_scope_family(current_scope, prev_scope)1573        or (current_launch != "n/a" and current_launch == prev_launch)1574        or is_neighbor_dependency_like(current, prev_event)1575    )1576    next_risk = next_tight and (1577        same_scope_family(current_scope, next_scope)1578        or (current_launch != "n/a" and current_launch == next_launch)1579        or is_neighbor_dependency_like(current, next_event)1580    )1581 1582    prev_unclear = (1583        prev_tight1584        and not prev_risk1585        and (current_scope == "unmapped" or prev_scope == "unmapped")1586    )1587    next_unclear = (1588        next_tight1589        and not next_risk1590        and (current_scope == "unmapped" or next_scope == "unmapped")1591    )1592 1593    if prev_risk and next_risk:1594        signal = "both-side serial risk"1595    elif prev_risk:1596        signal = "prev-side serial risk"1597    elif next_risk:1598        signal = "next-side serial risk"1599    elif prev_unclear or next_unclear:1600        signal = "adjacency unclear"1601    else:1602        signal = "serial risk low"1603 1604    prev_desc = describe_neighbor(prev_event, prev_gap, source_map)1605    next_desc = describe_neighbor(next_event, next_gap, source_map)1606    return signal, prev_desc, next_desc1607 1608 1609def dependency_risk_label(signal: str) -> str:1610    mapping = {1611        "serial risk low": "low",1612        "prev-side serial risk": "high",1613        "next-side serial risk": "high",1614        "both-side serial risk": "high",1615        "adjacency unclear": "unclear",1616    }1617    return mapping.get(signal, signal)1618 1619 1620def build_priority_and_recommendation(1621    verdict: str,1622    category: str,1623    dependency_signal: str,1624    stats: AggregateStats,1625    share_pct: float,1626) -> Tuple[str, str]:1627    dep_label = dependency_risk_label(dependency_signal)1628    if share_pct < 1.0:1629        return "P5", "skip"1630 1631    if verdict == "headroom":1632        if dep_label == "low":1633            if category == "communication":1634                return "P1", "try overlap"1635            return "P1", "try fusion"1636        return "P2", "check deps"1637 1638    if verdict == "low-roi-hidden":1639        return "P4", "skip"1640 1641    if stats.exclusive_ratio >= 0.85 and dep_label == "low":1642        return "P3", "defer"1643    if stats.hidden_ratio >= 0.7:1644        return "P5", "skip"1645    if dep_label == "high":1646        return "P4", "check deps"1647    if dep_label == "unclear":1648        return "P4", "inspect"1649    return "P4", "defer"1650 1651 1652def make_action_row(1653    stats: AggregateStats,1654    verdict: str,1655    suggestion: str,1656    source_map: Dict[str, KernelSourceStats],1657    formal_events: Sequence[KernelEvent],1658    neighbor_index: Dict[int, Tuple[Optional[KernelEvent], Optional[KernelEvent]]],1659    total_busy_us: float,1660) -> ActionRow:1661    source = relaxed_source_stats_lookup(source_map, stats.name)1662    representative_idx = stats.representative_idx1663    dependency_signal = "adjacency unclear"1664    prev_neighbor = "none"1665    next_neighbor = "none"1666    share_pct = (stats.total_us / total_busy_us * 100.0) if total_busy_us > 0 else 0.01667    if representative_idx is not None:1668        current_event = next(1669            (event for event in formal_events if event.idx == representative_idx), None1670        )1671        if current_event is not None:1672            prev_event, next_event = neighbor_index.get(1673                representative_idx, (None, None)1674            )1675            dependency_signal, prev_neighbor, next_neighbor = (1676                classify_dependency_signal(1677                    current=current_event,1678                    source=source,1679                    prev_event=prev_event,1680                    next_event=next_event,1681                    source_map=source_map,1682                )1683            )1684    priority, recommendation = build_priority_and_recommendation(1685        verdict=verdict,1686        category=stats.category,1687        dependency_signal=dependency_signal,1688        stats=stats,1689        share_pct=share_pct,1690    )1691 1692    return ActionRow(1693        priority=priority,1694        verdict=verdict,1695        kernel=stats.name,1696        category=stats.category,1697        total_us=stats.total_us,1698        share_pct=share_pct,1699        exclusive_ratio=stats.exclusive_ratio,1700        hidden_ratio=stats.hidden_ratio,1701        python_scope=source.best_scope if source and source.best_scope else "unmapped",1702        launch_op=source.best_launch_op if source and source.best_launch_op else "n/a",1703        mapping_ratio=source.mapping_ratio if source else 0.0,1704        dependency_signal=dependency_signal,1705        prev_neighbor=prev_neighbor,1706        next_neighbor=next_neighbor,1707        recommendation=recommendation,1708        suggestion=suggestion,1709        representative_idx=representative_idx,1710    )1711 1712 1713def build_action_rows(1714    aggregates: Dict[Tuple[str, str], AggregateStats],1715    source_map: Dict[str, KernelSourceStats],1716    formal_events: Sequence[KernelEvent],1717    total_busy_us: float,1718    table_limit: int,1719) -> List[ActionRow]:1720    rows: List[ActionRow] = []1721    seen: set[str] = set()1722    neighbor_index = build_stream_neighbor_index(formal_events)1723 1724    for stats in top_overlap_opportunities(aggregates):1725        row = make_action_row(1726            stats=stats,1727            verdict="headroom",1728            suggestion=build_headroom_suggestion(stats),1729            source_map=source_map,1730            formal_events=formal_events,1731            neighbor_index=neighbor_index,1732            total_busy_us=total_busy_us,1733        )1734        if row.priority == "P5":1735            continue1736        rows.append(row)1737        seen.add(stats.name)1738 1739    for stats in top_hidden_low_roi(aggregates):1740        if stats.name in seen:1741            continue1742        rows.append(1743            make_action_row(1744                stats=stats,1745                verdict="low-roi-hidden",1746                suggestion=build_hidden_suggestion(stats),1747                source_map=source_map,1748                formal_events=formal_events,1749                neighbor_index=neighbor_index,1750                total_busy_us=total_busy_us,1751            )1752        )1753        seen.add(stats.name)1754 1755    if table_limit > 0:1756        return rows[:table_limit]1757    return rows1758