feat(context): bound per-tool output before it enters agent history

Cap the size of every tool result so a single verbose command (recursive
find, noisy scanner, full page dump) can't pin the conversation near the
model's context window for the rest of a scan.

- New ContextSettings config group with env-tunable caps.
- Default the SDK shell tools' max_output_tokens so exec_command /
  write_stdin truncate head+tail instead of returning unbounded output.
- Bound Strix's own FunctionTool/CustomTool results (line + UTF-8 byte
  head+tail preview with a truncation notice) and cap error strings.
This commit is contained in:
Ahmed Allam
2026-07-26 14:38:12 -07:00
committed by Ahmed Allam
parent d2fbcb726d
commit a70a87f272
6 changed files with 214 additions and 11 deletions
+18 -4
View File
@@ -9,6 +9,7 @@ import pytest
from agents.tool import FunctionTool
from strix.agents import factory
from strix.config import load_settings
def _capturing_exec_tool(captured: dict[str, str]) -> FunctionTool:
@@ -32,10 +33,23 @@ async def test_wrap_exec_command_defaults_shell_to_bash() -> None:
result = await wrapped.on_invoke_tool(cast("Any", None), json.dumps({"cmd": "source /tmp/env"}))
assert result == "ok"
assert json.loads(captured["raw_input"]) == {
"cmd": "source /tmp/env",
"shell": "bash",
}
parsed = json.loads(captured["raw_input"])
assert parsed["cmd"] == "source /tmp/env"
assert parsed["shell"] == "bash"
expected_cap = load_settings().context.tool_output_max_tokens
assert parsed["max_output_tokens"] == expected_cap
@pytest.mark.asyncio
async def test_wrap_exec_command_preserves_explicit_output_cap() -> None:
captured: dict[str, str] = {}
wrapped = factory._wrap_exec_command(_capturing_exec_tool(captured))
await wrapped.on_invoke_tool(
cast("Any", None), json.dumps({"cmd": "echo hi", "max_output_tokens": 42})
)
assert json.loads(captured["raw_input"])["max_output_tokens"] == 42
@pytest.mark.asyncio
+46
View File
@@ -0,0 +1,46 @@
"""Tests for per-tool-output bounding."""
from __future__ import annotations
from strix.tools.output_store import bound_text
def test_small_output_passes_through_unchanged() -> None:
text = "line 1\nline 2\nline 3"
assert bound_text(text, max_lines=100, max_bytes=10_000) == text
def test_line_limit_keeps_head_and_tail() -> None:
text = "\n".join(str(i) for i in range(1000))
bounded = bound_text(text, max_lines=10, max_bytes=1_000_000)
assert bounded.startswith("0\n1\n2\n3\n4")
assert bounded.rstrip().endswith("999")
assert "truncated" in bounded
# Head + tail only, far fewer than the original 1000 lines.
assert len(bounded.splitlines()) < 30
def test_byte_limit_enforced_on_single_long_line() -> None:
text = "x" * 100_000
bounded = bound_text(text, max_lines=2_000, max_bytes=1_000)
assert "truncated" in bounded
assert len(bounded.encode("utf-8")) < 3_000
def test_multibyte_characters_not_split() -> None:
text = "😀" * 50_000
bounded = bound_text(text, max_lines=2_000, max_bytes=1_000)
# Must remain valid UTF-8 (no broken surrogate halves from a mid-char cut).
assert bounded == bounded.encode("utf-8").decode("utf-8")
assert "truncated" in bounded
def test_notice_reports_dropped_counts() -> None:
text = "\n".join("y" * 10 for _ in range(500))
bounded = bound_text(text, max_lines=10, max_bytes=1_000_000)
assert "lines" in bounded
assert "bytes" in bounded