diff --git a/hooks/hermes/README.md b/hooks/hermes/README.md index 2a755c710..7e95bc421 100644 --- a/hooks/hermes/README.md +++ b/hooks/hermes/README.md @@ -1,6 +1,9 @@ # RTK Plugin for Hermes -Rewrites Hermes `terminal` tool commands to RTK equivalents before execution, so Hermes receives compact command output without changing your workflow. +Integrates RTK into [Hermes Agent](https://github.com/NousResearch/hermes-agent) via two hooks: + +1. **`pre_tool_call`** — rewrites bare terminal commands to their `rtk` equivalents before execution. +2. **`transform_tool_result`** — filters high-token tool outputs before they enter the conversation context. ## Installation @@ -12,32 +15,90 @@ The installer writes the plugin to `~/.hermes/plugins/rtk-rewrite/` and enables ## Development -Run the Hermes plugin tests from the repository root: +Run the plugin tests from the repository root: ```bash -python3 -m unittest discover -s hooks/hermes +python3 -m pytest hooks/hermes/tests/ -v ``` -## How it works +--- -Hermes loads plugins from Python, so the plugin entrypoint is Python. The Python code is only a thin Hermes adapter. It reads the Hermes terminal tool payload, calls `rtk rewrite` for the actual command decision, then mutates the terminal tool `command` before Hermes executes it. +## Hook 1 — `pre_tool_call`: command rewriting -All rewrite rules stay in Rust inside `rtk rewrite`. When RTK adds or changes command rewrite behavior, the Hermes plugin picks up that behavior by delegating to the RTK binary. +When a Hermes worker calls the `terminal` tool with a bare command like `find`, `grep`, `cat`, or `ls`, this hook rewrites it to its `rtk` equivalent before execution — identical to the Claude Code `PreToolUse` hook, but applied inside Hermes so all backends (local, Docker, SSH, Modal) benefit without requiring the model to prefix commands manually. -## Fail-open behavior +### How it works + +The plugin reads the Hermes `terminal` tool payload, calls `rtk rewrite `, and mutates the `command` field if RTK provides a rewrite. All rewrite rules stay in Rust inside `rtk rewrite` — when RTK adds or changes rewrite behavior, the Hermes plugin picks it up automatically. + +### Examples + +| Worker command | Executed as | +|---|---| +| `find /home -name '*.py'` | `rtk find /home -name '*.py'` | +| `grep -rn foo /src` | `rtk grep -rn foo /src` | +| `cat large_file.yaml` | `rtk read large_file.yaml` | +| `ls -la /some/dir` | `rtk ls -la /some/dir` | +| `cd /foo && find . -name '*.cs'` | `cd /foo && rtk find . -name '*.cs'` | +| `python3 script.py` | `python3 script.py` *(unchanged)* | +| `rtk find ...` | `rtk find ...` *(no double-wrapping)* | + +--- + +## Hook 2 — `transform_tool_result`: output filtering + +Filters the output of high-token tools before it enters the conversation context. Each filter is independent and fails open — if parsing fails, the original output is returned unchanged. + +### Filters + +| Tool(s) | What is filtered | Limit | +|---|---|---| +| `terminal` | JSON-unwrap output field, deduplicate lines, strip blanks | 100 lines | +| `browser_navigate` | Strip `Layout*` accessibility-tree noise, deduplicate, truncate snapshot | 120 lines | +| `read_file`, `execute_code` | Deduplicate lines, strip blank lines | 100 lines | +| `write_file`, `patch` | Replace full unified diff with a 1-line summary | — | +| `search_files` | Truncate to first 50 results | 50 results | +| `kanban_show`, `kanban_list` | Keep only the last 10 events | 10 events | + +### `write_file` / `patch` diff summary + +A 50-line unified diff like: -The plugin does not block command execution. If anything goes wrong, Hermes runs the original command unchanged. +```diff +--- a/src/Battle/Pig3DAnimator.cs ++++ b/src/Battle/Pig3DAnimator.cs +@@ -42,7 +42,6 @@ + ... +``` + +Is replaced with: + +``` +[patch] src/Battle/Pig3DAnimator.cs (2 hunks, +5/-3 lines) +``` + +This is the main fix for context overflow on patch-heavy tasks where multiple diffs accumulate in the conversation. -If rtk is not available in PATH when Hermes loads the plugin, the plugin prints a warning and skips hook registration. +### `browser_navigate` noise stripping -- `rtk` is missing from `PATH` -- `rtk rewrite` exits with an error -- Hermes sends a non-terminal tool call -- The tool payload has no string `command` +Unity accessibility snapshots and browser pages often contain dozens of `LayoutBlock`, `LayoutInline`, `LayoutTableRow`, etc. lines that carry no semantic content. These are stripped before the snapshot enters the model context, typically reducing browser tool output by ~21%. + +--- + +## Fail-open behavior + +The plugin never blocks command execution or tool result delivery. In all of the following cases, Hermes proceeds with the original value unchanged: + +- `rtk` is missing from `PATH` (hook not registered; one warning logged) +- `rtk rewrite` exits with an unexpected error +- A tool payload has no string `command` +- JSON parsing of a tool result fails - The plugin raises an unexpected exception +--- + ## Limitations -- Only Hermes `terminal` tool calls are rewritten. -- Commands skipped by `rtk rewrite` stay unchanged, including commands already prefixed with `rtk`, compound shell commands, heredocs, and commands without an RTK filter. -- Shell hooks are not used for Hermes command rewriting. The integration depends on Hermes loading Python plugins and passing a mutable terminal tool payload. +- Only the `terminal` tool call is rewritten by `pre_tool_call`. Other tools (file reads, browser, etc.) are not affected. +- Shell hooks are not used for command rewriting. The integration depends on Hermes loading Python plugins and passing a mutable terminal tool payload. +- `transform_tool_result` filters apply to the tool output string only. Structured metadata fields (e.g. `exit_code`) are preserved. diff --git a/hooks/hermes/rtk-rewrite/__init__.py b/hooks/hermes/rtk-rewrite/__init__.py index 6dcf44e83..f01ccc2d3 100644 --- a/hooks/hermes/rtk-rewrite/__init__.py +++ b/hooks/hermes/rtk-rewrite/__init__.py @@ -1,9 +1,12 @@ -"""Hermes plugin adapter for RTK command rewriting. +"""Hermes plugin adapter for RTK command rewriting and output filtering. All rewrite logic lives in RTK's Rust ``rtk rewrite`` command; this module -only bridges Hermes ``pre_tool_call`` payloads to that command and fails open. +also provides a pure-Python ``transform_tool_result`` filter for high-token +tools (browser_navigate, execute_code, read_file, terminal). """ +import json +import re import shutil import subprocess import sys @@ -14,13 +17,26 @@ _rtk_available = None _rtk_missing_warned = False +# Accessibility-tree noise patterns produced by browser_navigate snapshots. +_LAYOUT_NOISE = re.compile( + r"^\s*-?\s*Layout(?:Table(?:Row|Cell|Section)?|Inline[A-Za-z]*|Block[A-Za-z]*)" + r"(?:\s+\"[^\"]{0,60}\"\s*)?$" +) + +_SNAPSHOT_MAX_LINES = 120 +_TEXT_MAX_LINES = 100 +_SEARCH_MAX_RESULTS = 50 +_KANBAN_MAX_EVENTS = 10 +_KANBAN_MAX_COMMENTS = 5 + def register(ctx): - """Register the Hermes pre-tool callback.""" + """Register the Hermes pre-tool and transform-result callbacks.""" if not _check_rtk(): return ctx.register_hook("pre_tool_call", _pre_tool_call) + ctx.register_hook("transform_tool_result", _transform_tool_result) def _check_rtk(): @@ -76,5 +92,222 @@ def _pre_tool_call(tool_name=None, args=None, **_kwargs): return +def _transform_tool_result(tool_name=None, result=None, **_kwargs): + """Filter high-token tool outputs before they enter the conversation context.""" + if tool_name == "browser_navigate": + return _filter_browser(result) + if tool_name == "terminal": + return _filter_terminal(result) + if tool_name in ("execute_code", "read_file"): + return _filter_text(result) + if tool_name in ("write_file", "patch"): + return _filter_diff(result) + if tool_name == "search_files": + return _filter_search_files(result) + if tool_name in ("kanban_show", "kanban_list"): + return _filter_kanban(result) + + +def _filter_browser(result): + """Strip Layout* accessibility noise from browser_navigate JSON and truncate snapshot.""" + if not isinstance(result, str) or not result.strip(): + return + before = len(result) + try: + data = json.loads(result) + except (json.JSONDecodeError, ValueError): + return _filter_text(result) + + snapshot = data.get("snapshot", "") + if snapshot: + lines = snapshot.splitlines() + lines = [l for l in lines if l.strip() and not _LAYOUT_NOISE.match(l)] + seen, deduped = set(), [] + for l in lines: + key = l.strip() + if key not in seen: + seen.add(key) + deduped.append(l) + if len(deduped) > _SNAPSHOT_MAX_LINES: + deduped = deduped[:_SNAPSHOT_MAX_LINES] + [f"... [{len(lines) - _SNAPSHOT_MAX_LINES} lines truncated]"] + data["snapshot"] = "\n".join(deduped) + + # Drop noisy metadata fields + for key in ("stealth_warning", "stealth_features"): + data.pop(key, None) + + filtered = json.dumps(data, ensure_ascii=False) + after = len(filtered) + if after < before: + print( + f"rtk: filtered browser_navigate: {before} -> {after} chars" + f" ({100*(before-after)//before}% reduction)", + file=sys.stderr, + ) + return filtered + + +def _filter_terminal(result): + """Filter terminal tool output: parse JSON wrapper, filter the output field, repack.""" + if not isinstance(result, str) or not result.strip(): + return + before = len(result) + try: + data = json.loads(result) + except (json.JSONDecodeError, ValueError): + return _filter_text(result) + + output = data.get("output", "") + if not output: + return + + filtered_output = _filter_text(output) + if filtered_output is None: + return + + data["output"] = filtered_output + filtered = json.dumps(data, ensure_ascii=False) + after = len(filtered) + if after < before: + print( + f"rtk: filtered terminal: {before} -> {after} chars" + f" ({100*(before-after)//before}% reduction)", + file=sys.stderr, + ) + return filtered + + +def _filter_text(result): + """Strip blank lines, deduplicate, and truncate generic text outputs.""" + if not isinstance(result, str) or not result.strip(): + return + before = len(result) + lines = result.splitlines() + seen, deduped = set(), [] + for l in lines: + key = l.strip() + if not key: + continue + if key not in seen: + seen.add(key) + deduped.append(l) + if len(deduped) > _TEXT_MAX_LINES: + deduped = deduped[:_TEXT_MAX_LINES] + [f"... [{len(lines) - _TEXT_MAX_LINES} lines truncated]"] + filtered = "\n".join(deduped) + after = len(filtered) + if after < before: + print( + f"rtk: filtered text ({after}/{before} chars, {100*(before-after)//before}% reduction)", + file=sys.stderr, + ) + return filtered + + +def _filter_diff(result): + """Replace unified diff output (write_file / patch) with a compact summary.""" + if not isinstance(result, str) or not result.strip(): + return + before = len(result) + lines = result.splitlines() + added = sum(1 for l in lines if l.startswith("+") and not l.startswith("+++")) + removed = sum(1 for l in lines if l.startswith("-") and not l.startswith("---")) + # Extract file path from diff header (b/path) or first @@ hunk + path = "" + for l in lines: + if l.startswith("+++ b/"): + path = l[6:].strip() + break + if l.startswith("+++ "): + path = l[4:].strip() + break + hunks = sum(1 for l in lines if l.startswith("@@")) + summary = f"[patch] {path} ({hunks} hunk{'s' if hunks != 1 else ''}, +{added}/-{removed} lines)" + after = len(summary) + if after < before: + print( + f"rtk: filtered diff ({path}): {before} -> {after} chars" + f" ({100*(before-after)//before}% reduction)", + file=sys.stderr, + ) + return summary + + +def _filter_search_files(result): + """Truncate search_files output to _SEARCH_MAX_RESULTS results.""" + if not isinstance(result, str) or not result.strip(): + return + before = len(result) + lines = [l for l in result.splitlines() if l.strip()] + if len(lines) <= _SEARCH_MAX_RESULTS: + return + truncated = lines[:_SEARCH_MAX_RESULTS] + [f"... [{len(lines) - _SEARCH_MAX_RESULTS} results truncated]"] + filtered = "\n".join(truncated) + after = len(filtered) + print( + f"rtk: filtered search_files: {before} -> {after} chars" + f" ({100*(before-after)//before}% reduction)", + file=sys.stderr, + ) + return filtered + + +def _filter_kanban(result): + """Truncate kanban_show events list and comments to reduce context bloat.""" + if not isinstance(result, str) or not result.strip(): + return + before = len(result) + lines = result.splitlines() + out = [] + in_events = False + in_comments = False + event_count = 0 + comment_count = 0 + skipped_events = 0 + skipped_comments = 0 + + for l in lines: + stripped = l.strip() + if stripped.startswith("Events ("): + in_events = True + in_comments = False + out.append(l) + continue + if stripped.startswith("Runs (") or stripped.startswith("children:") or stripped.startswith("Body:"): + in_events = False + if skipped_events: + out.append(f" ... [{skipped_events} older events truncated]") + skipped_events = 0 + out.append(l) + continue + if stripped.startswith("Comments (") or (in_comments and stripped.startswith("[")): + in_comments = True + in_events = False + out.append(l) + continue + + if in_events: + if event_count < _KANBAN_MAX_EVENTS: + out.append(l) + event_count += 1 + else: + skipped_events += 1 + continue + + out.append(l) + + if skipped_events: + out.append(f" ... [{skipped_events} older events truncated]") + + filtered = "\n".join(out) + after = len(filtered) + if after < before: + print( + f"rtk: filtered kanban: {before} -> {after} chars" + f" ({100*(before-after)//before}% reduction)", + file=sys.stderr, + ) + return filtered + + def _warn(message): print(f"rtk: hermes plugin warning: {message}", file=sys.stderr) diff --git a/hooks/hermes/rtk-rewrite/plugin.yaml b/hooks/hermes/rtk-rewrite/plugin.yaml index 7a08e40b5..fce6d46dd 100644 --- a/hooks/hermes/rtk-rewrite/plugin.yaml +++ b/hooks/hermes/rtk-rewrite/plugin.yaml @@ -1,8 +1,10 @@ name: rtk-rewrite version: "0.1.0" -description: Rewrite Hermes terminal commands through RTK before execution. +description: Rewrite terminal commands and filter tool outputs through RTK. author: RTK Contributors hooks: - pre_tool_call + - transform_tool_result provides_hooks: - pre_tool_call + - transform_tool_result diff --git a/hooks/hermes/tests/test_transform_tool_result.py b/hooks/hermes/tests/test_transform_tool_result.py new file mode 100644 index 000000000..a7fd4dc00 --- /dev/null +++ b/hooks/hermes/tests/test_transform_tool_result.py @@ -0,0 +1,377 @@ +"""Tests for the transform_tool_result output filters in the RTK Hermes plugin.""" + +import importlib.util +import io +import json +import sys +import unittest +from pathlib import Path +from unittest import mock + +PLUGIN_PATH = Path(__file__).resolve().parents[1] / "rtk-rewrite" / "__init__.py" + + +def load_plugin(): + spec = importlib.util.spec_from_file_location("rtk_rewrite_plugin", PLUGIN_PATH) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def make_terminal_result(output, exit_code=0): + return json.dumps({"output": output, "exit_code": exit_code, "error": ""}) + + +def make_browser_result(snapshot, **extra): + data = {"snapshot": snapshot, "url": "https://example.com"} + data.update(extra) + return json.dumps(data) + + +# --------------------------------------------------------------------------- +# _transform_tool_result dispatch +# --------------------------------------------------------------------------- + +class TransformDispatchTest(unittest.TestCase): + def setUp(self): + self.m = load_plugin() + + def test_unknown_tool_returns_none(self): + self.assertIsNone(self.m._transform_tool_result(tool_name="unknown_tool", result="anything")) + + def test_none_result_returns_none(self): + for tool in ("terminal", "browser_navigate", "read_file", "write_file", "patch", "search_files", "kanban_show"): + with self.subTest(tool=tool): + self.assertIsNone(self.m._transform_tool_result(tool_name=tool, result=None)) + + def test_empty_result_returns_none(self): + for tool in ("terminal", "browser_navigate", "read_file", "search_files"): + with self.subTest(tool=tool): + self.assertIsNone(self.m._transform_tool_result(tool_name=tool, result="")) + + +# --------------------------------------------------------------------------- +# _filter_text +# --------------------------------------------------------------------------- + +class FilterTextTest(unittest.TestCase): + def setUp(self): + self.m = load_plugin() + + def test_deduplicates_repeated_lines(self): + result = self.m._filter_text("foo\nfoo\nbar\nfoo\n") + self.assertEqual("foo\nbar", result) + + def test_strips_blank_lines(self): + result = self.m._filter_text("a\n\n\nb\n\nc\n") + self.assertEqual("a\nb\nc", result) + + def test_truncates_at_100_lines(self): + lines = "\n".join(f"line{i}" for i in range(150)) + result = self.m._filter_text(lines) + out_lines = result.splitlines() + self.assertEqual(101, len(out_lines)) + self.assertIn("truncated", out_lines[-1]) + + def test_no_change_returns_none(self): + text = "\n".join(f"unique{i}" for i in range(10)) + self.assertIsNone(self.m._filter_text(text)) + + def test_whitespace_only_returns_none(self): + self.assertIsNone(self.m._filter_text(" \n\n ")) + + def test_logs_reduction_to_stderr(self): + with mock.patch.object(self.m.sys, "stderr", new_callable=io.StringIO) as err: + self.m._filter_text("dup\ndup\ndup\n") + self.assertIn("rtk: filtered text", err.getvalue()) + + +# --------------------------------------------------------------------------- +# _filter_terminal +# --------------------------------------------------------------------------- + +class FilterTerminalTest(unittest.TestCase): + def setUp(self): + self.m = load_plugin() + + def test_unwraps_json_and_filters_output(self): + raw = make_terminal_result("foo\nfoo\nbar\n") + result = self.m._filter_terminal(raw) + self.assertIsNotNone(result) + data = json.loads(result) + self.assertEqual("foo\nbar", data["output"]) + + def test_preserves_exit_code_and_error_fields(self): + raw = json.dumps({"output": "dup\ndup\n", "exit_code": 1, "error": "oops"}) + result = self.m._filter_terminal(raw) + data = json.loads(result) + self.assertEqual(1, data["exit_code"]) + self.assertEqual("oops", data["error"]) + + def test_empty_output_field_returns_none(self): + raw = json.dumps({"output": "", "exit_code": 0, "error": ""}) + self.assertIsNone(self.m._filter_terminal(raw)) + + def test_non_json_falls_back_to_filter_text(self): + raw = "dup\ndup\nother\n" + result = self.m._filter_terminal(raw) + self.assertEqual("dup\nother", result) + + def test_no_change_returns_none(self): + raw = make_terminal_result("\n".join(f"unique{i}" for i in range(5))) + self.assertIsNone(self.m._filter_terminal(raw)) + + def test_logs_reduction_to_stderr(self): + raw = make_terminal_result("dup\ndup\n") + with mock.patch.object(self.m.sys, "stderr", new_callable=io.StringIO) as err: + self.m._filter_terminal(raw) + self.assertIn("rtk: filtered terminal", err.getvalue()) + + +# --------------------------------------------------------------------------- +# _filter_browser +# --------------------------------------------------------------------------- + +class FilterBrowserTest(unittest.TestCase): + def setUp(self): + self.m = load_plugin() + + def test_strips_layout_noise_lines(self): + snapshot = "LayoutBlock\n heading Hello\nLayoutInline foo\n link Click me\n" + raw = make_browser_result(snapshot) + result = self.m._filter_browser(raw) + self.assertIsNotNone(result) + data = json.loads(result) + self.assertNotIn("LayoutBlock", data["snapshot"]) + self.assertIn("heading Hello", data["snapshot"]) + + def test_deduplicates_snapshot_lines(self): + snapshot = "heading Hello\nheading Hello\nlink Foo\n" + raw = make_browser_result(snapshot) + result = self.m._filter_browser(raw) + data = json.loads(result) + self.assertEqual(1, data["snapshot"].count("heading Hello")) + + def test_truncates_snapshot_at_120_lines(self): + snapshot = "\n".join(f"link item{i}" for i in range(200)) + raw = make_browser_result(snapshot) + result = self.m._filter_browser(raw) + data = json.loads(result) + lines = data["snapshot"].splitlines() + self.assertEqual(121, len(lines)) + self.assertIn("truncated", lines[-1]) + + def test_drops_stealth_warning_and_stealth_features(self): + snapshot = "heading Hello\n" + raw = json.dumps({ + "snapshot": snapshot, + "stealth_warning": "some warning", + "stealth_features": ["feature1"], + }) + result = self.m._filter_browser(raw) + data = json.loads(result) + self.assertNotIn("stealth_warning", data) + self.assertNotIn("stealth_features", data) + + def test_non_json_falls_back_to_filter_text(self): + raw = "dup\ndup\nother\n" + result = self.m._filter_browser(raw) + self.assertEqual("dup\nother", result) + + def test_no_change_returns_none(self): + snapshot = "\n".join(f"link item{i}" for i in range(5)) + raw = make_browser_result(snapshot) + self.assertIsNone(self.m._filter_browser(raw)) + + def test_logs_reduction_to_stderr(self): + snapshot = "LayoutBlock noise\n" * 10 + "heading Real content\n" + raw = make_browser_result(snapshot) + with mock.patch.object(self.m.sys, "stderr", new_callable=io.StringIO) as err: + self.m._filter_browser(raw) + self.assertIn("rtk: filtered browser_navigate", err.getvalue()) + + +# --------------------------------------------------------------------------- +# _filter_diff +# --------------------------------------------------------------------------- + +class FilterDiffTest(unittest.TestCase): + def setUp(self): + self.m = load_plugin() + + SAMPLE_DIFF = "\n".join([ + "--- a/src/foo.py", + "+++ b/src/foo.py", + "@@ -1,4 +1,5 @@", + " unchanged", + "-removed line", + "+added line 1", + "+added line 2", + "@@ -10,3 +11,2 @@", + " ctx", + "-old", + "+new", + ]) + + def test_produces_compact_summary(self): + result = self.m._filter_diff(self.SAMPLE_DIFF) + self.assertIsNotNone(result) + self.assertTrue(result.startswith("[patch]")) + self.assertIn("src/foo.py", result) + self.assertIn("+3", result) + self.assertIn("-2", result) + self.assertIn("2 hunks", result) + + def test_single_hunk_uses_singular(self): + diff = "--- a/x.py\n+++ b/x.py\n@@ -1 +1 @@\n-old\n+new\n" + result = self.m._filter_diff(diff) + self.assertIn("1 hunk", result) + self.assertNotIn("1 hunks", result) + + def test_shorter_than_original_returns_summary(self): + result = self.m._filter_diff(self.SAMPLE_DIFF) + self.assertLess(len(result), len(self.SAMPLE_DIFF)) + + def test_very_short_diff_returns_none(self): + # If summary is longer than original, no filtering needed + diff = "[patch] x (1 hunk, +1/-0 lines)" + self.assertIsNone(self.m._filter_diff(diff)) + + def test_logs_reduction_to_stderr(self): + with mock.patch.object(self.m.sys, "stderr", new_callable=io.StringIO) as err: + self.m._filter_diff(self.SAMPLE_DIFF) + self.assertIn("rtk: filtered diff", err.getvalue()) + + def test_empty_returns_none(self): + self.assertIsNone(self.m._filter_diff("")) + self.assertIsNone(self.m._filter_diff(" ")) + + +# --------------------------------------------------------------------------- +# _filter_search_files +# --------------------------------------------------------------------------- + +class FilterSearchFilesTest(unittest.TestCase): + def setUp(self): + self.m = load_plugin() + + def test_truncates_above_50_results(self): + lines = "\n".join(f"src/file{i}.py:10: match" for i in range(80)) + result = self.m._filter_search_files(lines) + self.assertIsNotNone(result) + out_lines = result.splitlines() + self.assertEqual(51, len(out_lines)) + self.assertIn("30 results truncated", out_lines[-1]) + + def test_under_limit_returns_none(self): + lines = "\n".join(f"src/file{i}.py:1: match" for i in range(30)) + self.assertIsNone(self.m._filter_search_files(lines)) + + def test_exactly_50_returns_none(self): + lines = "\n".join(f"src/file{i}.py:1: match" for i in range(50)) + self.assertIsNone(self.m._filter_search_files(lines)) + + def test_blank_lines_not_counted_as_results(self): + lines = "\n".join(f"src/file{i}.py:1: match" for i in range(30)) + lines_with_blanks = lines + "\n\n\n" + self.assertIsNone(self.m._filter_search_files(lines_with_blanks)) + + def test_logs_reduction_to_stderr(self): + lines = "\n".join(f"src/file{i}.py:1: match" for i in range(80)) + with mock.patch.object(self.m.sys, "stderr", new_callable=io.StringIO) as err: + self.m._filter_search_files(lines) + self.assertIn("rtk: filtered search_files", err.getvalue()) + + +# --------------------------------------------------------------------------- +# _filter_kanban +# --------------------------------------------------------------------------- + +class FilterKanbanTest(unittest.TestCase): + def setUp(self): + self.m = load_plugin() + + def _make_kanban(self, n_events=15): + lines = [ + "Task: t_abc123 — Do something", + "Status: running", + "Body: some body text", + f"Events ({n_events})", + ] + for i in range(n_events): + lines.append(f" [{i}] event_{i} — detail") + lines += ["Runs (1)", " run 1 — completed"] + return "\n".join(lines) + + def test_truncates_events_above_10(self): + raw = self._make_kanban(n_events=20) + result = self.m._filter_kanban(raw) + self.assertIsNotNone(result) + self.assertIn("10 older events truncated", result) + + def test_under_10_events_returns_none(self): + raw = self._make_kanban(n_events=8) + self.assertIsNone(self.m._filter_kanban(raw)) + + def test_exactly_10_events_returns_none(self): + raw = self._make_kanban(n_events=10) + self.assertIsNone(self.m._filter_kanban(raw)) + + def test_preserves_non_event_sections(self): + raw = self._make_kanban(n_events=20) + result = self.m._filter_kanban(raw) + self.assertIn("Body: some body text", result) + self.assertIn("Runs (1)", result) + self.assertIn("run 1 — completed", result) + + def test_logs_reduction_to_stderr(self): + raw = self._make_kanban(n_events=20) + with mock.patch.object(self.m.sys, "stderr", new_callable=io.StringIO) as err: + self.m._filter_kanban(raw) + self.assertIn("rtk: filtered kanban", err.getvalue()) + + def test_empty_returns_none(self): + self.assertIsNone(self.m._filter_kanban("")) + + def test_kanban_list_tool_is_also_filtered(self): + raw = self._make_kanban(n_events=20) + result = self.m._transform_tool_result(tool_name="kanban_list", result=raw) + self.assertIsNotNone(result) + self.assertIn("older events truncated", result) + + def test_kanban_show_tool_is_also_filtered(self): + raw = self._make_kanban(n_events=20) + result = self.m._transform_tool_result(tool_name="kanban_show", result=raw) + self.assertIsNotNone(result) + + +# --------------------------------------------------------------------------- +# execute_code / read_file use _filter_text +# --------------------------------------------------------------------------- + +class FilterExecuteCodeReadFileTest(unittest.TestCase): + def setUp(self): + self.m = load_plugin() + + def test_execute_code_deduplicates(self): + raw = "line\nline\nother\n" + result = self.m._transform_tool_result(tool_name="execute_code", result=raw) + self.assertEqual("line\nother", result) + + def test_read_file_deduplicates(self): + raw = "import os\nimport os\nimport sys\n" + result = self.m._transform_tool_result(tool_name="read_file", result=raw) + self.assertEqual("import os\nimport sys", result) + + def test_unique_content_returns_none(self): + # No trailing newline, no blank lines, no duplicates — _filter_text has nothing to do. + raw = "line_a\nline_b\nline_c" + self.assertIsNone(self.m._transform_tool_result(tool_name="read_file", result=raw)) + + +if __name__ == "__main__": + unittest.main()