Truncate logs (#571)
* feat: Add ratio-based text truncation * feat(log): Add automatic truncation for long log entries
The Master committed
Jul 9, 2025 at 20:23 UTC
d5ccc867be447d73e83f9496ad5ab444ef16c13e
2 files changed
+118
-8
python/helpers/log.py
+91
-7
@@ -3,6 +3,7 @@ import json
3
from typing import Any, Literal, Optional, Dict
4
import uuid
5
from collections import OrderedDict # Import OrderedDict
6
+from python.helpers.strings import truncate_text_by_ratio
7
8
Type = Literal[
9
"agent",
@@ -23,6 +24,69 @@ Type = Literal[
24
ProgressUpdate = Literal["persistent", "temporary", "none"]
25
26
27
+HEADING_MAX_LEN: int = 120
28
+CONTENT_MAX_LEN: int = 4000
29
+KEY_MAX_LEN: int = 60
30
+VALUE_MAX_LEN: int = 1000
31
+PROGRESS_MAX_LEN: int = 120
32
+
33
+
34
+def _truncate_heading(text: str | None) -> str:
35
+ if text is None:
36
+ return ""
37
+ return truncate_text_by_ratio(str(text), HEADING_MAX_LEN, "...", ratio=1.0)
38
+
39
+def _truncate_progress(text: str | None) -> str:
40
+ if text is None:
41
+ return ""
42
+ return truncate_text_by_ratio(str(text), PROGRESS_MAX_LEN, "...", ratio=1.0)
43
+
44
+def _truncate_key(text: str) -> str:
45
+ return truncate_text_by_ratio(str(text), KEY_MAX_LEN, "...", ratio=1.0)
46
+
47
+def _truncate_value(text: Any) -> Any:
48
+ # Convert non-str values to json for consistent length measurement
49
+ if isinstance(text, str):
50
+ raw = text
51
+ else:
52
+ try:
53
+ raw = json.dumps(text, ensure_ascii=False)
54
+ except Exception:
55
+ raw = str(text)
56
+
57
+ if len(raw) <= VALUE_MAX_LEN:
58
+ return text # No truncation needed, preserve original type
59
+
60
+ # Determine removed characters dynamically to build replacement string
61
+ removed = len(raw) - VALUE_MAX_LEN
62
+ while True:
63
+ replacement = f"\n\n<< {removed} Characters hidden >>\n\n"
64
+ truncated = truncate_text_by_ratio(raw, VALUE_MAX_LEN, replacement, ratio=0.3)
65
+ new_removed = len(raw) - (len(truncated) - len(replacement))
66
+ if new_removed == removed:
67
+ break
68
+ removed = new_removed
69
+
70
+ return truncated
71
+
72
+def _truncate_content(text: str | None) -> str:
73
+ if text is None:
74
+ return ""
75
+ raw = str(text)
76
+ if len(raw) <= CONTENT_MAX_LEN:
77
+ return raw
78
+
79
+ # Same dynamic replacement logic as value truncation
80
+ removed = len(raw) - CONTENT_MAX_LEN
81
+ while True:
82
+ replacement = f"\n\n<< {removed} Characters hidden >>\n\n"
83
+ truncated = truncate_text_by_ratio(raw, CONTENT_MAX_LEN, replacement, ratio=0.3)
84
+ new_removed = len(raw) - (len(truncated) - len(replacement))
85
+ if new_removed == removed:
86
+ break
87
+ removed = new_removed
88
+ return truncated
89
+
90
@dataclass
91
class LogItem:
92
log: "Log"
@@ -107,9 +171,22 @@ class Log:
171
id: Optional[str] = None, # Add id parameter
172
**kwargs,
173
) -> LogItem:
110
- # Use OrderedDict if kvps is provided
174
+ # Truncate heading and content
175
+ heading = _truncate_heading(heading)
176
+ content = _truncate_content(content)
177
+
178
+ # Truncate kvps
179
if kvps is not None:
112
- kvps = OrderedDict(kvps)
180
+ kvps = OrderedDict({
181
+ _truncate_key(k): _truncate_value(v) for k, v in kvps.items()
182
+ })
183
+ # Apply truncation to kwargs merged into kvps later
184
+ kwargs = { _truncate_key(k): _truncate_value(v) for k, v in (kwargs or {}).items() }
185
+
186
+ # Ensure kvps is OrderedDict even if None
187
+ if kvps is None:
188
+ kvps = OrderedDict()
189
+
190
item = LogItem(
191
log=self,
192
no=len(self.logs),
@@ -140,16 +217,23 @@ class Log:
217
**kwargs,
218
):
219
item = self.logs[no]
220
+ # Apply truncation where necessary
221
if type is not None:
222
item.type = type
223
+
224
if update_progress is not None:
225
item.update_progress = update_progress
226
+
227
if heading is not None:
148
- item.heading = heading
228
+ item.heading = _truncate_heading(heading)
229
+
230
if content is not None:
150
- item.content = content
231
+ item.content = _truncate_content(content)
232
+
233
if kvps is not None:
152
- item.kvps = OrderedDict(kvps) # Use OrderedDict to keep the order
234
+ item.kvps = OrderedDict({
235
+ _truncate_key(k): _truncate_value(v) for k, v in kvps.items()
236
+ }) # Ensure order
237
238
if temp is not None:
239
item.temp = temp
@@ -158,13 +242,13 @@ class Log:
242
if item.kvps is None:
243
item.kvps = OrderedDict() # Ensure kvps is an OrderedDict
244
for k, v in kwargs.items():
161
- item.kvps[k] = v
245
+ item.kvps[_truncate_key(k)] = _truncate_value(v)
246
247
self.updates += [item.no]
248
self._update_progress_from_item(item)
249
250
def set_progress(self, progress: str, no: int = 0, active: bool = True):
167
- self.progress = progress
251
+ self.progress = _truncate_progress(progress)
252
if not no:
253
no = len(self.logs)
254
self.progress_no = no
python/helpers/strings.py
+27
-1
@@ -129,4 +129,30 @@ def truncate_text(text: str, length: int, at_end: bool = True, replacement: str
129
return text[:length] + replacement
130
else:
131
return replacement + text[-length:]
132
-
\ No newline at end of file
132
+
133
+def truncate_text_by_ratio(text: str, threshold: int, replacement: str = "...", ratio: float = 0.5) -> str:
134
+ """Truncate text with replacement at a specified ratio position."""
135
+ threshold = int(threshold)
136
+ if not threshold or len(text) <= threshold:
137
+ return text
138
+
139
+ # Clamp ratio to valid range
140
+ ratio = max(0.0, min(1.0, float(ratio)))
141
+
142
+ # Calculate available space for original text after accounting for replacement
143
+ available_space = threshold - len(replacement)
144
+ if available_space <= 0:
145
+ return replacement[:threshold]
146
+
147
+ # Handle edge cases for efficiency
148
+ if ratio == 0.0:
149
+ # Replace from start: "...text"
150
+ return replacement + text[-available_space:]
151
+ elif ratio == 1.0:
152
+ # Replace from end: "text..."
153
+ return text[:available_space] + replacement
154
+ else:
155
+ # Replace in middle based on ratio
156
+ start_len = int(available_space * ratio)
157
+ end_len = available_space - start_len
158
+ return text[:start_len] + replacement + text[-end_len:]
\ No newline at end of file