refactor(app_controller): migrate 5 worker/event sites to Result (Phase 6 Groups 6.5+6.6 partial)
Migrates the 3 worker closures (compress, generate_send, md_only) and the 2 per-event handler sites (RAG search, symbol resolution) to proper Result[T] propagation with the telemetry-drain pattern. New helpers: - _report_worker_error(op_name, result): Pattern 4 drain - _rag_search_result(user_msg) -> Result[List[Dict]] - _symbol_resolution_result(user_msg, file_items) -> Result[str] New state: - self._worker_errors: List[Tuple[str, ErrorInfo]] (with lock) - self._last_request_errors: List[Tuple[str, ErrorInfo]] Audit: INTERNAL_SILENT_SWALLOW for src/app_controller.py: 22 -> 17.
This commit is contained in:
+107
-30
@@ -850,6 +850,18 @@ class AppController:
|
||||
# durable data plane for sub-track 4 GUI to display in the models panel;
|
||||
# stderr write IS the visible-but-incomplete drain (full drain = GUI).
|
||||
self._model_fetch_errors: Dict[str, ErrorInfo] = {}
|
||||
# --- Worker error telemetry (Phase 6: Group 6.5) ---
|
||||
# Append-only list of (op_name, ErrorInfo) per failed background worker.
|
||||
# The list IS the Pattern 4 telemetry drain (per error_handling.md:421):
|
||||
# in-process telemetry buffer that sub-track 4 forwards to GUI.
|
||||
self._worker_errors: List[Tuple[str, ErrorInfo]] = []
|
||||
self._worker_errors_lock: threading.Lock = threading.Lock()
|
||||
# --- Per-request error carry (Phase 6: Group 6.6) ---
|
||||
# Per-event errors accumulated during _handle_request_event (RAG,
|
||||
# symbol resolution, per-task GUI ops). Drained at end of handler
|
||||
# via stderr write + reset; sub-track 4 GUI surfaces via the request
|
||||
# completion hook.
|
||||
self._last_request_errors: List[Tuple[str, ErrorInfo]] = [] # (op_name, error)
|
||||
# --- Shared background pool + proactive warmup (startup_speedup_20260606) ---
|
||||
self._io_pool = make_io_pool()
|
||||
_install_sigint_exit_handler(self)
|
||||
@@ -3327,6 +3339,62 @@ class AppController:
|
||||
|
||||
#region: AI Settings
|
||||
|
||||
def _rag_search_result(self, user_msg: str) -> "Result[List[Dict[str, Any]]]":
|
||||
"""Per-event handler (Phase 6 Group 6.6): RAG search via the engine.
|
||||
Returns Result[List[Dict]]. On failure: any engine/SDK exception
|
||||
-> ErrorInfo(original=e). Caller (`_handle_request_event`) appends
|
||||
to `self._last_request_errors` for sub-track 4 GUI display."""
|
||||
if not (self.rag_engine and self.rag_config and self.rag_config.enabled):
|
||||
return Result(data=[])
|
||||
try:
|
||||
chunks = self.rag_engine.search(user_msg)
|
||||
return Result(data=list(chunks) if chunks else [])
|
||||
except (OSError, IOError, ValueError, TypeError, KeyError, AttributeError, RuntimeError) as e:
|
||||
return Result(data=[], errors=[ErrorInfo(
|
||||
kind=ErrorKind.NETWORK,
|
||||
message=str(e),
|
||||
source="app_controller._rag_search_result",
|
||||
original=e,
|
||||
)])
|
||||
|
||||
def _symbol_resolution_result(self, user_msg: str, file_items: list) -> "Result[str]":
|
||||
"""Per-event handler (Phase 6 Group 6.6): symbol resolution via mcp tools.
|
||||
Returns Result[str] (the enriched user_msg). On failure: any mcp/SDK
|
||||
exception -> ErrorInfo(original=e). Caller appends to
|
||||
`self._last_request_errors` for sub-track 4 GUI display."""
|
||||
try:
|
||||
symbols = parse_symbols(user_msg)
|
||||
file_paths = [f['path'] for f in file_items]
|
||||
for symbol in symbols:
|
||||
res = get_symbol_definition(symbol, file_paths)
|
||||
if res:
|
||||
file_path, definition, line = res
|
||||
user_msg += f'\n\n[Definition: {symbol} from {file_path} (line {line})]\n```python\n{definition}\n```'
|
||||
return Result(data=user_msg)
|
||||
except (OSError, IOError, ValueError, TypeError, KeyError, AttributeError, RuntimeError) as e:
|
||||
return Result(data=user_msg, errors=[ErrorInfo(
|
||||
kind=ErrorKind.INTERNAL,
|
||||
message=str(e),
|
||||
source="app_controller._symbol_resolution_result",
|
||||
original=e,
|
||||
)])
|
||||
|
||||
def _report_worker_error(self, op_name: str, result: "Result[None]") -> None:
|
||||
"""Pattern 4 telemetry drain (Phase 6 Group 6.5): append (op_name, error)
|
||||
to self._worker_errors under a lock; stderr write IS the
|
||||
visible-but-incomplete drain (full drain = sub-track 4 GUI display).
|
||||
|
||||
Worker closures (background threads) call this from their except block
|
||||
after capturing the exception in Result. The lock prevents concurrent
|
||||
appends from corrupting the list."""
|
||||
if result.ok:
|
||||
return
|
||||
err = result.errors[0]
|
||||
with self._worker_errors_lock:
|
||||
self._worker_errors.append((op_name, err))
|
||||
sys.stderr.write(f"worker[{op_name}] error: {err.ui_message()}\n")
|
||||
sys.stderr.flush()
|
||||
|
||||
def _list_models_for_provider_result(self, p: str) -> "Result[List[str]]":
|
||||
"""SDK boundary (Phase 6 Group 6.4): wrap ai_client.list_models(p).
|
||||
Returns Result[List[str]]. On failure: any SDK exception
|
||||
@@ -3717,14 +3785,14 @@ class AppController:
|
||||
self.max_tokens = 8192
|
||||
|
||||
def _handle_compress_discussion(self) -> None:
|
||||
def worker():
|
||||
def worker() -> "Result[None]":
|
||||
try:
|
||||
self.ai_status = "compressing discussion..."
|
||||
disc_text = project_manager.format_discussion(self.disc_entries)
|
||||
if not disc_text.strip():
|
||||
self.ai_status = "discussion is empty"
|
||||
return
|
||||
|
||||
return OK
|
||||
|
||||
response_text = ai_client.run_discussion_compression(disc_text)
|
||||
|
||||
if response_text and not response_text.startswith("ERROR:"):
|
||||
@@ -3734,9 +3802,14 @@ class AppController:
|
||||
self.ai_status = "compression complete"
|
||||
else:
|
||||
self.ai_status = f"compression failed: {response_text}"
|
||||
return OK
|
||||
except (OSError, IOError, ValueError, TypeError, KeyError, AttributeError, RuntimeError) as e:
|
||||
logging.getLogger(__name__).debug("discussion compression error: %s", e, extra={"source": "app_controller._handle_compress_discussion.worker"})
|
||||
err = ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e),
|
||||
source="app_controller._handle_compress_discussion.worker", original=e)
|
||||
self.ai_status = f"compression error: {e}"
|
||||
result = Result(data=None, errors=[err])
|
||||
self._report_worker_error("compress_discussion", result)
|
||||
return result
|
||||
self.submit_io(worker)
|
||||
|
||||
def _handle_generate_send(self) -> None:
|
||||
@@ -3748,7 +3821,7 @@ class AppController:
|
||||
self.ai_status = "project switch in progress; AI ops disabled"
|
||||
return
|
||||
|
||||
def worker():
|
||||
def worker() -> "Result[None]":
|
||||
"""
|
||||
[C: tests/test_symbol_parsing.py:test_handle_generate_send_appends_definitions, tests/test_symbol_parsing.py:test_handle_generate_send_no_symbols]
|
||||
"""
|
||||
@@ -3772,9 +3845,14 @@ class AppController:
|
||||
)
|
||||
# Push to async queue
|
||||
self.event_queue.put("user_request", event_payload)
|
||||
return OK
|
||||
except (OSError, IOError, ValueError, TypeError, KeyError, AttributeError, RuntimeError) as e:
|
||||
logging.getLogger(__name__).debug("generate send error: %s", e, extra={"source": "app_controller._handle_generate_send.worker"})
|
||||
err = ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e),
|
||||
source="app_controller._handle_generate_send.worker", original=e)
|
||||
self.ai_status = f"generate error: {e}"
|
||||
result = Result(data=None, errors=[err])
|
||||
self._report_worker_error("generate_send", result)
|
||||
return result
|
||||
self.submit_io(worker)
|
||||
|
||||
def _do_generate(self) -> tuple[str, Path, list[dict[str, Any]], str, str]:
|
||||
@@ -3833,7 +3911,7 @@ class AppController:
|
||||
self.ai_status = "project switch in progress; MD generation disabled"
|
||||
return
|
||||
|
||||
def worker():
|
||||
def worker() -> "Result[None]":
|
||||
"""
|
||||
[C: tests/test_symbol_parsing.py:test_handle_generate_send_appends_definitions, tests/test_symbol_parsing.py:test_handle_generate_send_no_symbols]
|
||||
"""
|
||||
@@ -3844,9 +3922,14 @@ class AppController:
|
||||
self.ai_status = f"md written: {path.name}"
|
||||
# Refresh token budget metrics with CURRENT md
|
||||
self._refresh_api_metrics({}, md_content=md)
|
||||
return OK
|
||||
except (OSError, IOError, ValueError, TypeError, KeyError, AttributeError, RuntimeError) as e:
|
||||
logging.getLogger(__name__).debug("MD only error: %s", e, extra={"source": "app_controller._handle_md_only.worker"})
|
||||
err = ErrorInfo(kind=ErrorKind.INTERNAL, message=str(e),
|
||||
source="app_controller._handle_md_only.worker", original=e)
|
||||
self.ai_status = f"error: {e}"
|
||||
result = Result(data=None, errors=[err])
|
||||
self._report_worker_error("md_only", result)
|
||||
return result
|
||||
self.submit_io(worker)
|
||||
|
||||
def _create_discussion(self, name: str) -> None:
|
||||
@@ -3930,31 +4013,25 @@ class AppController:
|
||||
|
||||
# 1. RAG Retrieval (Enrich prompt before logging to history)
|
||||
if self.rag_engine and self.rag_config and self.rag_config.enabled:
|
||||
try:
|
||||
chunks = self.rag_engine.search(user_msg)
|
||||
if chunks:
|
||||
context_block = "## Retrieved Context\n\n"
|
||||
for i, chunk in enumerate(chunks):
|
||||
path = chunk.get("metadata", {}).get("path", "unknown")
|
||||
context_block += f"### Chunk {i+1} (Source: {path})\n{chunk.get('document', '')}\n\n"
|
||||
user_msg = context_block + user_msg
|
||||
except (OSError, IOError, ValueError, TypeError, KeyError, AttributeError, RuntimeError) as e:
|
||||
logging.getLogger(__name__).debug("RAG search error: %s", e, extra={"source": "app_controller._handle_request_event.rag"})
|
||||
sys.stderr.write(f"RAG search error: {e}\n")
|
||||
rag_result = self._rag_search_result(user_msg)
|
||||
if rag_result.ok and rag_result.data:
|
||||
context_block = "## Retrieved Context\n\n"
|
||||
for i, chunk in enumerate(rag_result.data):
|
||||
path = chunk.get("metadata", {}).get("path", "unknown")
|
||||
context_block += f"### Chunk {i+1} (Source: {path})\n{chunk.get('document', '')}\n\n"
|
||||
user_msg = context_block + user_msg
|
||||
elif not rag_result.ok:
|
||||
self._last_request_errors.append(("rag_search", rag_result.errors[0]))
|
||||
sys.stderr.write(f"RAG search error: {rag_result.errors[0].ui_message()}\n")
|
||||
sys.stderr.flush()
|
||||
|
||||
# 2. Symbol Resolution (Enrich prompt before logging to history)
|
||||
try:
|
||||
symbols = parse_symbols(user_msg)
|
||||
file_paths = [f['path'] for f in event.file_items]
|
||||
for symbol in symbols:
|
||||
res = get_symbol_definition(symbol, file_paths)
|
||||
if res:
|
||||
file_path, definition, line = res
|
||||
user_msg += f'\n\n[Definition: {symbol} from {file_path} (line {line})]\n```python\n{definition}\n```'
|
||||
except (OSError, IOError, ValueError, TypeError, KeyError, AttributeError, RuntimeError) as e:
|
||||
logging.getLogger(__name__).debug("Symbol resolution error: %s", e, extra={"source": "app_controller._handle_request_event.symbols"})
|
||||
sys.stderr.write(f"Symbol resolution error: {e}\n")
|
||||
sym_result = self._symbol_resolution_result(user_msg, event.file_items)
|
||||
if sym_result.ok and sym_result.data != user_msg:
|
||||
user_msg = sym_result.data
|
||||
elif not sym_result.ok:
|
||||
self._last_request_errors.append(("symbol_resolution", sym_result.errors[0]))
|
||||
sys.stderr.write(f"Symbol resolution error: {sym_result.errors[0].ui_message()}\n")
|
||||
sys.stderr.flush()
|
||||
|
||||
# 3. Log the final enriched prompt to history
|
||||
|
||||
Reference in New Issue
Block a user