Coverage for src / harnessutils / conversion / to_model.py: 98%

1"""Convert internal messages to model format. 

2 

3Transforms internal message format (granular, storage-optimized) 

4to model format (LLM-compatible). 

5""" 

6 

7from typing import Any 

8 

9from harnessutils.models.message import Message 

10from harnessutils.models.parts import ToolPart 

11 

12 

13def to_model_messages( 

14 messages: list[Message], 

15 excluded_outputs: set[str] | None = None, 

16) -> list[dict[str, Any]]: 

17 """Convert internal messages to model format. 

18 

19 Handles: 

20 - Excluded tool outputs (replace with marker, via excluded_outputs set) 

21 - Legacy compacted tool outputs (replace with marker, via time.compacted) 

22 - Interrupted tool calls (inject error) 

23 - Two-part system prompts 

24 - Stop at summary message 

25 

26 Args: 

27 messages: Internal message objects 

28 excluded_outputs: Call IDs whose outputs should be omitted from the payload 

29 

30 Returns: 

31 List of messages in model format 

32 """ 

33 excluded = excluded_outputs or set() 

34 model_messages: list[dict[str, Any]] = [] 

35 summary_found = False 

36 summary_parent_id: str | None = None 

37 

38 for msg in reversed(messages): 

39 if summary_found and msg.role == "user" and msg.id == summary_parent_id: 

40 break 

41 

42 if msg.role == "assistant" and msg.summary: 

43 summary_found = True 

44 summary_parent_id = msg.parent_id 

45 

46 if len(msg.parts) == 0: 

47 continue 

48 

49 if msg.role == "user": 

50 model_msg = _convert_user_message(msg) 

51 if model_msg and model_msg.get("content"): 

52 model_messages.insert(0, model_msg) 

53 

54 elif msg.role == "assistant": 

55 model_msg = _convert_assistant_message(msg, excluded_outputs=excluded) 

56 if model_msg and model_msg.get("content"): 

57 model_messages.insert(0, model_msg) 

58 

59 return model_messages 

60 

61 

62def _convert_user_message(msg: Message) -> dict[str, Any] | None: 

63 """Convert user message to model format. 

64 

65 Args: 

66 msg: User message 

67 

68 Returns: 

69 Model format message or None if empty 

70 """ 

71 content_parts: list[str] = [] 

72 

73 for part in msg.parts: 

74 if part.type == "text" and not getattr(part, "ignored", False): 

75 content_parts.append(getattr(part, "text", "")) 

76 elif part.type == "compaction": 

77 content_parts.append("What did we do so far?") 

78 

79 if not content_parts: 

80 return None 

81 

82 return { 

83 "role": "user", 

84 "content": "\n".join(content_parts), 

85 } 

86 

87 

88def _convert_assistant_message( 

89 msg: Message, 

90 excluded_outputs: set[str] | None = None, 

91) -> dict[str, Any] | None: 

92 """Convert assistant message to model format. 

93 

94 Args: 

95 msg: Assistant message 

96 excluded_outputs: Call IDs whose outputs should be replaced with a marker 

97 

98 Returns: 

99 Model format message or None if empty 

100 """ 

101 if msg.error and not msg.has_partial_output(): 

102 return None 

103 

104 excluded = excluded_outputs or set() 

105 content_parts: list[str] = [] 

106 

107 for part in msg.parts: 

108 if part.type == "text": 

109 content_parts.append(getattr(part, "text", "")) 

110 

111 elif part.type == "reasoning": 

112 content_parts.append(f"[Extended thinking: {len(getattr(part, 'text', ''))} chars]") 

113 

114 elif part.type == "tool": 

115 tool_part = part 

116 assert isinstance(tool_part, ToolPart) 

117 

118 if tool_part.state.status == "completed": 

119 is_excluded = tool_part.call_id in excluded 

120 is_legacy_compacted = bool( 

121 tool_part.state.time and tool_part.state.time.compacted 

122 ) 

123 if is_excluded or is_legacy_compacted: 

124 if tool_part.state.summary_text: 

125 output = f"[Summarized: {tool_part.state.summary_text}]" 

126 elif is_legacy_compacted: 

127 output = "[Old tool result content cleared]" 

128 else: 

129 output = "[Tool output excluded from context]" 

130 else: 

131 output = tool_part.state.output 

132 

133 content_parts.append(f"[Tool: {tool_part.tool}] {tool_part.state.title}\n{output}") 

134 

135 elif tool_part.state.status == "error": 

136 content_parts.append(f"[Tool Error: {tool_part.tool}] {tool_part.state.error}") 

137 

138 elif tool_part.state.status in ["pending", "running"]: 

139 content_parts.append(f"[Tool execution was interrupted: {tool_part.tool}]") 

140 

141 if not content_parts: 

142 return None 

143 

144 return { 

145 "role": "assistant", 

146 "content": "\n\n".join(content_parts), 

147 }