Coverage for src / harnessutils / turn / hooks.py: 100%

1"""Hook definitions for turn processing. 

2 

3Hooks allow apps to inject custom behavior at key points 

4in the turn processing flow. 

5""" 

6 

7from collections.abc import Callable 

8from dataclasses import dataclass, field 

9from typing import Any 

10 

11 

12@dataclass 

13class TurnHooks: 

14 """Callbacks for turn processing events. 

15 

16 Apps can provide these callbacks to customize behavior during 

17 turn processing (e.g., LLM calls, tool execution, summarization). 

18 """ 

19 

20 on_llm_request: Callable[[dict[str, Any]], None] | None = None 

21 """Called before making LLM request with messages.""" 

22 

23 on_llm_response: Callable[[dict[str, Any]], None] | None = None 

24 """Called when LLM response is received.""" 

25 

26 on_tool_call: Callable[[str, str, dict[str, Any]], Any] | None = None 

27 """Called when tool should be executed. 

28 

29 Args: 

30 tool_name: Name of the tool to execute 

31 call_id: Unique call identifier 

32 input_data: Tool input parameters 

33 

34 Returns: 

35 Tool execution result (output string or dict) 

36 """ 

37 

38 on_tool_result: Callable[[str, Any], None] | None = None 

39 """Called when tool execution completes. 

40 

41 Args: 

42 call_id: Tool call identifier 

43 result: Tool execution result 

44 """ 

45 

46 on_tool_error: Callable[[str, Exception], None] | None = None 

47 """Called when tool execution fails. 

48 

49 Args: 

50 call_id: Tool call identifier 

51 error: Exception that occurred 

52 """ 

53 

54 on_summarization_request: Callable[[list[dict[str, Any]]], dict[str, Any]] | None = None 

55 """Called when summarization is needed. 

56 

57 Args: 

58 messages: Messages to summarize 

59 

60 Returns: 

61 Summarization result with content and usage 

62 """ 

63 

64 on_doom_loop: Callable[[str, dict[str, Any], int], bool] | None = None 

65 """Called when doom loop is detected. 

66 

67 Args: 

68 tool_name: Tool being called repeatedly 

69 input_data: Tool input 

70 count: Number of identical calls 

71 

72 Returns: 

73 True to continue, False to stop 

74 """ 

75 

76 on_snapshot: Callable[[str], str] | None = None 

77 """Called to capture state snapshot. 

78 

79 Args: 

80 event: Snapshot event type (start/finish) 

81 

82 Returns: 

83 Snapshot identifier 

84 """ 

85 

86 metadata: dict[str, Any] = field(default_factory=dict) 

87 """Custom metadata for hooks."""