feat(scripts): add audit_legacy_wrappers.py + Phase 2 wrapper inventory (9 P1 wrappers)
Phase 2 inventory results (vs spec claim of 8+ confirmed): - Total wrappers: 9 (all P1 drop-errors-via-.data; no P3 confirmed) - By file: mcp_client 1, ai_client 5, rag_engine 1, gui_2 2 Audit script revision: The spec's audit logic incorrectly flagged the proper _result helpers as wrappers (they contain _result( calls in their body when they call OTHER _result helpers). The fix: require the function name NOT to end in _result, AND the body must call (name + _result) specifically. This narrowed the finding from 111 (false-positive) to 9 (true legacy wrappers). Public MCP tool wrappers (search_files, list_directory, etc.) are NOT flagged: they ARE the protocol drain points, returning str per JSON-RPC wire format.
This commit is contained in:
@@ -0,0 +1,109 @@
|
||||
"""Audit legacy wrapper patterns in src/.
|
||||
|
||||
A legacy wrapper is `def _x(...):` that delegates to `_<x>_result(...).data`,
|
||||
dropping the .ok check and error context. This is a false drain: per
|
||||
error_handling.md the wrapper defeats the entire Result[T] migration.
|
||||
|
||||
Patterns scanned:
|
||||
P1 `def _x(...):` whose body is `return _x_result(...).data`
|
||||
(the primary false-drain; .data extraction drops .ok + .errors)
|
||||
P2 `def _x(...):` whose body checks `.ok` but does `pass` or only
|
||||
logs (softer false drain)
|
||||
P3 `def _x(...):` whose body is `return _x_result(...)`
|
||||
(returns the Result; caller doesn't get the ok check at the wrapper)
|
||||
"""
|
||||
import ast
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def is_legacy_wrapper(func: ast.FunctionDef) -> tuple[bool, str]:
|
||||
"""Return (is_legacy_wrapper, pattern_name) for a function.
|
||||
|
||||
A legacy wrapper is `def _x(...):` (NOT ending in `_result`) whose body
|
||||
delegates to its sibling `_x_result(...)` helper. The wrapper defeats the
|
||||
Result[T] migration by unwrapping .data (P1) or returning the Result
|
||||
unchanged (P3, extra layer of indirection but propagates errors).
|
||||
|
||||
Functions whose own name ends in `_result` are NOT wrappers — they ARE
|
||||
the Result-returning helpers. They may legitimately call OTHER helpers
|
||||
(also ending in `_result`), but that's a call graph not a wrapper pattern.
|
||||
"""
|
||||
name = func.name
|
||||
if not name.startswith("_"):
|
||||
return False, ""
|
||||
if name.endswith("_result"):
|
||||
return False, ""
|
||||
if "return " not in ast.unparse(func):
|
||||
return False, ""
|
||||
sibling = name + "_result"
|
||||
body_str = ast.unparse(func)
|
||||
if sibling + "(" not in body_str:
|
||||
return False, ""
|
||||
if ".data" in body_str:
|
||||
return True, "P1_drop_errors_via_dot_data"
|
||||
if ".ok" in body_str and "pass" in body_str:
|
||||
return True, "P2_pass_in_except_block"
|
||||
return True, "P3_returns_result_unchanged"
|
||||
|
||||
|
||||
def find_callers(func_name: str, src_dir: str = "src") -> list[tuple[str, int]]:
|
||||
"""Find all in-site callers of `func_name` in src/."""
|
||||
import re
|
||||
callers = []
|
||||
pattern = re.compile(rf"\b{re.escape(func_name)}\(")
|
||||
for py_file in Path(src_dir).glob("*.py"):
|
||||
try:
|
||||
text = py_file.read_text(encoding="utf-8")
|
||||
except (OSError, UnicodeDecodeError):
|
||||
continue
|
||||
for m in pattern.finditer(text):
|
||||
line_no = text[:m.start()].count("\n") + 1
|
||||
callers.append((str(py_file), line_no))
|
||||
return callers
|
||||
|
||||
|
||||
def audit_directory(src_dir: str = "src") -> list[dict]:
|
||||
findings = []
|
||||
for py_file in Path(src_dir).glob("*.py"):
|
||||
try:
|
||||
source = py_file.read_text(encoding="utf-8")
|
||||
except (OSError, UnicodeDecodeError):
|
||||
continue
|
||||
try:
|
||||
tree = ast.parse(source)
|
||||
except SyntaxError as e:
|
||||
continue
|
||||
for node in ast.walk(tree):
|
||||
if not isinstance(node, ast.FunctionDef):
|
||||
continue
|
||||
is_wrapper, pattern = is_legacy_wrapper(node)
|
||||
if not is_wrapper:
|
||||
continue
|
||||
findings.append({
|
||||
"file": str(py_file),
|
||||
"line": node.lineno,
|
||||
"name": node.name,
|
||||
"pattern": pattern,
|
||||
})
|
||||
return findings
|
||||
|
||||
|
||||
def main():
|
||||
findings = audit_directory("src")
|
||||
print(f"Found {len(findings)} legacy wrappers in src/:")
|
||||
print()
|
||||
for f in findings:
|
||||
print(f" {f['file']}:{f['line']:>5} {f['name']:<40} [{f['pattern']}]")
|
||||
print()
|
||||
print("Caller scan:")
|
||||
for f in findings:
|
||||
callers = find_callers(f["name"])
|
||||
print(f" {f['name']} ({len(callers)} in-site callers):")
|
||||
for cfile, cline in callers:
|
||||
print(f" {cfile}:{cline}")
|
||||
sys.exit(0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user