feat(eco): /eco — RTK-style token compression for Bash tool output

src/command_system/__init__.py

@@ -76,6 +76,7 @@
 from .output_style_command import OUTPUT_STYLE_COMMAND, OutputStyleCommand
 from .export_command import EXPORT_COMMAND, ExportCommand
 from .theme_command import THEME_COMMAND, ThemeCommand
+from .eco_command import ECO_COMMAND, eco_command_call
 from .effort_command import EFFORT_COMMAND, EffortCommand
 from .model_command import MODEL_COMMAND, ModelCommand
 from .logo_command import LOGO_COMMAND, LogoCommand
@@ -165,6 +166,8 @@
     "ExportCommand",
     "THEME_COMMAND",
     "ThemeCommand",
+    "ECO_COMMAND",
+    "eco_command_call",
     "EFFORT_COMMAND",
     "EffortCommand",
     "MODEL_COMMAND",

src/command_system/builtins.py

@@ -29,6 +29,7 @@
 from .security_review import SECURITY_REVIEW_COMMAND
 from .statusline import STATUSLINE_COMMAND
 from .theme_command import THEME_COMMAND
+from .eco_command import ECO_COMMAND
 from .effort_command import EFFORT_COMMAND
 from .model_command import MODEL_COMMAND
 from .logo_command import LOGO_COMMAND
@@ -1331,6 +1332,7 @@ def get_builtin_commands() -> list[Command]:
         OUTPUT_STYLE_COMMAND,
         EXPORT_COMMAND,
         THEME_COMMAND,
+        ECO_COMMAND,
         EFFORT_COMMAND,
         MODEL_COMMAND,
         LOGO_COMMAND,

src/command_system/eco_command.py

@@ -0,0 +1,100 @@
+"""/eco — toggle RTK-style token compression of Bash tool output.
+
+Session-scoped switch over :mod:`src.eco` (the same process-global session
+state shape as ``/effort ultracode``). When on, the wire rendering of Bash
+results is compressed by deterministic filters (test-runner failure focus,
+noise stripping, log dedup, recoverable head caps) with the raw output teed
+per session. The TUI transcript renders the same compact string the model
+sees (there is no separate raw display path for Bash); the raw
+stdout/stderr fields still ride the output dict and the tee files. Exit
+codes and error semantics are untouched.
+
+Grammar: bare ``/eco`` toggles; ``on`` / ``off`` are explicit;
+``status``/``stats`` reports the switch plus session savings. Unknown args →
+usage. All paths are headless-safe (no UI surface needed).
+"""
+
+from __future__ import annotations
+
+from .types import CommandContext, LocalCommand, LocalCommandResult
+
+_USAGE = (
+    "Usage: /eco [on|off|status]\n\n"
+    "Compresses Bash tool output before it reaches the model (60-90% fewer\n"
+    "tokens on test runs, installs, and noisy logs). The transcript shows\n"
+    "the same compact rendering; whenever content is capped or summarized,\n"
+    "the full raw output is saved per session and referenced via\n"
+    "[full output: ...] hints, so nothing is unrecoverable."
+)
+
+_ON_MSG = (
+    "Eco mode on: Bash output is compressed before reaching the model "
+    "(test failures kept, noise stripped, long output capped) — the "
+    "transcript shows the same compact rendering. Capped or summarized "
+    "content is saved per session and linked via [full output: ...] hints. "
+    "/eco off to disable."
+)
+
+_OFF_MSG = "Eco mode off: Bash output reaches the model unmodified."
+
+
+def _status_text() -> str:
+    from src.eco import eco_stats, is_eco_session
+
+    stats = eco_stats()
+    state = "on" if is_eco_session() else "off"
+    lines = [f"Eco mode: {state}"]
+    if stats.commands:
+        lines.append(
+            f"  Compressed {stats.commands} command output(s): "
+            f"~{stats.baseline_tokens:,} → ~{stats.eco_tokens:,} tokens "
+            f"(saved ~{stats.saved_tokens:,}, {stats.savings_pct:.0f}%)"
+        )
+        for name, (uses, saved) in sorted(
+            stats.by_filter.items(), key=lambda kv: -kv[1][1]
+        ):
+            lines.append(f"    {name}: {uses} use(s), ~{saved:,} tokens saved")
+    else:
+        lines.append("  No compressions recorded this session yet.")
+    return "\n".join(lines)
+
+
+def eco_command_call(args: str, context: CommandContext) -> LocalCommandResult:
+    """Handle /eco — toggle, explicit on/off, or status."""
+    from src.eco import is_eco_session, set_eco_session
+
+    arg = (args or "").strip().lower()
+
+    if arg in ("status", "stats"):
+        return LocalCommandResult(type="text", value=_status_text())
+    if arg in ("help", "-h", "--help"):
+        return LocalCommandResult(type="text", value=_USAGE)
+
+    if arg == "":
+        target = not is_eco_session()
+    elif arg in ("on", "enable", "true", "1"):
+        target = True
+    elif arg in ("off", "disable", "false", "0"):
+        target = False
+    else:
+        return LocalCommandResult(
+            type="text", value=f"Unknown argument: {args.strip()}\n\n{_USAGE}"
+        )
+
+    set_eco_session(target)
+    if target:
+        return LocalCommandResult(type="text", value=_ON_MSG)
+    # Turning off keeps the stats (still reportable via /eco status).
+    return LocalCommandResult(type="text", value=f"{_OFF_MSG}\n{_status_text()}")
+
+
+ECO_COMMAND = LocalCommand(
+    name="eco",
+    description="Toggle Bash-output token compression (RTK-style)",
+    argument_hint="[on|off|status]",
+    supports_non_interactive=True,
+)
+ECO_COMMAND.set_call(eco_command_call)
+
+
+__all__ = ["ECO_COMMAND", "eco_command_call"]

src/eco/__init__.py

@@ -0,0 +1,35 @@
+"""eco — RTK-style token compression for Bash tool output.
+
+Ported methods from the RTK reference analysis (my-docs/token-compression/RTK/):
+deterministic, filter-based compression of the *model-bound* rendering of Bash
+tool results, guarded so it can never make things worse and never lose data
+unrecoverably. Toggled per session with the ``/eco`` slash command.
+
+Layering (mirrors RTK's core): pure string filters (:mod:`filters`), a
+never-worse guard + chars/4 estimator (:mod:`guard`), raw-output tee recovery
+(:mod:`tee`), session state + savings stats (:mod:`state`), and the dispatch
+engine (:mod:`engine`). Nothing in this package imports tool_system — the Bash
+tool calls in, not the other way around.
+"""
+
+from .engine import EcoOutcome, compress_bash_output
+from .guard import estimate_tokens, never_worse
+from .state import (
+    eco_stats,
+    is_eco_session,
+    record_compression,
+    reset_eco,
+    set_eco_session,
+)
+
+__all__ = [
+    "EcoOutcome",
+    "compress_bash_output",
+    "estimate_tokens",
+    "never_worse",
+    "eco_stats",
+    "is_eco_session",
+    "record_compression",
+    "reset_eco",
+    "set_eco_session",
+]

src/eco/engine.py

@@ -0,0 +1,149 @@
+"""Dispatch engine: raw Bash output → compressed model-bound rendering.
+
+Order of operations per command (RTK runner/emit_guarded ported to our
+harness position):
+
+1. Normalize the *full* pre-truncation output (``\\r`` frames, ANSI) — this
+   processed text is both the filter input and the tee payload, so
+   ``tail -n +N`` hints line up exactly with what the filter counted.
+2. First matching filter wins; no match → ``None`` (passthrough — the caller
+   ships its baseline untouched and nothing is recorded).
+3. Loss accounting: ``safe_loss`` hits (pure-ceremony strips) ship as-is;
+   every other hit must tee the processed text and append a recovery hint —
+   tee unavailable → the hit is DISCARDED (RTK's never-lossy-without-recovery
+   rule, main.rs:1341).
+4. ``never_worse`` against the exact baseline the mapper would otherwise
+   emit; baseline wins → passthrough.
+5. Record savings (only real compressions — passthroughs never dilute stats).
+
+Any exception → passthrough (a compressor must never break the tool).
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass
+from pathlib import Path
+
+from .filters import FILTERS, normalize_carriage_returns, strip_ansi
+from .guard import estimate_tokens, never_worse
+from .state import record_compression
+from .tee import full_hint, tail_hint, tee_raw
+
+logger = logging.getLogger(__name__)
+
+# A "compressed" rendering larger than this is a filter bug; the existing
+# bash truncate_output contract stays intact above us.
+_MAX_ECO_CHARS = 30_000
+
+# Inputs beyond this skip eco entirely (RTK RAW_CAP spirit): the regex passes
+# would be pure cost, and the Step-11 <persisted-output> layer already gives
+# giant results a preview + full-file pointer.
+_MAX_INPUT_CHARS = 10_485_760
+
+# Conservative allowance for a recovery-hint line when pre-checking the guard
+# before writing the tee file (paths are ~60-120 chars → ~30 tokens).
+_HINT_TOKEN_ALLOWANCE = 30
+
+
+@dataclass(frozen=True)
+class EcoOutcome:
+    content: str
+    filter_name: str
+    baseline_tokens: int
+    eco_tokens: int
+
+    @property
+    def saved_tokens(self) -> int:
+        return max(0, self.baseline_tokens - self.eco_tokens)
+
+
+def _slug_for(command: str) -> str:
+    return "_".join(command.strip().split())[:40] or "cmd"
+
+
+def compress_bash_output(
+    command: str,
+    exit_code: int,
+    full_text: str,
+    baseline: str,
+    tee_dir: Path | None,
+) -> EcoOutcome | None:
+    """Compress one Bash result; ``None`` means "ship the baseline".
+
+    ``full_text`` is the pre-truncation stdout+stderr assembly (the mapper's
+    shape, but unbounded); ``baseline`` is exactly what the mapper would emit
+    without eco. ``tee_dir`` is the per-session eco directory (None → only
+    safe-loss filters can fire).
+    """
+    try:
+        if not baseline.strip():
+            return None
+        if len(full_text) > _MAX_INPUT_CHARS:
+            return None
+
+        processed = strip_ansi(normalize_carriage_returns(full_text))
+        baseline_tokens = estimate_tokens(baseline)
+
+        for filt in FILTERS:
+            try:
+                hit = filt(command, exit_code, processed)
+            except Exception:  # noqa: BLE001 — one bad filter must not break the chain
+                logger.debug("[eco] filter %r failed", filt, exc_info=True)
+                continue
+            if hit is None:
+                continue
+            if not hit.body.strip():
+                # Filters must never produce empty output (the downstream
+                # empty-content marker would misreport "no output").
+                continue
+
+            body = hit.body
+            if not hit.safe_loss:
+                if tee_dir is None:
+                    continue
+                # Guard pre-check BEFORE writing: if the body plus a typical
+                # hint can't beat the baseline, don't leave an orphan file.
+                if (
+                    estimate_tokens(body) + _HINT_TOKEN_ALLOWANCE
+                    > baseline_tokens
+                ):
+                    continue
+                path = tee_raw(processed, _slug_for(command), tee_dir)
+                if path is None:
+                    # Tiny content (< MIN_TEE_SIZE) or write failure: loss
+                    # would be unrecoverable → discard the hit.
+                    continue
+                hint = (
+                    tail_hint(path, hit.tail_offset)
+                    if hit.tail_offset is not None
+                    else full_hint(path)
+                )
+                body = f"{body}\n{hint}"
+                if (
+                    len(body) > _MAX_ECO_CHARS
+                    or never_worse(baseline, body) == baseline
+                ):
+                    # The real hint pushed it over after all — remove the
+                    # now-unreferenced tee file and pass through.
+                    try:
+                        path.unlink(missing_ok=True)
+                    except OSError:
+                        pass
+                    continue
+            else:
+                if len(body) > _MAX_ECO_CHARS or never_worse(baseline, body) == baseline:
+                    continue
+
+            eco_tokens = estimate_tokens(body)
+            record_compression(hit.name, baseline_tokens, eco_tokens)
+            return EcoOutcome(
+                content=body,
+                filter_name=hit.name,
+                baseline_tokens=baseline_tokens,
+                eco_tokens=eco_tokens,
+            )
+        return None
+    except Exception:  # noqa: BLE001 — the engine must never break the Bash tool
+        logger.debug("[eco] engine failed; passing through", exc_info=True)
+        return None