From c329c86931ce229e2c61484819f95219da00e0b7 Mon Sep 17 00:00:00 2001 From: Ed_ Date: Wed, 17 Jun 2026 19:22:56 -0400 Subject: [PATCH] refactor(src): narrow exception types in Phase 8 MEDIUM files (10 sites across 2 files) Migrates the MEDIUM files (session_logger, warmup) by narrowing the exception types from broad 'except Exception' to specific stdlib exceptions. session_logger.py (8 sites): 1. L99 - registry.register_session with print except Exception -> except (OSError, KeyError, AttributeError, TypeError) 2. L131 - registry.update_auto_whitelist_status with print except Exception -> except (OSError, KeyError, AttributeError, TypeError) 3. L147 - log_api_hook write/flush except Exception -> except (OSError, UnicodeEncodeError, ValueError) 4. L160 - log_comms json.dump except Exception -> except (OSError, TypeError, ValueError) 5. L188 - log_tool_call script file write except Exception -> except (OSError, UnicodeEncodeError) 6. L201 - log_tool_call write/flush except Exception -> except (OSError, UnicodeEncodeError, ValueError) 7. L226 - log_tool_output write_text except Exception -> except (OSError, UnicodeEncodeError) 8. L245 - log_cli_call write/flush except Exception -> except (OSError, TypeError, ValueError) warmup.py (2 sites): 1. L276 - _log_canary sys.stderr.write except Exception -> except OSError 2. L300 - _log_summary sys.stderr.write except Exception -> except OSError Decisions: - warmup.py L85: raise RuntimeError (validation raise) - keep as-is per spec - warmup.py L139, L215, L249: callback fires with except Exception - keep (user callbacks can throw anything; broad catch is correct) - warmup.py L175: _warmup_one with except BaseException - keep (intentional broad catch for module import failures) Tests verified: - tests/test_session_logging.py (1 test) PASS - tests/test_session_logger_reset.py (1 test) PASS - tests/test_session_logger_optimization.py (4 tests) PASS - tests/test_logging_e2e.py (1 test) PASS - tests/test_warmup.py (10 tests) PASS - tests/test_warmup_canaries.py (18 tests) PASS --- src/session_logger.py | 16 ++++++++-------- src/warmup.py | 14 +++++++------- 2 files changed, 15 insertions(+), 15 deletions(-) diff --git a/src/session_logger.py b/src/session_logger.py index 8b9c9b22..9f8ffe5f 100644 --- a/src/session_logger.py +++ b/src/session_logger.py @@ -96,7 +96,7 @@ def open_session(label: Optional[str] = None) -> None: registry = LogRegistry(str(paths.get_logs_dir() / "log_registry.toml")) registry.register_session(_session_id, str(_session_dir), datetime.datetime.now()) - except Exception as e: + except (OSError, KeyError, AttributeError, TypeError) as e: print(f"Warning: Could not register session in LogRegistry: {e}") atexit.register(close_session) @@ -128,7 +128,7 @@ def close_session() -> None: registry = LogRegistry(str(paths.get_logs_dir() / "log_registry.toml")) registry.update_auto_whitelist_status(_session_id) - except Exception as e: + except (OSError, KeyError, AttributeError, TypeError) as e: print(f"Warning: Could not update auto-whitelist on close: {e}") def reset_session(label: Optional[str] = None) -> None: @@ -144,7 +144,7 @@ def log_api_hook(method: str, path: str, payload: str) -> None: try: _api_fh.write(f"[{ts_entry}] {method} {path} - Payload: {payload}\n") _api_fh.flush() - except Exception: + except (OSError, UnicodeEncodeError, ValueError): pass def log_comms(entry: dict[str, Any]) -> None: @@ -157,7 +157,7 @@ def log_comms(entry: dict[str, Any]) -> None: return try: _comms_fh.write(json.dumps(entry, ensure_ascii=False, default=str) + "\n") - except Exception: + except (OSError, TypeError, ValueError): pass def log_tool_call(script: str, result: str, script_path: Optional[str]) -> Optional[str]: @@ -185,7 +185,7 @@ def log_tool_call(script: str, result: str, script_path: Optional[str]) -> Optio try: if ps1_path: ps1_path.write_text(script, encoding="utf-8") - except Exception as exc: + except (OSError, UnicodeEncodeError) as exc: ps1_path = None ps1_name = f"(write error: {exc})" @@ -198,7 +198,7 @@ def log_tool_call(script: str, result: str, script_path: Optional[str]) -> Optio f"---\n\n" ) _tool_fh.flush() - except Exception: + except (OSError, UnicodeEncodeError, ValueError): pass return str(ps1_path) if ps1_path else None @@ -223,7 +223,7 @@ def log_tool_output(content: str) -> Optional[str]: try: out_path.write_text(content, encoding="utf-8") return str(out_path) - except Exception: + except (OSError, UnicodeEncodeError): return None def log_cli_call(command: str, stdin_content: Optional[str], stdout_content: Optional[str], stderr_content: Optional[str], latency: float) -> None: @@ -242,5 +242,5 @@ def log_cli_call(command: str, stdin_content: Optional[str], stdout_content: Opt } _cli_fh.write(json.dumps(log_data, ensure_ascii=False, default=str) + "\n") _cli_fh.flush() - except Exception: + except (OSError, TypeError, ValueError): pass diff --git a/src/warmup.py b/src/warmup.py index 08922de4..054e1212 100644 --- a/src/warmup.py +++ b/src/warmup.py @@ -273,7 +273,7 @@ class WarmupManager: try: sys.stderr.write(line) sys.stderr.flush() - except Exception: pass + except OSError: pass def _log_summary(self) -> None: if not self._log_to_stderr: return @@ -292,12 +292,12 @@ 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 Exception: pass + 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: pass def _snapshot(self) -> dict[str, list[str]]: return {