tmux human text compression.

Author: joaopauloschuler

src/smolagents/bp_tools_tmux.py

@@ -23,6 +23,7 @@
 import subprocess
 import time
 
+from .bp_utils_readable_compress import readable_compress
 from .tools import Tool
 
 
@@ -205,14 +206,21 @@ class TmuxReadTool(Tool):
             "description": "If True (default), return only new lines since the last read. If False, return the full tail.",
             "nullable": True,
         },
+        "compress": {
+            "type": "boolean",
+            "description": "If True (default), apply readable compression to the output (strip ANSI, collapse duplicates, remove progress lines, strip timestamps). Defaults to True.",
+            "nullable": True,
+        },
     }
     output_type = "string"
 
-    def forward(self, session_name: str, lines: int | None = None, incremental: bool | None = None) -> str:
+    def forward(self, session_name: str, lines: int | None = None, incremental: bool | None = None, compress: bool | None = None) -> str:
         if lines is None:
             lines = 20
         if incremental is None:
             incremental = True
+        if compress is None:
+            compress = True
         lines = min(lines, _MAX_READ_LINES)
         full = _full_name(session_name)
         # Always capture the full scrollback; slice in Python afterwards.
@@ -263,7 +271,10 @@ def forward(self, session_name: str, lines: int | None = None, incremental: bool
         # Cap at max requested lines.
         new_lines = new_lines[-lines:]
         _last_read[session_name] = _make_fingerprint(all_lines)
-        return "\n".join(new_lines)
+        text = "\n".join(new_lines)
+        if compress:
+            text = readable_compress(text)
+        return text
 
 
 # ======================================================================

src/smolagents/bp_utils_readable_compress.py

@@ -0,0 +1,172 @@
+# BPSA - Beyond Python SmolAgents
+# https://github.com/joaopauloschuler/beyond-python-smolagents
+#
+# Copyright (c) 2024-2026 Joao Paulo Schwarz Schuler and others.
+# Refer to the git commit history for individual authorship.
+# Licensed under the Apache License, Version 2.0
+
+"""
+Readable text compression for agent output.
+
+Compresses text while keeping it human-readable. Designed for tmux
+scrollback and similar command output where verbosity wastes context
+tokens without adding information.
+
+Passes (in order):
+1. ANSI escape stripping
+2. Whitespace normalization (trailing spaces, blank-line runs)
+3. Exact consecutive duplicate line collapsing
+4. Progress line removal (keep only the final state)
+5. Timestamp prefix stripping
+"""
+
+import re
+
+# ---------------------------------------------------------------------------
+# 1. ANSI escape stripping
+# ---------------------------------------------------------------------------
+
+# Matches CSI sequences (e.g. colours, cursor moves) and OSC sequences.
+_ANSI_RE = re.compile(r"\x1b\[[0-9;]*[A-Za-z]|\x1b\][^\x07]*\x07|\x1b\[[\d;]*m")
+
+
+def _strip_ansi(text: str) -> str:
+    """Remove ANSI escape sequences."""
+    return _ANSI_RE.sub("", text)
+
+
+# ---------------------------------------------------------------------------
+# 2. Whitespace normalization
+# ---------------------------------------------------------------------------
+
+
+def _normalize_whitespace(lines: list[str]) -> list[str]:
+    """Strip trailing whitespace per line and collapse runs of blank lines."""
+    result: list[str] = []
+    prev_blank = False
+    for line in lines:
+        stripped = line.rstrip()
+        is_blank = stripped == ""
+        if is_blank and prev_blank:
+            continue
+        result.append(stripped)
+        prev_blank = is_blank
+    return result
+
+
+# ---------------------------------------------------------------------------
+# 3. Exact consecutive duplicate collapsing
+# ---------------------------------------------------------------------------
+
+
+def _collapse_repeated_lines(lines: list[str]) -> list[str]:
+    """Collapse runs of identical consecutive lines."""
+    if not lines:
+        return lines
+    result: list[str] = []
+    run_line = lines[0]
+    run_count = 1
+    for line in lines[1:]:
+        if line == run_line:
+            run_count += 1
+        else:
+            _flush_run(result, run_line, run_count)
+            run_line = line
+            run_count = 1
+    _flush_run(result, run_line, run_count)
+    return result
+
+
+def _flush_run(result: list[str], line: str, count: int) -> None:
+    """Append a run to *result*, summarising if count > 2."""
+    if count <= 2:
+        result.extend([line] * count)
+    else:
+        result.append(line)
+        result.append(f"  ... (repeated {count - 1} more times)")
+
+
+# ---------------------------------------------------------------------------
+# 4. Progress line removal
+# ---------------------------------------------------------------------------
+
+# Matches lines that look like progress indicators:
+#   - contain a percentage  (e.g. " 45%", "100%")
+#   - or a progress bar     (e.g. [####    ], [=====>  ])
+#   - or common spinners    (|, /, -, \)
+_PERCENT_RE = re.compile(r"\d{1,3}%")
+_PROGRESS_BAR_RE = re.compile(r"\[[\s#=\->]+\]")
+_SPINNER_RE = re.compile(r"^[\s]*[|/\\\-][\s]*$")
+
+
+def _collapse_progress_lines(lines: list[str]) -> list[str]:
+    """Remove intermediate progress lines, keeping only the last in a run."""
+    if not lines:
+        return lines
+    result: list[str] = []
+    progress_run: list[str] = []
+    for line in lines:
+        if _is_progress_line(line):
+            progress_run.append(line)
+        else:
+            if progress_run:
+                result.append(progress_run[-1])
+                progress_run = []
+            result.append(line)
+    if progress_run:
+        result.append(progress_run[-1])
+    return result
+
+
+def _is_progress_line(line: str) -> bool:
+    """Heuristic: does *line* look like a progress indicator?"""
+    if _SPINNER_RE.match(line):
+        return True
+    if _PERCENT_RE.search(line) and _PROGRESS_BAR_RE.search(line):
+        return True
+    return False
+
+
+# ---------------------------------------------------------------------------
+# 5. Timestamp prefix stripping
+# ---------------------------------------------------------------------------
+
+# Common log timestamp patterns at the start of a line:
+#   2026-04-10T12:00:01.123Z  |  2026-04-10 12:00:01  |  [12:00:01]  |
+#   Apr 10 12:00:01
+_TIMESTAMP_RE = re.compile(
+    r"^("
+    r"\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}[\.\d]*[Z]?\s*"
+    r"|\[\d{2}:\d{2}:\d{2}\]\s*"
+    r"|[A-Z][a-z]{2}\s+\d{1,2}\s+\d{2}:\d{2}:\d{2}\s*"
+    r")"
+)
+
+
+def _strip_timestamps(lines: list[str]) -> list[str]:
+    """Strip common timestamp prefixes and prepend a note if any were found."""
+    stripped: list[str] = []
+    count = 0
+    for line in lines:
+        new_line, n = _TIMESTAMP_RE.subn("", line, count=1)
+        count += n
+        stripped.append(new_line)
+    if count > 0:
+        stripped.insert(0, f"[timestamps stripped from {count} lines]")
+    return stripped
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+
+def readable_compress(text: str) -> str:
+    """Apply all readable compression passes and return compressed text."""
+    text = _strip_ansi(text)
+    lines = text.splitlines()
+    lines = _normalize_whitespace(lines)
+    lines = _collapse_repeated_lines(lines)
+    lines = _collapse_progress_lines(lines)
+    lines = _strip_timestamps(lines)
+    return "\n".join(lines)

tests/test_bp_utils_readable_compress.py

@@ -0,0 +1,175 @@
+#!/usr/bin/env python3
+"""
+Unit tests for bp_utils_readable_compress.
+"""
+
+import os
+import sys
+
+import pytest
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src'))
+
+from smolagents.bp_utils_readable_compress import (
+    readable_compress,
+    _strip_ansi,
+    _normalize_whitespace,
+    _collapse_repeated_lines,
+    _collapse_progress_lines,
+    _strip_timestamps,
+)
+
+
+# ── ANSI stripping ──────────────────────────────────────────────────────
+
+class TestStripAnsi:
+    def test_removes_colour_codes(self):
+        assert _strip_ansi("\x1b[31mERROR\x1b[0m") == "ERROR"
+
+    def test_removes_cursor_move(self):
+        assert _strip_ansi("\x1b[2Ahello") == "hello"
+
+    def test_plain_text_unchanged(self):
+        assert _strip_ansi("hello world") == "hello world"
+
+    def test_osc_sequence(self):
+        assert _strip_ansi("\x1b]0;title\x07rest") == "rest"
+
+
+# ── Whitespace normalization ────────────────────────────────────────────
+
+class TestNormalizeWhitespace:
+    def test_strips_trailing_spaces(self):
+        assert _normalize_whitespace(["hello   ", "world  "]) == ["hello", "world"]
+
+    def test_collapses_blank_runs(self):
+        lines = ["a", "", "", "", "b"]
+        assert _normalize_whitespace(lines) == ["a", "", "b"]
+
+    def test_single_blank_preserved(self):
+        lines = ["a", "", "b"]
+        assert _normalize_whitespace(lines) == ["a", "", "b"]
+
+
+# ── Duplicate collapsing ────────────────────────────────────────────────
+
+class TestCollapseRepeatedLines:
+    def test_no_repeats(self):
+        lines = ["a", "b", "c"]
+        assert _collapse_repeated_lines(lines) == ["a", "b", "c"]
+
+    def test_two_repeats_kept(self):
+        lines = ["a", "a", "b"]
+        assert _collapse_repeated_lines(lines) == ["a", "a", "b"]
+
+    def test_many_repeats_collapsed(self):
+        lines = ["x"] * 50
+        result = _collapse_repeated_lines(lines)
+        assert len(result) == 2
+        assert result[0] == "x"
+        assert "49 more" in result[1]
+
+    def test_empty_input(self):
+        assert _collapse_repeated_lines([]) == []
+
+    def test_mixed_runs(self):
+        lines = ["a", "a", "a", "b", "b", "c"]
+        result = _collapse_repeated_lines(lines)
+        assert result == ["a", "  ... (repeated 2 more times)", "b", "b", "c"]
+
+
+# ── Progress line removal ──────────────────────────────────────────────
+
+class TestCollapseProgressLines:
+    def test_percentage_with_bar(self):
+        lines = [
+            "Downloading [##        ] 20%",
+            "Downloading [#####     ] 50%",
+            "Downloading [##########] 100%",
+            "Done.",
+        ]
+        result = _collapse_progress_lines(lines)
+        assert result == ["Downloading [##########] 100%", "Done."]
+
+    def test_spinner_lines(self):
+        lines = ["|", "/", "-", "\\", "finished"]
+        result = _collapse_progress_lines(lines)
+        assert result == ["\\", "finished"]
+
+    def test_normal_lines_unchanged(self):
+        lines = ["compiling foo.c", "compiling bar.c"]
+        assert _collapse_progress_lines(lines) == lines
+
+    def test_percentage_without_bar_not_matched(self):
+        lines = ["Test passed: 100% coverage"]
+        assert _collapse_progress_lines(lines) == lines
+
+    def test_empty(self):
+        assert _collapse_progress_lines([]) == []
+
+
+# ── Timestamp stripping ─────────────────────────────────────────────────
+
+class TestStripTimestamps:
+    def test_iso_timestamp(self):
+        lines = ["2026-04-10T12:00:01Z hello", "2026-04-10T12:00:02Z world"]
+        result = _strip_timestamps(lines)
+        assert result[0].startswith("[timestamps stripped")
+        assert "hello" in result[1]
+        assert "world" in result[2]
+
+    def test_bracketed_time(self):
+        lines = ["[12:00:01] info message"]
+        result = _strip_timestamps(lines)
+        assert len(result) == 2
+        assert "info message" in result[1]
+
+    def test_syslog_timestamp(self):
+        lines = ["Apr 10 12:00:01 server msg"]
+        result = _strip_timestamps(lines)
+        assert "server msg" in result[1]
+
+    def test_no_timestamps(self):
+        lines = ["hello", "world"]
+        result = _strip_timestamps(lines)
+        assert result == ["hello", "world"]
+
+
+# ── Full pipeline ───────────────────────────────────────────────────────
+
+class TestReadableCompress:
+    def test_full_pipeline(self):
+        text = (
+            "\x1b[32m2026-04-10T12:00:01Z Starting build\x1b[0m\n"
+            "2026-04-10T12:00:02Z Compiling\n"
+            "Downloading [##        ] 20%\n"
+            "Downloading [#####     ] 50%\n"
+            "Downloading [##########] 100%\n"
+            "line\n"
+            "line\n"
+            "line\n"
+            "line\n"
+            "line\n"
+            "\n"
+            "\n"
+            "\n"
+            "done"
+        )
+        result = readable_compress(text)
+        # ANSI stripped
+        assert "\x1b" not in result
+        # Timestamps stripped
+        assert "[timestamps stripped" in result
+        # Progress collapsed — only 100% kept
+        assert "20%" not in result
+        assert "100%" in result
+        # Duplicate lines collapsed
+        assert "repeated" in result
+        # Blank line run collapsed
+        assert "\n\n\n" not in result
+        # Content preserved
+        assert "done" in result
+
+    def test_plain_text_passthrough(self):
+        text = "hello\nworld"
+        assert readable_compress(text) == "hello\nworld"