diff --git a/strix/agents/factory.py b/strix/agents/factory.py index 977d01f19..35e5812da 100644 --- a/strix/agents/factory.py +++ b/strix/agents/factory.py @@ -16,6 +16,8 @@ from pydantic import ValidationError from strix.agents.prompt import render_system_prompt +from strix.config import load_settings +from strix.core.tool_output import truncate_tool_output from strix.tools.agents_graph.tools import ( agent_finish, create_agent, @@ -217,7 +219,7 @@ async def invoke(ctx: Any, raw_input: str) -> Any: parsed["shell"] = "bash" raw_input = json.dumps(parsed) try: - return await invoke_tool(ctx, raw_input) + result = await invoke_tool(ctx, raw_input) except ValidationError as exc: return _format_validation_error(tool.name, exc) except InvalidManifestPathError as exc: @@ -227,6 +229,20 @@ async def invoke(ctx: Any, raw_input: str) -> Any: "(or omitted to use the turn's cwd). " f"Got: {rel!r}." ) + # Cap scanner dumps before they land in SQLite session history and + # get re-serialized into the next model request (context overflow). + if isinstance(result, str): + max_chars = load_settings().runtime.max_tool_output_chars + truncated = truncate_tool_output(result, max_chars) + if truncated != result: + logger.info( + "Truncated exec_command output from %d to %d chars (cap=%d)", + len(result), + len(truncated), + max_chars, + ) + return truncated + return result tool.on_invoke_tool = invoke return tool diff --git a/strix/config/settings.py b/strix/config/settings.py index 9d16e0836..36af409b4 100644 --- a/strix/config/settings.py +++ b/strix/config/settings.py @@ -56,6 +56,13 @@ class RuntimeSettings(BaseSettings): # on large repos). Above this, the user must bind-mount via ``--mount``. # Set to 0 (or less) to disable the pre-flight check entirely. max_local_copy_mb: int = Field(default=1024, alias="STRIX_MAX_LOCAL_COPY_MB") + # Cap shell/tool stdout stored in the agent session. Unbounded scanner + # dumps (e.g. semgrep JSON on a monorepo) otherwise re-enter the next LLM + # request and raise ContextWindowExceededError. Set to 0 to disable. + max_tool_output_chars: int = Field( + default=65_536, + alias="STRIX_MAX_TOOL_OUTPUT_CHARS", + ) class TelemetrySettings(BaseSettings): diff --git a/strix/core/execution.py b/strix/core/execution.py index 06dc3ddf5..addc603d8 100644 --- a/strix/core/execution.py +++ b/strix/core/execution.py @@ -15,9 +15,15 @@ from docker import errors as docker_errors # type: ignore[import-untyped, unused-ignore] from openai import APIError +from strix.config import load_settings from strix.core.hooks import BudgetExceededError from strix.core.inputs import child_initial_input -from strix.core.sessions import open_agent_session, strip_all_images_from_session +from strix.core.sessions import ( + open_agent_session, + strip_all_images_from_session, + truncate_large_outputs_in_session, +) +from strix.core.tool_output import is_context_window_error if TYPE_CHECKING: @@ -346,6 +352,7 @@ async def _run_cycle( # noqa: PLR0912, PLR0915 hooks: RunHooks[dict[str, Any]] | None, ) -> RunResultBase | None: image_strips = 0 + context_truncations = 0 while True: try: await coordinator.mark_running(agent_id) @@ -398,10 +405,50 @@ async def _run_cycle( # noqa: PLR0912, PLR0915 await coordinator.trigger_budget_stop() raise except Exception as exc: + # Context-window overflows often arrive as HTTP 400. Image-strip + # recovery cannot help (there are no images) and only burns + # retries — truncate oversized tool outputs instead. + if ( + context_truncations < 2 + and session is not None + and is_context_window_error(exc) + ): + max_chars = load_settings().runtime.max_tool_output_chars + try: + truncated = await truncate_large_outputs_in_session( + session, + max_chars=max_chars, + ) + except Exception: + logger.exception("context-window truncation recovery failed for %s", agent_id) + truncated = False + if truncated: + context_truncations += 1 + logger.warning( + "Truncated oversized tool outputs for %s after context-window " + "error; retrying (%d)", + agent_id, + context_truncations, + ) + input_data = [ + { + "role": "user", + "content": ( + "Your previous tool output exceeded the model context window " + "and was truncated in session history. Continue from the " + "truncated results: write full scanner dumps to files under " + "/workspace, then inspect samples with head/rg/jq instead of " + "printing entire monorepo findings." + ), + } + ] + continue + if ( image_strips < 3 and session is not None and getattr(exc, "status_code", None) in _INPUT_REJECTION_CODES + and not is_context_window_error(exc) ): try: stripped = await strip_all_images_from_session(session) @@ -421,7 +468,9 @@ async def _run_cycle( # noqa: PLR0912, PLR0915 raise if isinstance(exc, MaxTurnsExceeded): status: Status = "stopped" - elif isinstance(exc, UserError | AgentsException | APIError): + elif isinstance(exc, UserError | AgentsException | APIError) or is_context_window_error( + exc + ): status = "failed" else: status = "crashed" diff --git a/strix/core/sessions.py b/strix/core/sessions.py index 8e4f4142d..668afbf4f 100644 --- a/strix/core/sessions.py +++ b/strix/core/sessions.py @@ -3,10 +3,13 @@ from __future__ import annotations import contextlib +import logging from typing import TYPE_CHECKING, Any, cast from agents.memory import SQLiteSession +from strix.core.tool_output import DEFAULT_MAX_TOOL_OUTPUT_CHARS, truncate_tool_output + if TYPE_CHECKING: from pathlib import Path @@ -15,6 +18,9 @@ from agents.memory import Session +logger = logging.getLogger(__name__) + + def open_agent_session(agent_id: str, path: Path) -> SQLiteSession: path.parent.mkdir(parents=True, exist_ok=True) return SQLiteSession(session_id=agent_id, db_path=path) @@ -23,6 +29,32 @@ def open_agent_session(agent_id: str, path: Path) -> SQLiteSession: _IMAGE_REJECTED_TEXT = "[image rejected by the model]" +async def _rewrite_session_items( + session: Session, + rebuilt: list[Any], + *, + previous: list[Any], +) -> None: + """Replace session items with ``rebuilt``, restoring ``previous`` on failure. + + The SDK only exposes clear+add, so a failed write after clear would otherwise + wipe history. Always keep a snapshot of the prior items and re-insert it if + the new payload cannot be persisted. + """ + rebuilt_items = cast("list[TResponseInputItem]", rebuilt) + previous_items = cast("list[TResponseInputItem]", list(previous)) + await session.clear_session() + try: + await session.add_items(rebuilt_items) + except Exception: + logger.exception("Failed to write rebuilt session items; restoring prior history") + with contextlib.suppress(Exception): + await session.clear_session() + if previous_items: + await session.add_items(previous_items) + raise + + async def strip_all_images_from_session(session: Session) -> bool: items = await session.get_items() if not items: @@ -54,12 +86,72 @@ async def strip_all_images_from_session(session: Session) -> bool: if not changed: return False - rebuilt_items = cast("list[TResponseInputItem]", rebuilt) - await session.clear_session() - try: - await session.add_items(rebuilt_items) - except Exception: - with contextlib.suppress(Exception): - await session.add_items(rebuilt_items) - raise + await _rewrite_session_items(session, rebuilt, previous=items) + return True + + +def _truncate_function_output_value(output: Any, max_chars: int) -> tuple[Any, bool]: + """Return ``(new_output, changed)`` for one function_call_output payload.""" + if isinstance(output, str): + truncated = truncate_tool_output(output, max_chars) + return truncated, truncated != output + + if isinstance(output, list): + new_blocks: list[Any] = [] + changed = False + for block in output: + if isinstance(block, dict) and block.get("type") == "input_text": + text = block.get("text") + if isinstance(text, str): + truncated = truncate_tool_output(text, max_chars) + if truncated != text: + changed = True + new_blocks.append({**block, "text": truncated}) + continue + new_blocks.append(block) + return new_blocks, changed + + return output, False + + +async def truncate_large_outputs_in_session( + session: Session, + max_chars: int = DEFAULT_MAX_TOOL_OUTPUT_CHARS, +) -> bool: + """Truncate oversized ``function_call_output`` items already in the session. + + Used as recovery after a context-window rejection: huge tool results + (semgrep dumps, etc.) sit in SQLite and get resent on the next turn. + Returns True when at least one item was shortened. + """ + if max_chars <= 0: + return False + + items = await session.get_items() + if not items: + return False + + rebuilt: list[Any] = [] + changed = False + for item in items: + item_dict = cast("dict[str, Any]", item) if isinstance(item, dict) else None + if item_dict is not None and item_dict.get("type") == "function_call_output": + new_output, item_changed = _truncate_function_output_value( + item_dict.get("output"), + max_chars, + ) + if item_changed: + changed = True + rebuilt.append({**item_dict, "output": new_output}) + continue + rebuilt.append(item) + + if not changed: + return False + + await _rewrite_session_items(session, rebuilt, previous=items) + logger.info( + "Truncated oversized tool outputs in session (cap=%d chars)", + max_chars, + ) return True diff --git a/strix/core/tool_output.py b/strix/core/tool_output.py new file mode 100644 index 000000000..a596a2c46 --- /dev/null +++ b/strix/core/tool_output.py @@ -0,0 +1,211 @@ +"""Helpers for capping tool outputs that would blow the LLM context window.""" + +from __future__ import annotations + +import json +import logging +import re +from typing import Any + + +logger = logging.getLogger(__name__) + +# Defaults match the practical ceiling used in community fix PR #583. +DEFAULT_MAX_TOOL_OUTPUT_CHARS = 65_536 +DEFAULT_MAX_JSON_RECORDS = 50 +DEFAULT_MAX_LINES = 300 + +_CONTEXT_WINDOW_MARKERS = ( + "contextwindowexceeded", + "context window", + "context_length", + "maximum context", + "max context", + "prompt is too long", + "prompt too long", + "input is too long", + "too many tokens", + "token limit", + "maximum number of tokens", +) + + +def is_context_window_error(exc: BaseException) -> bool: + """Return True when ``exc`` indicates an LLM context-window overflow.""" + name = type(exc).__name__.lower() + if "contextwindow" in name.replace("_", ""): + return True + + # Walk the exception chain (LiteLLM often wraps provider errors). + seen: set[int] = set() + current: BaseException | None = exc + while current is not None and id(current) not in seen: + seen.add(id(current)) + text = str(current).lower() + if any(marker in text for marker in _CONTEXT_WINDOW_MARKERS): + return True + # "1,023,797 tokens > 1,000,000" style messages + if re.search(r"\d[\d,]*\s*tokens?\s*>\s*\d", text): + return True + current = current.__cause__ or current.__context__ # type: ignore[assignment] + return False + + +def truncate_tool_output( + text: str, + max_chars: int, + *, + max_json_records: int = DEFAULT_MAX_JSON_RECORDS, + max_lines: int = DEFAULT_MAX_LINES, +) -> str: + """Cap a tool result string so it can fit in subsequent model turns. + + ``max_chars <= 0`` disables truncation. Prefer structured JSON trimming + when the payload is a JSON array (typical of scanners like semgrep); + otherwise keep the first ``max_lines`` lines, then hard-cap by chars. + """ + if max_chars <= 0 or len(text) <= max_chars: + return text + + original_len = len(text) + stripped = text.lstrip() + if stripped.startswith(("[", "{")): + truncated_json = _truncate_json_payload( + stripped, + max_chars=max_chars, + max_json_records=max_json_records, + original_len=original_len, + ) + if truncated_json is not None: + return truncated_json + + return _truncate_text_payload( + text, + max_chars=max_chars, + max_lines=max_lines, + original_len=original_len, + ) + + +def _hard_cap(text: str, max_chars: int) -> str: + """Return ``text`` guaranteed to be at most ``max_chars`` characters.""" + if max_chars <= 0 or len(text) <= max_chars: + return text + marker = "\n...[truncated]" + if max_chars <= len(marker): + return text[:max_chars] + return text[: max_chars - len(marker)] + marker + + +def _truncate_json_array( + data: list[Any], + *, + max_chars: int, + max_json_records: int, + original_len: int, +) -> str: + total = len(data) + kept = data[: max(1, max_json_records)] + # Shrink further if even the first records exceed the char budget. + while kept: + body = json.dumps(kept, ensure_ascii=False, indent=2) + header = ( + f"[truncated JSON array: showing {len(kept)}/{total} records; " + f"original {original_len} chars. Re-run with a narrower path/" + f"rule filter or write results to a file and inspect samples.]\n" + ) + candidate = header + body + if len(candidate) <= max_chars: + return candidate + if len(kept) == 1: + return _hard_cap(candidate, max_chars) + kept = kept[: max(1, len(kept) // 2)] + return _hard_cap(json.dumps(data[:1], ensure_ascii=False), max_chars) + + +def _truncate_json_object( + data: dict[str, Any], + *, + max_chars: int, + max_json_records: int, + original_len: int, +) -> str: + body = json.dumps(data, ensure_ascii=False, indent=2) + if len(body) <= max_chars: + return body + header = ( + f"[truncated JSON object; original {original_len} chars. " + f"Write full output to a file and inspect targeted fields.]\n" + ) + slim: dict[str, Any] = {} + candidate = header + "{}" + for key, value in data.items(): + if isinstance(value, (str, int, float, bool)) or value is None: + slim[key] = value + elif isinstance(value, list): + slim[key] = value[: max(1, max_json_records)] + slim[f"{key}__truncated"] = True + else: + slim[key] = f"<{type(value).__name__} omitted>" + candidate = header + json.dumps(slim, ensure_ascii=False, indent=2) + if len(candidate) > max_chars: + break + return _hard_cap(candidate, max_chars) + + +def _truncate_json_payload( + text: str, + *, + max_chars: int, + max_json_records: int, + original_len: int, +) -> str | None: + try: + data = json.loads(text) + except (json.JSONDecodeError, TypeError, ValueError): + return None + + if isinstance(data, list): + return _truncate_json_array( + data, + max_chars=max_chars, + max_json_records=max_json_records, + original_len=original_len, + ) + if isinstance(data, dict): + return _truncate_json_object( + data, + max_chars=max_chars, + max_json_records=max_json_records, + original_len=original_len, + ) + return None + + +def _truncate_text_payload( + text: str, + *, + max_chars: int, + max_lines: int, + original_len: int, +) -> str: + lines = text.splitlines() + if len(lines) > max_lines: + head = "\n".join(lines[:max_lines]) + notice = ( + f"[truncated: showing first {max_lines}/{len(lines)} lines; " + f"original {original_len} chars. Pipe output to a file and read " + f"targeted sections instead of dumping everything.]\n" + ) + candidate = notice + head + else: + candidate = text + + if len(candidate) <= max_chars: + return candidate + notice = f"[truncated to {max_chars} chars; original {original_len} chars]\n" + marker = "\n...[truncated]" + budget = max_chars - len(notice) - len(marker) + if budget < 1: + return _hard_cap(candidate, max_chars) + return notice + candidate[:budget] + marker diff --git a/strix/interface/tui/app.py b/strix/interface/tui/app.py index 73a7ec0de..de814d4b9 100644 --- a/strix/interface/tui/app.py +++ b/strix/interface/tui/app.py @@ -1476,13 +1476,13 @@ def scan_target() -> None: logger.info("Scan stopped: --max-budget-usd limit reached") except (ConnectionError, TimeoutError) as e: logging.exception("Network error during scan") - self._scan_error = e + self._set_scan_error(e) except RuntimeError as e: logging.exception("Runtime error during scan") - self._scan_error = e + self._set_scan_error(e) except Exception as e: logging.exception("Unexpected error during scan") - self._scan_error = e + self._set_scan_error(e) finally: with contextlib.suppress(Exception): loop.run_until_complete( @@ -1498,6 +1498,27 @@ def scan_target() -> None: self._scan_thread = threading.Thread(target=scan_target, daemon=True) self._scan_thread.start() + def _set_scan_error(self, error: BaseException) -> None: + """Record a scan failure and surface it in-app immediately. + + Previously errors were only re-raised after the TUI exited, so users + saw a silent hang until they quit. Post a toast from the scan thread + via ``call_from_thread`` so the failure is visible while the app runs. + """ + self._scan_error = error + message = f"{type(error).__name__}: {error}" + if len(message) > 280: + message = message[:277] + "..." + try: + self.call_from_thread( + self.notify, + f"Scan failed — {message}", + severity="error", + ) + except RuntimeError: + # App not running / no active loop — exit path still raises. + logger.debug("Could not show scan-error toast", exc_info=True) + def _capture_sdk_event(self, agent_id: str, event: Any) -> None: try: self.call_from_thread(self._record_sdk_event, agent_id, event) diff --git a/tests/test_context_window_recovery.py b/tests/test_context_window_recovery.py new file mode 100644 index 000000000..084499f4a --- /dev/null +++ b/tests/test_context_window_recovery.py @@ -0,0 +1,217 @@ +"""Tests for context-window recovery branching in the agent run cycle.""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from strix.core.execution import _run_cycle + + +class ContextWindowExceededError(Exception): + status_code = 400 + + def __init__(self) -> None: + super().__init__("1,023,797 tokens > 1,000,000 max") + + +class ImageRejectedError(Exception): + status_code = 400 + + def __init__(self) -> None: + super().__init__("invalid image content") + + +class _FailOnceStream: + def __init__(self, exc: BaseException) -> None: + self._exc = exc + self.run_loop_exception = None + + async def stream_events(self) -> Any: + raise self._exc + yield # pragma: no cover — force async-generator type for `async for` + + +@pytest.mark.asyncio +async def test_context_window_error_uses_truncation_not_image_strip() -> None: + coordinator = MagicMock() + coordinator.mark_running = AsyncMock() + coordinator.attach_stream = AsyncMock() + coordinator.detach_stream = AsyncMock() + coordinator.set_status = AsyncMock() + coordinator._lock = AsyncMock() + coordinator._lock.__aenter__ = AsyncMock(return_value=None) + coordinator._lock.__aexit__ = AsyncMock(return_value=None) + coordinator.statuses = {"agent": "running"} + coordinator.parent_of = {} + coordinator.names = {"agent": "strix"} + coordinator.is_shutting_down = False + + session = MagicMock() + stream1 = _FailOnceStream(ContextWindowExceededError()) + stream2 = MagicMock() + stream2.run_loop_exception = None + + async def _empty_events() -> Any: + if False: # pragma: no cover + yield None + return + + stream2.stream_events = _empty_events + + with ( + patch("strix.core.execution.Runner") as runner_cls, + patch( + "strix.core.execution.truncate_large_outputs_in_session", + new_callable=AsyncMock, + return_value=True, + ) as trunc, + patch( + "strix.core.execution.strip_all_images_from_session", + new_callable=AsyncMock, + return_value=True, + ) as strip, + patch("strix.core.execution.load_settings") as settings, + ): + settings.return_value.runtime.max_tool_output_chars = 65_536 + runner_cls.run_streamed = MagicMock(side_effect=[stream1, stream2]) + + result = await _run_cycle( + agent=MagicMock(), + coordinator=coordinator, + agent_id="agent", + input_data=[], + run_config=MagicMock(), + context={"parent_id": None}, + max_turns=10, + session=session, + interactive=True, + event_sink=None, + hooks=None, + ) + + assert result is stream2 + trunc.assert_awaited() + strip.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_disabled_tool_output_cap_skips_truncation_recovery() -> None: + coordinator = MagicMock() + coordinator.mark_running = AsyncMock() + coordinator.attach_stream = AsyncMock() + coordinator.detach_stream = AsyncMock() + coordinator.set_status = AsyncMock() + coordinator._lock = AsyncMock() + coordinator._lock.__aenter__ = AsyncMock(return_value=None) + coordinator._lock.__aexit__ = AsyncMock(return_value=None) + coordinator.statuses = {"agent": "running"} + coordinator.parent_of = {"agent": None} + coordinator.names = {"agent": "strix"} + coordinator.is_shutting_down = False + coordinator.send = AsyncMock() + + session = MagicMock() + stream1 = _FailOnceStream(ContextWindowExceededError()) + + with ( + patch("strix.core.execution.Runner") as runner_cls, + patch( + "strix.core.execution.truncate_large_outputs_in_session", + new_callable=AsyncMock, + return_value=False, + ) as trunc, + patch( + "strix.core.execution.strip_all_images_from_session", + new_callable=AsyncMock, + return_value=False, + ) as strip, + patch("strix.core.execution.load_settings") as settings, + ): + settings.return_value.runtime.max_tool_output_chars = 0 + runner_cls.run_streamed = MagicMock(return_value=stream1) + + with pytest.raises(ContextWindowExceededError): + await _run_cycle( + agent=MagicMock(), + coordinator=coordinator, + agent_id="agent", + input_data=[], + run_config=MagicMock(), + context={"parent_id": None}, + max_turns=10, + session=session, + interactive=True, + event_sink=None, + hooks=None, + ) + + trunc.assert_awaited_once() + assert trunc.await_args.kwargs["max_chars"] == 0 + strip.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_image_rejection_still_uses_image_strip() -> None: + coordinator = MagicMock() + coordinator.mark_running = AsyncMock() + coordinator.attach_stream = AsyncMock() + coordinator.detach_stream = AsyncMock() + coordinator.set_status = AsyncMock() + coordinator._lock = AsyncMock() + coordinator._lock.__aenter__ = AsyncMock(return_value=None) + coordinator._lock.__aexit__ = AsyncMock(return_value=None) + coordinator.statuses = {"agent": "running"} + coordinator.parent_of = {} + coordinator.names = {"agent": "strix"} + coordinator.is_shutting_down = False + + session = MagicMock() + stream1 = _FailOnceStream(ImageRejectedError()) + stream2 = MagicMock() + stream2.run_loop_exception = None + + async def _empty_events() -> Any: + if False: # pragma: no cover + yield None + return + + stream2.stream_events = _empty_events + + with ( + patch("strix.core.execution.Runner") as runner_cls, + patch( + "strix.core.execution.truncate_large_outputs_in_session", + new_callable=AsyncMock, + return_value=False, + ) as trunc, + patch( + "strix.core.execution.strip_all_images_from_session", + new_callable=AsyncMock, + return_value=True, + ) as strip, + patch("strix.core.execution.load_settings") as settings, + ): + settings.return_value.runtime.max_tool_output_chars = 65_536 + runner_cls.run_streamed = MagicMock(side_effect=[stream1, stream2]) + + result = await _run_cycle( + agent=MagicMock(), + coordinator=coordinator, + agent_id="agent", + input_data=[], + run_config=MagicMock(), + context={"parent_id": None}, + max_turns=10, + session=session, + interactive=True, + event_sink=None, + hooks=None, + ) + + assert result is stream2 + # Image rejection is not a context-window error, so truncation path is skipped. + trunc.assert_not_awaited() + strip.assert_awaited() diff --git a/tests/test_tool_output.py b/tests/test_tool_output.py new file mode 100644 index 000000000..1838c722f --- /dev/null +++ b/tests/test_tool_output.py @@ -0,0 +1,168 @@ +"""Tests for context-window tool-output helpers.""" + +from __future__ import annotations + +import json +from typing import Any +from unittest.mock import AsyncMock + +import pytest + +from strix.core.sessions import truncate_large_outputs_in_session +from strix.core.tool_output import _hard_cap, is_context_window_error, truncate_tool_output + + +def test_is_context_window_error_by_class_name() -> None: + class ContextWindowExceededError(Exception): + status_code = 400 + + assert is_context_window_error(ContextWindowExceededError("boom")) + + +def test_is_context_window_error_by_message() -> None: + err = Exception("This model's maximum context length is 1000000 tokens") + assert is_context_window_error(err) + assert is_context_window_error(Exception("1,023,797 tokens > 1,000,000 max")) + + +def test_is_context_window_error_false_for_other_400() -> None: + class BadRequestError(Exception): + status_code = 400 + + assert not is_context_window_error(BadRequestError("invalid image")) + + +def test_truncate_disabled_when_max_non_positive() -> None: + text = "x" * 100_000 + assert truncate_tool_output(text, 0) == text + assert truncate_tool_output(text, -1) == text + + +def test_truncate_passes_small_text_through() -> None: + assert truncate_tool_output("hello", 100) == "hello" + + +def test_truncate_json_array_keeps_prefix_records() -> None: + payload = [{"id": i, "msg": "finding"} for i in range(200)] + raw = json.dumps(payload) + out = truncate_tool_output(raw, max_chars=4_000, max_json_records=50) + assert "truncated JSON array" in out + assert "showing" in out + # Must be valid-ish truncated payload under the cap. + assert len(out) <= 4_000 or out.endswith("...[truncated]") + assert "finding" in out + + +def test_truncate_text_by_lines() -> None: + # Each line is long enough that the full dump exceeds max_chars, so the + # line-cap path runs (short lines under the char budget are left alone). + lines = [f"line-{i}-" + ("x" * 80) for i in range(1_000)] + raw = "\n".join(lines) + out = truncate_tool_output(raw, max_chars=20_000, max_lines=300) + assert "truncated" in out + assert "line-0-" in out + assert "line-999-" not in out + assert len(out) < len(raw) + assert len(out) <= 20_000 or out.endswith("...[truncated]") + + +def test_truncate_hard_char_cap() -> None: + raw = "a" * 10_000 + out = truncate_tool_output(raw, max_chars=500, max_lines=10_000) + assert len(out) <= 500 + assert "truncated" in out + + +def test_hard_cap_never_exceeds_limit() -> None: + huge = "x" * 10_000 + for cap in (1, 5, 16, 100, 500): + out = _hard_cap(huge, cap) + assert len(out) <= cap + + +@pytest.mark.asyncio +async def test_truncate_session_restores_history_on_write_failure() -> None: + huge = "z" * 100_000 + original = [ + { + "type": "function_call_output", + "call_id": "c1", + "output": huge, + } + ] + session = _FakeSession(original) + session.add_items = AsyncMock(side_effect=[RuntimeError("disk full"), None]) + + with pytest.raises(RuntimeError, match="disk full"): + await truncate_large_outputs_in_session(session, max_chars=1_000) # type: ignore[arg-type] + + # clear (wipe) + failed add + clear (restore prep) + restore previous + assert session.clear_session.await_count == 2 + assert session.add_items.await_count == 2 + restored = session.add_items.await_args_list[1].args[0] + assert restored == original + + +class _FakeSession: + def __init__(self, items: list[Any]) -> None: + self._items = items + self.clear_session = AsyncMock() + self.add_items = AsyncMock() + + async def get_items(self) -> list[Any]: + return list(self._items) + + +@pytest.mark.asyncio +async def test_truncate_large_outputs_in_session_rewrites_string_output() -> None: + huge = "z" * 100_000 + session = _FakeSession( + [ + { + "type": "function_call_output", + "call_id": "c1", + "output": huge, + }, + {"type": "message", "role": "user", "content": "ok"}, + ] + ) + changed = await truncate_large_outputs_in_session(session, max_chars=1_000) # type: ignore[arg-type] + assert changed is True + session.clear_session.assert_awaited_once() + session.add_items.assert_awaited_once() + rewritten = session.add_items.await_args.args[0] + assert rewritten[0]["type"] == "function_call_output" + assert len(rewritten[0]["output"]) < len(huge) + assert "truncated" in rewritten[0]["output"] + + +@pytest.mark.asyncio +async def test_truncate_large_outputs_noop_when_small() -> None: + session = _FakeSession( + [ + { + "type": "function_call_output", + "call_id": "c1", + "output": "tiny", + } + ] + ) + changed = await truncate_large_outputs_in_session(session, max_chars=1_000) # type: ignore[arg-type] + assert changed is False + session.clear_session.assert_not_awaited() + + +def test_is_context_window_error_walks_cause_chain() -> None: + root = Exception("context window exceeded by model") + wrapped = RuntimeError("provider failed") + wrapped.__cause__ = root + assert is_context_window_error(wrapped) + + +def test_execution_skips_image_strip_predicate() -> None: + """Sanity: 400 alone is not enough to classify as context overflow.""" + + class Fake400Error(Exception): + status_code = 400 + + assert not is_context_window_error(Fake400Error("image rejected"))