aaronrene / gabriel-muse public
_framework.py python
702 lines 23.7 KB
Raw
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago
1 """Framework entry-point detection for the code-domain symbol graph.
2
3 Problem
4 -------
5 Static call-graph analysis only records *explicit* call edges — ``A()`` inside
6 ``B`` produces edge ``B→A``. Framework entry points (FastAPI route handlers,
7 Flask views, Celery tasks, …) are wired at *runtime* through decorators and DI
8 containers. No explicit call site exists in user code, so they appear dead to
9 :func:`~muse.cli.commands.dead.run` and have empty blast-radius in
10 :func:`~muse.cli.commands.impact.run`.
11
12 Solution
13 --------
14 This module adds a complementary edge type — :class:`ImplicitEntryEdge` — that
15 represents "this symbol is externally reachable via the framework named
16 ``framework_id``." Edges are synthesised by :class:`FrameworkPlugin`
17 implementations, one per framework, during a single linear pass over the
18 manifest. The resulting :data:`ImplicitEdgeGraph` is consumed by the impact
19 and dead-code commands to:
20
21 * Surface entry-point annotations instead of "no callers detected".
22 * Exclude entry-point symbols from dead-code results.
23
24 Adding a new framework
25 ----------------------
26 1. Create a class that satisfies the :class:`FrameworkPlugin` protocol.
27 2. Append an instance to :data:`BUILTIN_PLUGINS`.
28 3. Write tests in ``tests/test_framework_plugins.py``.
29
30 No changes to the core graph, impact, or dead-code modules are required.
31
32 Configuration
33 -------------
34 ``.muse/code_config.toml`` is the single configuration file for all code-domain
35 intelligence. It has two independent sections:
36
37 ``[code]`` — controls which tree-sitter language grammars are active::
38
39 [code]
40 # Restrict to Python + TypeScript + shell; all other grammars become
41 # file-level fallbacks even if the grammar packages are installed.
42 languages = ["python", "typescript", "tsx", "css", "bash"]
43
44 ``[framework_detection]`` — controls entry-point detection::
45
46 [framework_detection]
47 auto_detect = true
48 disabled_plugins = [] # e.g. ["celery"] to silence false positives
49
50 [[framework_detection.custom_entry_points]]
51 language = "Python"
52 kind = "custom-rpc"
53 decorator_names = ["rpc_handler", "grpc_handler"]
54
55 Both sections are optional; omitting either reproduces the zero-config defaults.
56 """
57
58 from __future__ import annotations
59
60 import ast
61 import logging
62 import pathlib
63 import tomllib
64 from dataclasses import dataclass, field
65 from typing import Protocol, runtime_checkable
66
67 from muse.core._types import Manifest
68 from muse.core.object_store import read_object
69 from muse.core.symbol_cache import SymbolCache
70 from muse.core.validation import MAX_AST_BYTES
71 from muse.plugins.code.ast_parser import SymbolTree, parse_symbols
72
73 logger = logging.getLogger(__name__)
74
75 # ---------------------------------------------------------------------------
76 # Core data types
77 # ---------------------------------------------------------------------------
78
79 #: String-keyed metadata dict for per-framework entry-point annotations.
80 #: Keys vary by framework (e.g. "method", "path" for HTTP; "hook" for Flask).
81 type _Metadata = dict[str, str]
82
83
84 @dataclass(frozen=True, slots=True)
85 class ImplicitEntryEdge:
86 """One framework-implied reachability edge for a symbol.
87
88 Attributes:
89 framework_id: Stable lowercase identifier, e.g. ``"fastapi"``.
90 symbol_address: Full symbol address, e.g.
91 ``"app/routers/runs.py::create_run"``.
92 kind: Semantic category of the entry point:
93 ``"http-route"``, ``"task"``, ``"cli-command"``, …
94 metadata: Arbitrary per-framework annotations. Common keys:
95 ``method`` (HTTP verb), ``path`` (URL pattern),
96 ``queue`` (Celery queue name).
97 """
98
99 framework_id: str
100 symbol_address: str
101 kind: str
102 metadata: _Metadata = field(default_factory=dict)
103
104
105 #: Maps ``symbol_address → [ImplicitEntryEdge, …]`` for every framework-wired
106 #: symbol in a snapshot. Analogous to :data:`~._callgraph.ReverseGraph` but
107 #: for implicit framework callers rather than explicit call sites.
108 ImplicitEdgeGraph = dict[str, list[ImplicitEntryEdge]]
109
110
111 # ---------------------------------------------------------------------------
112 # FrameworkPlugin protocol
113 # ---------------------------------------------------------------------------
114
115
116 @runtime_checkable
117 class FrameworkPlugin(Protocol):
118 """Protocol every framework entry-point plugin must satisfy.
119
120 Implementations should be stateless singletons. The contract is:
121
122 * ``language`` — the primary language this plugin targets. Used to skip
123 files whose suffix does not match, avoiding unnecessary parses.
124 * ``framework_id`` — a stable lowercase slug, e.g. ``"fastapi"``.
125 * ``detect_entry_points`` — pure, thread-safe, returns an empty list when
126 the file contains no framework-wired symbols.
127 """
128
129 @property
130 def language(self) -> str:
131 """Primary language, e.g. ``"Python"`` or ``"TypeScript"``."""
132 ...
133
134 @property
135 def framework_id(self) -> str:
136 """Stable lowercase slug, e.g. ``"fastapi"``."""
137 ...
138
139 def detect_entry_points(
140 self,
141 file_path: str,
142 sym_tree: SymbolTree,
143 source: bytes,
144 ) -> list[ImplicitEntryEdge]:
145 """Return implicit entry edges for framework-wired symbols in *file_path*.
146
147 Args:
148 file_path: Workspace-relative POSIX path, e.g.
149 ``"app/routers/runs.py"``.
150 sym_tree: Pre-parsed symbol tree for the file. Plugins may
151 use this to correlate decorator findings with known
152 symbol addresses without re-parsing.
153 source: Raw file bytes. Plugins that need full AST detail
154 (e.g. decorator arguments) must call
155 ``ast.parse(source)`` themselves.
156
157 Returns:
158 A (possibly empty) list of :class:`ImplicitEntryEdge` objects.
159 One edge per framework-wired symbol per entry-point type.
160 """
161 ...
162
163
164 # ---------------------------------------------------------------------------
165 # Configuration
166 # ---------------------------------------------------------------------------
167
168
169 @dataclass
170 class _CustomEntryPointRule:
171 """One user-defined entry-point pattern from ``.muse/code_config.toml``."""
172
173 language: str
174 kind: str
175 decorator_names: list[str] = field(default_factory=list)
176
177
178 @dataclass
179 class FrameworkConfig:
180 """Parsed ``[framework_detection]`` section from ``.muse/code_config.toml``.
181
182 Defaults reflect the "zero-config" happy path: all built-in plugins active,
183 no custom rules.
184 """
185
186 auto_detect: bool = True
187 disabled_plugins: frozenset[str] = field(default_factory=frozenset)
188 custom_entry_points: list[_CustomEntryPointRule] = field(default_factory=list)
189
190
191 def load_framework_config(root: pathlib.Path) -> FrameworkConfig:
192 """Read ``[framework_detection]`` from ``.muse/code_config.toml``.
193
194 Returns default :class:`FrameworkConfig` if the file is absent or the
195 section is missing — callers should not need to handle ``None``.
196 """
197 config_path = root / ".muse" / "code_config.toml"
198 if not config_path.exists():
199 return FrameworkConfig()
200
201 try:
202 with open(config_path, "rb") as fh:
203 raw = tomllib.load(fh)
204 except Exception as exc: # noqa: BLE001
205 logger.warning("⚠️ Could not parse .muse/code_config.toml: %s", exc)
206 return FrameworkConfig()
207
208 section = raw.get("framework_detection", {})
209 if not isinstance(section, dict):
210 return FrameworkConfig()
211
212 auto_detect: bool = bool(section.get("auto_detect", True))
213 disabled: list[str] = section.get("disabled_plugins", [])
214 custom: list[_CustomEntryPointRule] = []
215 for rule in section.get("custom_entry_points", []):
216 if not isinstance(rule, dict):
217 continue
218 lang = rule.get("language", "Python")
219 kind = rule.get("kind", "custom")
220 names = rule.get("decorator_names", [])
221 if isinstance(names, list):
222 custom.append(_CustomEntryPointRule(language=lang, kind=kind, decorator_names=names))
223
224 return FrameworkConfig(
225 auto_detect=auto_detect,
226 disabled_plugins=frozenset(disabled),
227 custom_entry_points=custom,
228 )
229
230
231 # ---------------------------------------------------------------------------
232 # Custom pattern plugin (reads user-defined rules from FrameworkConfig)
233 # ---------------------------------------------------------------------------
234
235
236 class _CustomPatternPlugin:
237 """Synthesises entry edges from ``[[framework_detection.custom_entry_points]]`` rules."""
238
239 language = "Python"
240 framework_id = "custom"
241
242 def __init__(self, rules: list[_CustomEntryPointRule]) -> None:
243 self._rules = [r for r in rules if r.language == "Python"]
244
245 def detect_entry_points(
246 self,
247 file_path: str,
248 sym_tree: SymbolTree,
249 source: bytes,
250 ) -> list[ImplicitEntryEdge]:
251 if not self._rules or len(source) > MAX_AST_BYTES:
252 return []
253 try:
254 tree = ast.parse(source)
255 except SyntaxError:
256 return []
257
258 decorator_to_kind = {} # bare_name → kind string
259 for rule in self._rules:
260 for name in rule.decorator_names:
261 decorator_to_kind[name] = rule.kind
262
263 if not decorator_to_kind:
264 return []
265
266 edges: list[ImplicitEntryEdge] = []
267 for node in ast.walk(tree):
268 if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
269 continue
270 for dec in node.decorator_list:
271 dec_name = _decorator_bare_name(dec)
272 if dec_name and dec_name in decorator_to_kind:
273 addr = _resolve_address(file_path, node.name, sym_tree)
274 if addr:
275 edges.append(ImplicitEntryEdge(
276 framework_id=self.framework_id,
277 symbol_address=addr,
278 kind=decorator_to_kind[dec_name],
279 metadata={"decorator": dec_name},
280 ))
281 return edges
282
283
284 # ---------------------------------------------------------------------------
285 # Built-in plugins
286 # ---------------------------------------------------------------------------
287
288
289 class _FastAPIPlugin:
290 """Detects FastAPI / APIRouter route handlers.
291
292 Recognises the decorator forms::
293
294 @app.get("/path")
295 @router.post("/path", ...)
296 @api_router.put("/path")
297
298 Any variable name is accepted as the router/app object; only the HTTP
299 *method attribute* is checked (``get``, ``post``, ``put``, ``delete``,
300 ``patch``, ``head``, ``options``, ``trace``).
301 """
302
303 language = "Python"
304 framework_id = "fastapi"
305
306 _HTTP_METHODS: frozenset[str] = frozenset(
307 {"get", "post", "put", "delete", "patch", "head", "options", "trace"}
308 )
309
310 def detect_entry_points(
311 self,
312 file_path: str,
313 sym_tree: SymbolTree,
314 source: bytes,
315 ) -> list[ImplicitEntryEdge]:
316 if len(source) > MAX_AST_BYTES:
317 return []
318 try:
319 tree = ast.parse(source)
320 except SyntaxError:
321 return []
322
323 edges: list[ImplicitEntryEdge] = []
324 for node in ast.walk(tree):
325 if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
326 continue
327 for dec in node.decorator_list:
328 edge = self._edge_from_decorator(dec, node.name, file_path, sym_tree)
329 if edge is not None:
330 edges.append(edge)
331 return edges
332
333 def _edge_from_decorator(
334 self,
335 dec: ast.expr,
336 func_name: str,
337 file_path: str,
338 sym_tree: SymbolTree,
339 ) -> ImplicitEntryEdge | None:
340 # Must be a Call: @router.get("/path") or @router.get("/path", ...)
341 if not isinstance(dec, ast.Call):
342 return None
343 func = dec.func
344 # Must be an attribute: <anything>.get / .post / etc.
345 if not isinstance(func, ast.Attribute):
346 return None
347 method = func.attr.lower()
348 if method not in self._HTTP_METHODS:
349 return None
350
351 # Extract the path from the first positional argument (optional).
352 path = ""
353 if dec.args and isinstance(dec.args[0], ast.Constant):
354 path = str(dec.args[0].value)
355
356 addr = _resolve_address(file_path, func_name, sym_tree)
357 if addr is None:
358 return None
359
360 return ImplicitEntryEdge(
361 framework_id=self.framework_id,
362 symbol_address=addr,
363 kind="http-route",
364 metadata={"method": method.upper(), "path": path},
365 )
366
367
368 class _FlaskPlugin:
369 """Detects Flask and Blueprint route handlers and lifecycle hooks.
370
371 Recognises::
372
373 @app.route("/path", methods=["GET", "POST"])
374 @bp.route("/path")
375 @app.before_request / @app.after_request / @app.teardown_appcontext
376 @app.errorhandler(404)
377 @app.cli.command()
378 """
379
380 language = "Python"
381 framework_id = "flask"
382
383 _ROUTE_ATTRS: frozenset[str] = frozenset({"route"})
384 _LIFECYCLE_ATTRS: frozenset[str] = frozenset(
385 {"before_request", "after_request", "teardown_appcontext",
386 "before_app_request", "after_app_request"}
387 )
388 _ERROR_ATTRS: frozenset[str] = frozenset({"errorhandler"})
389
390 def detect_entry_points(
391 self,
392 file_path: str,
393 sym_tree: SymbolTree,
394 source: bytes,
395 ) -> list[ImplicitEntryEdge]:
396 if len(source) > MAX_AST_BYTES:
397 return []
398 try:
399 tree = ast.parse(source)
400 except SyntaxError:
401 return []
402
403 edges: list[ImplicitEntryEdge] = []
404 for node in ast.walk(tree):
405 if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
406 continue
407 for dec in node.decorator_list:
408 edge = self._edge_from_decorator(dec, node.name, file_path, sym_tree)
409 if edge is not None:
410 edges.append(edge)
411 return edges
412
413 def _edge_from_decorator(
414 self,
415 dec: ast.expr,
416 func_name: str,
417 file_path: str,
418 sym_tree: SymbolTree,
419 ) -> ImplicitEntryEdge | None:
420 addr = _resolve_address(file_path, func_name, sym_tree)
421 if addr is None:
422 return None
423
424 # @app.route("/path", methods=[...])
425 if isinstance(dec, ast.Call) and isinstance(dec.func, ast.Attribute):
426 attr = dec.func.attr
427 if attr in self._ROUTE_ATTRS:
428 path = ""
429 if dec.args and isinstance(dec.args[0], ast.Constant):
430 path = str(dec.args[0].value)
431 methods = _extract_methods_kwarg(dec) or ["GET"]
432 return ImplicitEntryEdge(
433 framework_id=self.framework_id,
434 symbol_address=addr,
435 kind="http-route",
436 metadata={"method": ",".join(methods), "path": path},
437 )
438 if attr in self._LIFECYCLE_ATTRS:
439 return ImplicitEntryEdge(
440 framework_id=self.framework_id,
441 symbol_address=addr,
442 kind="lifecycle-hook",
443 metadata={"hook": attr},
444 )
445 if attr in self._ERROR_ATTRS:
446 return ImplicitEntryEdge(
447 framework_id=self.framework_id,
448 symbol_address=addr,
449 kind="error-handler",
450 metadata={},
451 )
452
453 # Bare @app.before_request (no call parens)
454 if isinstance(dec, ast.Attribute) and dec.attr in self._LIFECYCLE_ATTRS:
455 return ImplicitEntryEdge(
456 framework_id=self.framework_id,
457 symbol_address=addr,
458 kind="lifecycle-hook",
459 metadata={"hook": dec.attr},
460 )
461
462 return None
463
464
465 class _CeleryPlugin:
466 """Detects Celery task functions.
467
468 Recognises::
469
470 @app.task
471 @celery.task
472 @shared_task
473 @app.task(bind=True)
474 @app.periodic_task(run_every=…)
475 """
476
477 language = "Python"
478 framework_id = "celery"
479
480 _TASK_BARE: frozenset[str] = frozenset({"shared_task"})
481 _TASK_ATTRS: frozenset[str] = frozenset({"task", "periodic_task"})
482
483 def detect_entry_points(
484 self,
485 file_path: str,
486 sym_tree: SymbolTree,
487 source: bytes,
488 ) -> list[ImplicitEntryEdge]:
489 if len(source) > MAX_AST_BYTES:
490 return []
491 try:
492 tree = ast.parse(source)
493 except SyntaxError:
494 return []
495
496 edges: list[ImplicitEntryEdge] = []
497 for node in ast.walk(tree):
498 if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
499 continue
500 for dec in node.decorator_list:
501 edge = self._edge_from_decorator(dec, node.name, file_path, sym_tree)
502 if edge is not None:
503 edges.append(edge)
504 return edges
505
506 def _edge_from_decorator(
507 self,
508 dec: ast.expr,
509 func_name: str,
510 file_path: str,
511 sym_tree: SymbolTree,
512 ) -> ImplicitEntryEdge | None:
513 addr = _resolve_address(file_path, func_name, sym_tree)
514 if addr is None:
515 return None
516
517 # @shared_task (bare Name)
518 if isinstance(dec, ast.Name) and dec.id in self._TASK_BARE:
519 return ImplicitEntryEdge(
520 framework_id=self.framework_id,
521 symbol_address=addr,
522 kind="task",
523 metadata={},
524 )
525
526 # @shared_task(...) (called Name)
527 if isinstance(dec, ast.Call) and isinstance(dec.func, ast.Name):
528 if dec.func.id in self._TASK_BARE:
529 return ImplicitEntryEdge(
530 framework_id=self.framework_id,
531 symbol_address=addr,
532 kind="task",
533 metadata={},
534 )
535
536 # @app.task / @celery.task (bare Attribute)
537 if isinstance(dec, ast.Attribute) and dec.attr in self._TASK_ATTRS:
538 kind = "periodic-task" if dec.attr == "periodic_task" else "task"
539 return ImplicitEntryEdge(
540 framework_id=self.framework_id,
541 symbol_address=addr,
542 kind=kind,
543 metadata={},
544 )
545
546 # @app.task(...) / @app.periodic_task(...) (called Attribute)
547 if isinstance(dec, ast.Call) and isinstance(dec.func, ast.Attribute):
548 if dec.func.attr in self._TASK_ATTRS:
549 kind = "periodic-task" if dec.func.attr == "periodic_task" else "task"
550 return ImplicitEntryEdge(
551 framework_id=self.framework_id,
552 symbol_address=addr,
553 kind=kind,
554 metadata={},
555 )
556
557 return None
558
559
560 #: Default plugin instances — extend this list to add new frameworks.
561 BUILTIN_PLUGINS: list[FrameworkPlugin] = [
562 _FastAPIPlugin(),
563 _FlaskPlugin(),
564 _CeleryPlugin(),
565 ]
566
567
568 # ---------------------------------------------------------------------------
569 # Graph construction
570 # ---------------------------------------------------------------------------
571
572
573 _PY_SUFFIXES: frozenset[str] = frozenset({".py", ".pyi"})
574
575
576 def build_implicit_edge_graph(
577 root: pathlib.Path,
578 manifest: Manifest,
579 cache: SymbolCache | None = None,
580 config: FrameworkConfig | None = None,
581 ) -> ImplicitEdgeGraph:
582 """Build the implicit entry-edge graph from *manifest*.
583
584 Runs all active :data:`BUILTIN_PLUGINS` (plus any custom rules from
585 *config*) over every file in the manifest and aggregates the results.
586
587 Args:
588 root: Repository root — used to read blobs from the object store.
589 manifest: Snapshot manifest mapping file path → SHA-256 object ID.
590 cache: Optional :class:`~muse.core.symbol_cache.SymbolCache`.
591 When provided, symbol trees are served from cache,
592 avoiding redundant parses.
593 config: Framework configuration. When ``None``,
594 :func:`load_framework_config` is called with *root*.
595
596 Returns:
597 :data:`ImplicitEdgeGraph` — ``{symbol_address: [ImplicitEntryEdge, …]}``.
598 Symbols with no implicit callers are absent from the mapping.
599 """
600 if config is None:
601 config = load_framework_config(root)
602
603 if not config.auto_detect:
604 return {}
605
606 active_plugins: list[FrameworkPlugin] = [
607 p for p in BUILTIN_PLUGINS
608 if p.framework_id not in config.disabled_plugins
609 ]
610 if config.custom_entry_points:
611 active_plugins.append(_CustomPatternPlugin(config.custom_entry_points))
612
613 if not active_plugins:
614 return {}
615
616 # Partition plugins by language so we only run Python plugins on .py files.
617 py_plugins = [p for p in active_plugins if p.language == "Python"]
618
619 graph: ImplicitEdgeGraph = {}
620
621 for file_path, obj_id in manifest.items():
622 suffix = pathlib.PurePosixPath(file_path).suffix.lower()
623
624 plugins_for_file = py_plugins if suffix in _PY_SUFFIXES else []
625 if not plugins_for_file:
626 continue
627
628 raw = read_object(root, obj_id)
629 if raw is None or len(raw) > MAX_AST_BYTES:
630 continue
631
632 sym_tree: SymbolTree = (
633 cache.get(obj_id) or parse_symbols(raw, file_path)
634 if cache is not None
635 else parse_symbols(raw, file_path)
636 )
637
638 for plugin in plugins_for_file:
639 try:
640 edges = plugin.detect_entry_points(file_path, sym_tree, raw)
641 except Exception as exc: # noqa: BLE001
642 logger.warning(
643 "⚠️ %s plugin raised on %s: %s",
644 plugin.framework_id, file_path, exc,
645 )
646 continue
647 for edge in edges:
648 graph.setdefault(edge.symbol_address, []).append(edge)
649
650 return graph
651
652
653 # ---------------------------------------------------------------------------
654 # Internal helpers
655 # ---------------------------------------------------------------------------
656
657
658 def _resolve_address(
659 file_path: str,
660 func_name: str,
661 sym_tree: SymbolTree,
662 ) -> str | None:
663 """Return the full symbol address for *func_name* within *file_path*.
664
665 Tries exact match first (``file::name``), then a scan for method forms
666 (``file::ClassName.name``). Returns ``None`` if the name is not in the
667 symbol tree — this prevents synthesising edges for lambdas or inline
668 functions that the AST parser does not index.
669 """
670 direct = f"{file_path}::{func_name}"
671 if direct in sym_tree:
672 return direct
673 # Method match: "file::SomeClass.func_name"
674 for addr in sym_tree:
675 if addr.endswith(f".{func_name}") and addr.startswith(file_path + "::"):
676 return addr
677 return None
678
679
680 def _decorator_bare_name(dec: ast.expr) -> str | None:
681 """Return the bare decorator name for simple ``@name`` or ``@name(...)`` forms."""
682 if isinstance(dec, ast.Name):
683 return dec.id
684 if isinstance(dec, ast.Call) and isinstance(dec.func, ast.Name):
685 return dec.func.id
686 if isinstance(dec, ast.Attribute):
687 return dec.attr
688 if isinstance(dec, ast.Call) and isinstance(dec.func, ast.Attribute):
689 return dec.func.attr
690 return None
691
692
693 def _extract_methods_kwarg(call_node: ast.Call) -> list[str] | None:
694 """Extract HTTP methods from a ``methods=[...]`` keyword in a decorator call."""
695 for kw in call_node.keywords:
696 if kw.arg == "methods" and isinstance(kw.value, ast.List):
697 methods: list[str] = []
698 for elt in kw.value.elts:
699 if isinstance(elt, ast.Constant):
700 methods.append(str(elt.value).upper())
701 return methods or None
702 return None
File History 1 commit
sha256:cf6265cea8c21d9228d90dec13ef6ec2dab5103d466db9cc4590681832de4bf8 docs(KD-STAGING): sync governance after KD-6b DONE Human 13 days ago