aaronrene / gabriel-muse public
cli_hooks.py python
322 lines 11.4 KB
Raw
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago
1 """Thin CLI dispatchers for Phase 3.3 ``muse code`` commands on vaults.
2
3 Each function in this module is a single-purpose runner invoked by the
4 corresponding ``muse code <command>`` CLI when the active domain is
5 ``knowtation``. They build the right inputs (commits, snapshot loader, link
6 index) and delegate to :mod:`muse.plugins.knowtation.note_metrics`.
7
8 Routing pattern in each CLI command (``hotspots.py``, ``gravity.py``,
9 ``dead.py``, ``clones.py``, ``entangle.py``, ``velocity.py``)::
10
11 from muse.plugins.registry import read_domain
12 from muse.plugins.knowtation.cli_hooks import run_<command>_for_vault
13
14 def run(args):
15 root = require_repo()
16 try:
17 domain = read_domain(root)
18 except Exception:
19 domain = "code"
20 if domain == "knowtation":
21 run_<command>_for_vault(root, args)
22 return
23 # … existing code-domain logic …
24
25 Each runner prints JSON when ``args.as_json`` is True; otherwise emits a
26 brief human-readable rendering. All runners exit via ``SystemExit`` on
27 user-facing errors (e.g. missing vault, malformed args).
28 """
29
30 from __future__ import annotations
31
32 import json
33 import logging
34 import pathlib
35 import sys
36 from typing import Any
37
38 from muse.core.errors import ExitCode
39 from muse.core.repo import read_repo_id
40 from muse.core.store import (
41 get_commit_snapshot_manifest,
42 read_current_branch,
43 resolve_commit_ref,
44 )
45 from muse.core.validation import sanitize_display
46 from muse.plugins.code._query import walk_commits_bfs
47
48 logger = logging.getLogger(__name__)
49
50
51 # ---------------------------------------------------------------------------
52 # Shared helpers
53 # ---------------------------------------------------------------------------
54
55
56 def _walk_commits(root: pathlib.Path, max_commits: int = 500) -> list:
57 """BFS-walk the commit DAG from HEAD and return CommitRecord list.
58
59 Args:
60 root: Repository root.
61 max_commits: BFS safety cap.
62
63 Returns:
64 Commits sorted newest-first. Empty list on any failure.
65 """
66 try:
67 repo_id = read_repo_id(root)
68 branch = read_current_branch(root)
69 head = resolve_commit_ref(root, repo_id, branch, "HEAD")
70 if head is None:
71 return []
72 commits, _ = walk_commits_bfs(root, head.commit_id, max_commits=max_commits)
73 return commits
74 except (Exception, SystemExit) as exc:
75 logger.debug("cli_hooks: walk_commits failed: %s", exc)
76 return []
77
78
79 def _snapshot_loader(root: pathlib.Path):
80 """Return a closure ``commit_id → manifest`` for the metrics helpers."""
81
82 def loader(commit_id: str) -> dict[str, str]:
83 try:
84 return get_commit_snapshot_manifest(root, commit_id) or {}
85 except Exception:
86 return {}
87
88 return loader
89
90
91 def _emit(result: dict[str, Any], *, as_json: bool, text_renderer) -> None:
92 """Emit *result* as JSON or via *text_renderer*.
93
94 Args:
95 result: Result dict (must be JSON-serialisable).
96 as_json: If True, dump to stdout as JSON and return.
97 text_renderer: Callable taking the result dict and printing text.
98 """
99 if as_json:
100 result["domain"] = "knowtation"
101 print(json.dumps(result, indent=2))
102 return
103 text_renderer(result)
104
105
106 # ---------------------------------------------------------------------------
107 # hotspots
108 # ---------------------------------------------------------------------------
109
110
111 def run_hotspots_for_vault(root: pathlib.Path, args) -> None:
112 """muse code hotspots — vault note churn leaderboard."""
113 from muse.plugins.knowtation.note_metrics import note_hotspots
114
115 commits = _walk_commits(root)
116 loader = _snapshot_loader(root)
117 top = getattr(args, "top", 20) or 20
118 min_changes = getattr(args, "min_changes", None) or getattr(args, "min", 1) or 1
119 result = note_hotspots(commits, loader, top=top, min_changes=min_changes)
120
121 def text(r: dict[str, Any]) -> None:
122 print(f"\nNote churn — top {top} most-changed notes")
123 print(f"Commits analysed: {r['commits_analysed']}")
124 print("─" * 62)
125 if not r["ranking"]:
126 print(" (no churn detected)")
127 return
128 for i, entry in enumerate(r["ranking"], 1):
129 print(f" {i:>3} {sanitize_display(entry['path']):<48} {entry['changes']:>3} change(s)")
130
131 _emit(result, as_json=getattr(args, "as_json", False), text_renderer=text)
132
133
134 # ---------------------------------------------------------------------------
135 # gravity
136 # ---------------------------------------------------------------------------
137
138
139 def run_gravity_for_vault(root: pathlib.Path, args) -> None:
140 """muse code gravity — vault note structural weight."""
141 from muse.plugins.knowtation.link_index import build_link_index
142 from muse.plugins.knowtation.note_metrics import note_gravity
143
144 try:
145 index = build_link_index(root)
146 except ValueError as exc:
147 print(f"❌ Could not index vault: {exc}", file=sys.stderr)
148 raise SystemExit(ExitCode.USER_ERROR)
149
150 top = getattr(args, "top", 20) or 20
151 result = note_gravity(index, top=top)
152
153 def text(r: dict[str, Any]) -> None:
154 print(f"\nNote gravity — top {top} most-depended-upon notes")
155 print(f"Total notes in vault: {r['total_notes']}")
156 print("─" * 62)
157 if not r["ranking"]:
158 print(" (no inbound links in vault)")
159 return
160 for i, entry in enumerate(r["ranking"], 1):
161 grav_pct = entry["gravity"] * 100
162 print(
163 f" {i:>3} {sanitize_display(entry['path']):<48} "
164 f"{entry['inbound']:>3} inbound ({grav_pct:.2f}%)"
165 )
166
167 _emit(result, as_json=getattr(args, "as_json", False), text_renderer=text)
168
169
170 # ---------------------------------------------------------------------------
171 # dead
172 # ---------------------------------------------------------------------------
173
174
175 def run_dead_for_vault(root: pathlib.Path, args) -> None:
176 """muse code dead — vault dead-note detector."""
177 from muse.plugins.knowtation.link_index import build_link_index
178 from muse.plugins.knowtation.note_metrics import note_dead
179
180 try:
181 index = build_link_index(root)
182 except ValueError as exc:
183 print(f"❌ Could not index vault: {exc}", file=sys.stderr)
184 raise SystemExit(ExitCode.USER_ERROR)
185
186 use_mtime = getattr(args, "use_mtime", True)
187 days = getattr(args, "mtime_days", 180) or 180
188 result = note_dead(
189 index, root=root, use_mtime=use_mtime, mtime_days_threshold=days
190 )
191
192 top = getattr(args, "top", None)
193 if top:
194 result["dead_notes"] = result["dead_notes"][:top]
195
196 def text(r: dict[str, Any]) -> None:
197 print(f"\nDead notes (zero inbound links{'/old' if use_mtime else ''})")
198 print(f"Indexed: {r['total_notes_indexed']} Dead: {r['dead_count']}")
199 print("─" * 62)
200 if not r["dead_notes"]:
201 print(" (no dead notes found)")
202 return
203 for entry in r["dead_notes"]:
204 print(f" {sanitize_display(entry['path'])}")
205
206 _emit(result, as_json=getattr(args, "as_json", False), text_renderer=text)
207
208
209 # ---------------------------------------------------------------------------
210 # clones
211 # ---------------------------------------------------------------------------
212
213
214 def run_clones_for_vault(root: pathlib.Path, args) -> None:
215 """muse code clones — vault duplicate-note detector."""
216 from muse.plugins.knowtation.note_metrics import note_clones
217
218 try:
219 repo_id = read_repo_id(root)
220 branch = read_current_branch(root)
221 head = resolve_commit_ref(root, repo_id, branch, "HEAD")
222 if head is None:
223 print("❌ No HEAD commit found.", file=sys.stderr)
224 raise SystemExit(ExitCode.USER_ERROR)
225 manifest = get_commit_snapshot_manifest(root, head.commit_id) or {}
226 except Exception as exc:
227 print(f"❌ Could not read HEAD manifest: {exc}", file=sys.stderr)
228 raise SystemExit(ExitCode.USER_ERROR)
229
230 result = note_clones(manifest)
231
232 def text(r: dict[str, Any]) -> None:
233 print("\nDuplicate notes (exact content hash matches)")
234 print(f"Total notes: {r['total_notes']} Clones: {r['total_clones']}")
235 print("─" * 62)
236 if not r["clone_groups"]:
237 print(" (no duplicate notes found)")
238 return
239 for group in r["clone_groups"]:
240 print(f"\n hash {group['hash'][:12]}…:")
241 for path in group["paths"]:
242 print(f" {sanitize_display(path)}")
243
244 _emit(result, as_json=getattr(args, "as_json", False), text_renderer=text)
245
246
247 # ---------------------------------------------------------------------------
248 # entangle
249 # ---------------------------------------------------------------------------
250
251
252 def run_entangle_for_vault(root: pathlib.Path, args) -> None:
253 """muse code entangle — vault note co-change pair detector.
254
255 Phase 3.5: passes the LinkIndex so mist-attachment co-references add to
256 the entanglement score (notes that share a mist artefact become entangled).
257 """
258 from muse.plugins.knowtation.link_index import build_link_index
259 from muse.plugins.knowtation.note_metrics import note_entangle
260
261 commits = _walk_commits(root)
262 loader = _snapshot_loader(root)
263 top = getattr(args, "top", 20) or 20
264 min_co_changes = getattr(args, "min", 2) or 2
265
266 try:
267 link_idx = build_link_index(root)
268 except Exception: # noqa: BLE001
269 link_idx = None
270
271 result = note_entangle(
272 commits, loader,
273 min_co_changes=min_co_changes,
274 top=top,
275 link_index=link_idx,
276 )
277
278 def text(r: dict[str, Any]) -> None:
279 print(f"\nNote entanglement — top {top} co-changing pairs")
280 print(f"Commits analysed: {r['commits_analysed']}")
281 print("─" * 62)
282 if not r["ranking"]:
283 print(" (no co-changing note pairs found)")
284 return
285 for entry in r["ranking"]:
286 a, b = entry["pair"]
287 print(
288 f" {sanitize_display(a)} ↔ {sanitize_display(b)} "
289 f"({entry['co_changes']} co-changes)"
290 )
291
292 _emit(result, as_json=getattr(args, "as_json", False), text_renderer=text)
293
294
295 # ---------------------------------------------------------------------------
296 # velocity
297 # ---------------------------------------------------------------------------
298
299
300 def run_velocity_for_vault(root: pathlib.Path, args) -> None:
301 """muse code velocity — vault note add/remove/modify velocity."""
302 from muse.plugins.knowtation.note_metrics import note_velocity
303
304 commits = _walk_commits(root)
305 loader = _snapshot_loader(root)
306 window = getattr(args, "window", 10) or 10
307 try:
308 result = note_velocity(commits, loader, window_size=window)
309 except ValueError as exc:
310 print(f"❌ {exc}", file=sys.stderr)
311 raise SystemExit(ExitCode.USER_ERROR)
312
313 def text(r: dict[str, Any]) -> None:
314 print(f"\nVault velocity — window size {r['window_size']} commit(s)")
315 print("─" * 62)
316 rw = r["recent_window"]
317 ow = r["older_window"]
318 print(f" Recent window: +{rw['added']} / -{rw['removed']} / ~{rw['modified']} (net {rw['net']:+d})")
319 print(f" Older window: +{ow['added']} / -{ow['removed']} / ~{ow['modified']} (net {ow['net']:+d})")
320 print(f" Acceleration: {r['acceleration']:+d} notes")
321
322 _emit(result, as_json=getattr(args, "as_json", False), text_renderer=text)
File History 1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago