1#!/usr/bin/env python3
2"""Analyze an Xcode Instruments .trace file and emit JSON + markdown.
3 
4Primary modes:
5  (default)           Full four-lane analysis + cross-lane correlations.
6  --list-logs         Dump os_log entries (optionally filtered) as JSON so an
7                      agent can locate a focus window by log content.
8  --list-signposts    Dump os_signpost intervals + point events as JSON.
9 
10Windowing:
11  --window START_MS:END_MS restricts every lane to that slice of the trace.
12"""
13from __future__ import annotations
14 
15import argparse
16import json
17import sys
18from pathlib import Path
19 
20from instruments_parser import (
21    causes,
22    correlate,
23    events,
24    hangs,
25    hitches,
26    summary,
27    swiftui,
28    time_profiler,
29    xctrace,
30)
31 
32 
33def main(argv: list[str] | None = None) -> int:
34    parser = argparse.ArgumentParser(
35        description="Analyze an Instruments .trace file.",
36    )
37    parser.add_argument("--trace", required=True, type=Path)
38    parser.add_argument(
39        "--output",
40        type=Path,
41        help="Base path; writes <output>.json and <output>.md",
42    )
43    parser.add_argument("--top", type=int, default=10, help="Top-N per lane")
44    parser.add_argument(
45        "--top-hitches",
46        type=int,
47        default=5,
48        help="Correlate only the N worst hitches (avoid flooding output).",
49    )
50    parser.add_argument(
51        "--window",
52        type=str,
53        default=None,
54        help="Restrict analysis to a time slice, e.g. --window 10400:11700 (ms).",
55    )
56    parser.add_argument(
57        "--run",
58        type=int,
59        default=None,
60        help="Which run to analyze (1-based). Required for traces with >1 run.",
61    )
62    parser.add_argument(
63        "--list-runs", action="store_true",
64        help="Emit per-run metadata as JSON (use this to discover available runs).",
65    )
66 
67    # Mode flags (mutually exclusive with full analysis)
68    mode_group = parser.add_argument_group("Discovery modes")
69    mode_group.add_argument(
70        "--list-logs", action="store_true",
71        help="Emit os_log entries as JSON (use filter flags below).",
72    )
73    mode_group.add_argument(
74        "--list-signposts", action="store_true",
75        help="Emit os_signpost intervals + events as JSON.",
76    )
77    mode_group.add_argument("--log-subsystem", type=str, default=None)
78    mode_group.add_argument("--log-category", type=str, default=None)
79    mode_group.add_argument(
80        "--log-type", type=str, default=None,
81        help="e.g. Fault, Error, Default, Info, Debug",
82    )
83    mode_group.add_argument(
84        "--log-message-contains", type=str, default=None,
85        help="Case-insensitive substring match on the message / format string.",
86    )
87    mode_group.add_argument(
88        "--log-limit", type=int, default=None,
89        help="Cap number of log entries returned (applied after all filters).",
90    )
91    mode_group.add_argument(
92        "--signpost-name-contains", type=str, default=None,
93        help="Case-insensitive substring match on signpost name.",
94    )
95    mode_group.add_argument("--signpost-subsystem", type=str, default=None)
96    mode_group.add_argument("--signpost-category", type=str, default=None)
97    mode_group.add_argument(
98        "--fanin-for", type=str, default=None,
99        help="Emit incoming cause-graph sources for destinations whose fmt "
100             "contains this substring. Case-insensitive.",
101    )
102 
103    fmt_group = parser.add_mutually_exclusive_group()
104    fmt_group.add_argument("--json-only", action="store_true")
105    fmt_group.add_argument("--markdown-only", action="store_true")
106 
107    args = parser.parse_args(argv)
108 
109    # The discovery modes aren't in a mutually_exclusive_group because they
110    # live alongside their sub-filters in the same argparse group; enforce the
111    # constraint by hand so an agent gets a clear error instead of silent
112    # precedence.
113    active_modes = sum([
114        args.list_runs,
115        args.list_logs,
116        args.list_signposts,
117        bool(args.fanin_for),
118    ])
119    if active_modes > 1:
120        parser.error(
121            "--list-runs, --list-logs, --list-signposts, and --fanin-for are "
122            "mutually exclusive; pick one per invocation."
123        )
124 
125    trace = args.trace
126    if not trace.exists():
127        print(f"error: trace not found: {trace}", file=sys.stderr)
128        return 2
129 
130    info = xctrace.toc(trace)
131    window_ns = _parse_window(args.window)
132 
133    if args.list_runs:
134        sys.stdout.write(json.dumps({
135            "xctrace_version": info.xctrace_version,
136            "runs": [
137                {
138                    "number": r.number,
139                    "template": r.template_name,
140                    "duration_s": r.duration_s,
141                    "start_date": r.start_date,
142                    "end_date": r.end_date,
143                    "schemas": sorted(r.schemas),
144                }
145                for r in info.runs
146            ],
147        }, indent=2))
148        sys.stdout.write("\n")
149        return 0
150 
151    run_info = _resolve_run(info, args.run)
152    if run_info is None:
153        return 2
154    run_number = run_info.number
155 
156    if args.list_logs:
157        out = events.list_logs(
158            trace, run_info.schemas,
159            subsystem=args.log_subsystem,
160            category=args.log_category,
161            message_contains=args.log_message_contains,
162            message_type=args.log_type,
163            limit=args.log_limit,
164            window_ns=window_ns,
165            run=run_number,
166        )
167        sys.stdout.write(json.dumps({"logs": out, "count": len(out)}, indent=2))
168        sys.stdout.write("\n")
169        return 0
170 
171    if args.list_signposts:
172        sp = events.list_signposts(
173            trace, run_info.schemas,
174            name_contains=args.signpost_name_contains,
175            subsystem=args.signpost_subsystem,
176            category=args.signpost_category,
177            window_ns=window_ns,
178            run=run_number,
179        )
180        sys.stdout.write(json.dumps(sp, indent=2))
181        sys.stdout.write("\n")
182        return 0
183 
184    if args.fanin_for:
185        fanin = causes.fanin_for(
186            trace, run_info.schemas,
187            destination_contains=args.fanin_for,
188            top_k=args.top,
189            window=window_ns,
190            run=run_number,
191        )
192        sys.stdout.write(json.dumps(fanin, indent=2))
193        sys.stdout.write("\n")
194        return 0
195 
196    # Full five-lane analysis
197    schemas = run_info.schemas
198    lanes_out = {
199        "time-profiler":  time_profiler.analyze(trace, schemas, top_n=args.top, window=window_ns, run=run_number),
200        "hangs":          hangs.analyze(trace, schemas, top_n=args.top, window=window_ns, run=run_number),
201        "hitches":        hitches.analyze(trace, schemas, top_n=args.top, window=window_ns, run=run_number),
202        "swiftui":        swiftui.analyze(trace, schemas, top_n=args.top, window=window_ns, run=run_number),
203        "swiftui-causes": causes.analyze(trace, schemas, top_n=args.top, window=window_ns, run=run_number),
204    }
205    correlations = correlate.build(
206        lanes_out, top_hitches=args.top_hitches, top_symbols=5
207    )
208    public_lanes = [_strip_internal(l) for l in lanes_out.values()]
209 
210    result: dict = {
211        "trace": str(trace),
212        "xctrace_version": info.xctrace_version,
213        "run": run_number,
214        "runs_available": [r.number for r in info.runs],
215        "template": run_info.template_name,
216        "duration_s": run_info.duration_s,
217        "start_date": run_info.start_date,
218        "end_date": run_info.end_date,
219        "schemas_available": sorted(run_info.schemas),
220        "lanes": public_lanes,
221        "correlations": correlations,
222    }
223    if window_ns is not None:
224        result["window_ms"] = {
225            "start": window_ns[0] / 1_000_000,
226            "end": window_ns[1] / 1_000_000,
227        }
228 
229    md = summary.render(result)
230 
231    if args.output:
232        json_path = args.output.with_suffix(".json")
233        md_path = args.output.with_suffix(".md")
234        json_path.write_text(json.dumps(result, indent=2))
235        md_path.write_text(md)
236        print(f"wrote {json_path}")
237        print(f"wrote {md_path}")
238        return 0
239 
240    if args.markdown_only:
241        sys.stdout.write(md)
242    elif args.json_only:
243        sys.stdout.write(json.dumps(result, indent=2))
244        sys.stdout.write("\n")
245    else:
246        sys.stdout.write(json.dumps(result, indent=2))
247        sys.stdout.write("\n---\n")
248        sys.stdout.write(md)
249    return 0
250 
251 
252def _resolve_run(info, requested: int | None):
253    """Pick a run from the trace.
254 
255    If `requested` is given, return that run or None on miss (with a friendly
256    error). If unset and the trace has exactly one run, default to it. If
257    unset and there are multiple runs, error out so the agent picks
258    explicitly — silently picking run 1 lost data for the user.
259    """
260    if not info.runs:
261        print("error: trace has no runs", file=sys.stderr)
262        return None
263    if requested is not None:
264        try:
265            return info.get_run(requested)
266        except KeyError as e:
267            print(f"error: {e}", file=sys.stderr)
268            return None
269    if len(info.runs) == 1:
270        return info.runs[0]
271    available = ", ".join(str(r.number) for r in info.runs)
272    print(
273        f"error: trace has {len(info.runs)} runs ({available}); pass --run N. "
274        f"Use --list-runs to see per-run metadata.",
275        file=sys.stderr,
276    )
277    return None
278 
279 
280def _parse_window(spec: str | None) -> tuple[int, int] | None:
281    if not spec:
282        return None
283    if ":" not in spec:
284        raise SystemExit(f"--window expects START_MS:END_MS, got {spec!r}")
285    start_s, end_s = spec.split(":", 1)
286    try:
287        start_ms = float(start_s)
288        end_ms = float(end_s)
289    except ValueError as e:
290        raise SystemExit(f"--window: {e}")
291    if end_ms < start_ms:
292        raise SystemExit("--window: end_ms must be >= start_ms")
293    return (int(start_ms * 1_000_000), int(end_ms * 1_000_000))
294 
295 
296def _strip_internal(lane: dict) -> dict:
297    return {k: v for k, v in lane.items() if not k.startswith("_")}
298 
299 
300if __name__ == "__main__":
301    sys.exit(main())
302