refactor(mcp_client): migrate L1358 derive_code_path to Result[T] (Phase 7 site 3)
Added derive_code_path_result inside Result Variants region. Legacy derive_code_path (str) now delegates to it. The nested trace function is now inside the _result variant; its inner try/except captures ErrorInfo correctly.
This commit is contained in:
+74
-58
@@ -936,6 +936,70 @@ def py_get_docstring_result(path: str, name: str) -> Result[str]:
|
||||
return Result(data=f"No docstring found for '{name}'.")
|
||||
except Exception as e:
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="mcp.py_get_docstring_result", original=e)])
|
||||
|
||||
def derive_code_path_result(target: str, max_depth: int = 5) -> Result[str]:
|
||||
from src.file_cache import ASTParser
|
||||
parser = ASTParser("python")
|
||||
found_path, found_code = None, None
|
||||
parts = target.split(".")
|
||||
symbol_name = parts[-1]
|
||||
if len(parts) > 1:
|
||||
possible_file = Path(*parts[:-1]).with_suffix(".py")
|
||||
if possible_file.exists(): found_path = str(possible_file)
|
||||
if not found_path:
|
||||
for root in ["src", "simulation"]:
|
||||
for p in Path(root).rglob("*.py"):
|
||||
if not _is_allowed(p): continue
|
||||
code = p.read_text(encoding="utf-8")
|
||||
if f"def {symbol_name}" in code or f"class {symbol_name}" in code:
|
||||
try:
|
||||
tree = ast.parse(code)
|
||||
if _get_symbol_node(tree, symbol_name):
|
||||
found_path, found_code = str(p), code
|
||||
break
|
||||
except Exception: continue
|
||||
if found_path: break
|
||||
if not found_path:
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.NOT_FOUND, message=f"could not find definition for '{target}'", source="mcp.derive_code_path_result")])
|
||||
if not found_code: found_code = Path(found_path).read_text(encoding="utf-8")
|
||||
visited, output = set(), [f"Code Path for: {target}", "=" * (11 + len(target)), ""]
|
||||
def trace(name, path, code, depth, indent):
|
||||
if depth > max_depth or (name, path) in visited: return
|
||||
visited.add((name, path))
|
||||
defn = parser.get_definition(code, name, path=path)
|
||||
if defn.startswith("ERROR:"):
|
||||
output.append(f"{indent}[!] {name} (Definition not found in {path})")
|
||||
return
|
||||
output.append(f"{indent}-> {name} ({path})")
|
||||
try:
|
||||
node = ast.parse(defn)
|
||||
calls = []
|
||||
for n in ast.walk(node):
|
||||
if isinstance(n, ast.Call):
|
||||
if isinstance(n.func, ast.Name): calls.append(n.func.id)
|
||||
elif isinstance(n.func, ast.Attribute): calls.append(n.func.attr)
|
||||
for call in sorted(set(calls)):
|
||||
if call in ("print", "len", "str", "int", "list", "dict", "set", "range", "enumerate", "isinstance", "getattr", "setattr", "hasattr"): continue
|
||||
c_path, c_code = None, None
|
||||
full_tree = ast.parse(code)
|
||||
if _get_symbol_node(full_tree, call): c_path, c_code = path, code
|
||||
else:
|
||||
for r in ["src", "simulation"]:
|
||||
for p in Path(r).rglob("*.py"):
|
||||
if not _is_allowed(p): continue
|
||||
f_code = p.read_text(encoding="utf-8")
|
||||
if f"def {call}" in f_code:
|
||||
c_path, c_code = str(p), f_code
|
||||
break
|
||||
if c_path: break
|
||||
if c_path: trace(call, c_path, c_code, depth + 1, indent + " ")
|
||||
except Exception as e:
|
||||
output.append(f"{indent} [!] Error parsing calls for {name}: {e}")
|
||||
try:
|
||||
trace(symbol_name, found_path, found_code, 0, "")
|
||||
return Result(data="\n".join(output))
|
||||
except Exception as e:
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source="mcp.derive_code_path_result", original=e)])
|
||||
#endregion: Result Variants
|
||||
|
||||
def edit_file(path: str, old_string: str, new_string: str, replace_all: bool = False) -> str:
|
||||
@@ -1351,64 +1415,16 @@ def py_get_docstring(path: str, name: str) -> str:
|
||||
return "; ".join(e.ui_message() for e in resolved.errors)
|
||||
|
||||
def derive_code_path(target: str, max_depth: int = 5) -> str:
|
||||
"""Recursively traces the execution path of a specific function or method."""
|
||||
from src.file_cache import ASTParser
|
||||
parser = ASTParser("python")
|
||||
found_path, found_code = None, None
|
||||
parts = target.split(".")
|
||||
symbol_name = parts[-1]
|
||||
if len(parts) > 1:
|
||||
possible_file = Path(*parts[:-1]).with_suffix(".py")
|
||||
if possible_file.exists(): found_path = str(possible_file)
|
||||
if not found_path:
|
||||
for root in ["src", "simulation"]:
|
||||
for p in Path(root).rglob("*.py"):
|
||||
if not _is_allowed(p): continue
|
||||
code = p.read_text(encoding="utf-8")
|
||||
if f"def {symbol_name}" in code or f"class {symbol_name}" in code:
|
||||
try:
|
||||
tree = ast.parse(code)
|
||||
if _get_symbol_node(tree, symbol_name):
|
||||
found_path, found_code = str(p), code
|
||||
break
|
||||
except Exception: continue
|
||||
if found_path: break
|
||||
if not found_path: return f"ERROR: could not find definition for '{target}'"
|
||||
if not found_code: found_code = Path(found_path).read_text(encoding="utf-8")
|
||||
visited, output = set(), [f"Code Path for: {target}", "=" * (11 + len(target)), ""]
|
||||
def trace(name, path, code, depth, indent):
|
||||
if depth > max_depth or (name, path) in visited: return
|
||||
visited.add((name, path))
|
||||
defn = parser.get_definition(code, name, path=path)
|
||||
if defn.startswith("ERROR:"):
|
||||
output.append(f"{indent}[!] {name} (Definition not found in {path})")
|
||||
return
|
||||
output.append(f"{indent}-> {name} ({path})")
|
||||
try:
|
||||
node = ast.parse(defn)
|
||||
calls = []
|
||||
for n in ast.walk(node):
|
||||
if isinstance(n, ast.Call):
|
||||
if isinstance(n.func, ast.Name): calls.append(n.func.id)
|
||||
elif isinstance(n.func, ast.Attribute): calls.append(n.func.attr)
|
||||
for call in sorted(set(calls)):
|
||||
if call in ("print", "len", "str", "int", "list", "dict", "set", "range", "enumerate", "isinstance", "getattr", "setattr", "hasattr"): continue
|
||||
c_path, c_code = None, None
|
||||
full_tree = ast.parse(code)
|
||||
if _get_symbol_node(full_tree, call): c_path, c_code = path, code
|
||||
else:
|
||||
for r in ["src", "simulation"]:
|
||||
for p in Path(r).rglob("*.py"):
|
||||
if not _is_allowed(p): continue
|
||||
f_code = p.read_text(encoding="utf-8")
|
||||
if f"def {call}" in f_code:
|
||||
c_path, c_code = str(p), f_code
|
||||
break
|
||||
if c_path: break
|
||||
if c_path: trace(call, c_path, c_code, depth + 1, indent + " ")
|
||||
except Exception as e: output.append(f"{indent} [!] Error parsing calls for {name}: {e}")
|
||||
trace(symbol_name, found_path, found_code, 0, "")
|
||||
return "\n".join(output)
|
||||
"""Recursively traces the execution path of a specific function or method.
|
||||
|
||||
Thin wrapper over derive_code_path_result; the legacy str shape is
|
||||
preserved for backward compatibility, but the try/except Exception
|
||||
lives in the Result variant.
|
||||
"""
|
||||
resolved = derive_code_path_result(target, max_depth)
|
||||
if resolved.ok:
|
||||
return resolved.data
|
||||
return "; ".join(e.ui_message() for e in resolved.errors)
|
||||
|
||||
#endregion Python AST
|
||||
|
||||
|
||||
Reference in New Issue
Block a user