1def read_hook_input() -> dict:
 2    """
 3    從 stdin 讀取 Hook 輸入
 4
 5    Returns:
 6        dict: 解析後的 JSON 資料，解析失敗時返回空字典
 7    """
 8    try:
 9        return json.load(sys.stdin)
10    except json.JSONDecodeError:
11        return {}
12    except Exception:
13        return {}

1def process_hook():
2    input_data = read_hook_input()
3    if not input_data:
4        # 問題：不知道是 JSON 解析失敗還是 stdin 讀取失敗
5        # 無法給出有意義的錯誤訊息
6        return {"error": "unknown error"}

1def validate_hook_config(config: dict) -> None:
2    """驗證 Hook 配置"""
3    if "tool_name" not in config:
4        raise ValueError("missing tool_name")  # 第一個錯誤後就停止
5    if "tool_input" not in config:
6        raise ValueError("missing tool_input")  # 永遠不會執行到

1def handle_hook_error(error: Exception) -> dict:
2    """處理 Hook 錯誤"""
3    # 問題：只能用 isinstance 檢查，很難擴展
4    if isinstance(error, json.JSONDecodeError):
5        return {"error": "invalid json"}
6    elif isinstance(error, FileNotFoundError):
7        return {"error": "file not found"}
8    else:
9        return {"error": str(error)}

  1"""
  2Hook 異常階層設計
  3
  4Exception hierarchy:
  5    HookError (base)
  6    ├── HookConfigError (configuration issues)
  7    │   ├── ConfigNotFoundError
  8    │   ├── ConfigParseError
  9    │   └── ConfigValidationError
 10    ├── HookInputError (input processing)
 11    │   ├── InputReadError
 12    │   └── InputValidationError
 13    └── HookExecutionError (runtime issues)
 14        ├── ToolNotFoundError
 15        └── PermissionDeniedError
 16"""
 17
 18class HookError(Exception):
 19    """
 20    Hook 異常基礎類別
 21
 22    所有 Hook 相關的異常都繼承自這個類別，
 23    讓呼叫者可以用 `except HookError` 捕獲所有 Hook 錯誤。
 24    """
 25
 26    def __init__(self, message: str, *, context: dict | None = None):
 27        super().__init__(message)
 28        self.context = context or {}
 29
 30    def __str__(self) -> str:
 31        if self.context:
 32            ctx = ", ".join(f"{k}={v!r}" for k, v in self.context.items())
 33            return f"{self.args[0]} ({ctx})"
 34        return self.args[0]
 35
 36# === Configuration Errors ===
 37
 38class HookConfigError(HookError):
 39    """配置相關錯誤的基礎類別"""
 40    pass
 41
 42class ConfigNotFoundError(HookConfigError):
 43    """配置檔案不存在"""
 44
 45    def __init__(self, config_name: str, search_paths: list[str] | None = None):
 46        self.config_name = config_name
 47        self.search_paths = search_paths or []
 48        super().__init__(
 49            f"Config '{config_name}' not found",
 50            context={"config_name": config_name, "search_paths": self.search_paths}
 51        )
 52
 53class ConfigParseError(HookConfigError):
 54    """配置檔案解析失敗"""
 55
 56    def __init__(self, config_name: str, line: int | None = None, detail: str = ""):
 57        self.config_name = config_name
 58        self.line = line
 59        self.detail = detail
 60        msg = f"Failed to parse config '{config_name}'"
 61        if line:
 62            msg += f" at line {line}"
 63        if detail:
 64            msg += f": {detail}"
 65        super().__init__(msg, context={"config_name": config_name, "line": line})
 66
 67class ConfigValidationError(HookConfigError):
 68    """配置內容驗證失敗"""
 69
 70    def __init__(self, config_name: str, field: str, reason: str):
 71        self.config_name = config_name
 72        self.field = field
 73        self.reason = reason
 74        super().__init__(
 75            f"Invalid config '{config_name}': field '{field}' {reason}",
 76            context={"config_name": config_name, "field": field}
 77        )
 78
 79# === Input Errors ===
 80
 81class HookInputError(HookError):
 82    """輸入相關錯誤的基礎類別"""
 83    pass
 84
 85class InputReadError(HookInputError):
 86    """讀取輸入失敗"""
 87
 88    def __init__(self, source: str = "stdin", cause: Exception | None = None):
 89        self.source = source
 90        self.cause = cause
 91        super().__init__(
 92            f"Failed to read from {source}",
 93            context={"source": source}
 94        )
 95        if cause:
 96            self.__cause__ = cause
 97
 98class InputValidationError(HookInputError):
 99    """輸入驗證失敗"""
100
101    def __init__(self, field: str, expected: str, actual: str | None = None):
102        self.field = field
103        self.expected = expected
104        self.actual = actual
105        msg = f"Invalid input: field '{field}' {expected}"
106        if actual:
107            msg += f", got {actual!r}"
108        super().__init__(msg, context={"field": field, "expected": expected})
109
110# === Execution Errors ===
111
112class HookExecutionError(HookError):
113    """執行時錯誤的基礎類別"""
114    pass
115
116class ToolNotFoundError(HookExecutionError):
117    """工具不存在"""
118
119    def __init__(self, tool_name: str, available_tools: list[str] | None = None):
120        self.tool_name = tool_name
121        self.available_tools = available_tools or []
122        super().__init__(
123            f"Tool '{tool_name}' not found",
124            context={"tool_name": tool_name, "available": self.available_tools}
125        )
126
127class PermissionDeniedError(HookExecutionError):
128    """權限不足"""
129
130    def __init__(self, action: str, resource: str, reason: str = ""):
131        self.action = action
132        self.resource = resource
133        self.reason = reason
134        msg = f"Permission denied: cannot {action} '{resource}'"
135        if reason:
136            msg += f" ({reason})"
137        super().__init__(msg, context={"action": action, "resource": resource})

  1from dataclasses import dataclass, field
  2from datetime import datetime
  3from typing import Any
  4
  5@dataclass
  6class ErrorContext:
  7    """
  8    錯誤上下文資訊
  9
 10    收集錯誤發生時的環境資訊，
 11    幫助除錯和錯誤報告。
 12    """
 13    timestamp: datetime = field(default_factory=datetime.now)
 14    hook_name: str = ""
 15    tool_name: str = ""
 16    tool_input: dict = field(default_factory=dict)
 17    environment: dict = field(default_factory=dict)
 18    stack_info: str = ""
 19
 20    def to_dict(self) -> dict[str, Any]:
 21        """轉換為字典，方便序列化"""
 22        return {
 23            "timestamp": self.timestamp.isoformat(),
 24            "hook_name": self.hook_name,
 25            "tool_name": self.tool_name,
 26            "tool_input": self.tool_input,
 27            "environment": self.environment,
 28        }
 29
 30class RichHookError(HookError):
 31    """
 32    帶有豐富上下文的 Hook 異常
 33
 34    提供詳細的錯誤資訊，包含：
 35    - 錯誤發生的時間
 36    - 相關的 Hook 和工具資訊
 37    - 環境變數
 38    - 建議的修復方式
 39    """
 40
 41    def __init__(
 42        self,
 43        message: str,
 44        *,
 45        error_code: str = "",
 46        error_context: ErrorContext | None = None,
 47        suggestions: list[str] | None = None,
 48        recoverable: bool = False,
 49    ):
 50        super().__init__(message)
 51        self.error_code = error_code
 52        self.error_context = error_context or ErrorContext()
 53        self.suggestions = suggestions or []
 54        self.recoverable = recoverable
 55
 56    def format_message(self) -> str:
 57        """格式化完整錯誤訊息"""
 58        lines = [f"[{self.error_code}] {self.args[0]}"]
 59
 60        if self.error_context.hook_name:
 61            lines.append(f"  Hook: {self.error_context.hook_name}")
 62        if self.error_context.tool_name:
 63            lines.append(f"  Tool: {self.error_context.tool_name}")
 64
 65        if self.suggestions:
 66            lines.append("\nSuggestions:")
 67            for i, suggestion in enumerate(self.suggestions, 1):
 68                lines.append(f"  {i}. {suggestion}")
 69
 70        return "\n".join(lines)
 71
 72    def to_response(self) -> dict[str, Any]:
 73        """轉換為 Hook 回應格式"""
 74        return {
 75            "error": True,
 76            "error_code": self.error_code,
 77            "message": str(self.args[0]),
 78            "context": self.error_context.to_dict(),
 79            "suggestions": self.suggestions,
 80            "recoverable": self.recoverable,
 81        }
 82
 83# Example: specific error with rich context
 84class HookValidationError(RichHookError):
 85    """
 86    Hook 驗證錯誤
 87
 88    當 Hook 輸入驗證失敗時拋出，
 89    提供詳細的錯誤資訊和修復建議。
 90    """
 91
 92    def __init__(
 93        self,
 94        field: str,
 95        reason: str,
 96        *,
 97        actual_value: Any = None,
 98        error_context: ErrorContext | None = None,
 99    ):
100        self.field = field
101        self.reason = reason
102        self.actual_value = actual_value
103
104        message = f"Validation failed for '{field}': {reason}"
105        suggestions = self._generate_suggestions(field, reason, actual_value)
106
107        super().__init__(
108            message,
109            error_code="HOOK_VALIDATION_ERROR",
110            error_context=error_context,
111            suggestions=suggestions,
112            recoverable=True,
113        )
114
115    def _generate_suggestions(
116        self,
117        field: str,
118        reason: str,
119        actual_value: Any
120    ) -> list[str]:
121        """根據錯誤類型生成修復建議"""
122        suggestions = []
123
124        if "required" in reason.lower():
125            suggestions.append(f"Add the required field '{field}' to your input")
126        if "type" in reason.lower():
127            suggestions.append(f"Check the type of '{field}' - expected type may differ")
128        if actual_value is not None:
129            suggestions.append(f"Current value: {actual_value!r}")
130
131        return suggestions

  1from typing import Callable
  2
  3class ValidationCollector:
  4    """
  5    驗證錯誤收集器
  6
  7    收集多個驗證錯誤，最後用 ExceptionGroup 一次報告。
  8    這樣可以讓使用者一次看到所有錯誤，而不是修完一個才看到下一個。
  9    """
 10
 11    def __init__(self, context_name: str = "validation"):
 12        self.context_name = context_name
 13        self.errors: list[Exception] = []
 14
 15    def add_error(self, error: Exception) -> None:
 16        """新增一個錯誤"""
 17        self.errors.append(error)
 18
 19    def check(self, condition: bool, error: Exception) -> None:
 20        """
 21        檢查條件，失敗時收集錯誤
 22
 23        Args:
 24            condition: 要檢查的條件
 25            error: 條件為 False 時要收集的錯誤
 26        """
 27        if not condition:
 28            self.errors.append(error)
 29
 30    def has_errors(self) -> bool:
 31        """是否有錯誤"""
 32        return len(self.errors) > 0
 33
 34    def raise_if_errors(self) -> None:
 35        """
 36        如果有錯誤，拋出 ExceptionGroup
 37
 38        Raises:
 39            ExceptionGroup: 包含所有收集到的錯誤
 40        """
 41        if self.errors:
 42            raise ExceptionGroup(
 43                f"{self.context_name} failed with {len(self.errors)} error(s)",
 44                self.errors
 45            )
 46
 47def validate_hook_input(data: dict) -> dict:
 48    """
 49    驗證 Hook 輸入，收集所有錯誤
 50
 51    使用 ExceptionGroup 一次報告所有驗證錯誤，
 52    而不是只報告第一個錯誤。
 53
 54    Args:
 55        data: 待驗證的輸入資料
 56
 57    Returns:
 58        驗證後的資料
 59
 60    Raises:
 61        ExceptionGroup: 包含所有驗證錯誤
 62    """
 63    collector = ValidationCollector("Hook input validation")
 64
 65    # Check required fields
 66    collector.check(
 67        "tool_name" in data,
 68        InputValidationError("tool_name", "is required")
 69    )
 70    collector.check(
 71        "tool_input" in data,
 72        InputValidationError("tool_input", "is required")
 73    )
 74
 75    # Check types (only if fields exist)
 76    if "tool_name" in data:
 77        collector.check(
 78            isinstance(data["tool_name"], str),
 79            InputValidationError(
 80                "tool_name",
 81                "must be a string",
 82                actual=type(data["tool_name"]).__name__
 83            )
 84        )
 85
 86    if "tool_input" in data:
 87        collector.check(
 88            isinstance(data["tool_input"], dict),
 89            InputValidationError(
 90                "tool_input",
 91                "must be a dict",
 92                actual=type(data["tool_input"]).__name__
 93            )
 94        )
 95
 96    # Check tool-specific validation
 97    if data.get("tool_name") == "Write":
 98        collector.check(
 99            "file_path" in data.get("tool_input", {}),
100            InputValidationError("tool_input.file_path", "is required for Write tool")
101        )
102        collector.check(
103            "content" in data.get("tool_input", {}),
104            InputValidationError("tool_input.content", "is required for Write tool")
105        )
106
107    # Raise all errors at once
108    collector.raise_if_errors()
109
110    return data
111
112# Using except* to handle ExceptionGroup (Python 3.11+)
113def handle_validation_errors_demo():
114    """示範如何用 except* 處理 ExceptionGroup"""
115    try:
116        validate_hook_input({
117            "tool_name": 123,  # Wrong type
118            # Missing tool_input
119        })
120    except* InputValidationError as eg:
121        # Handle all InputValidationError instances
122        print(f"Found {len(eg.exceptions)} validation errors:")
123        for error in eg.exceptions:
124            print(f"  - {error.field}: {error.expected}")
125    except* HookError as eg:
126        # Handle other HookError instances
127        print(f"Found {len(eg.exceptions)} hook errors")

  1from abc import ABC, abstractmethod
  2from typing import TypeVar, Generic
  3
  4T = TypeVar("T")
  5
  6class RecoveryStrategy(ABC, Generic[T]):
  7    """
  8    錯誤恢復策略抽象基礎類別
  9
 10    定義錯誤恢復的介面，讓不同類型的錯誤
 11    可以有不同的恢復方式。
 12    """
 13
 14    @abstractmethod
 15    def can_recover(self, error: Exception) -> bool:
 16        """判斷是否可以恢復此錯誤"""
 17        pass
 18
 19    @abstractmethod
 20    def recover(self, error: Exception) -> T:
 21        """執行恢復邏輯，返回恢復後的結果"""
 22        pass
 23
 24class DefaultValueRecovery(RecoveryStrategy[dict]):
 25    """
 26    預設值恢復策略
 27
 28    當錯誤發生時，返回預設值。
 29    """
 30
 31    def __init__(self, default: dict):
 32        self.default = default
 33
 34    def can_recover(self, error: Exception) -> bool:
 35        return isinstance(error, (ConfigNotFoundError, InputReadError))
 36
 37    def recover(self, error: Exception) -> dict:
 38        return self.default.copy()
 39
 40class RetryRecovery(RecoveryStrategy[dict]):
 41    """
 42    重試恢復策略
 43
 44    當錯誤是暫時性的，嘗試重試。
 45    """
 46
 47    def __init__(
 48        self,
 49        operation: Callable[[], dict],
 50        max_retries: int = 3,
 51        recoverable_errors: tuple[type[Exception], ...] = (IOError,)
 52    ):
 53        self.operation = operation
 54        self.max_retries = max_retries
 55        self.recoverable_errors = recoverable_errors
 56
 57    def can_recover(self, error: Exception) -> bool:
 58        return isinstance(error, self.recoverable_errors)
 59
 60    def recover(self, error: Exception) -> dict:
 61        import time
 62
 63        last_error = error
 64        for attempt in range(self.max_retries):
 65            try:
 66                time.sleep(0.1 * (2 ** attempt))  # Exponential backoff
 67                return self.operation()
 68            except self.recoverable_errors as e:
 69                last_error = e
 70                continue
 71
 72        # All retries failed, re-raise the last error
 73        raise last_error
 74
 75class ErrorRecoveryChain:
 76    """
 77    錯誤恢復鏈
 78
 79    按順序嘗試多個恢復策略，
 80    直到找到可以處理的策略。
 81    """
 82
 83    def __init__(self):
 84        self.strategies: list[RecoveryStrategy] = []
 85
 86    def add_strategy(self, strategy: RecoveryStrategy) -> "ErrorRecoveryChain":
 87        """新增恢復策略"""
 88        self.strategies.append(strategy)
 89        return self
 90
 91    def try_recover(self, error: Exception) -> tuple[bool, Any]:
 92        """
 93        嘗試恢復錯誤
 94
 95        Args:
 96            error: 要恢復的錯誤
 97
 98        Returns:
 99            (是否成功恢復, 恢復結果或 None)
100        """
101        for strategy in self.strategies:
102            if strategy.can_recover(error):
103                try:
104                    result = strategy.recover(error)
105                    return (True, result)
106                except Exception:
107                    continue
108        return (False, None)
109
110def read_hook_input_with_recovery() -> dict:
111    """
112    讀取 Hook 輸入，帶有錯誤恢復機制
113
114    使用恢復鏈處理不同類型的錯誤。
115    """
116    import json
117    import sys
118
119    # Set up recovery chain
120    recovery = ErrorRecoveryChain()
121    recovery.add_strategy(DefaultValueRecovery({"tool_name": "", "tool_input": {}}))
122
123    try:
124        data = json.load(sys.stdin)
125        return validate_hook_input(data)
126
127    except json.JSONDecodeError as e:
128        # JSON parsing error - try to recover
129        error = InputReadError("stdin", cause=e)
130        recovered, result = recovery.try_recover(error)
131        if recovered:
132            return result
133        raise error from e
134
135    except ExceptionGroup as eg:
136        # Multiple validation errors
137        # Check if all errors are recoverable
138        all_recoverable = all(
139            isinstance(e, (InputValidationError,))
140            for e in eg.exceptions
141        )
142        if all_recoverable:
143            return {"tool_name": "", "tool_input": {}, "validation_errors": len(eg.exceptions)}
144        raise
145
146    except Exception as e:
147        # Unknown error
148        error = InputReadError("stdin", cause=e)
149        recovered, result = recovery.try_recover(error)
150        if recovered:
151            return result
152        raise error from e

  1#!/usr/bin/env python3
  2"""
  3Hook Exception Hierarchy - Complete Example
  4
  5Demonstrates how to design a clear exception hierarchy
  6and use ExceptionGroup for multiple error handling.
  7"""
  8
  9from __future__ import annotations
 10
 11import json
 12import sys
 13from abc import ABC, abstractmethod
 14from dataclasses import dataclass, field
 15from datetime import datetime
 16from typing import Any, Callable, Generic, TypeVar
 17
 18T = TypeVar("T")
 19
 20# ===== Exception Hierarchy =====
 21
 22class HookError(Exception):
 23    """Base class for all Hook-related exceptions."""
 24
 25    def __init__(self, message: str, *, context: dict | None = None):
 26        super().__init__(message)
 27        self.context = context or {}
 28
 29    def __str__(self) -> str:
 30        if self.context:
 31            ctx = ", ".join(f"{k}={v!r}" for k, v in self.context.items())
 32            return f"{self.args[0]} ({ctx})"
 33        return self.args[0]
 34
 35class HookConfigError(HookError):
 36    """Configuration-related errors."""
 37    pass
 38
 39class ConfigNotFoundError(HookConfigError):
 40    """Config file not found."""
 41
 42    def __init__(self, config_name: str, search_paths: list[str] | None = None):
 43        self.config_name = config_name
 44        self.search_paths = search_paths or []
 45        super().__init__(
 46            f"Config '{config_name}' not found",
 47            context={"config_name": config_name, "search_paths": self.search_paths}
 48        )
 49
 50class ConfigParseError(HookConfigError):
 51    """Failed to parse config file."""
 52
 53    def __init__(self, config_name: str, line: int | None = None, detail: str = ""):
 54        self.config_name = config_name
 55        self.line = line
 56        self.detail = detail
 57        msg = f"Failed to parse config '{config_name}'"
 58        if line:
 59            msg += f" at line {line}"
 60        if detail:
 61            msg += f": {detail}"
 62        super().__init__(msg, context={"config_name": config_name, "line": line})
 63
 64class HookInputError(HookError):
 65    """Input-related errors."""
 66    pass
 67
 68class InputReadError(HookInputError):
 69    """Failed to read input."""
 70
 71    def __init__(self, source: str = "stdin", cause: Exception | None = None):
 72        self.source = source
 73        self.cause = cause
 74        super().__init__(f"Failed to read from {source}", context={"source": source})
 75        if cause:
 76            self.__cause__ = cause
 77
 78class InputValidationError(HookInputError):
 79    """Input validation failed."""
 80
 81    def __init__(self, field: str, expected: str, actual: str | None = None):
 82        self.field = field
 83        self.expected = expected
 84        self.actual = actual
 85        msg = f"Invalid input: field '{field}' {expected}"
 86        if actual:
 87            msg += f", got {actual!r}"
 88        super().__init__(msg, context={"field": field, "expected": expected})
 89
 90class HookExecutionError(HookError):
 91    """Execution-related errors."""
 92    pass
 93
 94class ToolNotFoundError(HookExecutionError):
 95    """Tool not found."""
 96
 97    def __init__(self, tool_name: str, available_tools: list[str] | None = None):
 98        self.tool_name = tool_name
 99        self.available_tools = available_tools or []
100        super().__init__(
101            f"Tool '{tool_name}' not found",
102            context={"tool_name": tool_name, "available": self.available_tools}
103        )
104
105class PermissionDeniedError(HookExecutionError):
106    """Permission denied."""
107
108    def __init__(self, action: str, resource: str, reason: str = ""):
109        self.action = action
110        self.resource = resource
111        self.reason = reason
112        msg = f"Permission denied: cannot {action} '{resource}'"
113        if reason:
114            msg += f" ({reason})"
115        super().__init__(msg, context={"action": action, "resource": resource})
116
117# ===== Error Context =====
118
119@dataclass
120class ErrorContext:
121    """Rich context information for errors."""
122
123    timestamp: datetime = field(default_factory=datetime.now)
124    hook_name: str = ""
125    tool_name: str = ""
126    tool_input: dict = field(default_factory=dict)
127    environment: dict = field(default_factory=dict)
128
129    def to_dict(self) -> dict[str, Any]:
130        return {
131            "timestamp": self.timestamp.isoformat(),
132            "hook_name": self.hook_name,
133            "tool_name": self.tool_name,
134            "tool_input": self.tool_input,
135            "environment": self.environment,
136        }
137
138# ===== Validation Collector =====
139
140class ValidationCollector:
141    """Collects validation errors and raises ExceptionGroup."""
142
143    def __init__(self, context_name: str = "validation"):
144        self.context_name = context_name
145        self.errors: list[Exception] = []
146
147    def add_error(self, error: Exception) -> None:
148        self.errors.append(error)
149
150    def check(self, condition: bool, error: Exception) -> None:
151        if not condition:
152            self.errors.append(error)
153
154    def has_errors(self) -> bool:
155        return len(self.errors) > 0
156
157    def raise_if_errors(self) -> None:
158        if self.errors:
159            raise ExceptionGroup(
160                f"{self.context_name} failed with {len(self.errors)} error(s)",
161                self.errors
162            )
163
164# ===== Recovery Strategies =====
165
166class RecoveryStrategy(ABC, Generic[T]):
167    """Abstract base class for error recovery strategies."""
168
169    @abstractmethod
170    def can_recover(self, error: Exception) -> bool:
171        pass
172
173    @abstractmethod
174    def recover(self, error: Exception) -> T:
175        pass
176
177class DefaultValueRecovery(RecoveryStrategy[dict]):
178    """Returns a default value when error occurs."""
179
180    def __init__(self, default: dict):
181        self.default = default
182
183    def can_recover(self, error: Exception) -> bool:
184        return isinstance(error, (ConfigNotFoundError, InputReadError))
185
186    def recover(self, error: Exception) -> dict:
187        return self.default.copy()
188
189class ErrorRecoveryChain:
190    """Chain of recovery strategies."""
191
192    def __init__(self):
193        self.strategies: list[RecoveryStrategy] = []
194
195    def add_strategy(self, strategy: RecoveryStrategy) -> "ErrorRecoveryChain":
196        self.strategies.append(strategy)
197        return self
198
199    def try_recover(self, error: Exception) -> tuple[bool, Any]:
200        for strategy in self.strategies:
201            if strategy.can_recover(error):
202                try:
203                    result = strategy.recover(error)
204                    return (True, result)
205                except Exception:
206                    continue
207        return (False, None)
208
209# ===== Main Functions =====
210
211def validate_hook_input(data: dict) -> dict:
212    """
213    Validate hook input, collecting all errors.
214
215    Uses ExceptionGroup to report all validation errors at once.
216    """
217    collector = ValidationCollector("Hook input validation")
218
219    # Required fields
220    collector.check(
221        "tool_name" in data,
222        InputValidationError("tool_name", "is required")
223    )
224    collector.check(
225        "tool_input" in data,
226        InputValidationError("tool_input", "is required")
227    )
228
229    # Type checks
230    if "tool_name" in data:
231        collector.check(
232            isinstance(data["tool_name"], str),
233            InputValidationError(
234                "tool_name",
235                "must be a string",
236                actual=type(data["tool_name"]).__name__
237            )
238        )
239
240    if "tool_input" in data:
241        collector.check(
242            isinstance(data["tool_input"], dict),
243            InputValidationError(
244                "tool_input",
245                "must be a dict",
246                actual=type(data["tool_input"]).__name__
247            )
248        )
249
250    collector.raise_if_errors()
251    return data
252
253def read_hook_input_safe() -> dict:
254    """
255    Read hook input with error recovery.
256
257    Enhanced version of the original read_hook_input()
258    with proper exception handling.
259    """
260    recovery = ErrorRecoveryChain()
261    recovery.add_strategy(DefaultValueRecovery({"tool_name": "", "tool_input": {}}))
262
263    try:
264        data = json.load(sys.stdin)
265        return validate_hook_input(data)
266
267    except json.JSONDecodeError as e:
268        error = InputReadError("stdin", cause=e)
269        recovered, result = recovery.try_recover(error)
270        if recovered:
271            return result
272        raise error from e
273
274    except ExceptionGroup:
275        # Re-raise validation errors as-is
276        raise
277
278    except Exception as e:
279        error = InputReadError("stdin", cause=e)
280        recovered, result = recovery.try_recover(error)
281        if recovered:
282            return result
283        raise error from e
284
285def create_error_response(error: HookError) -> dict:
286    """Create a standardized error response."""
287    return {
288        "decision": "deny",
289        "reason": str(error),
290        "error": {
291            "type": type(error).__name__,
292            "message": str(error.args[0]),
293            "context": error.context,
294        }
295    }
296
297# ===== Demo =====
298
299if __name__ == "__main__":
300    print("=== Exception Hierarchy Demo ===\n")
301
302    # Demo 1: Basic exception
303    print("1. Basic exception:")
304    try:
305        raise ConfigNotFoundError("agents", ["/path/1", "/path/2"])
306    except HookConfigError as e:
307        print(f"   Caught: {e}")
308        print(f"   Context: {e.context}")
309
310    # Demo 2: Exception chaining
311    print("\n2. Exception chaining:")
312    try:
313        try:
314            raise json.JSONDecodeError("Unexpected EOF", "", 0)
315        except json.JSONDecodeError as e:
316            raise InputReadError("stdin", cause=e) from e
317    except InputReadError as e:
318        print(f"   Caught: {e}")
319        print(f"   Caused by: {e.__cause__}")
320
321    # Demo 3: ExceptionGroup
322    print("\n3. ExceptionGroup (multiple errors):")
323    try:
324        validate_hook_input({"tool_name": 123})  # Wrong type, missing tool_input
325    except ExceptionGroup as eg:
326        print(f"   Caught ExceptionGroup: {eg}")
327        for i, err in enumerate(eg.exceptions, 1):
328            print(f"   Error {i}: {err}")
329
330    # Demo 4: except* syntax (Python 3.11+)
331    print("\n4. Using except* to handle specific error types:")
332    try:
333        validate_hook_input({})  # Missing both fields
334    except* InputValidationError as eg:
335        print(f"   Caught {len(eg.exceptions)} InputValidationError(s)")
336        for err in eg.exceptions:
337            print(f"   - {err.field}: {err.expected}")
338
339    # Demo 5: Error recovery
340    print("\n5. Error recovery:")
341    recovery = ErrorRecoveryChain()
342    recovery.add_strategy(DefaultValueRecovery({"default": True}))
343
344    error = ConfigNotFoundError("missing_config")
345    recovered, result = recovery.try_recover(error)
346    print(f"   Recovered: {recovered}, Result: {result}")
347
348    print("\n=== Demo Complete ===")

 1from hook_exceptions import (
 2    HookError,
 3    ConfigNotFoundError,
 4    InputValidationError,
 5    create_error_response,
 6)
 7
 8def process_hook():
 9    """處理 Hook 請求"""
10    try:
11        # Load configuration
12        config = load_config("agents")
13
14        # Validate input
15        data = validate_hook_input(read_input())
16
17        # Process...
18        return {"decision": "allow"}
19
20    except ConfigNotFoundError as e:
21        # Specific handling for missing config
22        print(f"Warning: {e}, using defaults")
23        return {"decision": "allow", "warning": str(e)}
24
25    except InputValidationError as e:
26        # Input validation error
27        return create_error_response(e)
28
29    except HookError as e:
30        # Catch-all for other hook errors
31        return create_error_response(e)

 1def validate_batch_inputs(inputs: list[dict]) -> list[dict]:
 2    """
 3    驗證多個輸入，收集所有錯誤
 4
 5    即使某些輸入失敗，仍然處理其他輸入。
 6    """
 7    results = []
 8    all_errors = []
 9
10    for i, data in enumerate(inputs):
11        try:
12            validated = validate_hook_input(data)
13            results.append({"index": i, "data": validated, "valid": True})
14        except ExceptionGroup as eg:
15            # Collect errors from this input
16            all_errors.extend(eg.exceptions)
17            results.append({
18                "index": i,
19                "errors": [str(e) for e in eg.exceptions],
20                "valid": False
21            })
22        except HookError as e:
23            all_errors.append(e)
24            results.append({"index": i, "error": str(e), "valid": False})
25
26    # Report summary
27    if all_errors:
28        print(f"Validation completed with {len(all_errors)} total errors")
29
30    return results
31
32# Using except* to handle different error types differently
33def handle_mixed_errors():
34    """示範用 except* 分別處理不同類型的錯誤"""
35    try:
36        # This might raise ExceptionGroup with mixed errors
37        process_complex_operation()
38
39    except* InputValidationError as eg:
40        # Handle validation errors - maybe log and continue
41        for err in eg.exceptions:
42            log_validation_error(err)
43
44    except* PermissionDeniedError as eg:
45        # Handle permission errors - need to notify admin
46        for err in eg.exceptions:
47            notify_admin(err)
48
49    except* HookExecutionError as eg:
50        # Handle execution errors - maybe retry
51        for err in eg.exceptions:
52            schedule_retry(err)

 1def load_config_with_chain(name: str) -> dict:
 2    """
 3    載入配置，保留原始錯誤資訊
 4
 5    使用 `raise ... from ...` 保留異常鏈，
 6    讓除錯時可以看到完整的錯誤來源。
 7    """
 8    import yaml
 9
10    config_path = f"/config/{name}.yaml"
11
12    try:
13        with open(config_path) as f:
14            return yaml.safe_load(f)
15
16    except FileNotFoundError as e:
17        # Wrap in domain-specific exception, preserve original
18        raise ConfigNotFoundError(name, [config_path]) from e
19
20    except yaml.YAMLError as e:
21        # Extract line number if available
22        line = getattr(e, "problem_mark", None)
23        line_num = line.line if line else None
24        raise ConfigParseError(name, line=line_num, detail=str(e)) from e
25
26    except PermissionError as e:
27        # Convert to domain exception
28        raise PermissionDeniedError("read", config_path, str(e)) from e
29
30# When catching, you can access the full chain
31def debug_config_error():
32    try:
33        config = load_config_with_chain("agents")
34    except HookConfigError as e:
35        print(f"Error: {e}")
36        print(f"Original cause: {e.__cause__}")
37
38        # Print full traceback including cause
39        import traceback
40        traceback.print_exception(type(e), e, e.__traceback__)

 1# 目標 API
 2def validate_form(data: dict) -> dict:
 3    """驗證表單，收集所有錯誤"""
 4    collector = ValidationCollector("form")
 5
 6    collector.check(
 7        len(data.get("username", "")) >= 3,
 8        FieldError("username", "must be at least 3 characters")
 9    )
10    collector.check(
11        "@" in data.get("email", ""),
12        FieldError("email", "must contain @")
13    )
14
15    collector.raise_if_errors()
16    return data

 1# 目標 API
 2try:
 3    validate_config(config)
 4except ConfigError as e:
 5    print(e.format_message())
 6    # Output:
 7    # [CONFIG_001] Invalid config 'agents.yaml'
 8    #   Field: known_agents
 9    #   Error: expected list, got string
10    #
11    # Suggestions:
12    #   1. Change 'known_agents: "basil"' to 'known_agents: ["basil"]'
13    #   2. See: https://docs.example.com/config#known_agents