Coverage for src / harness_utils / compaction / truncation.py: 96%

1"""Tier 1: Output truncation at tool execution boundary. 

2 

3Prevents large outputs from entering context by truncating at source. 

4Cost: Free, Latency: 0ms. 

5""" 

6 

7from dataclasses import dataclass 

8from typing import Literal 

9 

10from harness_utils.config import TruncationConfig 

11 

12 

13@dataclass 

14class TruncationResult: 

15 """Result of truncation operation.""" 

16 

17 content: str 

18 truncated: bool 

19 output_path: str | None = None 

20 bytes_removed: int = 0 

21 

22 

23def truncate_output( 

24 output: str, 

25 config: TruncationConfig, 

26 output_id: str | None = None, 

27) -> TruncationResult: 

28 """Truncate tool output if it exceeds limits. 

29 

30 Args: 

31 output: The tool output to potentially truncate 

32 config: Truncation configuration 

33 output_id: ID for saving full output (if None, full output not saved) 

34 

35 Returns: 

36 TruncationResult with content and metadata 

37 """ 

38 lines = output.split("\n") 

39 total_bytes = len(output.encode("utf-8")) 

40 

41 if len(lines) <= config.max_lines and total_bytes <= config.max_bytes: 

42 return TruncationResult( 

43 content=output, 

44 truncated=False, 

45 ) 

46 

47 preview_lines: list[str] = [] 

48 bytes_accumulated = 0 

49 

50 if config.direction == "head": 

51 for i, line in enumerate(lines): 

52 if i >= config.max_lines: 

53 break 

54 line_bytes = len(line.encode("utf-8")) + 1 # +1 for newline 

55 if bytes_accumulated + line_bytes > config.max_bytes: 

56 break 

57 preview_lines.append(line) 

58 bytes_accumulated += line_bytes 

59 else: # tail 

60 for i in range(len(lines) - 1, -1, -1): 

61 if len(preview_lines) >= config.max_lines: 

62 break 

63 line = lines[i] 

64 line_bytes = len(line.encode("utf-8")) + 1 # +1 for newline 

65 if bytes_accumulated + line_bytes > config.max_bytes: 

66 break 

67 preview_lines.insert(0, line) 

68 bytes_accumulated += line_bytes 

69 

70 preview = "\n".join(preview_lines) 

71 bytes_removed = total_bytes - bytes_accumulated 

72 

73 message = _format_truncated_message( 

74 preview, 

75 bytes_removed, 

76 output_id, 

77 config.direction, 

78 ) 

79 

80 return TruncationResult( 

81 content=message, 

82 truncated=True, 

83 output_path=output_id, 

84 bytes_removed=bytes_removed, 

85 ) 

86 

87 

88def _format_truncated_message( 

89 preview: str, 

90 bytes_removed: int, 

91 output_path: str | None, 

92 direction: Literal["head", "tail"], 

93) -> str: 

94 """Format the truncated output message. 

95 

96 Args: 

97 preview: Preview content (head or tail) 

98 bytes_removed: Number of bytes that were removed 

99 output_path: Path where full output was saved 

100 direction: Direction of truncation 

101 

102 Returns: 

103 Formatted message string 

104 """ 

105 parts = [preview] 

106 

107 if bytes_removed > 0: 

108 parts.append("") 

109 parts.append(f"...{bytes_removed} bytes truncated...") 

110 parts.append("") 

111 

112 if output_path: 

113 parts.append(f"Full output saved to: {output_path}") 

114 parts.append("Use search tools to query the full content or read specific sections.") 

115 parts.append("Delegate large file processing to specialized exploration agents.") 

116 

117 return "\n".join(parts)