Coverage for merco/tools/processors/truncation.py: 29%

1"""TruncationProcessor — truncates large tool results.""" 

2from __future__ import annotations 

3 

4import json 

5import os 

6import time 

7import logging 

8from pathlib import Path 

9 

10from merco.core.pipeline import Processor, ProcessContext, _walk_truncate, _is_reading_trunc_file 

11 

12logger = logging.getLogger("merco.pipeline") 

13 

14 

15class TruncationProcessor(Processor): 

16 """通用结果截断：任意工具结果 → 写入完整 JSON 文件 → 返回递归截断版。 

17 

18 对标 OpenCode truncate.ts 的缓存策略： 

19 

20 - **触发条件**：整个 result dict 序列化为 JSON，超过 max_bytes 则截断 

21 - **缓存位置**：~/.merco/trunc/ 

22 - **文件命名**：{timestamp_ms}_{safe_tool_name}.json（毫秒时间戳防冲突） 

23 - **缓存格式**：完整 JSON（保留所有字段结构） 

24 - **文件上限**：单文件最大 max_file_bytes（默认 50 MB），超限拒绝 

25 - **清理策略**：7 天过期自动删除（OpenCode 同款） 

26 - **清理频率**：懒清理 — 每次新写入时检查，距上次清理 > 1 小时才执行 

27 - **可配置**:: 

28 

29 pipeline.disable("truncation") # 调试时关闭 

30 TruncationProcessor(max_bytes=8000) # 调大上下文窗口 

31 TruncationProcessor(retention_days=3) # 加快清理 

32 """ 

33 

34 name = "truncation" 

35 _last_cleanup = 0.0 # 上次清理时间戳（类级，所有实例共享） 

36 _cleanup_interval = 3600 # 清理间隔（秒），1 小时 

37 

38 def __init__(self, max_bytes: int = 4000, *, 

39 retention_days: int = 7, 

40 max_file_bytes: int = 50 * 1024 * 1024, 

41 trunc_dir: str | None = None): 

42 self.max_bytes = max_bytes 

43 self.retention_days = retention_days 

44 self.max_file_bytes = max_file_bytes 

45 self._trunc_dir = trunc_dir 

46 

47 @property 

48 def trunc_dir(self) -> str: 

49 if self._trunc_dir is None: 

50 self._trunc_dir = os.path.expanduser("~/.merco/trunc") 

51 return self._trunc_dir 

52 

53 async def process(self, ctx: ProcessContext) -> bool: 

54 # 序列化整个结果判断是否需截断 

55 try: 

56 serialized = json.dumps(ctx.result, ensure_ascii=False) 

57 except (TypeError, ValueError): 

58 return False 

59 

60 if len(serialized) <= self.max_bytes: 

61 return False 

62 

63 reading_trunc_file = _is_reading_trunc_file(ctx, self.trunc_dir) 

64 per_value = max(int(self.max_bytes * 0.5), 500) 

65 total_chars = len(serialized) 

66 total_pages = (total_chars + per_value - 1) // per_value 

67 

68 # 文件大小安全上限（超限不写文件） 

69 if total_chars > self.max_file_bytes: 

70 ctx.result["_truncated"] = True 

71 ctx.result["_pagination"] = { 

72 "total_chars": total_chars, 

73 "page_size": per_value, 

74 "total_pages": total_pages, 

75 "current_page": 1, 

76 "next_page_offset": per_value, 

77 } 

78 ctx.result["_hint"] = ( 

79 f"结果过长（{total_chars:,} 字符，超过缓存上限 {self.max_file_bytes:,}），" 

80 f"已截断且未缓存至本地。可缩小请求范围重新执行。" 

81 ) 

82 ctx.result = _walk_truncate(ctx.result, per_value, "[超出缓存上限，未保存]") 

83 return False 

84 

85 # 分页元数据 

86 pagination = { 

87 "total_chars": total_chars, 

88 "page_size": per_value, 

89 "total_pages": total_pages, 

90 "current_page": 1, 

91 "next_page_offset": per_value, 

92 } 

93 

94 if reading_trunc_file: 

95 # 防套娃：读截断缓存文件 → 截断内容但不写新文件 

96 ctx.result["_truncated"] = True 

97 ctx.result["_pagination"] = pagination 

98 ctx.result["_hint"] = ( 

99 f"结果过长（{total_chars:,} 字符，第 1/{total_pages} 页）。" 

100 f"这是截断缓存文件，请用 read_file 的 offset/limit 翻页，" 

101 f"或用 bash grep 搜索关键信息。" 

102 ) 

103 ctx.result = _walk_truncate(ctx.result, per_value, 

104 "[截断缓存文件]", pagination) 

105 return False 

106 

107 # 正常截断：写缓存文件 + 分页 

108 os.makedirs(self.trunc_dir, exist_ok=True) 

109 self._maybe_cleanup() 

110 

111 ts = int(time.time() * 1000) 

112 safe_name = ctx.tool_name.replace("/", "_") 

113 filepath = os.path.join(self.trunc_dir, f"{ts}_{safe_name}.json") 

114 try: 

115 Path(filepath).write_text(serialized, encoding="utf-8") 

116 except OSError: 

117 logger.warning("截断文件写入失败: %s", filepath) 

118 return False 

119 

120 ctx.result["_truncated"] = True 

121 ctx.result["_full_output_path"] = filepath 

122 ctx.result["_pagination"] = pagination 

123 lines_per_page = max(1, per_value // 60) 

124 ctx.result["_hint"] = ( 

125 f"结果过长（{total_chars:,} 字符，共 {total_pages} 页）。" 

126 f"当前第 1 页。" 

127 f"➡️ 下一页: read_file {filepath} offset={lines_per_page + 1} limit={lines_per_page}" 

128 f" 或 bash grep 搜索关键信息。" 

129 ) 

130 ctx.result = _walk_truncate(ctx.result, per_value, filepath, pagination) 

131 

132 return False 

133 

134 def _maybe_cleanup(self) -> None: 

135 """懒清理：距上次清理超过间隔才执行，删除超过 retention_days 的文件。""" 

136 now = time.time() 

137 if now - TruncationProcessor._last_cleanup < TruncationProcessor._cleanup_interval: 

138 return 

139 

140 TruncationProcessor._last_cleanup = now 

141 cutoff = now - self.retention_days * 86400 # 秒 

142 removed = 0 

143 

144 try: 

145 for entry in Path(self.trunc_dir).iterdir(): 

146 if not entry.is_file(): 

147 continue 

148 if not entry.name.endswith(".json"): 

149 continue 

150 try: 

151 if entry.stat().st_mtime < cutoff: 

152 entry.unlink() 

153 removed += 1 

154 except OSError: 

155 pass 

156 except OSError: 

157 pass # 目录不存在等 

158 

159 if removed: 

160 logger.info("截断缓存清理: 删除 %d 个过期文件 (> %d 天)", removed, self.retention_days)