Coverage for src / harness_utils / 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( 

76 self, 

77 conversation_id: str, 

78 message_id: str, 

79 data: dict[str, Any] 

80 ) -> None: 

81 """Save message metadata. 

82 

83 Args: 

84 conversation_id: Conversation the message belongs to 

85 message_id: Unique message identifier 

86 data: Message data to save 

87 """ 

88 ... 

89 

90 def load_message( 

91 self, 

92 conversation_id: str, 

93 message_id: str 

94 ) -> dict[str, Any]: 

95 """Load message metadata. 

96 

97 Args: 

98 conversation_id: Conversation the message belongs to 

99 message_id: Unique message identifier 

100 

101 Returns: 

102 Message data 

103 

104 Raises: 

105 FileNotFoundError: If message doesn't exist 

106 """ 

107 ... 

108 

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

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

111 

112 Args: 

113 conversation_id: Conversation to list messages for 

114 

115 Returns: 

116 List of message IDs in chronological order 

117 """ 

118 ... 

119 

120 def save_part( 

121 self, 

122 message_id: str, 

123 part_id: str, 

124 data: dict[str, Any] 

125 ) -> None: 

126 """Save message part. 

127 

128 Args: 

129 message_id: Message the part belongs to 

130 part_id: Unique part identifier 

131 data: Part data to save 

132 """ 

133 ... 

134 

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

136 """Load message part. 

137 

138 Args: 

139 message_id: Message the part belongs to 

140 part_id: Unique part identifier 

141 

142 Returns: 

143 Part data 

144 

145 Raises: 

146 FileNotFoundError: If part doesn't exist 

147 """ 

148 ... 

149 

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

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

152 

153 Args: 

154 message_id: Message to list parts for 

155 

156 Returns: 

157 List of part IDs in order 

158 """ 

159 ... 

160 

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

162 """Save full output that was truncated. 

163 

164 Args: 

165 output_id: Unique output identifier 

166 content: Full output content 

167 """ 

168 ... 

169 

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

171 """Load full truncated output. 

172 

173 Args: 

174 output_id: Unique output identifier 

175 

176 Returns: 

177 Full output content 

178 

179 Raises: 

180 FileNotFoundError: If output doesn't exist 

181 """ 

182 ... 

183 

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

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

186 

187 Args: 

188 retention_days: Number of days to retain outputs 

189 

190 Returns: 

191 Number of outputs deleted 

192 """ 

193 ...