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
This commit is contained in:
@@ -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
|
||||
|
||||
+7
-7
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user