Coverage for src / harnessutils / types.py: 100%

1"""Protocol definitions for harness-utils. 

2 

3This module defines the interfaces that applications must implement 

4to integrate with harness-utils. 

5""" 

6 

7from typing import Any, Protocol, runtime_checkable 

8 

9 

10@runtime_checkable 

11class LLMClient(Protocol): 

12 """Protocol for LLM client implementations. 

13 

14 Applications must provide an implementation of this protocol to enable 

15 LLM-powered summarization (Tier 3 compaction). 

16 

17 This is a callback-based design where the application owns the LLM client 

18 and the library requests LLM operations through this interface. 

19 """ 

20 

21 def invoke( 

22 self, 

23 messages: list[dict[str, Any]], 

24 system: list[str] | None = None, 

25 model: str | None = None, 

26 ) -> dict[str, Any]: 

27 """Invoke the LLM with the given messages. 

28 

29 Args: 

30 messages: List of messages in model format (role, content) 

31 system: Optional system prompt parts 

32 model: Optional model identifier to use 

33 

34 Returns: 

35 Dictionary containing: 

36 - content: The LLM response text 

37 - usage: Token usage information (input, output, cache, reasoning) 

38 - model: The model that was used 

39 """ 

40 ... 

41 

42 

43@runtime_checkable 

44class StorageBackend(Protocol): 

45 """Protocol for storage backend implementations. 

46 

47 The library provides default implementations (filesystem, in-memory), 

48 but applications can provide custom implementations for different 

49 storage strategies (e.g., cloud storage, databases). 

50 """ 

51 

52 def save_conversation(self, conversation_id: str, data: dict[str, Any]) -> None: 

53 """Save conversation metadata. 

54 

55 Args: 

56 conversation_id: Unique conversation identifier 

57 data: Conversation data to save 

58 """ 

59 ... 

60 

61 def load_conversation(self, conversation_id: str) -> dict[str, Any]: 

62 """Load conversation metadata. 

63 

64 Args: 

65 conversation_id: Unique conversation identifier 

66 

67 Returns: 

68 Conversation data 

69 

70 Raises: 

71 FileNotFoundError: If conversation doesn't exist 

72 """ 

73 ... 

74 

75 def save_message(self, conversation_id: str, message_id: str, data: dict[str, Any]) -> None: 

76 """Save message metadata. 

77 

78 Args: 

79 conversation_id: Conversation the message belongs to 

80 message_id: Unique message identifier 

81 data: Message data to save 

82 """ 

83 ... 

84 

85 def load_message(self, conversation_id: str, message_id: str) -> dict[str, Any]: 

86 """Load message metadata. 

87 

88 Args: 

89 conversation_id: Conversation the message belongs to 

90 message_id: Unique message identifier 

91 

92 Returns: 

93 Message data 

94 

95 Raises: 

96 FileNotFoundError: If message doesn't exist 

97 """ 

98 ... 

99 

100 def list_messages(self, conversation_id: str) -> list[str]: 

101 """List all message IDs for a conversation. 

102 

103 Args: 

104 conversation_id: Conversation to list messages for 

105 

106 Returns: 

107 List of message IDs in chronological order 

108 """ 

109 ... 

110 

111 def save_part(self, message_id: str, part_id: str, data: dict[str, Any]) -> None: 

112 """Save message part. 

113 

114 Args: 

115 message_id: Message the part belongs to 

116 part_id: Unique part identifier 

117 data: Part data to save 

118 """ 

119 ... 

120 

121 def load_part(self, message_id: str, part_id: str) -> dict[str, Any]: 

122 """Load message part. 

123 

124 Args: 

125 message_id: Message the part belongs to 

126 part_id: Unique part identifier 

127 

128 Returns: 

129 Part data 

130 

131 Raises: 

132 FileNotFoundError: If part doesn't exist 

133 """ 

134 ... 

135 

136 def list_parts(self, message_id: str) -> list[str]: 

137 """List all part IDs for a message. 

138 

139 Args: 

140 message_id: Message to list parts for 

141 

142 Returns: 

143 List of part IDs in order 

144 """ 

145 ... 

146 

147 def save_truncated_output(self, output_id: str, content: str) -> None: 

148 """Save full output that was truncated. 

149 

150 Args: 

151 output_id: Unique output identifier 

152 content: Full output content 

153 """ 

154 ... 

155 

156 def load_truncated_output(self, output_id: str) -> str: 

157 """Load full truncated output. 

158 

159 Args: 

160 output_id: Unique output identifier 

161 

162 Returns: 

163 Full output content 

164 

165 Raises: 

166 FileNotFoundError: If output doesn't exist 

167 """ 

168 ... 

169 

170 def cleanup_old_outputs(self, retention_days: int) -> int: 

171 """Clean up truncated outputs older than retention period. 

172 

173 Args: 

174 retention_days: Number of days to retain outputs 

175 

176 Returns: 

177 Number of outputs deleted 

178 """ 

179 ...