From 4c42bd05457183c2f2776d896780864a0bcf26d6 Mon Sep 17 00:00:00 2001 From: Ed_ Date: Thu, 18 Jun 2026 00:06:11 -0400 Subject: [PATCH] refactor(src): warmup.py Phase 11.3.1 - FULL Result[T] migration (5 sites) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase 11.3.1 (REJECT Phase 10's sliming). Per the user's explicit direction: every try/except site that can fail MUST return Result[T]. No 'user callback' excuse; the user callbacks in WarmupManager are Callable[[dict], None] and stay as-is. The MANAGER's INTERNAL methods return Result[T]. Changes: - on_complete() returns Result[bool]; fires callback via _fire_callback helper that captures user-callback exceptions as ErrorInfo. - _record_success() returns Result[bool]; aggregates per-callback errors. - _record_failure() returns Result[bool]; same pattern. - _log_canary() returns Result[None]; uses _log_stderr helper. - _log_summary() returns Result[None]; uses _log_stderr helper. - _warmup_one() (io_pool callback) returns Result[bool]; delegates to _record_success/_record_failure. - _log_stderr() (new helper) returns Result[None]; captures OSError. - _fire_callback() (new helper) returns Result[bool]; captures user-callback exceptions. Audit post-migration: - L319 (_log_stderr) = INTERNAL_COMPLIANT (Heuristic A) ✓ - L337 (_fire_callback) = INTERNAL_COMPLIANT (Heuristic A) ✓ - L185 (_warmup_one) = INTERNAL_BROAD_CATCH (known limitation: indirect return via 'return self._record_failure(...)' is not detected by Heuristic A which matches 'return Result(...)' directly) - L96 (submit raise RuntimeError) = INTERNAL_RETHROW (programmer error, not a runtime failure; acceptable) Tests: 16/16 pass (test_api_hooks_warmup.py, test_gui_warmup_indicator.py). Per conductor/tracks/result_migration_small_files_20260617/plan.md section 11.3.1. --- src/warmup.py | 130 ++++++++++++++++++++++++++++++++++---------------- 1 file changed, 90 insertions(+), 40 deletions(-) diff --git a/src/warmup.py b/src/warmup.py index 8fba3dc4..ef18004d 100644 --- a/src/warmup.py +++ b/src/warmup.py @@ -11,7 +11,7 @@ Public API on the manager (and exposed on AppController via delegation): mgr.status() - {pending, completed, failed} mgr.is_done() - bool mgr.wait(timeout) - block until done - mgr.on_complete(callback) - register completion callback + mgr.on_complete(callback) - register completion callback (returns Result[bool]) mgr.canaries() - list[dict] of per-module canary records (observability) mgr.reset() - clear state (for re-warmup, e.g. in tests) @@ -26,6 +26,15 @@ Canary records (one per submitted module) carry: elapsed_ms: (end_ts - start_ts) * 1000 status: "running" | "completed" | "failed" | "cancelled" error: error message string if status == "failed", else None + +Phase 11.3.1 (2026-06-17): FULL Result[T] migration. Every method that +can fail returns `Result[T]` with structured `ErrorInfo`. User callbacks +remain `Callable[[dict], None]` (the convention says external callbacks +cannot be Result-typed); the MANAGER wraps each user-callback fire and +returns `Result[bool]` indicating whether all callbacks succeeded. +io_pool completion handler threads the Result through. Reference +implementation: src/hot_reloader.py:reload()/reload_all() and +src/hot_reloader.py's io_pool wiring. """ import importlib @@ -35,6 +44,8 @@ import time from concurrent.futures import Future, ThreadPoolExecutor from typing import Callable, Optional +from src.result_types import ErrorInfo, ErrorKind, Result + CompletionCallback = Callable[[dict], None] @@ -125,19 +136,18 @@ class WarmupManager: def wait(self, timeout: Optional[float] = None) -> bool: return self._done_event.wait(timeout=timeout) - def on_complete(self, callback: CompletionCallback) -> None: + def on_complete(self, callback: CompletionCallback) -> Result[bool]: fire_now = False + snap: Optional[dict] = None with self._lock: if self._done_event.is_set(): fire_now = True snap = self._snapshot() else: self._callbacks.append(callback) - if fire_now: - try: - callback(snap) - except Exception as e: - sys.stderr.write(f"[WarmupManager] on_complete callback raised: {e}\n") + if fire_now and snap is not None: + return self._fire_callback(callback, snap, "on_complete") + return Result(data=True) def reset(self) -> None: with self._lock: @@ -157,7 +167,7 @@ class WarmupManager: if c.get("start_ts") and c["elapsed_ms"] is None: c["elapsed_ms"] = (c["end_ts"] - c["start_ts"]) * 1000 - def _warmup_one(self, name: str) -> None: + def _warmup_one(self, name: str) -> Result[bool]: start_ts = time.time() thread = threading.current_thread() thread_name = thread.name @@ -174,12 +184,11 @@ class WarmupManager: importlib.import_module(name) except BaseException as e: end_ts = time.time() - self._record_failure(name, e, end_ts) - else: - end_ts = time.time() - self._record_success(name, end_ts) + return self._record_failure(name, e, end_ts) + end_ts = time.time() + return self._record_success(name, end_ts) - def _record_success(self, name: str, end_ts: Optional[float] = None) -> None: + def _record_success(self, name: str, end_ts: Optional[float] = None) -> Result[bool]: if end_ts is None: end_ts = time.time() callbacks: list[CompletionCallback] = [] canary_snapshot: Optional[dict] = None @@ -209,15 +218,16 @@ class WarmupManager: self._log_canary(canary_snapshot) if all_done: self._log_summary() + cb_errors: list[ErrorInfo] = [] for cb in callbacks: - try: - cb(self._snapshot()) - except Exception as e: - sys.stderr.write(f"[WarmupManager] _record_success callback raised: {e}\n") + cb_result = self._fire_callback(cb, self._snapshot(), "_record_success") + if not cb_result.ok: + cb_errors.extend(cb_result.errors) if all_done: self._done_event.set() + return Result(data=len(cb_errors) == 0, errors=cb_errors) - def _record_failure(self, name: str, _err: BaseException, end_ts: Optional[float] = None) -> None: + def _record_failure(self, name: str, _err: BaseException, end_ts: Optional[float] = None) -> Result[bool]: if end_ts is None: end_ts = time.time() callbacks: list[CompletionCallback] = [] canary_snapshot: Optional[dict] = None @@ -243,16 +253,17 @@ class WarmupManager: self._log_canary(canary_snapshot) if all_done: self._log_summary() + cb_errors: list[ErrorInfo] = [] for cb in callbacks: - try: - cb(self._snapshot()) - except Exception as e: - sys.stderr.write(f"[WarmupManager] _record_failure callback raised: {e}\n") + cb_result = self._fire_callback(cb, self._snapshot(), "_record_failure") + if not cb_result.ok: + cb_errors.extend(cb_result.errors) if all_done: self._done_event.set() + return Result(data=len(cb_errors) == 0, errors=cb_errors) - def _log_canary(self, canary: dict) -> None: - if not self._log_to_stderr: return + def _log_canary(self, canary: dict) -> Result[None]: + if not self._log_to_stderr: return Result(data=None) cid = canary["canary_id"] module = canary["module"] thread_name = canary.get("thread_name") or "?" @@ -270,17 +281,13 @@ class WarmupManager: line = f"[warmup {cid}] FAILED {module} on {thread_name} (id={thread_id}): {err}{main_tag}\n" else: line = f"[warmup {cid}] {status.upper()} {module} on {thread_name} (id={thread_id}){main_tag}\n" - try: - sys.stderr.write(line) - sys.stderr.flush() - except OSError as e: - sys.stderr.write(f"[WarmupManager] canary log write failed: {e}\n") + return self._log_stderr(line, source="warmup._log_canary") - def _log_summary(self) -> None: - if not self._log_to_stderr: return + def _log_summary(self) -> Result[None]: + if not self._log_to_stderr: return Result(data=None) with self._lock: canaries = list(self._canaries) - if not canaries: return + if not canaries: return Result(data=None) total = len(canaries) completed = sum(1 for c in canaries if c["status"] == "completed") failed = sum(1 for c in canaries if c["status"] == "failed") @@ -293,17 +300,60 @@ class WarmupManager: if failed: parts.append(f"{failed} failed") if cancelled: parts.append(f"{cancelled} cancelled") with self._log_lock: - try: - sys.stderr.write(f"[warmup done] {total} modules: {', '.join(parts)} (sum of per-module elapsed: {total_ms:.1f}ms)\n") - if main_thread_violations: - sys.stderr.write(f"[warmup WARNING] {len(main_thread_violations)} module(s) loaded on the MAIN THREAD (violates main thread purity invariant): {', '.join(main_thread_violations)}\n") - sys.stderr.flush() - except OSError as e: - sys.stderr.write(f"[WarmupManager] summary log write failed: {e}\n") + main_line = "" + if main_thread_violations: + main_line = f"[warmup WARNING] {len(main_thread_violations)} module(s) loaded on the MAIN THREAD (violates main thread purity invariant): {', '.join(main_thread_violations)}\n" + summary_line = f"[warmup done] {total} modules: {', '.join(parts)} (sum of per-module elapsed: {total_ms:.1f}ms)\n" + r1 = self._log_stderr(summary_line, source="warmup._log_summary") + if main_line: + r2 = self._log_stderr(main_line, source="warmup._log_summary") + return r1 if not r1.ok else r1.with_errors(r2.errors) + return r1 + + def _log_stderr(self, line: str, source: str) -> Result[None]: + """Best-effort stderr write. Returns Result[None]; caller decides what to do.""" + try: + sys.stderr.write(line) + sys.stderr.flush() + return Result(data=None) + except OSError as e: + return Result(data=None, errors=[ErrorInfo( + kind=ErrorKind.INTERNAL, + message=f"stderr write failed: {e}", + source=source, + original=e, + )]) + + def _fire_callback(self, cb: CompletionCallback, snap: dict, source: str) -> Result[bool]: + """Fire a user callback and capture any exception as Result[bool]. + + The user callback signature is `Callable[[dict], None]` (per the public API). + If it raises, we convert to ErrorInfo and best-effort log to stderr; the + Result captures the failure so the manager can thread it. + """ + try: + cb(snap) + return Result(data=True) + except Exception as e: + err_msg = f"[WarmupManager] {source} callback raised: {e}" + log_result = self._log_stderr(err_msg + "\n", source=f"warmup.{source}") + if not log_result.ok: + return Result(data=False, errors=[ErrorInfo( + kind=ErrorKind.INTERNAL, + message=f"{source} callback raised: {e}; log also failed: {log_result.errors[0].message}", + source=f"warmup.{source}", + original=e, + )]) + return Result(data=False, errors=[ErrorInfo( + kind=ErrorKind.INTERNAL, + message=f"{source} callback raised: {e}", + source=f"warmup.{source}", + original=e, + )]) def _snapshot(self) -> dict[str, list[str]]: return { "pending": list(self._pending), "completed": list(self._completed), "failed": list(self._failed), - } + } \ No newline at end of file