1"""SwiftUI cause-graph lane (`swiftui-causes` schema).
2 
3Instruments emits one row per edge in SwiftUI's dependency graph: every time
4a source node (a state change, user defaults observer, system event, etc.)
5propagates to a destination node (a body evaluation, layout, creation), a
6row is written with both endpoints as metadata values.
7 
8This lane aggregates those edges two ways:
9 
10- **By source node** — which attribute graph nodes are driving the most
11  updates overall. The canonical "why is my app thrashing?" view; a
12  `UserDefaultObserver.send()` showing up with 11k outgoing edges is a
13  feedback storm.
14- **By destination node** — which views/modifiers receive the most
15  invalidations, and from whom. Use this to trace a hot view back to the
16  source that keeps poking it.
17 
18The analyzer's main lane (`swiftui`) tells you *what* updates are
19expensive; this lane tells you *why* they keep happening.
20"""
21from __future__ import annotations
22 
23from collections import Counter, defaultdict
24from pathlib import Path
25from typing import Any
26 
27from . import xctrace, xml_utils
28 
29SCHEMA = "swiftui-causes"
30 
31# Metadata nodes render as space-separated field dumps ("A gray icon n/a n/a").
32# We aggregate on the full fmt string so callers can spot specific edges like
33# "@AppStorage TextStyleModifier.fontOption", but also expose the short head
34# ("@AppStorage", "Creation of App", ...) for coarser grouping.
35 
36 
37def analyze(
38    trace_path: Path,
39    toc_schemas: frozenset[str],
40    top_n: int = 10,
41    top_k_per_node: int = 5,
42    window: tuple[int, int] | None = None,
43    run: int = 1,
44) -> dict[str, Any]:
45    if SCHEMA not in toc_schemas:
46        return {
47            "lane": "swiftui-causes",
48            "available": False,
49            "notes": [
50                "SwiftUI causes data not present (requires SwiftUI template on a real device).",
51            ],
52        }
53 
54    xml_bytes = xctrace.export_schema(trace_path, SCHEMA, run=run)
55    stream = xml_utils.RowStream(xml_bytes)
56 
57    source_edges: Counter[str] = Counter()
58    destination_edges: Counter[str] = Counter()
59    fanout: dict[str, Counter[str]] = defaultdict(Counter)
60    fanin: dict[str, Counter[str]] = defaultdict(Counter)
61    label_counts: Counter[str] = Counter()
62    total_edges = 0
63 
64    for row in stream:
65        time_el = xml_utils.first_present(row, "timestamp", "time")
66        if time_el is not None:
67            t_ns = xml_utils.int_text(stream.resolve(time_el))
68            if t_ns is not None and not xml_utils.in_window(t_ns, window):
69                continue
70 
71        src = _fmt(row, stream, "source-node")
72        dst = _fmt(row, stream, "destination-node")
73        if not src or not dst:
74            continue
75 
76        source_edges[src] += 1
77        destination_edges[dst] += 1
78        fanout[src][dst] += 1
79        fanin[dst][src] += 1
80 
81        label = _fmt(row, stream, "label")
82        if label:
83            label_counts[label] += 1
84 
85        total_edges += 1
86 
87    top_sources = [
88        {
89            "source": src,
90            "edges": count,
91            "top_destinations": [
92                {"destination": d, "edges": c}
93                for d, c in fanout[src].most_common(top_k_per_node)
94            ],
95        }
96        for src, count in source_edges.most_common(top_n)
97    ]
98 
99    top_destinations = [
100        {
101            "destination": dst,
102            "edges": count,
103            "top_sources": [
104                {"source": s, "edges": c}
105                for s, c in fanin[dst].most_common(top_k_per_node)
106            ],
107        }
108        for dst, count in destination_edges.most_common(top_n)
109    ]
110 
111    return {
112        "lane": "swiftui-causes",
113        "available": True,
114        "schema_used": SCHEMA,
115        "metrics": {
116            "total_edges": total_edges,
117            "unique_sources": len(source_edges),
118            "unique_destinations": len(destination_edges),
119            "top_labels": dict(label_counts.most_common(top_n)),
120        },
121        "top_sources": top_sources,
122        "top_destinations": top_destinations,
123        "notes": [],
124    }
125 
126 
127def fanin_for(
128    trace_path: Path,
129    toc_schemas: frozenset[str],
130    destination_contains: str,
131    top_k: int = 10,
132    window: tuple[int, int] | None = None,
133    run: int = 1,
134) -> dict[str, Any]:
135    """Return the top source nodes feeding any destination whose fmt string
136    contains `destination_contains` (case-insensitive substring).
137 
138    Used when the agent has a suspect view from the `swiftui` lane and wants
139    to know *who keeps invalidating it*. Does a full pass over the causes
140    schema each time — cheap enough at typical trace sizes.
141    """
142    if SCHEMA not in toc_schemas:
143        return {"available": False, "matches": []}
144 
145    needle = destination_contains.lower()
146    xml_bytes = xctrace.export_schema(trace_path, SCHEMA, run=run)
147    stream = xml_utils.RowStream(xml_bytes)
148 
149    matches: dict[str, Counter[str]] = defaultdict(Counter)
150    totals: Counter[str] = Counter()
151 
152    for row in stream:
153        time_el = xml_utils.first_present(row, "timestamp", "time")
154        if time_el is not None:
155            t_ns = xml_utils.int_text(stream.resolve(time_el))
156            if t_ns is not None and not xml_utils.in_window(t_ns, window):
157                continue
158 
159        dst = _fmt(row, stream, "destination-node")
160        if not dst or needle not in dst.lower():
161            continue
162        src = _fmt(row, stream, "source-node")
163        if not src:
164            continue
165 
166        matches[dst][src] += 1
167        totals[dst] += 1
168 
169    out = []
170    for dst, count in totals.most_common(top_k):
171        out.append({
172            "destination": dst,
173            "total_incoming_edges": count,
174            "top_sources": [
175                {"source": s, "edges": c}
176                for s, c in matches[dst].most_common(top_k)
177            ],
178        })
179    return {"available": True, "matches": out}
180 
181 
182def _fmt(row, stream, key: str) -> str | None:
183    el = row.get(key)
184    if el is None:
185        return None
186    resolved = stream.resolve(el)
187    return resolved.get("fmt") or xml_utils.str_text(resolved)
188