feat(tui): fix lint, add robot/benchmark/changelog for pruning

- Fix ruff format violations in app.py, conversation.py, and
  tui_conversation_pruning_steps.py (3 files reformatted)
- Fix ruff E501 line-too-long in helper_tui_conversation_pruning.py
- Expand ConversationSettings docstring to explain hysteresis design
- Add _note_active invariant comment in ConversationStream.__init__
- Add full docstring to load_conversation_settings (config path, keys,
  fallback behaviour)
- Narrow bare except Exception to (OSError, json.JSONDecodeError) and
  (ValueError, TypeError) in load_conversation_settings
- Add Robot Framework integration tests (tui_conversation_pruning.robot
  + helper) covering prune-trigger, note-inserted, protected-blocks,
  clear-resets-state, settings-defaults
- Add ASV benchmarks (conversation_stream_bench.py) covering add_block
  baseline, heavy-load pruning, render after prune, clear+repopulate
- Add CHANGELOG.md entry for feat(tui) conversation content pruning

ISSUES CLOSED: #6350
This commit is contained in:
2026-05-31 13:12:47 -04:00
committed by drew
parent b5a3b3c84f
commit c5df00c6d3
7 changed files with 355 additions and 20 deletions
+99
View File
@@ -0,0 +1,99 @@
"""ASV benchmarks for ConversationStream pruning performance.
Measures the performance of:
- ConversationStream.add_block() with no pruning (baseline)
- ConversationStream.add_block() under heavy load (pruning fires on every block)
- ConversationStream.render() on a pruned stream
- ConversationStream.clear() followed by repopulation
"""
from __future__ import annotations
import sys
from pathlib import Path
try:
from cleveragents.tui.conversation import ConversationSettings, ConversationStream
except ModuleNotFoundError:
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from cleveragents.tui.conversation import ConversationSettings, ConversationStream
def _make_stream(
prune_low_mark: int = 1_500,
prune_excess: int = 1_000,
preserve_recent_lines: int = 500,
) -> ConversationStream:
settings = ConversationSettings(
prune_low_mark=prune_low_mark,
prune_excess=prune_excess,
preserve_recent_lines=preserve_recent_lines,
)
return ConversationStream(settings=settings)
_BLOCK_10_LINES = "\n".join(f"bench line {i}" for i in range(10))
_BLOCK_50_LINES = "\n".join(f"bench line {i}" for i in range(50))
class ConversationStreamAddBlockSuite:
"""Benchmark ConversationStream.add_block under varying load."""
def setup(self):
# Pre-built stream near (but not at) the default trigger threshold.
self.stream_near_threshold = _make_stream()
for i in range(240):
self.stream_near_threshold.add_block(_BLOCK_10_LINES, block_type="message")
# (240 * 10 = 2400 lines, trigger = 2500 — still below threshold)
def time_add_block_no_prune(self):
"""Baseline: add a single block to an empty stream (no pruning)."""
stream = _make_stream()
stream.add_block(_BLOCK_10_LINES, block_type="message")
def time_add_block_triggers_prune(self):
"""Add one block past the threshold so pruning fires once."""
stream = _make_stream()
for i in range(241):
stream.add_block(_BLOCK_10_LINES, block_type="message")
def time_add_block_heavy_load(self):
"""Add 500 blocks to a tight-threshold stream (pruning fires repeatedly)."""
stream = _make_stream(prune_low_mark=100, prune_excess=50)
for _ in range(500):
stream.add_block(_BLOCK_10_LINES, block_type="message")
def time_add_block_large_blocks(self):
"""Add 50-line blocks; fewer blocks needed to trigger pruning."""
stream = _make_stream(prune_low_mark=100, prune_excess=50)
for _ in range(100):
stream.add_block(_BLOCK_50_LINES, block_type="message")
class ConversationStreamRenderSuite:
"""Benchmark ConversationStream.render() after pruning."""
def setup(self):
self.stream = _make_stream(prune_low_mark=100, prune_excess=50)
for i in range(200):
self.stream.add_block(_BLOCK_10_LINES, block_type="message")
def time_render_pruned_stream(self):
"""Benchmark render() on a stream that has been pruned."""
self.stream.render()
class ConversationStreamClearSuite:
"""Benchmark ConversationStream.clear() and repopulation."""
def setup(self):
self.block = _BLOCK_10_LINES
def time_clear_and_repopulate(self):
"""Benchmark clear() followed by adding 50 blocks."""
stream = _make_stream()
for _ in range(50):
stream.add_block(self.block, block_type="message")
stream.clear()
for _ in range(50):
stream.add_block(self.block, block_type="message")