1"""Hangs lane parser (schema `potential-hangs`).
2 
3The schema lacks inline backtraces — stacks come from Time Profiler samples
4that overlap each hang's window. Correlation is done later in correlate.py.
5"""
6from __future__ import annotations
7 
8from pathlib import Path
9from typing import Any
10 
11from . import xctrace, xml_utils
12 
13PREFERRED_SCHEMAS = ("potential-hangs",)
14FALLBACK_SCHEMAS = ("main-thread-hang", "hang", "hangs")
15 
16 
17def analyze(
18    trace_path: Path,
19    toc_schemas: frozenset[str],
20    top_n: int = 10,
21    window: tuple[int, int] | None = None,
22    run: int = 1,
23) -> dict[str, Any]:
24    schema = _pick_schema(toc_schemas)
25    if schema is None:
26        return {
27            "lane": "hangs",
28            "available": False,
29            "notes": ["Hangs data not present in trace."],
30        }
31 
32    xml_bytes = xctrace.export_schema(trace_path, schema, run=run)
33    stream = xml_utils.RowStream(xml_bytes)
34 
35    hangs: list[dict] = []
36    for row in stream:
37        start_el = row.get("start")
38        dur_el = row.get("duration")
39        type_el = row.get("hang-type")
40        thread_el = row.get("thread")
41        if start_el is None or dur_el is None:
42            continue
43        start_ns = xml_utils.int_text(stream.resolve(start_el))
44        duration_ns = xml_utils.int_text(stream.resolve(dur_el))
45        if start_ns is None or duration_ns is None:
46            continue
47        if not xml_utils.event_overlaps_window(start_ns, start_ns + duration_ns, window):
48            continue
49        hang_type = xml_utils.str_text(stream.resolve(type_el)) if type_el is not None else None
50        thread = xml_utils.extract_thread(thread_el, stream) if thread_el is not None else None
51 
52        hangs.append({
53            "start_ns": start_ns,
54            "duration_ns": duration_ns,
55            "end_ns": start_ns + duration_ns,
56            "duration_ms": round(duration_ns / 1_000_000, 2),
57            "start_ms": round(start_ns / 1_000_000, 2),
58            "hang_type": hang_type or "Hang",
59            "thread": thread,
60        })
61 
62    hangs.sort(key=lambda h: h["duration_ns"], reverse=True)
63 
64    total_ms = sum(h["duration_ms"] for h in hangs)
65    worst = hangs[0] if hangs else None
66 
67    # Severity buckets per Apple docs (Microhang: 250ms–500ms, Hang: ≥500ms).
68    # We bucket by raw duration so the agent can reason about it.
69    buckets = {"lt_250ms": 0, "250ms_1s": 0, "gt_1s": 0}
70    for h in hangs:
71        if h["duration_ms"] < 250:
72            buckets["lt_250ms"] += 1
73        elif h["duration_ms"] < 1000:
74            buckets["250ms_1s"] += 1
75        else:
76            buckets["gt_1s"] += 1
77 
78    top_offenders = [
79        {
80            "start_ms": h["start_ms"],
81            "duration_ms": h["duration_ms"],
82            "hang_type": h["hang_type"],
83            "thread": (h["thread"] or {}).get("name", ""),
84        }
85        for h in hangs[:top_n]
86    ]
87 
88    return {
89        "lane": "hangs",
90        "available": True,
91        "schema_used": schema,
92        "metrics": {
93            "count": len(hangs),
94            "total_duration_ms": round(total_ms, 2),
95            "worst_duration_ms": worst["duration_ms"] if worst else 0,
96            "severity_buckets": buckets,
97        },
98        "top_offenders": top_offenders,
99        "notes": [],
100        "_events": hangs,  # retained for correlation
101    }
102 
103 
104def _pick_schema(available: frozenset[str]) -> str | None:
105    for s in PREFERRED_SCHEMAS + FALLBACK_SCHEMAS:
106        if s in available:
107            return s
108    return None
109