From 3b8980c47ba1777d4237ca8ccc5c458e8c79f9de Mon Sep 17 00:00:00 2001 From: Ahmed Allam Date: Sat, 25 Jul 2026 21:02:08 +0000 Subject: [PATCH 1/6] 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. --- strix/agents/factory.py | 65 +++++++++++++++++++++++++--- strix/config/__init__.py | 2 + strix/config/settings.py | 18 ++++++++ strix/tools/output_store.py | 72 +++++++++++++++++++++++++++++++ tests/test_agent_factory_shell.py | 22 ++++++++-- tests/test_output_store.py | 46 ++++++++++++++++++++ 6 files changed, 214 insertions(+), 11 deletions(-) create mode 100644 strix/tools/output_store.py create mode 100644 tests/test_output_store.py diff --git a/strix/agents/factory.py b/strix/agents/factory.py index 977d01f19..ab3d3fe39 100644 --- a/strix/agents/factory.py +++ b/strix/agents/factory.py @@ -16,6 +16,7 @@ from pydantic import ValidationError from strix.agents.prompt import render_system_prompt +from strix.config import load_settings from strix.tools.agents_graph.tools import ( agent_finish, create_agent, @@ -33,6 +34,7 @@ list_notes, update_note, ) +from strix.tools.output_store import bound_text from strix.tools.proxy.tools import ( list_requests, list_sitemap, @@ -103,8 +105,41 @@ def _extract_custom_input(tool: CustomTool, raw_input: str | dict[str, Any]) -> return value if isinstance(value, str) else "" +def _tool_output_limits() -> tuple[int, int]: + context = load_settings().context + return context.tool_output_max_lines, context.tool_output_max_bytes + + +def _bound_result(result: Any) -> Any: + if not isinstance(result, str): + return result + max_lines, max_bytes = _tool_output_limits() + return bound_text(result, max_lines=max_lines, max_bytes=max_bytes) + + def _format_tool_error(exc: Exception) -> str: - return str(exc) or exc.__class__.__name__ + message = str(exc) or exc.__class__.__name__ + max_lines, max_bytes = _tool_output_limits() + return bound_text(message, max_lines=max_lines, max_bytes=max_bytes) + + +def _with_bounded_result(tool: FunctionTool) -> FunctionTool: + """Cap the size of a tool's result before it enters agent history. + + Idempotent: base tools are shared singletons reused across every agent, so + the guard prevents stacking the wrapper on repeated ``build_strix_agent`` + calls. + """ + if getattr(tool, "_strix_bounded", False): + return tool + invoke_tool = tool.on_invoke_tool + + async def invoke(ctx: Any, raw_input: str) -> Any: + return _bound_result(await invoke_tool(ctx, raw_input)) + + tool.on_invoke_tool = invoke + tool._strix_bounded = True # type: ignore[attr-defined] + return tool def _function_tool_with_error_result(tool: FunctionTool) -> FunctionTool: @@ -112,7 +147,7 @@ def _function_tool_with_error_result(tool: FunctionTool) -> FunctionTool: async def invoke(ctx: Any, raw_input: str) -> Any: try: - return await invoke_tool(ctx, raw_input) + return _bound_result(await invoke_tool(ctx, raw_input)) except Exception as exc: # noqa: BLE001 - tool errors should be model-visible results. logger.debug("Tool %s failed; returning error as result", tool.name, exc_info=True) return _format_tool_error(exc) @@ -127,7 +162,7 @@ async def invoke(ctx: Any, raw_input: str) -> Any: if not custom_input: return f"`{_custom_tool_input_field(tool)}` must be a non-empty string." try: - return await tool.on_invoke_tool(ctx, custom_input) + return _bound_result(await tool.on_invoke_tool(ctx, custom_input)) except Exception as exc: # noqa: BLE001 - matches SDK CustomTool error-as-result behavior. logger.debug("Tool %s failed; returning error as result", tool.name, exc_info=True) return _format_tool_error(exc) @@ -205,6 +240,15 @@ def _format_validation_error(tool_name: str, exc: ValidationError) -> str: return f"{tool_name}: invalid arguments — " + "; ".join(parts) +def _apply_shell_output_cap(parsed: dict[str, Any]) -> None: + """Default the SDK shell tools' own token cap so a single command can't + dump unbounded output into history. The SDK truncates head+tail when the + model omits the field; respect an explicit model-supplied value. + """ + if parsed.get("max_output_tokens") is None: + parsed["max_output_tokens"] = load_settings().context.tool_output_max_tokens + + def _wrap_exec_command(tool: FunctionTool) -> FunctionTool: invoke_tool = tool.on_invoke_tool @@ -213,8 +257,10 @@ async def invoke(ctx: Any, raw_input: str) -> Any: parsed = json.loads(raw_input) except (json.JSONDecodeError, TypeError): parsed = None - if isinstance(parsed, dict) and "shell" not in parsed: - parsed["shell"] = "bash" + if isinstance(parsed, dict): + if "shell" not in parsed: + parsed["shell"] = "bash" + _apply_shell_output_cap(parsed) raw_input = json.dumps(parsed) try: return await invoke_tool(ctx, raw_input) @@ -240,8 +286,10 @@ async def invoke(ctx: Any, raw_input: str) -> Any: parsed = json.loads(raw_input) except json.JSONDecodeError: parsed = None - if isinstance(parsed, dict) and isinstance(parsed.get("chars"), str): - parsed["chars"] = _decode_chars_escape(parsed["chars"]) + if isinstance(parsed, dict): + if isinstance(parsed.get("chars"), str): + parsed["chars"] = _decode_chars_escape(parsed["chars"]) + _apply_shell_output_cap(parsed) raw_input = json.dumps(parsed) try: return await invoke_tool(ctx, raw_input) @@ -440,6 +488,9 @@ def build_strix_agent( else: tools = [*_BASE_TOOLS, *agent_tools, agent_finish] _ensure_unique_tool_names(tools) + tools = [ + _with_bounded_result(tool) if isinstance(tool, FunctionTool) else tool for tool in tools + ] logger.info( "Built %s agent '%s' (skills=%d, tools=%d, scan_mode=%s, whitebox=%s)", diff --git a/strix/config/__init__.py b/strix/config/__init__.py index 6e9dded17..f21fdab6b 100644 --- a/strix/config/__init__.py +++ b/strix/config/__init__.py @@ -17,6 +17,7 @@ persist_current, ) from strix.config.settings import ( + ContextSettings, DedupeSettings, IntegrationSettings, LlmSettings, @@ -27,6 +28,7 @@ __all__ = [ + "ContextSettings", "DedupeSettings", "IntegrationSettings", "LlmSettings", diff --git a/strix/config/settings.py b/strix/config/settings.py index 98d68934d..edb0ab099 100644 --- a/strix/config/settings.py +++ b/strix/config/settings.py @@ -55,6 +55,23 @@ class DedupeSettings(BaseSettings): api_base: str | None = Field(default=None, alias="DEDUPE_LLM_API_BASE") +class ContextSettings(BaseSettings): + """Context-window management: per-tool-output caps and history compaction.""" + + model_config = _BASE_CONFIG + + auto_compact: bool = Field(default=True, alias="STRIX_CONTEXT_AUTO_COMPACT") + compact_buffer_tokens: int = Field(default=20_000, gt=0, alias="STRIX_CONTEXT_BUFFER_TOKENS") + keep_tokens: int = Field(default=8_000, gt=0, alias="STRIX_CONTEXT_KEEP_TOKENS") + fallback_context_tokens: int = Field( + default=200_000, gt=0, alias="STRIX_CONTEXT_FALLBACK_TOKENS" + ) + summary_max_tokens: int = Field(default=4_096, gt=0, alias="STRIX_CONTEXT_SUMMARY_TOKENS") + tool_output_max_tokens: int = Field(default=8_000, gt=0, alias="STRIX_TOOL_OUTPUT_MAX_TOKENS") + tool_output_max_lines: int = Field(default=2_000, gt=0, alias="STRIX_TOOL_OUTPUT_MAX_LINES") + tool_output_max_bytes: int = Field(default=50 * 1024, gt=0, alias="STRIX_TOOL_OUTPUT_MAX_BYTES") + + class RuntimeSettings(BaseSettings): model_config = _BASE_CONFIG @@ -99,6 +116,7 @@ class Settings(BaseSettings): llm: LlmSettings = Field(default_factory=LlmSettings) dedupe: DedupeSettings = Field(default_factory=DedupeSettings) runtime: RuntimeSettings = Field(default_factory=RuntimeSettings) + context: ContextSettings = Field(default_factory=ContextSettings) telemetry: TelemetrySettings = Field(default_factory=TelemetrySettings) integrations: IntegrationSettings = Field(default_factory=IntegrationSettings) viewer: ViewerSettings = Field(default_factory=ViewerSettings) diff --git a/strix/tools/output_store.py b/strix/tools/output_store.py new file mode 100644 index 000000000..44a9fa9c9 --- /dev/null +++ b/strix/tools/output_store.py @@ -0,0 +1,72 @@ +"""Bound oversized tool results before they enter agent history. + +A single verbose tool result (a recursive ``find``, a noisy scanner, a full +page dump) can otherwise pin the whole conversation near the model's context +limit for the rest of the scan. This keeps a head + tail slice of the output +and drops the middle, mirroring how the shell capability truncates its own +output — the agent still sees the start and end plus how much was removed. +""" + +from __future__ import annotations + + +_TRUNCATION_NOTICE = "[... {lines} lines ({bytes} bytes) truncated ...]" + + +def _byte_len(text: str) -> int: + return len(text.encode("utf-8")) + + +def _take_prefix(text: str, max_bytes: int) -> str: + budget = 0 + out: list[str] = [] + for char in text: + size = len(char.encode("utf-8")) + if budget + size > max_bytes: + break + out.append(char) + budget += size + return "".join(out) + + +def _take_suffix(text: str, max_bytes: int) -> str: + budget = 0 + out: list[str] = [] + for char in reversed(text): + size = len(char.encode("utf-8")) + if budget + size > max_bytes: + break + out.append(char) + budget += size + out.reverse() + return "".join(out) + + +def bound_text(text: str, *, max_lines: int, max_bytes: int) -> str: + """Return ``text`` unchanged when small, else a head+tail preview. + + Truncation happens on whichever limit is hit first (line count or UTF-8 + byte size). The removed middle is replaced with a notice recording how + many lines and bytes were dropped so the agent knows output was elided. + """ + lines = text.split("\n") + total_bytes = _byte_len(text) + if len(lines) <= max_lines and total_bytes <= max_bytes: + return text + + head_lines = max(1, max_lines // 2) + tail_lines = max_lines - head_lines + head = "\n".join(lines[:head_lines]) + tail = "\n".join(lines[len(lines) - tail_lines :]) if tail_lines > 0 else "" + + # Enforce the byte budget even when the line count alone was fine. + half_bytes = max(1, max_bytes // 2) + if _byte_len(head) > half_bytes: + head = _take_prefix(head, half_bytes) + if tail and _byte_len(tail) > half_bytes: + tail = _take_suffix(tail, half_bytes) + + dropped_lines = max(0, len(lines) - head_lines - tail_lines) + dropped_bytes = max(0, total_bytes - _byte_len(head) - _byte_len(tail)) + notice = _TRUNCATION_NOTICE.format(lines=dropped_lines, bytes=dropped_bytes) + return f"{head}\n\n{notice}\n\n{tail}" if tail else f"{head}\n\n{notice}" diff --git a/tests/test_agent_factory_shell.py b/tests/test_agent_factory_shell.py index 6de932190..302765620 100644 --- a/tests/test_agent_factory_shell.py +++ b/tests/test_agent_factory_shell.py @@ -9,6 +9,7 @@ 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 diff --git a/tests/test_output_store.py b/tests/test_output_store.py new file mode 100644 index 000000000..23bcdf301 --- /dev/null +++ b/tests/test_output_store.py @@ -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 From dab93bcc1239c99ec165cc1e9cadb83c6ab53cdb Mon Sep 17 00:00:00 2001 From: Ahmed Allam Date: Sat, 25 Jul 2026 22:32:56 +0000 Subject: [PATCH 2/6] fix(context): clamp shell output cap and count byte-trimmed dropped lines Treat tool_output_max_tokens as a ceiling so an explicit model-supplied cap can't exceed it, and derive the truncation notice's dropped-line count from the lines actually kept after the byte-trim pass. Also cast the pygments fallback lexer so it satisfies the resolve_lexer return type under the pre-commit mypy hook. --- strix/agents/factory.py | 16 +++++++++++----- strix/report/writer.py | 6 +++--- strix/tools/output_store.py | 6 +++++- tests/test_agent_factory_shell.py | 16 +++++++++++++++- tests/test_output_store.py | 17 +++++++++++++++++ 5 files changed, 51 insertions(+), 10 deletions(-) diff --git a/strix/agents/factory.py b/strix/agents/factory.py index ab3d3fe39..b9c79bfec 100644 --- a/strix/agents/factory.py +++ b/strix/agents/factory.py @@ -241,12 +241,18 @@ def _format_validation_error(tool_name: str, exc: ValidationError) -> str: def _apply_shell_output_cap(parsed: dict[str, Any]) -> None: - """Default the SDK shell tools' own token cap so a single command can't - dump unbounded output into history. The SDK truncates head+tail when the - model omits the field; respect an explicit model-supplied value. + """Bound the SDK shell tools' own token cap so a single command can't dump + unbounded output into history. The SDK truncates head+tail from this value. + + The configured cap is a ceiling: a missing value defaults to it, and a + larger model-supplied value is clamped down to it. A smaller explicit value + is respected, so the model can still ask for less. """ - if parsed.get("max_output_tokens") is None: - parsed["max_output_tokens"] = load_settings().context.tool_output_max_tokens + ceiling = load_settings().context.tool_output_max_tokens + requested = parsed.get("max_output_tokens") + parsed["max_output_tokens"] = ( + ceiling if not isinstance(requested, int) or requested > ceiling else requested + ) def _wrap_exec_command(tool: FunctionTool) -> FunctionTool: diff --git a/strix/report/writer.py b/strix/report/writer.py index 32fc961fb..1b0a8a1df 100644 --- a/strix/report/writer.py +++ b/strix/report/writer.py @@ -10,7 +10,7 @@ import tempfile from datetime import UTC, datetime from pathlib import Path -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, cast from pygments.lexers import PythonLexer, get_lexer_by_name, guess_lexer from pygments.lexers.special import TextLexer @@ -74,10 +74,10 @@ def resolve_lexer(language: str | None, code: str) -> Lexer: try: lexer = guess_lexer(code) except ClassNotFound: - return PythonLexer() + return cast("Lexer", PythonLexer()) # ``guess_lexer`` returns the plain-text lexer when it can't detect anything. if isinstance(lexer, TextLexer): - return PythonLexer() + return cast("Lexer", PythonLexer()) return lexer diff --git a/strix/tools/output_store.py b/strix/tools/output_store.py index 44a9fa9c9..92390ce9a 100644 --- a/strix/tools/output_store.py +++ b/strix/tools/output_store.py @@ -66,7 +66,11 @@ def bound_text(text: str, *, max_lines: int, max_bytes: int) -> str: if tail and _byte_len(tail) > half_bytes: tail = _take_suffix(tail, half_bytes) - dropped_lines = max(0, len(lines) - head_lines - tail_lines) + # Count kept lines from the final slices: the byte pass above may have + # dropped whole lines from head/tail, so deriving this from the original + # head_lines/tail_lines would undercount what was actually removed. + kept_lines = len(head.split("\n")) + (len(tail.split("\n")) if tail else 0) + dropped_lines = max(0, len(lines) - kept_lines) dropped_bytes = max(0, total_bytes - _byte_len(head) - _byte_len(tail)) notice = _TRUNCATION_NOTICE.format(lines=dropped_lines, bytes=dropped_bytes) return f"{head}\n\n{notice}\n\n{tail}" if tail else f"{head}\n\n{notice}" diff --git a/tests/test_agent_factory_shell.py b/tests/test_agent_factory_shell.py index 302765620..7165378c3 100644 --- a/tests/test_agent_factory_shell.py +++ b/tests/test_agent_factory_shell.py @@ -41,7 +41,7 @@ async def test_wrap_exec_command_defaults_shell_to_bash() -> None: @pytest.mark.asyncio -async def test_wrap_exec_command_preserves_explicit_output_cap() -> None: +async def test_wrap_exec_command_preserves_smaller_explicit_output_cap() -> None: captured: dict[str, str] = {} wrapped = factory._wrap_exec_command(_capturing_exec_tool(captured)) @@ -52,6 +52,20 @@ async def test_wrap_exec_command_preserves_explicit_output_cap() -> None: assert json.loads(captured["raw_input"])["max_output_tokens"] == 42 +@pytest.mark.asyncio +async def test_wrap_exec_command_clamps_oversized_explicit_output_cap() -> None: + captured: dict[str, str] = {} + wrapped = factory._wrap_exec_command(_capturing_exec_tool(captured)) + ceiling = load_settings().context.tool_output_max_tokens + + await wrapped.on_invoke_tool( + cast("Any", None), + json.dumps({"cmd": "echo hi", "max_output_tokens": ceiling * 100}), + ) + + assert json.loads(captured["raw_input"])["max_output_tokens"] == ceiling + + @pytest.mark.asyncio @pytest.mark.parametrize("shell", ["/bin/zsh", ""]) async def test_wrap_exec_command_preserves_explicit_shell(shell: str) -> None: diff --git a/tests/test_output_store.py b/tests/test_output_store.py index 23bcdf301..2dc1906fd 100644 --- a/tests/test_output_store.py +++ b/tests/test_output_store.py @@ -2,6 +2,8 @@ from __future__ import annotations +import re + from strix.tools.output_store import bound_text @@ -44,3 +46,18 @@ def test_notice_reports_dropped_counts() -> None: assert "lines" in bounded assert "bytes" in bounded + + +def test_dropped_line_count_accounts_for_byte_trimming() -> None: + # A tight byte budget forces the byte pass to drop whole lines from the + # head/tail slices; the notice must count those, not just the middle. + text = "\n".join(f"line-{i}" for i in range(200)) + bounded = bound_text(text, max_lines=20, max_bytes=40) + + match = re.search(r"\[\.\.\. (\d+) lines", bounded) + assert match is not None, bounded + dropped = int(match.group(1)) + kept = [ln for ln in bounded.splitlines() if ln and "truncated" not in ln] + assert dropped == 200 - len(kept) + # The naive middle-only count (max_lines split evenly) would under-report. + assert dropped > 200 - 20 From ce358aa8794b343f3945e6bab044f91b3e8f67b6 Mon Sep 17 00:00:00 2001 From: Ahmed Allam Date: Sat, 25 Jul 2026 22:50:50 +0000 Subject: [PATCH 3/6] fix(context): bound native filesystem tool output in Responses mode Chat-completions mode converts filesystem CustomTools to FunctionTools (which bounds their result), but the Responses-API path kept them native and unbounded, so a large read_file could still exhaust the context window. Always configure the Filesystem capability to head+tail bound tool output in both modes. --- strix/agents/factory.py | 43 ++++++++++++++++++++++++++----- tests/test_agent_factory_shell.py | 35 ++++++++++++++++++++++++- 2 files changed, 71 insertions(+), 7 deletions(-) diff --git a/strix/agents/factory.py b/strix/agents/factory.py index b9c79bfec..4b880de42 100644 --- a/strix/agents/factory.py +++ b/strix/agents/factory.py @@ -194,12 +194,43 @@ async def approve(ctx: Any, args: dict[str, Any], call_id: str) -> bool: ) -def _configure_chat_completions_filesystem_tools(toolset: Any) -> None: +def _bound_custom_tool(tool: CustomTool) -> CustomTool: + """Bound a native ``CustomTool`` result in place. + + Chat-completions mode converts filesystem ``CustomTool``s to ``FunctionTool``s + (which bounds the result), but the Responses path keeps them native, so a + large ``read_file``/directory listing would otherwise append unbounded text + to history. Wrap ``on_invoke_tool`` so the same head+tail bound applies. + """ + invoke_tool = tool.on_invoke_tool + + async def invoke(ctx: Any, raw_input: str) -> Any: + return _bound_result(await invoke_tool(ctx, raw_input)) + + tool.on_invoke_tool = invoke + return tool + + +def _configure_filesystem_tools(toolset: Any, *, chat_completions: bool) -> None: for name, tool in vars(toolset).items(): - if isinstance(tool, CustomTool): - setattr(toolset, name, _custom_tool_as_function_tool(tool)) + if chat_completions: + if isinstance(tool, CustomTool): + setattr(toolset, name, _custom_tool_as_function_tool(tool)) + elif isinstance(tool, FunctionTool): + setattr(toolset, name, _function_tool_with_error_result(tool)) + # Responses-API path: keep tools native but still bound their output so + # filesystem reads can't exhaust the context window on later turns. + elif isinstance(tool, CustomTool): + setattr(toolset, name, _bound_custom_tool(tool)) elif isinstance(tool, FunctionTool): - setattr(toolset, name, _function_tool_with_error_result(tool)) + setattr(toolset, name, _with_bounded_result(tool)) + + +def _make_filesystem_configurator(*, chat_completions: bool) -> Any: + def configure(toolset: Any) -> None: + _configure_filesystem_tools(toolset, chat_completions=chat_completions) + + return configure _CHARS_ESCAPE_RE = re.compile(r"\\(?:u[0-9a-fA-F]{4}|x[0-9a-fA-F]{2}|[0abtnvfr\\])") @@ -516,8 +547,8 @@ def build_strix_agent( model=None, capabilities=[ Filesystem( - configure_tools=( - _configure_chat_completions_filesystem_tools if chat_completions_tools else None + configure_tools=_make_filesystem_configurator( + chat_completions=chat_completions_tools, ), ), Shell( diff --git a/tests/test_agent_factory_shell.py b/tests/test_agent_factory_shell.py index 7165378c3..22b662c74 100644 --- a/tests/test_agent_factory_shell.py +++ b/tests/test_agent_factory_shell.py @@ -3,10 +3,11 @@ from __future__ import annotations import json +from types import SimpleNamespace from typing import Any, cast import pytest -from agents.tool import FunctionTool +from agents.tool import CustomTool, FunctionTool from strix.agents import factory from strix.config import load_settings @@ -77,3 +78,35 @@ async def test_wrap_exec_command_preserves_explicit_shell(shell: str) -> None: ) assert json.loads(captured["raw_input"])["shell"] == shell + + +@pytest.mark.asyncio +async def test_responses_filesystem_custom_tool_output_is_bounded() -> None: + # In Responses-API mode filesystem tools stay native CustomTools; a large + # read must still be head+tail bounded before it enters history. + async def invoke(_ctx: Any, _inp: str) -> str: + return "line\n" * 50_000 + + toolset = SimpleNamespace( + read_file=CustomTool(name="read_file", description="read", on_invoke_tool=invoke) + ) + factory._configure_filesystem_tools(toolset, chat_completions=False) + + assert isinstance(toolset.read_file, CustomTool) + result = await toolset.read_file.on_invoke_tool(cast("Any", None), "{}") + + assert "truncated" in result + assert len(result) < len("line\n" * 50_000) + + +@pytest.mark.asyncio +async def test_chat_completions_filesystem_custom_tool_becomes_function_tool() -> None: + async def invoke(_ctx: Any, _inp: str) -> str: + return "ok" + + toolset = SimpleNamespace( + read_file=CustomTool(name="read_file", description="read", on_invoke_tool=invoke) + ) + factory._configure_filesystem_tools(toolset, chat_completions=True) + + assert isinstance(toolset.read_file, FunctionTool) From aac59de1e50e527c29e9ded7550acdf78216a8b6 Mon Sep 17 00:00:00 2001 From: Ahmed Allam Date: Sat, 25 Jul 2026 23:06:40 +0000 Subject: [PATCH 4/6] fix(context): reserve notice budget so bounded output honors max_bytes The head+tail slices could each take half of max_bytes, then the truncation notice and its separators were appended on top, so the value persisted to history could exceed the configured maximum. Reserve an upper bound for the notice (and separators) out of the byte budget before slicing so the whole joined result stays within max_bytes. --- strix/tools/output_store.py | 11 ++++++++++- tests/test_output_store.py | 4 +++- 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/strix/tools/output_store.py b/strix/tools/output_store.py index 92390ce9a..3e622c080 100644 --- a/strix/tools/output_store.py +++ b/strix/tools/output_store.py @@ -48,19 +48,28 @@ def bound_text(text: str, *, max_lines: int, max_bytes: int) -> str: Truncation happens on whichever limit is hit first (line count or UTF-8 byte size). The removed middle is replaced with a notice recording how many lines and bytes were dropped so the agent knows output was elided. + ``max_bytes`` bounds the *entire* joined result, notice and separators + included. """ lines = text.split("\n") total_bytes = _byte_len(text) if len(lines) <= max_lines and total_bytes <= max_bytes: return text + # Reserve room for the notice and its two blank-line separators so the + # head+tail slices can't consume the whole budget and push the persisted + # value over max_bytes. Upper-bound the notice with the largest possible + # counts; the real notice is never longer. ``+ 4`` covers the separators. + notice_overhead = _byte_len(_TRUNCATION_NOTICE.format(lines=len(lines), bytes=total_bytes)) + 4 + byte_budget = max(2, max_bytes - notice_overhead) + head_lines = max(1, max_lines // 2) tail_lines = max_lines - head_lines head = "\n".join(lines[:head_lines]) tail = "\n".join(lines[len(lines) - tail_lines :]) if tail_lines > 0 else "" # Enforce the byte budget even when the line count alone was fine. - half_bytes = max(1, max_bytes // 2) + half_bytes = max(1, byte_budget // 2) if _byte_len(head) > half_bytes: head = _take_prefix(head, half_bytes) if tail and _byte_len(tail) > half_bytes: diff --git a/tests/test_output_store.py b/tests/test_output_store.py index 2dc1906fd..14bff19c1 100644 --- a/tests/test_output_store.py +++ b/tests/test_output_store.py @@ -28,7 +28,9 @@ def test_byte_limit_enforced_on_single_long_line() -> None: bounded = bound_text(text, max_lines=2_000, max_bytes=1_000) assert "truncated" in bounded - assert len(bounded.encode("utf-8")) < 3_000 + # The whole joined result (head + tail + notice + separators) honours the + # configured maximum, not just the head/tail slices. + assert len(bounded.encode("utf-8")) <= 1_000 def test_multibyte_characters_not_split() -> None: From 95046a6cea3bf73b6248f69ad0da94e44654c1e6 Mon Sep 17 00:00:00 2001 From: Ahmed Allam Date: Sat, 25 Jul 2026 23:46:11 +0000 Subject: [PATCH 5/6] fix(context): reject tool-output byte ceilings below the notice size A configured tool_output_max_bytes smaller than the truncation notice itself can't fit a bounded preview, so a persisted result could exceed the ceiling. Enforce a config floor (ge=1024) so nonsensical values are rejected at load time instead of being worked around at runtime. --- strix/config/settings.py | 6 +++++- strix/tools/output_store.py | 3 ++- tests/test_config_loader.py | 19 ++++++++++++++++++- 3 files changed, 25 insertions(+), 3 deletions(-) diff --git a/strix/config/settings.py b/strix/config/settings.py index edb0ab099..d7c795b62 100644 --- a/strix/config/settings.py +++ b/strix/config/settings.py @@ -69,7 +69,11 @@ class ContextSettings(BaseSettings): summary_max_tokens: int = Field(default=4_096, gt=0, alias="STRIX_CONTEXT_SUMMARY_TOKENS") tool_output_max_tokens: int = Field(default=8_000, gt=0, alias="STRIX_TOOL_OUTPUT_MAX_TOKENS") tool_output_max_lines: int = Field(default=2_000, gt=0, alias="STRIX_TOOL_OUTPUT_MAX_LINES") - tool_output_max_bytes: int = Field(default=50 * 1024, gt=0, alias="STRIX_TOOL_OUTPUT_MAX_BYTES") + # Floor comfortably above the truncation-notice size so a preview + # (head+tail+notice) always fits within the configured ceiling. + tool_output_max_bytes: int = Field( + default=50 * 1024, ge=1024, alias="STRIX_TOOL_OUTPUT_MAX_BYTES" + ) class RuntimeSettings(BaseSettings): diff --git a/strix/tools/output_store.py b/strix/tools/output_store.py index 3e622c080..6e6e62026 100644 --- a/strix/tools/output_store.py +++ b/strix/tools/output_store.py @@ -49,7 +49,8 @@ def bound_text(text: str, *, max_lines: int, max_bytes: int) -> str: byte size). The removed middle is replaced with a notice recording how many lines and bytes were dropped so the agent knows output was elided. ``max_bytes`` bounds the *entire* joined result, notice and separators - included. + included, and must be large enough to hold the notice itself (guaranteed by + the ``tool_output_max_bytes`` config floor). """ lines = text.split("\n") total_bytes = _byte_len(text) diff --git a/tests/test_config_loader.py b/tests/test_config_loader.py index be0bfb8f7..0d37e1e89 100644 --- a/tests/test_config_loader.py +++ b/tests/test_config_loader.py @@ -6,10 +6,11 @@ from typing import TYPE_CHECKING import pytest -from pydantic import AliasChoices, Field +from pydantic import AliasChoices, Field, ValidationError from pydantic.fields import FieldInfo from strix.config import loader +from strix.config.settings import ContextSettings if TYPE_CHECKING: @@ -120,6 +121,22 @@ def test_read_json_overrides_uses_json_when_no_alias_in_environ(tmp_path: Path) assert loader._read_json_overrides(path) == {"llm": {"api_key": "sk-file"}} +# --------------------------------------------------------------------------- # +# ContextSettings validation +# --------------------------------------------------------------------------- # + + +def test_tool_output_max_bytes_rejects_sub_notice_values() -> None: + # A ceiling below the truncation notice can't fit a bounded preview, so it + # is rejected at load time rather than producing over-cap persisted output. + with pytest.raises(ValidationError): + ContextSettings(STRIX_TOOL_OUTPUT_MAX_BYTES=64) + + +def test_tool_output_max_bytes_accepts_floor() -> None: + assert ContextSettings(STRIX_TOOL_OUTPUT_MAX_BYTES=1024).tool_output_max_bytes == 1024 + + # --------------------------------------------------------------------------- # # _aliases_for # --------------------------------------------------------------------------- # From 10376b412b5848490c8aa35b002243c2873bf2b0 Mon Sep 17 00:00:00 2001 From: Ahmed Allam Date: Sun, 26 Jul 2026 19:26:51 +0000 Subject: [PATCH 6/6] refactor(context): trim verbose comments --- strix/agents/factory.py | 26 ++++---------------------- strix/config/settings.py | 3 +-- strix/tools/output_store.py | 26 +++++++------------------- tests/test_agent_factory_shell.py | 2 -- tests/test_config_loader.py | 7 ------- tests/test_output_store.py | 9 ++------- 6 files changed, 14 insertions(+), 59 deletions(-) diff --git a/strix/agents/factory.py b/strix/agents/factory.py index 4b880de42..cb43ed6aa 100644 --- a/strix/agents/factory.py +++ b/strix/agents/factory.py @@ -124,12 +124,7 @@ def _format_tool_error(exc: Exception) -> str: def _with_bounded_result(tool: FunctionTool) -> FunctionTool: - """Cap the size of a tool's result before it enters agent history. - - Idempotent: base tools are shared singletons reused across every agent, so - the guard prevents stacking the wrapper on repeated ``build_strix_agent`` - calls. - """ + """Cap a tool's result size before it enters history (idempotent).""" if getattr(tool, "_strix_bounded", False): return tool invoke_tool = tool.on_invoke_tool @@ -195,13 +190,7 @@ async def approve(ctx: Any, args: dict[str, Any], call_id: str) -> bool: def _bound_custom_tool(tool: CustomTool) -> CustomTool: - """Bound a native ``CustomTool`` result in place. - - Chat-completions mode converts filesystem ``CustomTool``s to ``FunctionTool``s - (which bounds the result), but the Responses path keeps them native, so a - large ``read_file``/directory listing would otherwise append unbounded text - to history. Wrap ``on_invoke_tool`` so the same head+tail bound applies. - """ + """Bound a native ``CustomTool`` result in place (Responses path).""" invoke_tool = tool.on_invoke_tool async def invoke(ctx: Any, raw_input: str) -> Any: @@ -218,8 +207,6 @@ def _configure_filesystem_tools(toolset: Any, *, chat_completions: bool) -> None setattr(toolset, name, _custom_tool_as_function_tool(tool)) elif isinstance(tool, FunctionTool): setattr(toolset, name, _function_tool_with_error_result(tool)) - # Responses-API path: keep tools native but still bound their output so - # filesystem reads can't exhaust the context window on later turns. elif isinstance(tool, CustomTool): setattr(toolset, name, _bound_custom_tool(tool)) elif isinstance(tool, FunctionTool): @@ -272,13 +259,8 @@ def _format_validation_error(tool_name: str, exc: ValidationError) -> str: def _apply_shell_output_cap(parsed: dict[str, Any]) -> None: - """Bound the SDK shell tools' own token cap so a single command can't dump - unbounded output into history. The SDK truncates head+tail from this value. - - The configured cap is a ceiling: a missing value defaults to it, and a - larger model-supplied value is clamped down to it. A smaller explicit value - is respected, so the model can still ask for less. - """ + """Clamp the SDK shell tools' ``max_output_tokens`` to the configured + ceiling; a smaller explicit value is respected.""" ceiling = load_settings().context.tool_output_max_tokens requested = parsed.get("max_output_tokens") parsed["max_output_tokens"] = ( diff --git a/strix/config/settings.py b/strix/config/settings.py index d7c795b62..0d910e60f 100644 --- a/strix/config/settings.py +++ b/strix/config/settings.py @@ -69,8 +69,7 @@ class ContextSettings(BaseSettings): summary_max_tokens: int = Field(default=4_096, gt=0, alias="STRIX_CONTEXT_SUMMARY_TOKENS") tool_output_max_tokens: int = Field(default=8_000, gt=0, alias="STRIX_TOOL_OUTPUT_MAX_TOKENS") tool_output_max_lines: int = Field(default=2_000, gt=0, alias="STRIX_TOOL_OUTPUT_MAX_LINES") - # Floor comfortably above the truncation-notice size so a preview - # (head+tail+notice) always fits within the configured ceiling. + # Floor above the truncation-notice size so a preview always fits. tool_output_max_bytes: int = Field( default=50 * 1024, ge=1024, alias="STRIX_TOOL_OUTPUT_MAX_BYTES" ) diff --git a/strix/tools/output_store.py b/strix/tools/output_store.py index 6e6e62026..ed65dfa8c 100644 --- a/strix/tools/output_store.py +++ b/strix/tools/output_store.py @@ -1,10 +1,7 @@ """Bound oversized tool results before they enter agent history. -A single verbose tool result (a recursive ``find``, a noisy scanner, a full -page dump) can otherwise pin the whole conversation near the model's context -limit for the rest of the scan. This keeps a head + tail slice of the output -and drops the middle, mirroring how the shell capability truncates its own -output — the agent still sees the start and end plus how much was removed. +Keeps a head + tail slice and drops the middle, replacing it with a notice of +how much was removed. """ from __future__ import annotations @@ -45,22 +42,16 @@ def _take_suffix(text: str, max_bytes: int) -> str: def bound_text(text: str, *, max_lines: int, max_bytes: int) -> str: """Return ``text`` unchanged when small, else a head+tail preview. - Truncation happens on whichever limit is hit first (line count or UTF-8 - byte size). The removed middle is replaced with a notice recording how - many lines and bytes were dropped so the agent knows output was elided. - ``max_bytes`` bounds the *entire* joined result, notice and separators - included, and must be large enough to hold the notice itself (guaranteed by - the ``tool_output_max_bytes`` config floor). + Truncates on whichever limit is hit first (line count or UTF-8 byte size). + ``max_bytes`` bounds the entire joined result, notice and separators + included. """ lines = text.split("\n") total_bytes = _byte_len(text) if len(lines) <= max_lines and total_bytes <= max_bytes: return text - # Reserve room for the notice and its two blank-line separators so the - # head+tail slices can't consume the whole budget and push the persisted - # value over max_bytes. Upper-bound the notice with the largest possible - # counts; the real notice is never longer. ``+ 4`` covers the separators. + # Reserve notice + separator bytes up front; ``+ 4`` covers the two "\n\n". notice_overhead = _byte_len(_TRUNCATION_NOTICE.format(lines=len(lines), bytes=total_bytes)) + 4 byte_budget = max(2, max_bytes - notice_overhead) @@ -69,16 +60,13 @@ def bound_text(text: str, *, max_lines: int, max_bytes: int) -> str: head = "\n".join(lines[:head_lines]) tail = "\n".join(lines[len(lines) - tail_lines :]) if tail_lines > 0 else "" - # Enforce the byte budget even when the line count alone was fine. half_bytes = max(1, byte_budget // 2) if _byte_len(head) > half_bytes: head = _take_prefix(head, half_bytes) if tail and _byte_len(tail) > half_bytes: tail = _take_suffix(tail, half_bytes) - # Count kept lines from the final slices: the byte pass above may have - # dropped whole lines from head/tail, so deriving this from the original - # head_lines/tail_lines would undercount what was actually removed. + # Count from the final slices; the byte pass may have dropped whole lines. kept_lines = len(head.split("\n")) + (len(tail.split("\n")) if tail else 0) dropped_lines = max(0, len(lines) - kept_lines) dropped_bytes = max(0, total_bytes - _byte_len(head) - _byte_len(tail)) diff --git a/tests/test_agent_factory_shell.py b/tests/test_agent_factory_shell.py index 22b662c74..814da60c8 100644 --- a/tests/test_agent_factory_shell.py +++ b/tests/test_agent_factory_shell.py @@ -82,8 +82,6 @@ async def test_wrap_exec_command_preserves_explicit_shell(shell: str) -> None: @pytest.mark.asyncio async def test_responses_filesystem_custom_tool_output_is_bounded() -> None: - # In Responses-API mode filesystem tools stay native CustomTools; a large - # read must still be head+tail bounded before it enters history. async def invoke(_ctx: Any, _inp: str) -> str: return "line\n" * 50_000 diff --git a/tests/test_config_loader.py b/tests/test_config_loader.py index 0d37e1e89..7e14662ae 100644 --- a/tests/test_config_loader.py +++ b/tests/test_config_loader.py @@ -121,14 +121,7 @@ def test_read_json_overrides_uses_json_when_no_alias_in_environ(tmp_path: Path) assert loader._read_json_overrides(path) == {"llm": {"api_key": "sk-file"}} -# --------------------------------------------------------------------------- # -# ContextSettings validation -# --------------------------------------------------------------------------- # - - def test_tool_output_max_bytes_rejects_sub_notice_values() -> None: - # A ceiling below the truncation notice can't fit a bounded preview, so it - # is rejected at load time rather than producing over-cap persisted output. with pytest.raises(ValidationError): ContextSettings(STRIX_TOOL_OUTPUT_MAX_BYTES=64) diff --git a/tests/test_output_store.py b/tests/test_output_store.py index 14bff19c1..4db8ccf75 100644 --- a/tests/test_output_store.py +++ b/tests/test_output_store.py @@ -19,7 +19,6 @@ def test_line_limit_keeps_head_and_tail() -> None: 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 @@ -28,8 +27,6 @@ def test_byte_limit_enforced_on_single_long_line() -> None: bounded = bound_text(text, max_lines=2_000, max_bytes=1_000) assert "truncated" in bounded - # The whole joined result (head + tail + notice + separators) honours the - # configured maximum, not just the head/tail slices. assert len(bounded.encode("utf-8")) <= 1_000 @@ -37,7 +34,7 @@ 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). + # Must remain valid UTF-8 (no mid-character cut). assert bounded == bounded.encode("utf-8").decode("utf-8") assert "truncated" in bounded @@ -51,8 +48,7 @@ def test_notice_reports_dropped_counts() -> None: def test_dropped_line_count_accounts_for_byte_trimming() -> None: - # A tight byte budget forces the byte pass to drop whole lines from the - # head/tail slices; the notice must count those, not just the middle. + # Tight byte budget drops whole lines from head/tail; the notice must count them. text = "\n".join(f"line-{i}" for i in range(200)) bounded = bound_text(text, max_lines=20, max_bytes=40) @@ -61,5 +57,4 @@ def test_dropped_line_count_accounts_for_byte_trimming() -> None: dropped = int(match.group(1)) kept = [ln for ln in bounded.splitlines() if ln and "truncated" not in ln] assert dropped == 200 - len(kept) - # The naive middle-only count (max_lines split evenly) would under-report. assert dropped > 200 - 20