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

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,16 +105,44 @@ 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 a tool's result size before it enters history (idempotent)."""
+    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:
     invoke_tool = tool.on_invoke_tool
 
     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 +157,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)
@@ -159,12 +189,35 @@ 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 (Responses path)."""
+    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))
+        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\\])")
@@ -205,6 +258,16 @@ 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:
+    """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"] = (
+        ceiling if not isinstance(requested, int) or requested > ceiling else requested
+    )
+
+
 def _wrap_exec_command(tool: FunctionTool) -> FunctionTool:
     invoke_tool = tool.on_invoke_tool
 
@@ -213,8 +276,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 +305,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 +507,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)",
@@ -459,8 +529,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(

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",

strix/config/settings.py

@@ -55,6 +55,26 @@ 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")
+    # 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"
+    )
+
+
 class RuntimeSettings(BaseSettings):
     model_config = _BASE_CONFIG
 
@@ -99,6 +119,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)

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
 
 

strix/tools/output_store.py

@@ -0,0 +1,74 @@
+"""Bound oversized tool results before they enter agent history.
+
+Keeps a head + tail slice and drops the middle, replacing it with a notice of
+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.
+
+    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 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)
+
+    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 ""
+
+    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 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))
+    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}"