fix(core): backstop single-chunk overflow; add helper tests

Code review findings:

- factory._wrap_exec_command: add post-compression backstop that calls
  truncate_exec_result when compress_exec_result returns an oversized
  result (happens when _split produces 1 chunk, e.g. a single very
  large JSON record — compress_large_output correctly returns it
  unchanged, but the result was never guarded against threshold after).

- tests/agents/test_factory_helpers.py (new): unit tests for
  _resolve_model (run_config vs settings fallback, None/whitespace
  error cases, non-string Model object guard) and _extract_task_hint
  (valid cmd, missing cmd, non-string cmd, invalid JSON, non-dict JSON).
This commit is contained in:
Mads Hvelplund 2026-06-19 13:17:37 +02:00
parent eac81f8693
commit 0640e5c9ea
No known key found for this signature in database
GPG key ID: 08208EA18E154F00
4 changed files with 102 additions and 1 deletions

View file

@ -268,6 +268,10 @@ def _wrap_exec_command(tool: FunctionTool) -> FunctionTool:
exc_info=True,
)
result = truncate_exec_result(result, threshold=threshold)
# Backstop: a single oversized chunk returns unchanged from
# compress_large_output; truncate it so the session stays safe.
if len(result) > threshold:
result = truncate_exec_result(result, threshold=threshold)
return result
tool.on_invoke_tool = invoke

0
tests/agents/__init__.py Normal file
View file

View file

@ -0,0 +1,93 @@
"""Tests for helper functions added to strix/agents/factory.py."""
from __future__ import annotations
import json
from unittest.mock import MagicMock, patch
import pytest
from strix.agents.factory import _extract_task_hint, _resolve_model
# ---------------------------------------------------------------------------
# _resolve_model
# ---------------------------------------------------------------------------
def test_resolve_model_uses_run_config_model() -> None:
"""Prefers ctx.run_config.model when it is a non-empty string."""
ctx = MagicMock()
ctx.run_config.model = "claude-sonnet-4-6"
assert _resolve_model(ctx) == "claude-sonnet-4-6"
def test_resolve_model_strips_whitespace() -> None:
ctx = MagicMock()
ctx.run_config.model = " claude-haiku-4-5 "
assert _resolve_model(ctx) == "claude-haiku-4-5"
def test_resolve_model_skips_model_object_falls_back_to_settings() -> None:
"""A non-string run_config.model (e.g. Model object) must be ignored."""
ctx = MagicMock()
ctx.run_config.model = object() # not a str
with patch("strix.agents.factory.load_settings") as mock_settings:
mock_settings.return_value.llm.model = "anthropic/claude-3-5-haiku"
result = _resolve_model(ctx)
assert result == "anthropic/claude-3-5-haiku"
def test_resolve_model_falls_back_to_settings_when_run_config_is_none() -> None:
ctx = MagicMock()
ctx.run_config = None
with patch("strix.agents.factory.load_settings") as mock_settings:
mock_settings.return_value.llm.model = "gpt-4o"
result = _resolve_model(ctx)
assert result == "gpt-4o"
def test_resolve_model_raises_when_both_none() -> None:
ctx = MagicMock()
ctx.run_config.model = None
with patch("strix.agents.factory.load_settings") as mock_settings:
mock_settings.return_value.llm.model = None
with pytest.raises(RuntimeError, match="No LLM model configured"):
_resolve_model(ctx)
def test_resolve_model_raises_when_settings_model_is_whitespace() -> None:
ctx = MagicMock()
ctx.run_config.model = None
with patch("strix.agents.factory.load_settings") as mock_settings:
mock_settings.return_value.llm.model = " "
with pytest.raises(RuntimeError, match="No LLM model configured"):
_resolve_model(ctx)
# ---------------------------------------------------------------------------
# _extract_task_hint
# ---------------------------------------------------------------------------
def test_extract_task_hint_returns_cmd_field() -> None:
raw = json.dumps({"cmd": "semgrep --json --config auto /workspace"})
assert _extract_task_hint(raw) == "semgrep --json --config auto /workspace"
def test_extract_task_hint_returns_empty_for_invalid_json() -> None:
assert _extract_task_hint("not json at all") == ""
def test_extract_task_hint_returns_empty_when_cmd_missing() -> None:
raw = json.dumps({"workdir": "/workspace"})
assert _extract_task_hint(raw) == ""
def test_extract_task_hint_returns_empty_when_cmd_is_not_string() -> None:
raw = json.dumps({"cmd": 42})
assert _extract_task_hint(raw) == ""
def test_extract_task_hint_returns_empty_for_non_object_json() -> None:
assert _extract_task_hint(json.dumps(["a", "b"])) == ""

View file

@ -143,7 +143,11 @@ async def test_summarise_returns_placeholder_on_timeout() -> None:
async def test_compress_single_chunk_returns_output_unchanged() -> None:
"""≤ CHUNK_SIZE_RECORDS records → 1 chunk → returned unchanged, no LLM call."""
"""≤ CHUNK_SIZE_RECORDS records → 1 chunk → returned unchanged, no LLM call.
The caller (factory._wrap_exec_command) owns the post-compression backstop
for the single-oversized-chunk case; compress_large_output itself is a no-op.
"""
records = [{"id": i} for i in range(50)]
output = json.dumps(records)
with patch(