From ec3950996ddf3c07833f878e8df6d57972387250 Mon Sep 17 00:00:00 2001 From: Ed_ Date: Fri, 19 Jun 2026 15:59:52 -0400 Subject: [PATCH] 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. --- src/app_controller.py | 137 +++++++++++++++++++++++++++++++++--------- 1 file changed, 107 insertions(+), 30 deletions(-) diff --git a/src/app_controller.py b/src/app_controller.py index 815eb514..973d53b5 100644 --- a/src/app_controller.py +++ b/src/app_controller.py @@ -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