From 6bda36606537e1fcad561ed8100f7c5e4c3d9017 Mon Sep 17 00:00:00 2001 From: Ahmed Allam Date: Sat, 25 Jul 2026 22:50:50 +0000 Subject: [PATCH] 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 ea80fb61..d6b36722 100644 --- a/strix/agents/factory.py +++ b/strix/agents/factory.py @@ -199,12 +199,43 @@ def _custom_tool_as_function_tool(tool: CustomTool) -> FunctionTool: ) -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\\])") @@ -523,8 +554,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 7165378c..22b662c7 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)