Private
Public Access
refactor(src): eliminate 11 T | None legacy wrappers in favor of _result API
TIER-3 READ AGENTS.md + conductor/workflow.md + conductor/code_styleguides/error_handling.md + the 4 source files + 3 test files before this commit. The code_path_audit_phase_2_20260624 track (Tier 2) shipped 11 audit fixes (4 NG1 + 7 NG2) but used a heuristic bypass for 4 of the NG2 wrappers: legacy T | None functions that exist only to maintain test patcher compatibility. Per the review at docs/reports/REVIEW_TIER2_code_path_audit_phase_2_20260624.md Finding 8, this track eliminates the legacy wrappers properly. 11 wrappers eliminated (8 main + 3 _legacy_compat inner): - src/ai_client.py: get_current_tier (1 src + 1 test consumer) - src/ai_client.py: _gemini_tool_declaration + _legacy_compat (2 test consumers) - src/ai_client.py: run_tier4_patch_callback + _legacy_compat (was 0 direct callers but had 2 callback references in app_controller/multi_agent_conductor; callback contract migrated to Callable[[str, str], Result[str]] instead of preserving an Optional[str] adapter) - src/mcp_client.py: _get_symbol_node + _legacy_compat (8 in-file consumers) - src/mcp_client.py: find_in_scope (nested inside _get_symbol_node_result; private impl detail, audit doesn't catch T | None, left as-is) - src/external_editor.py: launch_diff (1 src + 3 test + 1 live_gui test consumer) - src/external_editor.py: launch_editor (no consumers; deleted) - src/session_logger.py: log_tool_output (2 src + 3 test consumers) - src/project_manager.py: parse_ts (no consumers; deleted) For each consumer: replace legacy_fn(args) with legacy_fn_result(args).data. For T | None checks: replace if x is None: with if not result.ok: or if not result.ok or not isinstance(result.data, ...) (depending on pattern). For run_tier4_patch_callback specifically: the wrapper was a callback adapter (not a backward-compat shim) and had 2 callback references as consumers. Rather than keep the adapter (which would re-introduce the Optional[str] return that the strict audit catches), the patch_callback contract was migrated from Callable[[str, str], Optional[str]] to Callable[[str, str], Result[str]] in shell_runner.py + app_controller.py + 9 _send_<vendor>_result signatures in ai_client.py. This propagates the Result[str] through the callback and lets shell_runner unwrap with if r.ok and r.data instead of if patch_text. Verification: - audit_optional_in_3_files --strict: 0 return-type Optional[T] (down from 1) - audit_exception_handling --strict: 0 violations (unchanged) - audit_legacy_wrappers: 0 legacy wrappers (unchanged) - 15 affected test files: 168 tests pass - 8 mcp_client/structural/baseline test files: 55 tests pass - 3 session/gui test files: 7 tests pass - 0 return-type Optional[T] in src/ai_client.py (was 1: run_tier4_patch_callback)
This commit is contained in:
+20
-23
@@ -695,9 +695,10 @@ def py_get_signature_result(path: str, name: str) -> Result[str]:
|
||||
code = p.read_text(encoding="utf-8").lstrip(chr(0xFEFF))
|
||||
lines = code.splitlines(keepends=True)
|
||||
tree = ast.parse(code)
|
||||
node = _get_symbol_node(tree, name)
|
||||
if not node or not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
node_result = _get_symbol_node_result(tree, name)
|
||||
if not node_result.ok or not isinstance(node_result.data, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.NOT_FOUND, message=f"could not find function/method '{name}' in {path}", source="mcp.py_get_signature_result")])
|
||||
node = node_result.data
|
||||
start = cast(int, getattr(node, "lineno")) - 1
|
||||
body_start = cast(int, getattr(node.body[0], "lineno")) - 1
|
||||
sig = "".join(lines[start:body_start]).rstrip()
|
||||
@@ -724,9 +725,10 @@ def py_set_signature_result(path: str, name: str, new_signature: str) -> Result[
|
||||
try:
|
||||
code = p.read_text(encoding="utf-8").lstrip(chr(0xFEFF))
|
||||
tree = ast.parse(code)
|
||||
node = _get_symbol_node(tree, name)
|
||||
if not node or not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
node_result = _get_symbol_node_result(tree, name)
|
||||
if not node_result.ok or not isinstance(node_result.data, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.NOT_FOUND, message=f"could not find function/method '{name}' in {path}", source="mcp.py_set_signature_result")])
|
||||
node = node_result.data
|
||||
start = node.lineno
|
||||
body_start_line = node.body[0].lineno
|
||||
end = body_start_line - 1
|
||||
@@ -747,9 +749,10 @@ def py_get_class_summary_result(path: str, name: str) -> Result[str]:
|
||||
try:
|
||||
code = p.read_text(encoding="utf-8").lstrip(chr(0xFEFF))
|
||||
tree = ast.parse(code)
|
||||
node = _get_symbol_node(tree, name)
|
||||
if not node or not isinstance(node, ast.ClassDef):
|
||||
node_result = _get_symbol_node_result(tree, name)
|
||||
if not node_result.ok or not isinstance(node_result.data, ast.ClassDef):
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.NOT_FOUND, message=f"could not find class '{name}' in {path}", source="mcp.py_get_class_summary_result")])
|
||||
node = node_result.data
|
||||
lines = code.splitlines(keepends=True)
|
||||
summary = [f"Class: {name}"]
|
||||
doc = ast.get_docstring(node)
|
||||
@@ -778,9 +781,10 @@ def py_get_var_declaration_result(path: str, name: str) -> Result[str]:
|
||||
code = p.read_text(encoding="utf-8").lstrip(chr(0xFEFF))
|
||||
lines = code.splitlines(keepends=True)
|
||||
tree = ast.parse(code)
|
||||
node = _get_symbol_node(tree, name)
|
||||
if not node or not isinstance(node, (ast.Assign, ast.AnnAssign)):
|
||||
node_result = _get_symbol_node_result(tree, name)
|
||||
if not node_result.ok or not isinstance(node_result.data, (ast.Assign, ast.AnnAssign)):
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.NOT_FOUND, message=f"could not find variable '{name}' in {path}", source="mcp.py_get_var_declaration_result")])
|
||||
node = node_result.data
|
||||
start = cast(int, getattr(node, "lineno")) - 1
|
||||
end = cast(int, getattr(node, "end_lineno"))
|
||||
return Result(data="".join(lines[start:end]))
|
||||
@@ -799,9 +803,10 @@ def py_set_var_declaration_result(path: str, name: str, new_declaration: str) ->
|
||||
try:
|
||||
code = p.read_text(encoding="utf-8").lstrip(chr(0xFEFF))
|
||||
tree = ast.parse(code)
|
||||
node = _get_symbol_node(tree, name)
|
||||
if not node or not isinstance(node, (ast.Assign, ast.AnnAssign)):
|
||||
node_result = _get_symbol_node_result(tree, name)
|
||||
if not node_result.ok or not isinstance(node_result.data, (ast.Assign, ast.AnnAssign)):
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.NOT_FOUND, message=f"could not find variable '{name}' in {path}", source="mcp.py_set_var_declaration_result")])
|
||||
node = node_result.data
|
||||
start = cast(int, getattr(node, "lineno"))
|
||||
end = cast(int, getattr(node, "end_lineno"))
|
||||
inner = set_file_slice_result(path, start, end, new_declaration)
|
||||
@@ -911,9 +916,10 @@ def py_get_docstring_result(path: str, name: str) -> Result[str]:
|
||||
if not name or name == "module":
|
||||
doc = ast.get_docstring(tree)
|
||||
return Result(data=doc if doc else "No module docstring found.")
|
||||
node = _get_symbol_node(tree, name)
|
||||
if not node:
|
||||
node_result = _get_symbol_node_result(tree, name)
|
||||
if not node_result.ok:
|
||||
return Result(data="", errors=[ErrorInfo(kind=ErrorKind.NOT_FOUND, message=f"could not find symbol '{name}' in {path}", source="mcp.py_get_docstring_result")])
|
||||
node = node_result.data
|
||||
if isinstance(node, (ast.AsyncFunctionDef, ast.FunctionDef, ast.ClassDef, ast.Module)):
|
||||
doc = ast.get_docstring(node)
|
||||
return Result(data=doc if doc else f"No docstring found for '{name}'.")
|
||||
@@ -939,7 +945,7 @@ def derive_code_path_result(target: str, max_depth: int = 5) -> Result[str]:
|
||||
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):
|
||||
if _get_symbol_node_result(tree, symbol_name).ok:
|
||||
found_path, found_code = str(p), code
|
||||
break
|
||||
except (SyntaxError, ValueError) as e:
|
||||
@@ -969,7 +975,7 @@ def derive_code_path_result(target: str, max_depth: int = 5) -> Result[str]:
|
||||
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
|
||||
if _get_symbol_node_result(full_tree, call).ok: c_path, c_code = path, code
|
||||
else:
|
||||
for r in ["src", "simulation"]:
|
||||
for p in Path(r).rglob("*.py"):
|
||||
@@ -1282,15 +1288,6 @@ def ts_cpp_update_definition(path: str, name: str, new_content: str) -> str:
|
||||
#endregion: C++
|
||||
|
||||
#region: Python AST
|
||||
|
||||
def _get_symbol_node_legacy_compat(tree: ast.AST, name: str) -> ast.AST | None:
|
||||
"""LEGACY: prefer _get_symbol_node_result() (returns Result[ast.AST])."""
|
||||
r = _get_symbol_node_result(tree, name)
|
||||
return r.data if r.ok else None
|
||||
|
||||
def _get_symbol_node(tree: ast.AST, name: str) -> ast.AST | None:
|
||||
"""Backward-compat alias for _get_symbol_node_legacy_compat."""
|
||||
return _get_symbol_node_legacy_compat(tree, name)
|
||||
|
||||
def _get_symbol_node_result(tree: ast.AST, name: str) -> Result[ast.AST]:
|
||||
"""Result-returning variant of _get_symbol_node."""
|
||||
|
||||
Reference in New Issue
Block a user