Coverage for src / harnessutils / models / parts.py: 96%

1"""Part types for message decomposition. 

2 

3Parts are the granular units that make up messages. This enables 

4selective compaction where tool outputs can be cleared while 

5preserving text and metadata. 

6""" 

7 

8from dataclasses import dataclass, field 

9from typing import Any, Literal 

10 

11 

12@dataclass 

13class TimeInfo: 

14 """Timing information for parts.""" 

15 

16 start: int # Unix timestamp in milliseconds 

17 end: int | None = None 

18 compacted: int | None = None # When output was compacted 

19 

20 

21@dataclass 

22class ToolState: 

23 """State information for tool execution.""" 

24 

25 status: Literal["pending", "running", "completed", "error"] 

26 input: dict[str, Any] = field(default_factory=dict) 

27 output: str = "" 

28 title: str = "" 

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

30 error: str | None = None 

31 time: TimeInfo | None = None 

32 attachments: list[dict[str, Any]] = field(default_factory=list) 

33 

34 

35@dataclass 

36class Part: 

37 """Base class for message parts.""" 

38 

39 type: str = field(init=False) 

40 time: TimeInfo | None = None 

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

42 

43 

44@dataclass 

45class TextPart(Part): 

46 """Text content part.""" 

47 

48 text: str = "" 

49 ignored: bool = False # If true, skip when converting to model format 

50 

51 def __post_init__(self) -> None: 

52 self.type = "text" 

53 

54 

55@dataclass 

56class ReasoningPart(Part): 

57 """Extended thinking/reasoning content part.""" 

58 

59 text: str = "" 

60 

61 def __post_init__(self) -> None: 

62 self.type = "reasoning" 

63 

64 

65@dataclass 

66class ToolPart(Part): 

67 """Tool execution part.""" 

68 

69 tool: str = "" 

70 call_id: str = "" 

71 state: ToolState = field(default_factory=lambda: ToolState(status="pending")) 

72 

73 def __post_init__(self) -> None: 

74 self.type = "tool" 

75 

76 

77@dataclass 

78class StepStartPart(Part): 

79 """Step boundary marker - start of LLM turn.""" 

80 

81 snapshot: str = "" # State snapshot ID 

82 

83 def __post_init__(self) -> None: 

84 self.type = "step-start" 

85 

86 

87@dataclass 

88class StepFinishPart(Part): 

89 """Step boundary marker - end of LLM turn.""" 

90 

91 reason: Literal["stop", "tool-calls", "length"] = "stop" 

92 snapshot: str = "" # State snapshot ID 

93 tokens: dict[str, Any] = field(default_factory=dict) 

94 cost: float = 0.0 

95 

96 def __post_init__(self) -> None: 

97 self.type = "step-finish" 

98 

99 

100@dataclass 

101class CompactionPart(Part): 

102 """Marker for compaction request.""" 

103 

104 auto: bool = False # Was this auto-triggered? 

105 

106 def __post_init__(self) -> None: 

107 self.type = "compaction" 

108 

109 

110@dataclass 

111class PatchPart(Part): 

112 """Code change marker.""" 

113 

114 hash: str = "" # Diff hash 

115 files: list[str] = field(default_factory=list) 

116 

117 def __post_init__(self) -> None: 

118 self.type = "patch" 

119 

120 

121@dataclass 

122class SubtaskPart(Part): 

123 """Subtask invocation marker.""" 

124 

125 prompt: str = "" 

126 description: str = "" 

127 agent: str = "" 

128 model: dict[str, str] = field(default_factory=dict) 

129 

130 def __post_init__(self) -> None: 

131 self.type = "subtask"