Fix blocking history compression edge cases
Detect stalled automatic history compression so the prompt-prep wait loop cannot spin forever when no further reduction is possible. Split large manual chat compaction input by verified token budget instead of line midpoint, covering single-line 85k+ character histories. Add regression tests for stalled compression, max-pass bailout, and large single-line compaction chunking.
Alessandro committed
May 12, 2026 at 04:47 UTC
6de7073bf9c06e7796a22e39bb3b5f9722ab6bd9
4 files changed
+312
-8
extensions/python/message_loop_prompts_before/_90_organize_history_wait.py
+36
-3
@@ -3,6 +3,8 @@ from agent import LoopData
3
from extensions.python.message_loop_end._10_organize_history import DATA_NAME_TASK
4
from helpers.defer import DeferredTask, THREAD_BACKGROUND
5
6
+MAX_SYNC_COMPRESSION_PASSES = 64
7
+
8
9
class OrganizeHistoryWait(Extension):
10
async def execute(self, loop_data: LoopData = LoopData(), **kwargs):
@@ -10,9 +12,13 @@ class OrganizeHistoryWait(Extension):
12
return
13
14
# sync action only required if the history is too large, otherwise leave it in background
15
+ passes = 0
16
while self.agent.history.is_over_limit():
17
+ passes += 1
18
+ before_tokens = self.agent.history.get_tokens()
19
+
20
# get task
15
- task: DeferredTask|None = self.agent.get_data(DATA_NAME_TASK)
21
+ task: DeferredTask | None = self.agent.get_data(DATA_NAME_TASK)
22
23
# Check if the task is already done
24
if task:
@@ -20,12 +26,39 @@ class OrganizeHistoryWait(Extension):
26
self.agent.context.log.set_progress("Compressing history...")
27
28
# Wait for the task to complete
23
- await task.result()
29
+ compressed = bool(await task.result())
30
31
# Clear the coroutine data after it's done
32
self.agent.set_data(DATA_NAME_TASK, None)
33
else:
34
# no task was running, start and wait
35
self.agent.context.log.set_progress("Compressing history...")
30
- await self.agent.history.compress()
36
+ compressed = await self.agent.history.compress()
37
+
38
+ after_tokens = self.agent.history.get_tokens()
39
+ if not compressed or after_tokens >= before_tokens:
40
+ self._log_compression_stalled(before_tokens, after_tokens)
41
+ break
42
+
43
+ if passes >= MAX_SYNC_COMPRESSION_PASSES:
44
+ self._log_compression_stalled(
45
+ before_tokens, after_tokens, max_passes=True
46
+ )
47
+ break
48
+
49
+ def _log_compression_stalled(
50
+ self, before_tokens: int, after_tokens: int, max_passes: bool = False
51
+ ) -> None:
52
+ if not self.agent:
53
+ return
54
55
+ detail = (
56
+ f"History compression stopped after {MAX_SYNC_COMPRESSION_PASSES} passes"
57
+ if max_passes
58
+ else "History compression could not reduce the prompt history further"
59
+ )
60
+ self.agent.context.log.log(
61
+ type="warning",
62
+ heading="History compression stalled",
63
+ content=f"{detail}. Tokens before: {before_tokens}; after: {after_tokens}.",
64
+ )
plugins/_chat_compaction/helpers/compactor.py
+91
-5
@@ -1,5 +1,6 @@
1
"""Core compaction logic for the compaction plugin."""
2
import os
3
+from collections import deque
4
from datetime import datetime
5
6
import models as models_module
@@ -15,6 +16,9 @@ from helpers.persist_chat import (
16
from helpers.state_monitor_integration import mark_dirty_all
17
18
MIN_COMPACTION_TOKENS = 1000
19
+COMPACTION_CHUNK_TARGET_RATIO = 0.9
20
+COMPACTION_CHUNK_VERIFY_RATIO = 0.98
21
+
22
from plugins._model_config.helpers.model_config import (
23
get_chat_model_config,
24
get_utility_model_config,
@@ -199,14 +203,11 @@ async def _compact_large_history(
203
agent, full_text: str, token_count: int, max_input_tokens: int, log_item, model
204
) -> str:
205
"""Handle large histories by splitting into chunks and summarizing iteratively."""
206
+ chunks = _split_text_for_compaction(agent, full_text, token_count, max_input_tokens)
207
log_item.update(
203
- content=f"History is large (~{token_count} tokens). Splitting into chunks...",
208
+ content=f"History is large (~{token_count} tokens). Splitting into {len(chunks)} chunks...",
209
)
210
206
- lines = full_text.split('\n')
207
- mid = len(lines) // 2
208
- chunks = ['\n'.join(lines[:mid]), '\n'.join(lines[mid:])]
209
-
211
summaries = []
212
for i, chunk in enumerate(chunks, 1):
213
log_item.update(content=f"Summarizing part {i}/{len(chunks)}...")
@@ -241,6 +242,91 @@ async def _compact_large_history(
242
return final_summary
243
244
245
+def _split_text_for_compaction(
246
+ agent, full_text: str, token_count: int, max_input_tokens: int
247
+) -> list[str]:
248
+ """Split large compaction input into prompt-safe chunks.
249
+
250
+ The previous line-midpoint split left a single-line payload as one empty
251
+ chunk plus one still-oversized chunk. This splitter derives a conservative
252
+ character target from the measured token density, then verifies each prompt
253
+ and keeps splitting any chunk that still exceeds the model input budget.
254
+ """
255
+ text = full_text or ""
256
+ if not text:
257
+ return []
258
+
259
+ prompt_overhead = _compaction_input_tokens(agent, "")
260
+ usable_tokens = max(max_input_tokens - prompt_overhead, 1)
261
+ target_tokens = max(int(usable_tokens * COMPACTION_CHUNK_TARGET_RATIO), 1)
262
+
263
+ if token_count <= target_tokens:
264
+ return [text]
265
+
266
+ chars_per_token = max(len(text) / max(token_count, 1), 0.01)
267
+ target_chars = max(int(target_tokens * chars_per_token), 1)
268
+ chunks = _split_text_by_chars(text, target_chars)
269
+
270
+ verified: list[str] = []
271
+ max_verified_tokens = max(int(max_input_tokens * COMPACTION_CHUNK_VERIFY_RATIO), 1)
272
+ pending = deque(chunk for chunk in chunks if chunk)
273
+
274
+ while pending:
275
+ chunk = pending.popleft()
276
+ if not chunk:
277
+ continue
278
+
279
+ if (
280
+ len(chunk) <= 1
281
+ or _compaction_input_tokens(agent, chunk) <= max_verified_tokens
282
+ ):
283
+ verified.append(chunk)
284
+ continue
285
+
286
+ split_chunks = _split_text_by_chars(chunk, max(len(chunk) // 2, 1))
287
+ if len(split_chunks) <= 1:
288
+ verified.append(chunk)
289
+ else:
290
+ pending.extendleft(reversed(split_chunks))
291
+
292
+ return verified
293
+
294
+
295
+def _compaction_input_tokens(agent, conversation: str) -> int:
296
+ system_prompt = agent.read_prompt("compact.sys.md")
297
+ user_prompt = agent.read_prompt("compact.msg.md", conversation=conversation)
298
+ return tokens.approximate_tokens(system_prompt) + tokens.approximate_tokens(
299
+ user_prompt
300
+ )
301
+
302
+
303
+def _split_text_by_chars(text: str, target_chars: int) -> list[str]:
304
+ if not text:
305
+ return []
306
+
307
+ target_chars = max(int(target_chars), 1)
308
+ chunks: list[str] = []
309
+ start = 0
310
+ length = len(text)
311
+
312
+ while start < length:
313
+ end = min(start + target_chars, length)
314
+ if end < length:
315
+ floor = start + max((end - start) // 2, 1)
316
+ split_at = text.rfind("\n", floor, end)
317
+ if split_at == -1:
318
+ split_at = text.rfind(" ", floor, end)
319
+ if split_at > start:
320
+ end = split_at + 1
321
+
322
+ chunk = text[start:end]
323
+ if chunk:
324
+ chunks.append(chunk)
325
+ start = end
326
+
327
+ return chunks
328
+
329
+
330
async def get_compaction_stats(context) -> dict:
331
"""
332
Get statistics about the current chat for the confirmation modal.
tests/test_chat_compaction.py
new
+89
@@ -0,0 +1,89 @@
1
+import sys
2
+from pathlib import Path
3
+
4
+import pytest
5
+
6
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
7
+if str(PROJECT_ROOT) not in sys.path:
8
+ sys.path.insert(0, str(PROJECT_ROOT))
9
+
10
+from plugins._chat_compaction.helpers import compactor
11
+
12
+
13
+class _FakeAgent:
14
+ def read_prompt(self, name: str, **kwargs):
15
+ if name == "compact.sys.md":
16
+ return "system"
17
+ if name == "compact.msg.md":
18
+ return kwargs.get("conversation", "")
19
+ raise AssertionError(f"Unexpected prompt: {name}")
20
+
21
+
22
+class _FakeLog:
23
+ def __init__(self):
24
+ self.updates = []
25
+ self.streams = []
26
+
27
+ def update(self, **kwargs):
28
+ self.updates.append(kwargs)
29
+
30
+ def stream(self, **kwargs):
31
+ self.streams.append(kwargs)
32
+
33
+
34
+class _RecordingModel:
35
+ def __init__(self):
36
+ self.user_messages = []
37
+
38
+ async def unified_call(self, system_message, user_message, response_callback=None):
39
+ self.user_messages.append(user_message)
40
+ if response_callback:
41
+ await response_callback("done", "done")
42
+ return f"summary-{len(self.user_messages)}", None
43
+
44
+
45
+def test_compaction_splitter_wraps_single_line_85k_payload(monkeypatch):
46
+ monkeypatch.setattr(
47
+ compactor.tokens, "approximate_tokens", lambda text: len(text or "")
48
+ )
49
+
50
+ agent = _FakeAgent()
51
+ chunks = compactor._split_text_for_compaction(
52
+ agent,
53
+ "x" * 85_000,
54
+ token_count=85_000,
55
+ max_input_tokens=10_000,
56
+ )
57
+
58
+ assert len(chunks) > 2
59
+ assert all(chunks)
60
+ assert "".join(chunks) == "x" * 85_000
61
+ assert all(
62
+ compactor._compaction_input_tokens(agent, chunk) <= 10_000
63
+ for chunk in chunks
64
+ )
65
+
66
+
67
+@pytest.mark.asyncio
68
+async def test_large_compaction_does_not_send_unsplit_single_line_payload(monkeypatch):
69
+ monkeypatch.setattr(
70
+ compactor.tokens, "approximate_tokens", lambda text: len(text or "")
71
+ )
72
+
73
+ agent = _FakeAgent()
74
+ model = _RecordingModel()
75
+
76
+ summary = await compactor._compact_large_history(
77
+ agent,
78
+ "x" * 85_000,
79
+ token_count=85_000,
80
+ max_input_tokens=10_000,
81
+ log_item=_FakeLog(),
82
+ model=model,
83
+ )
84
+
85
+ chunk_messages = model.user_messages[:-1]
86
+ assert summary == f"summary-{len(model.user_messages)}"
87
+ assert len(chunk_messages) > 2
88
+ assert all(chunk_messages)
89
+ assert all(len(message) <= 10_000 for message in chunk_messages)
tests/test_history_compression_wait.py
new
+96
@@ -0,0 +1,96 @@
1
+import sys
2
+from pathlib import Path
3
+
4
+import pytest
5
+
6
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
7
+if str(PROJECT_ROOT) not in sys.path:
8
+ sys.path.insert(0, str(PROJECT_ROOT))
9
+
10
+from extensions.python.message_loop_prompts_before._90_organize_history_wait import (
11
+ MAX_SYNC_COMPRESSION_PASSES,
12
+ OrganizeHistoryWait,
13
+)
14
+
15
+
16
+class _StalledHistory:
17
+ def __init__(self):
18
+ self.compress_calls = 0
19
+
20
+ def is_over_limit(self):
21
+ return self.compress_calls < 2
22
+
23
+ def get_tokens(self):
24
+ return 1234
25
+
26
+ async def compress(self):
27
+ self.compress_calls += 1
28
+ return False
29
+
30
+
31
+class _MaxPassHistory:
32
+ def __init__(self):
33
+ self.compress_calls = 0
34
+ self.tokens = 2000
35
+
36
+ def is_over_limit(self):
37
+ return True
38
+
39
+ def get_tokens(self):
40
+ return self.tokens
41
+
42
+ async def compress(self):
43
+ self.compress_calls += 1
44
+ self.tokens -= 1
45
+ return True
46
+
47
+
48
+class _FakeLog:
49
+ def __init__(self):
50
+ self.entries = []
51
+
52
+ def set_progress(self, *args, **kwargs):
53
+ pass
54
+
55
+ def log(self, **kwargs):
56
+ self.entries.append(kwargs)
57
+
58
+
59
+class _FakeAgent:
60
+ def __init__(self, history=None):
61
+ self.data = {}
62
+ self.history = history or _StalledHistory()
63
+ self.context = type("Context", (), {"log": _FakeLog()})()
64
+
65
+ def get_data(self, key):
66
+ return self.data.get(key)
67
+
68
+ def set_data(self, key, value):
69
+ self.data[key] = value
70
+
71
+
72
+@pytest.mark.asyncio
73
+async def test_history_wait_stops_when_compression_makes_no_progress():
74
+ agent = _FakeAgent()
75
+
76
+ await OrganizeHistoryWait(agent).execute()
77
+
78
+ assert agent.history.compress_calls == 1
79
+ assert agent.context.log.entries
80
+ assert agent.context.log.entries[-1]["heading"] == "History compression stalled"
81
+
82
+
83
+@pytest.mark.asyncio
84
+async def test_history_wait_stops_after_max_sync_compression_passes():
85
+ history = _MaxPassHistory()
86
+ agent = _FakeAgent(history)
87
+
88
+ await OrganizeHistoryWait(agent).execute()
89
+
90
+ assert history.compress_calls == MAX_SYNC_COMPRESSION_PASSES
91
+ assert agent.context.log.entries
92
+ assert agent.context.log.entries[-1]["heading"] == "History compression stalled"
93
+ assert (
94
+ f"stopped after {MAX_SYNC_COMPRESSION_PASSES} passes"
95
+ in agent.context.log.entries[-1]["content"]
96
+ )