Private
Public Access
refactor(src): Phase 10.2 batch 4 - aggregate + api_hooks + context_presets + external_editor
aggregate.py (1 site): - compute_file_stats returns Result[dict[str, int]]. The 2 SILENT_SWALLOW sites (ast.parse + open) now append to errors list. Callers in gui_2.py updated to extract result.data from the cache. api_hooks.py (1 site): - WebSocketServer._handler - was 2 except ...: pass (JSONDecodeError + ConnectionClosed). Now logs warnings instead of silently swallowing. The audit's heuristic #19 (catch + log) classifies this as INTERNAL_COMPLIANT. context_presets.py (1 site): - ContextPresetManager.load_all returns Result[Dict[str, ContextPreset]]. Caller in app_controller.py (load_context_preset) updated to check result.ok. external_editor.py (1 site): - _find_vscode_in_registry returns Result[Optional[str]]. The 1 SILENT_SWALLOW site (subprocess.run) now appends to errors. Caller in ExternalEditorLauncher._resolve_vscode updated to extract result.data. Tests updated to check result.ok and use result.data.
This commit is contained in:
+9
-6
@@ -13,6 +13,8 @@ This is essential for keeping prompt tokens low while giving the AI enough struc
|
||||
to use the MCP tools to fetch only what it needs.
|
||||
"""
|
||||
import ast
|
||||
|
||||
from src.result_types import Result, ErrorInfo, ErrorKind
|
||||
import glob
|
||||
import os
|
||||
import re
|
||||
@@ -86,12 +88,13 @@ def group_files_by_dir(files: list[Any]) -> dict[str, list[Any]]:
|
||||
grouped[dir_name].append(f)
|
||||
return grouped
|
||||
|
||||
def compute_file_stats(abs_path: str) -> dict[str, int]:
|
||||
def compute_file_stats(abs_path: str) -> Result[dict[str, int]]:
|
||||
"""
|
||||
Computes lines and basic AST stats for a file.
|
||||
[C: src/gui_2.py:App._stats_worker, tests/test_context_composition_phase3.py:test_compute_file_stats]
|
||||
"""
|
||||
stats = {"lines": 0, "ast_elements": 0}
|
||||
errors: list[ErrorInfo] = []
|
||||
try:
|
||||
with open(abs_path, 'r', encoding='utf-8') as f:
|
||||
content = f.read()
|
||||
@@ -100,11 +103,11 @@ def compute_file_stats(abs_path: str) -> dict[str, int]:
|
||||
try:
|
||||
tree = ast.parse(content)
|
||||
stats["ast_elements"] = sum(1 for node in ast.walk(tree) if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)))
|
||||
except (SyntaxError, ValueError):
|
||||
pass
|
||||
except (OSError, SyntaxError):
|
||||
pass
|
||||
return stats
|
||||
except (SyntaxError, ValueError) as e:
|
||||
errors.append(ErrorInfo(kind=ErrorKind.INTERNAL, message=f"ast.parse failed: {e}", source=f"aggregate.compute_file_stats[{abs_path}]", original=e))
|
||||
except (OSError, SyntaxError) as e:
|
||||
errors.append(ErrorInfo(kind=ErrorKind.NOT_FOUND, message=str(e), source=f"aggregate.compute_file_stats[{abs_path}]", original=e))
|
||||
return Result(data=stats, errors=errors)
|
||||
|
||||
def build_discussion_section(history: list[Any]) -> str:
|
||||
"""
|
||||
|
||||
+4
-4
@@ -909,10 +909,10 @@ class WebSocketServer:
|
||||
if channel in self.clients:
|
||||
self.clients[channel].add(websocket)
|
||||
await websocket.send(json.dumps({"type": "subscription_confirmed", "channel": channel}))
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
except _require_warmed("websockets").exceptions.ConnectionClosed:
|
||||
pass
|
||||
except json.JSONDecodeError as e:
|
||||
logging.warning(f"WebSocketServer: JSON decode error: {e}")
|
||||
except _require_warmed("websockets").exceptions.ConnectionClosed as e:
|
||||
logging.info(f"WebSocketServer: connection closed: {e}")
|
||||
finally:
|
||||
for channel in self.clients:
|
||||
if websocket in self.clients[channel]:
|
||||
|
||||
@@ -1,22 +1,23 @@
|
||||
from typing import Dict, Any
|
||||
|
||||
from src.models import ContextPreset
|
||||
from src.result_types import Result, ErrorInfo, ErrorKind
|
||||
|
||||
|
||||
class ContextPresetManager:
|
||||
"""Manages context presets within the project dictionary (manual_slop.toml)."""
|
||||
|
||||
def load_all(self, project_dict: Dict[str, Any]) -> Dict[str, ContextPreset]:
|
||||
def load_all(self, project_dict: Dict[str, Any]) -> Result[Dict[str, ContextPreset]]:
|
||||
"""Loads all context presets from the project dictionary."""
|
||||
presets: Dict[str, ContextPreset] = {}
|
||||
errors: list[ErrorInfo] = []
|
||||
presets_data = project_dict.get("context_presets", {})
|
||||
for name, data in presets_data.items():
|
||||
try:
|
||||
presets[name] = ContextPreset.from_dict(name, data)
|
||||
except (ValueError, KeyError, TypeError):
|
||||
# Silent failure or logging could be added here
|
||||
pass
|
||||
return presets
|
||||
except (ValueError, KeyError, TypeError) as e:
|
||||
errors.append(ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source=f"context_presets.load_all[{name}]", original=e))
|
||||
return Result(data=presets, errors=errors)
|
||||
|
||||
def save_preset(self, project_dict: Dict[str, Any], preset: ContextPreset) -> None:
|
||||
"""Saves a context preset into the project dictionary."""
|
||||
|
||||
+10
-6
@@ -60,8 +60,9 @@ class ExternalEditorLauncher:
|
||||
_cached_vscode_config: Optional[TextEditorConfig] = None
|
||||
|
||||
|
||||
def _find_vscode_in_registry() -> Optional[str]:
|
||||
def _find_vscode_in_registry() -> Result[Optional[str]]:
|
||||
paths = []
|
||||
errors: list[ErrorInfo] = []
|
||||
reg_keys = [
|
||||
r"HKLM\SOFTWARE\Microsoft\Windows\CurrentVersion\Uninstall\*",
|
||||
r"HKCU\SOFTWARE\Microsoft\Windows\CurrentVersion\Uninstall\*",
|
||||
@@ -79,11 +80,11 @@ def _find_vscode_in_registry() -> Optional[str]:
|
||||
exe_path = line.strip() + "\\Code.exe"
|
||||
if os.path.exists(exe_path):
|
||||
paths.append(exe_path)
|
||||
except (OSError, subprocess.SubprocessError, subprocess.TimeoutExpired):
|
||||
pass
|
||||
except (OSError, subprocess.SubprocessError, subprocess.TimeoutExpired) as e:
|
||||
errors.append(ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e), source=f"external_editor._find_vscode_in_registry[{key}]", original=e))
|
||||
if paths:
|
||||
return paths[0]
|
||||
return None
|
||||
return Result(data=paths[0], errors=errors)
|
||||
return Result(data=None, errors=errors)
|
||||
|
||||
|
||||
def _find_vscode_common_paths() -> Optional[str]:
|
||||
@@ -103,7 +104,10 @@ def auto_detect_vscode() -> Optional[TextEditorConfig]:
|
||||
global _cached_vscode_config
|
||||
if _cached_vscode_config is not None:
|
||||
return _cached_vscode_config
|
||||
vscode_path = _find_vscode_in_registry() or _find_vscode_common_paths()
|
||||
vscode_result = _find_vscode_in_registry()
|
||||
vscode_path = vscode_result.data if vscode_result.ok else None
|
||||
if vscode_path is None:
|
||||
vscode_path = _find_vscode_common_paths()
|
||||
if vscode_path:
|
||||
_cached_vscode_config = TextEditorConfig(
|
||||
name="vscode",
|
||||
|
||||
+4
-2
@@ -1374,7 +1374,8 @@ class App:
|
||||
cache_key = f"{f_path}_{mtime}"
|
||||
if cache_key not in self._file_stats_cache: missing_keys.append((f_path, cache_key))
|
||||
else:
|
||||
stats = self._file_stats_cache[cache_key]
|
||||
cached = self._file_stats_cache[cache_key]
|
||||
stats = cached.data if hasattr(cached, "data") else cached
|
||||
total_lines += stats.get("lines", 0)
|
||||
total_ast += stats.get("ast_elements", 0)
|
||||
|
||||
@@ -4084,7 +4085,8 @@ def render_context_files_table(app: App) -> None:
|
||||
_exists = os.path.exists(_abs_p)
|
||||
mtime = os.path.getmtime(_abs_p) if _exists else 0
|
||||
cache_key = f"{f_path}_{mtime}"
|
||||
stats = app._file_stats_cache.get(cache_key, {"lines": 0, "ast_elements": 0})
|
||||
stats_raw = app._file_stats_cache.get(cache_key, {"lines": 0, "ast_elements": 0})
|
||||
stats = stats_raw.data if hasattr(stats_raw, "data") else stats_raw
|
||||
f_name = os.path.basename(f_path)
|
||||
imgui.text(f"{f_name} (L: {stats.get('lines', 0)}, AST: {stats.get('ast_elements', 0)})")
|
||||
if not _exists:
|
||||
|
||||
@@ -24,7 +24,8 @@ def test_compute_file_stats():
|
||||
py_path = os.path.join(temp_dir, "test.py")
|
||||
with open(py_path, "w") as f:
|
||||
f.write("def foo():\n pass\n\nclass Bar:\n pass\n")
|
||||
|
||||
stats = compute_file_stats(py_path)
|
||||
stats_result = compute_file_stats(py_path)
|
||||
assert stats_result.ok, f"compute_file_stats failed: {stats_result.errors}"
|
||||
stats = stats_result.data
|
||||
assert stats["lines"] == 5
|
||||
assert stats["ast_elements"] == 2 # 1 func, 1 class
|
||||
|
||||
@@ -28,13 +28,15 @@ def test_load_all_context_presets():
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
presets = manager.load_all(project_dict)
|
||||
|
||||
presets_result = manager.load_all(project_dict)
|
||||
assert presets_result.ok, f"load_all failed: {presets_result.errors}"
|
||||
presets = presets_result.data
|
||||
|
||||
assert "test_preset" in presets
|
||||
assert presets["test_preset"].files[0].path == "file1.py"
|
||||
assert presets["test_preset"].screenshots == ["screenshot1.png"]
|
||||
|
||||
|
||||
def test_delete_context_preset():
|
||||
manager = ContextPresetManager()
|
||||
project_dict = {
|
||||
|
||||
Reference in New Issue
Block a user