aaronrene / gabriel-muse public
gravity.py python
784 lines 27.2 KB
Raw
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 12 days ago
1 """muse code gravity — structural weight of every symbol.
2
3 "Gravity" answers the question every architect dreads:
4
5 **"If I change the contract of this symbol, how much of the codebase breaks?"**
6
7 Not just the direct callers — the full transitive closure. If ``read_object``
8 changes, that breaks ``read_commit``, which breaks ``resolve_commit_ref``, which
9 breaks every route handler, which breaks every CLI command that touches the
10 repo. That is gravity: the fraction of the production codebase that lives
11 downstream of a symbol in the call graph.
12
13 Why this matters
14 ----------------
15 * High-gravity symbols are **structural pillars**. Treat their public
16 contracts as APIs — design changes carefully, communicate broadly, bump
17 MAJOR.
18 * Low-gravity symbols are **safe to refactor**. They sit at the edge of the
19 dependency graph with few downstream effects.
20 * **Gravity ≠ hotspot.** A frequently-changed symbol with low gravity is a
21 quick iteration loop. A rarely-changed symbol with gravity 90% is a time
22 bomb — the moment its contract shifts, everything breaks.
23
24 How it works
25 ------------
26 1. Loads the committed HEAD snapshot and builds the production call graph
27 (AST-based, same approach as ``blast-risk`` and ``semantic-test-coverage``).
28 2. Inverts the graph: for each symbol, BFS through the reverse call graph to
29 find every symbol that transitively depends on it.
30 3. ``gravity_pct = transitive_dependents / total_production_symbols × 100``.
31
32 Test files are excluded from the dependency graph by default (a function being
33 called by 50 test functions does not mean it has production structural weight).
34 Pass ``--include-tests`` to count test callers as well.
35
36 Usage::
37
38 muse code gravity
39 muse code gravity --top 20
40 muse code gravity --min-gravity 50
41 muse code gravity --kind function
42 muse code gravity --file core/store.py
43 muse code gravity --sort depth
44 muse code gravity --depth 3
45 muse code gravity --explain "muse/core/object_store.py::read_object"
46 muse code gravity --include-tests
47 muse code gravity --json
48
49 Default output::
50
51 Symbol gravity — HEAD (378 production symbols)
52 Structural weight: fraction of production codebase that transitively depends on each symbol.
53
54 # ADDRESS GRAVITY DIRECT DEPTH
55 1 muse/core/object_store.py::read_object 94.2% 28 8
56 2 muse/core/store.py::resolve_commit_ref 87.1% 47 6
57 3 muse/core/errors.py::ExitCode 75.4% 31 5
58
59 ⚠️ High-gravity symbols are structural pillars. Treat their contracts as APIs.
60
61 ``--explain`` output::
62
63 Gravity breakdown — muse/core/object_store.py::read_object
64
65 Kind: function
66 Total gravity: 356 / 378 (94.2%)
67 Direct callers: 28
68 Max depth: 8
69
70 Depth distribution:
71 depth 1 (direct): 28 callers ████████████████████
72 depth 2: 89 callers ████████████████████████████████████████
73 depth 3: 127 callers ████████████████████████████████████████████████████
74 depth 4: 82 callers ██████████████████████████████████
75 depth 5+: 30 callers ████████████
76
77 Deepest callers (depth 8):
78 muse/api/routes/wire.py::push_handler
79 muse/api/routes/wire.py::pull_handler
80 muse/cli/commands/log.py::run
81
82 A breaking change here propagates through 8 layers of the codebase.
83
84 JSON output (``--json``)::
85
86 {
87 "ref": "HEAD",
88 "snapshot_id": "a3f2c9e1ab2c",
89 "total_production_symbols": 378,
90 "max_depth": 0,
91 "include_tests": false,
92 "filters": { "kind": null, "file": null, "min_gravity": 0.0, "top": 20 },
93 "symbols": [
94 {
95 "address": "muse/core/object_store.py::read_object",
96 "name": "read_object",
97 "kind": "function",
98 "file": "muse/core/object_store.py",
99 "gravity_pct": 94.2,
100 "direct_dependents": 28,
101 "transitive_dependents": 356,
102 "max_depth": 8,
103 "depth_distribution": { "1": 28, "2": 89, "3": 127, "4": 82, "5": 30 }
104 }
105 ]
106 }
107
108 Security note
109 -------------
110 Symbol addresses are validated to contain ``::`` before any store access.
111 File paths from the manifest are not passed to the shell — all access goes
112 through the object store's content-addressed lookup.
113 """
114
115 from __future__ import annotations
116
117 import argparse
118 import ast
119 import json
120 import logging
121 import pathlib
122 import re
123 import sys
124 from typing import Literal, TypedDict
125
126 from muse.core.errors import ExitCode
127 from muse.core.repo import read_repo_id, require_repo
128 from muse.core.store import (
129 Manifest,
130 get_commit_snapshot_manifest,
131 read_current_branch,
132 resolve_commit_ref,
133 )
134 from muse.plugins.code._callgraph import (
135 ForwardGraph,
136 ReverseGraph,
137 call_name,
138 find_func_node,
139 transitive_callers,
140 )
141 from muse.plugins.code._query import symbols_for_snapshot
142 from muse.plugins.code.ast_parser import SymbolRecord, parse_symbols
143 from muse.core.validation import clamp_int, MAX_AST_BYTES, sanitize_display
144
145 type _BlobMap = dict[str, bytes]
146 type _SymbolIndex = dict[str, SymbolRecord]
147 type _CounterMap = dict[str, int]
148
149 logger = logging.getLogger(__name__)
150
151 # ── Constants ──────────────────────────────────────────────────────────────────
152
153 _DEFAULT_TOP = 20
154 _DEFAULT_DEPTH = 0 # 0 = unlimited
155 _PY_SUFFIXES: frozenset[str] = frozenset({".py", ".pyi"})
156
157 _TEST_PATTERNS: tuple[re.Pattern[str], ...] = (
158 re.compile(r"(^|/)test_"),
159 re.compile(r"_test\.py$"),
160 re.compile(r"(^|/)tests/"),
161 re.compile(r"(^|/)spec/"),
162 re.compile(r"(^|/)conftest\.py$"),
163 )
164
165 _IMPORT_KIND = "import"
166 _TRACKED_KINDS: frozenset[str] = frozenset(
167 {"function", "async_function", "method", "async_method", "class"}
168 )
169
170 SortKey = Literal["gravity", "direct", "depth"]
171 _SORT_CHOICES: tuple[SortKey, ...] = ("gravity", "direct", "depth")
172
173 _BAR_MAX_WIDTH = 40
174
175
176 # ── TypedDicts ─────────────────────────────────────────────────────────────────
177
178
179 class _SymbolGravity(TypedDict):
180 """Gravity record for a single production symbol."""
181
182 address: str
183 name: str
184 kind: str
185 file: str
186 gravity_pct: float
187 direct_dependents: int
188 transitive_dependents: int
189 max_depth: int
190 depth_distribution: _CounterMap
191
192
193 class _FilterSpec(TypedDict):
194 kind: str | None
195 file: str | None
196 min_gravity: float
197 top: int
198
199
200 class _JsonOut(TypedDict):
201 ref: str
202 snapshot_id: str
203 total_production_symbols: int
204 max_depth: int
205 include_tests: bool
206 filters: _FilterSpec
207 symbols: list[_SymbolGravity]
208
209
210 # ── Helpers ────────────────────────────────────────────────────────────────────
211
212
213
214 def _is_test_file(file_path: str) -> bool:
215 """Return True if *file_path* looks like a test or conftest file."""
216 return any(p.search(file_path) for p in _TEST_PATTERNS)
217
218
219 def _load_py_blobs(
220 root: pathlib.Path,
221 manifest: Manifest,
222 include_tests: bool,
223 ) -> _BlobMap:
224 """Load Python file blobs from the object store into memory.
225
226 Args:
227 root: Repository root.
228 manifest: Snapshot manifest.
229 include_tests: When False (default), test files are excluded.
230
231 Returns:
232 Mapping ``{file_path: raw_bytes}`` for qualifying Python files.
233 """
234 from muse.core.object_store import read_object
235
236 blobs: _BlobMap = {}
237 for file_path, obj_id in manifest.items():
238 if pathlib.PurePosixPath(file_path).suffix.lower() not in _PY_SUFFIXES:
239 continue
240 if not include_tests and _is_test_file(file_path):
241 continue
242 raw = read_object(root, obj_id)
243 if raw is not None:
244 blobs[file_path] = raw
245 return blobs
246
247
248 def _build_forward_graph(
249 blobs: _BlobMap,
250 ) -> ForwardGraph:
251 """Build the forward call graph from pre-loaded blobs.
252
253 Scans every Python file in *blobs*, walking each callable symbol's body
254 for ``ast.Call`` nodes. Returns ``{caller_address: frozenset[callee_bare_name]}``.
255
256 Building from pre-loaded blobs avoids a second object-store round-trip.
257 """
258 graph: ForwardGraph = {}
259 for file_path, raw in blobs.items():
260 try:
261 if len(raw) > MAX_AST_BYTES:
262 return {}
263 tree = ast.parse(raw)
264 except SyntaxError:
265 continue
266 sym_tree = parse_symbols(raw, file_path)
267 for addr, rec in sym_tree.items():
268 if rec["kind"] not in {"function", "async_function", "method", "async_method"}:
269 continue
270 func_node = find_func_node(tree.body, rec["qualified_name"].split("."))
271 if func_node is None:
272 continue
273 callees: set[str] = set()
274 for node in ast.walk(func_node):
275 if isinstance(node, ast.Call):
276 n = call_name(node.func)
277 if n:
278 callees.add(n)
279 graph[addr] = frozenset(callees)
280 return graph
281
282
283 def _invert_graph(forward: ForwardGraph) -> ReverseGraph:
284 """Invert the forward call graph to get ``{callee_bare_name: [caller_addr, ...]}``.
285
286 Produces a sorted list of callers for each callee name so output is
287 deterministic across runs.
288 """
289 reverse: ReverseGraph = {}
290 for caller_addr, callee_names in forward.items():
291 for name in callee_names:
292 reverse.setdefault(name, []).append(caller_addr)
293 for name in reverse:
294 reverse[name].sort()
295 return reverse
296
297
298 def _gravity_bar(count: int, max_count: int, width: int = _BAR_MAX_WIDTH) -> str:
299 """Return a unicode block bar proportional to *count* / *max_count*."""
300 if max_count <= 0:
301 return ""
302 filled = round(count / max_count * width)
303 return "█" * filled
304
305
306 # ── Core algorithm ─────────────────────────────────────────────────────────────
307
308
309 def _compute_gravity_for_symbol(
310 bare_name: str,
311 reverse: ReverseGraph,
312 max_depth: int,
313 ) -> tuple[dict[int, list[str]], int]:
314 """Compute transitive dependents for a symbol with the given bare name.
315
316 Uses the existing ``transitive_callers`` BFS from ``_callgraph``.
317
318 Args:
319 bare_name: Bare callee name (last component of the symbol address).
320 reverse: Reverse call graph from ``_invert_graph``.
321 max_depth: BFS depth limit; ``0`` = unlimited.
322
323 Returns:
324 ``(depth_map, max_reached_depth)`` where depth_map is
325 ``{depth: [caller_address, ...]}``.
326 """
327 depth_map = transitive_callers(bare_name, reverse, max_depth=max_depth)
328 max_reached = max(depth_map) if depth_map else 0
329 return depth_map, max_reached
330
331
332 def _build_gravity_records(
333 prod_symbols: _SymbolIndex,
334 reverse: ReverseGraph,
335 max_depth: int,
336 file_filter: str | None,
337 kind_filter: str | None,
338 min_gravity: float,
339 top: int,
340 sort_key: SortKey,
341 ) -> tuple[list[_SymbolGravity], int]:
342 """Compute gravity for every qualifying production symbol.
343
344 Args:
345 prod_symbols: Flat ``{address: SymbolRecord}`` for all production symbols.
346 reverse: Reverse call graph.
347 max_depth: BFS depth cap (``0`` = unlimited).
348 file_filter: Optional path suffix filter.
349 kind_filter: Optional symbol kind filter.
350 min_gravity: Minimum gravity percentage threshold.
351 top: Maximum number of records to return (``0`` = unlimited).
352 sort_key: Sort dimension.
353
354 Returns:
355 ``(records, total_prod_symbols)`` — records are sorted and truncated
356 to *top*. ``total_prod_symbols`` is the unfiltered production symbol
357 count used as the denominator for gravity %.
358 """
359 total = len(prod_symbols)
360 if total == 0:
361 return [], 0
362
363 records: list[_SymbolGravity] = []
364
365 for addr, rec in prod_symbols.items():
366 kind = rec["kind"]
367 if kind == _IMPORT_KIND or kind not in _TRACKED_KINDS:
368 continue
369 file_path = addr.split("::")[0]
370 if file_filter and file_filter not in file_path:
371 continue
372 if kind_filter and kind != kind_filter:
373 continue
374
375 bare_name = rec["name"]
376 depth_map, max_reached = _compute_gravity_for_symbol(bare_name, reverse, max_depth)
377
378 direct = len(depth_map.get(1, []))
379 total_trans = sum(len(v) for v in depth_map.values())
380 denom = max(1, total - 1) # exclude self
381 pct = round(total_trans / denom * 100, 1)
382
383 if pct < min_gravity:
384 continue
385
386 # Flatten depth distribution to str-keyed dict for JSON.
387 dist: _CounterMap = {str(d): len(addrs) for d, addrs in depth_map.items()}
388
389 records.append(
390 _SymbolGravity(
391 address=addr,
392 name=bare_name,
393 kind=kind,
394 file=file_path,
395 gravity_pct=pct,
396 direct_dependents=direct,
397 transitive_dependents=total_trans,
398 max_depth=max_reached,
399 depth_distribution=dist,
400 )
401 )
402
403 # Sort.
404 if sort_key == "direct":
405 records.sort(key=lambda r: (r["direct_dependents"], r["gravity_pct"]), reverse=True)
406 elif sort_key == "depth":
407 records.sort(key=lambda r: (r["max_depth"], r["gravity_pct"]), reverse=True)
408 else: # gravity (default)
409 records.sort(key=lambda r: (r["gravity_pct"], r["transitive_dependents"]), reverse=True)
410
411 if top > 0:
412 records = records[:top]
413
414 return records, total
415
416
417 def _find_deepest_callers(
418 bare_name: str,
419 reverse: ReverseGraph,
420 max_depth: int,
421 show_count: int = 5,
422 ) -> tuple[list[str], int]:
423 """Return the addresses of the deepest callers (longest dependency chains).
424
425 Args:
426 bare_name: Target symbol's bare name.
427 reverse: Reverse call graph.
428 max_depth: BFS depth cap; 0 = unlimited.
429 show_count: Maximum number of deepest-caller addresses to return.
430
431 Returns:
432 ``(addresses, depth)`` — the deepest addresses found, and the depth at
433 which they were found.
434 """
435 depth_map, max_reached = _compute_gravity_for_symbol(bare_name, reverse, max_depth)
436 if not depth_map:
437 return [], 0
438 deepest = depth_map[max_reached]
439 return deepest[:show_count], max_reached
440
441
442 # ── Formatters ─────────────────────────────────────────────────────────────────
443
444
445 def _print_leaderboard(
446 records: list[_SymbolGravity],
447 total_prod: int,
448 max_depth_cap: int,
449 sort_key: SortKey,
450 include_tests: bool,
451 ) -> None:
452 """Print the human-readable gravity leaderboard."""
453 scope = "all symbols" if include_tests else "production symbols"
454 cap_str = f" · depth cap: {max_depth_cap}" if max_depth_cap > 0 else ""
455 print(f"\n Symbol gravity — HEAD ({total_prod} {scope}{cap_str})")
456 print(
457 f" Structural weight: fraction of {scope} that transitively depends"
458 " on each symbol.\n"
459 )
460
461 if not records:
462 print(" No symbols match the current filters.\n")
463 return
464
465 # Column widths.
466 addr_w = min(60, max(len(r["address"]) for r in records) + 2)
467 print(
468 f" {'#':>4} {'ADDRESS':<{addr_w}} {'GRAVITY':>8} {'DIRECT':>7} {'DEPTH':>6}"
469 )
470 print(f" {'─' * 4} {'─' * addr_w} {'─' * 8} {'─' * 7} {'─' * 6}")
471
472 for i, rec in enumerate(records, 1):
473 addr = rec["address"]
474 if len(addr) > addr_w:
475 addr = "…" + addr[-(addr_w - 1):]
476 pct_str = f"{rec['gravity_pct']:5.1f}%"
477 print(
478 f" {i:>4} {addr:<{addr_w}} {pct_str:>8}"
479 f" {rec['direct_dependents']:>7}"
480 f" {rec['max_depth']:>6}"
481 )
482
483 print()
484 if records:
485 top = records[0]
486 print(
487 f" ⚠️ {top['name']} carries {top['gravity_pct']:.1f}% structural weight"
488 " — treat its contract as an API."
489 )
490 print()
491
492
493 def _print_explain(
494 rec: _SymbolGravity,
495 reverse: ReverseGraph,
496 max_depth: int,
497 total_prod: int,
498 ) -> None:
499 """Print the detailed gravity breakdown for a single symbol."""
500 print(f"\n Gravity breakdown — {sanitize_display(rec['address'])}\n")
501 print(f" Kind: {sanitize_display(rec['kind'])}")
502 print(f" File: {sanitize_display(rec['file'])}")
503 print(
504 f" Total gravity: {rec['transitive_dependents']} / {total_prod}"
505 f" ({rec['gravity_pct']:.1f}%)"
506 )
507 print(f" Direct callers: {rec['direct_dependents']}")
508 print(f" Max depth: {rec['max_depth']}")
509 print()
510
511 if rec["depth_distribution"]:
512 # Re-fetch depth_map for the distribution bars.
513 depth_map, _ = _compute_gravity_for_symbol(rec["name"], reverse, max_depth)
514 max_at_level = max(len(v) for v in depth_map.values()) if depth_map else 1
515
516 print(" Depth distribution:")
517 for depth in sorted(depth_map):
518 callers = depth_map[depth]
519 count = len(callers)
520 bar = _gravity_bar(count, max_at_level, width=30)
521 label = f"depth {depth} (direct):" if depth == 1 else f"depth {depth}: "
522 print(f" {label} {count:>5} callers {bar}")
523 print()
524
525 # Deepest callers.
526 deepest_addrs, deepest_depth = _find_deepest_callers(rec["name"], reverse, max_depth)
527 if deepest_addrs:
528 print(f" Deepest callers (depth {deepest_depth}):")
529 for addr in deepest_addrs:
530 print(f" {sanitize_display(addr)}")
531 print()
532 print(
533 f" A breaking change here propagates through"
534 f" {deepest_depth} layers of the codebase."
535 )
536 print()
537
538
539 # ── Entry point ────────────────────────────────────────────────────────────────
540
541
542 def run(args: argparse.Namespace) -> None:
543 """Entry point for ``muse code gravity``."""
544 root = require_repo()
545
546 from muse.plugins.registry import read_domain
547 try:
548 if read_domain(root) == "knowtation":
549 from muse.plugins.knowtation.cli_hooks import run_gravity_for_vault
550 run_gravity_for_vault(root, args)
551 return
552 except Exception: # noqa: BLE001
553 pass
554
555 # ── Argument validation ────────────────────────────────────────────────────
556 top: int = clamp_int(args.top, 0, 10000, 'top')
557 if top < 0:
558 print("❌ --top must be >= 0 (0 = unlimited).", file=sys.stderr)
559 raise SystemExit(ExitCode.USER_ERROR)
560
561 max_depth: int = clamp_int(args.depth, 0, 50, 'depth')
562 if max_depth < 0:
563 print("❌ --depth must be >= 0 (0 = unlimited).", file=sys.stderr)
564 raise SystemExit(ExitCode.USER_ERROR)
565
566 min_gravity: float = args.min_gravity
567 if not (0.0 <= min_gravity <= 100.0):
568 print("❌ --min-gravity must be between 0.0 and 100.0.", file=sys.stderr)
569 raise SystemExit(ExitCode.USER_ERROR)
570
571 sort_key: SortKey = args.sort
572 kind_filter: str | None = args.kind or None
573 file_filter: str | None = args.file or None
574 include_tests: bool = args.include_tests
575 explain_addr: str | None = args.explain or None
576
577 if explain_addr is not None and "::" not in explain_addr:
578 print(
579 "❌ --explain ADDRESS must be in ``file::Symbol`` format.",
580 file=sys.stderr,
581 )
582 raise SystemExit(ExitCode.USER_ERROR)
583
584 # ── Resolve HEAD ───────────────────────────────────────────────────────────
585 repo_id = read_repo_id(root)
586 branch = read_current_branch(root)
587
588 head = resolve_commit_ref(root, repo_id, branch, None)
589 if head is None:
590 print("❌ HEAD commit not found — is this an empty repository?", file=sys.stderr)
591 raise SystemExit(ExitCode.USER_ERROR)
592
593 manifest = get_commit_snapshot_manifest(root, head.commit_id) or {}
594
595 # ── Single bulk read of all qualifying Python blobs ────────────────────────
596 blobs = _load_py_blobs(root, manifest, include_tests)
597
598 # ── Symbol extraction ──────────────────────────────────────────────────────
599 all_trees = symbols_for_snapshot(root, manifest)
600
601 # Partition: test files vs. production (for symbol scope).
602 prod_trees = (
603 all_trees
604 if include_tests
605 else {fp: t for fp, t in all_trees.items() if not _is_test_file(fp)}
606 )
607
608 # Flat production symbol index (imports excluded).
609 prod_symbols: _SymbolIndex = {}
610 for sym_tree in prod_trees.values():
611 for addr, rec in sym_tree.items():
612 if rec["kind"] != _IMPORT_KIND and rec["kind"] in _TRACKED_KINDS:
613 prod_symbols[addr] = rec
614
615 if not prod_symbols:
616 print(" No production symbols found in HEAD snapshot.\n")
617 return
618
619 # ── Build call graphs ──────────────────────────────────────────────────────
620 forward = _build_forward_graph(blobs)
621 reverse = _invert_graph(forward)
622
623 # ── --explain: single-symbol deep dive ────────────────────────────────────
624 if explain_addr is not None:
625 if explain_addr not in prod_symbols:
626 print(
627 f"❌ {explain_addr!r} not found in HEAD production symbols.",
628 file=sys.stderr,
629 )
630 print(
631 " Use ``muse code symbols`` to list available addresses.",
632 file=sys.stderr,
633 )
634 raise SystemExit(ExitCode.USER_ERROR)
635
636 rec_explain = prod_symbols[explain_addr]
637 bare = rec_explain["name"]
638 depth_map, max_reached = _compute_gravity_for_symbol(bare, reverse, max_depth)
639 direct = len(depth_map.get(1, []))
640 total_trans = sum(len(v) for v in depth_map.values())
641 denom = max(1, len(prod_symbols) - 1)
642 pct = round(total_trans / denom * 100, 1)
643 dist = {str(d): len(addrs) for d, addrs in depth_map.items()}
644
645 single_rec = _SymbolGravity(
646 address=explain_addr,
647 name=bare,
648 kind=rec_explain["kind"],
649 file=explain_addr.split("::")[0],
650 gravity_pct=pct,
651 direct_dependents=direct,
652 transitive_dependents=total_trans,
653 max_depth=max_reached,
654 depth_distribution=dist,
655 )
656
657 if args.json:
658 print(json.dumps(single_rec, indent=2))
659 return
660
661 _print_explain(single_rec, reverse, max_depth, len(prod_symbols))
662 return
663
664 # ── Leaderboard ────────────────────────────────────────────────────────────
665 records, total_prod = _build_gravity_records(
666 prod_symbols,
667 reverse,
668 max_depth,
669 file_filter,
670 kind_filter,
671 min_gravity,
672 top,
673 sort_key,
674 )
675
676 if args.json:
677 out: _JsonOut = _JsonOut(
678 ref="HEAD",
679 snapshot_id=head.commit_id[:12],
680 total_production_symbols=total_prod,
681 max_depth=max_depth,
682 include_tests=include_tests,
683 filters=_FilterSpec(
684 kind=kind_filter,
685 file=file_filter,
686 min_gravity=min_gravity,
687 top=top,
688 ),
689 symbols=records,
690 )
691 print(json.dumps(out, indent=2))
692 return
693
694 _print_leaderboard(records, total_prod, max_depth, sort_key, include_tests)
695
696
697 # ── CLI registration ───────────────────────────────────────────────────────────
698
699
700 def register(
701 sub: argparse._SubParsersAction[argparse.ArgumentParser],
702 ) -> None:
703 """Register ``gravity`` under the ``code`` subcommand group.
704
705 Args:
706 sub: The subparser action from the ``code`` command group.
707 """
708 p = sub.add_parser(
709 "gravity",
710 help=(
711 "Structural weight — how much of the production codebase"
712 " transitively depends on each symbol."
713 ),
714 description=__doc__,
715 formatter_class=argparse.RawDescriptionHelpFormatter,
716 )
717 p.add_argument(
718 "--explain",
719 metavar="ADDRESS",
720 help=(
721 "Deep-dive into a specific symbol's gravity:"
722 " depth distribution and deepest callers."
723 ),
724 )
725 p.add_argument(
726 "--top",
727 type=int,
728 default=_DEFAULT_TOP,
729 metavar="N",
730 help=f"Show only the top N symbols by gravity (default: {_DEFAULT_TOP}; 0 = all).",
731 )
732 p.add_argument(
733 "--sort",
734 metavar="KEY",
735 choices=list(_SORT_CHOICES),
736 default="gravity",
737 help=(
738 "Sort dimension: ``gravity`` (default, transitive %%)"
739 ", ``direct`` (direct callers)"
740 ", ``depth`` (max dependency chain length)."
741 ),
742 )
743 p.add_argument(
744 "--depth",
745 type=int,
746 default=_DEFAULT_DEPTH,
747 metavar="D",
748 help=(
749 f"Maximum BFS depth for transitive analysis (default: {_DEFAULT_DEPTH} = unlimited)."
750 " Use a small value (e.g. 3) for large repos to bound runtime."
751 ),
752 )
753 p.add_argument(
754 "--kind",
755 metavar="KIND",
756 choices=["function", "async_function", "method", "async_method", "class"],
757 help="Filter symbols by kind.",
758 )
759 p.add_argument(
760 "--file",
761 metavar="SUFFIX",
762 help="Scope analysis to symbols in files whose path contains SUFFIX.",
763 )
764 p.add_argument(
765 "--min-gravity",
766 type=float,
767 default=0.0,
768 metavar="PCT",
769 help="Show only symbols with gravity >= PCT%%.",
770 )
771 p.add_argument(
772 "--include-tests",
773 action="store_true",
774 help=(
775 "Count test-file callers in the dependency graph"
776 " (default: test callers are excluded)."
777 ),
778 )
779 p.add_argument(
780 "--json",
781 action="store_true",
782 help="Emit JSON instead of human-readable text.",
783 )
784 p.set_defaults(func=run)
File History 1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 12 days ago