fix: harden remaining display-layer surfaces against non-string tool args

Consolidation follow-up on top of @liuwei666888's compressor fix (#52431)
and @Frowtek's ACP render guard (#62588) — completes the class fix across
every display-layer consumer of tool-call arguments:

- agent/context_compressor.py: wrap _summarize_tool_result in a
  never-raises backstop (same pattern as get_cute_tool_message and the
  ACP guard). The per-branch _str_arg coercions keep summaries
  informative; the wrapper guarantees compression can never crash-loop
  on a summary branch we didn't anticipate. Also reject non-dict parsed
  args (a JSON list/scalar would crash every args.get call).
- agent/display.py: build_tool_preview's process branch sliced
  session_id/data without coercion — crashed build_tool_preview and
  build_tool_label on the live tool-progress callback (probed live:
  TypeError 'int' object is not subscriptable). Coerce like the
  sibling branches.
- tests: backstop fuzz matrix (16 tools x 6 hostile value shapes),
  fallback-shape contracts, display process-preview regressions,
  non-dict args guard.
This commit is contained in:
Teknium 2026-07-14 22:12:54 -07:00
parent 05473428cb
commit 3804df5b36
3 changed files with 102 additions and 3 deletions

View file

@ -613,11 +613,30 @@ def _summarize_tool_result(tool_name: str, tool_args: str, tool_content: str) ->
[terminal] ran `npm test` -> exit 0, 47 lines output
[read_file] read config.py from line 1 (1,200 chars)
[search_files] content search for 'compress' in agent/ -> 12 matches
Never raises: models sometimes emit non-string argument values (bool,
int, None) and the args here come from persisted session history, so a
single malformed historical call must not crash compression which
retries on the same history and would crash-loop. Individual branches
coerce the values they slice/measure (keeping summaries informative);
this wrapper is the backstop for anything they miss.
"""
try:
return _summarize_tool_result_unguarded(tool_name, tool_args, tool_content)
except Exception as exc: # noqa: BLE001 — a summary must never crash compression
logger.debug("Tool-result summary failed for %s: %s", tool_name, exc)
_len = len(tool_content) if isinstance(tool_content, str) else 0
return f"[{tool_name}] ({_len:,} chars result)"
def _summarize_tool_result_unguarded(tool_name: str, tool_args: str, tool_content: str) -> str:
"""Build the summary line (unguarded; see ``_summarize_tool_result``)."""
try:
args = json.loads(tool_args) if tool_args else {}
except (json.JSONDecodeError, TypeError):
args = {}
if not isinstance(args, dict):
args = {}
content = tool_content or ""
content_len = len(content)

View file

@ -462,13 +462,14 @@ def build_tool_preview(tool_name: str, args: dict, max_len: int | None = None) -
sid = args.get("session_id", "")
data = args.get("data", "")
timeout_val = args.get("timeout")
parts = [action]
parts = [str(action) if action else ""]
if sid:
parts.append(sid[:16])
parts.append(str(sid)[:16])
if data:
parts.append(f'"{_oneline(data[:20])}"')
parts.append(f'"{_oneline(str(data)[:20])}"')
if timeout_val and action == "wait":
parts.append(f"{timeout_val}s")
parts = [p for p in parts if p]
return " ".join(parts) if parts else None
if tool_name == "todo":

View file

@ -187,6 +187,13 @@ class TestEdgeCases:
result = _summarize_tool_result("terminal", None, "output")
assert "terminal" in result
def test_non_dict_json_args(self):
"""Args that parse to a non-dict (list/scalar) should not crash."""
result = _summarize_tool_result("terminal", json.dumps([1, 2]), "output")
assert "terminal" in result
result = _summarize_tool_result("terminal", json.dumps("bare"), "output")
assert "terminal" in result
def test_unknown_tool_name(self):
"""Unknown tool name should return generic summary."""
args = json.dumps({"foo": "bar"})
@ -205,3 +212,75 @@ class TestEdgeCases:
result = _summarize_tool_result("terminal", args, '{"exit_code": 0}')
assert "terminal" in result
assert "ls" in result
class TestBackstopWrapper:
"""The outer guard: NO input shape may raise out of _summarize_tool_result.
Compression retries on the same persisted history, so an escaping
exception here becomes a crash loop. The wrapper returns a minimal
'[tool] (N chars result)' summary when a branch fails.
"""
def test_never_raises_matrix(self):
"""Fuzz the per-tool branches with hostile value shapes."""
hostile_values = [None, True, 42, 3.14, ["a"], {"k": "v"}]
tools = [
"terminal", "read_file", "write_file", "search_files", "patch",
"browser_navigate", "web_search", "web_extract", "delegate_task",
"execute_code", "skill_view", "vision_analyze", "memory",
"cronjob", "process", "totally_unknown_tool",
]
keys = ["command", "path", "content", "pattern", "url", "query",
"urls", "goal", "code", "name", "question", "action",
"target", "session_id", "mode", "offset", "ref"]
for tool in tools:
for value in hostile_values:
args = json.dumps({k: value for k in keys})
result = _summarize_tool_result(tool, args, "x" * 250)
assert isinstance(result, str) and result, (tool, value)
def test_backstop_fallback_shape(self):
"""When a branch does fail, the fallback names the tool and size."""
from unittest.mock import patch as _patch
with _patch(
"agent.context_compressor._summarize_tool_result_unguarded",
side_effect=TypeError("boom"),
):
result = _summarize_tool_result("terminal", "{}", "y" * 300)
assert result == "[terminal] (300 chars result)"
def test_backstop_handles_non_string_content(self):
from unittest.mock import patch as _patch
with _patch(
"agent.context_compressor._summarize_tool_result_unguarded",
side_effect=TypeError("boom"),
):
result = _summarize_tool_result("terminal", "{}", None)
assert result == "[terminal] (0 chars result)"
class TestDisplayPreviewTypeSafety:
"""Sibling site: agent/display.py previews run on the live
tool-progress callback and crashed on non-string process args."""
def test_process_preview_non_string_session_id(self):
from agent.display import build_tool_preview
assert build_tool_preview("process", {"action": "poll", "session_id": 123}) == "poll 123"
def test_process_preview_non_string_data(self):
from agent.display import build_tool_preview
result = build_tool_preview(
"process", {"action": "submit", "session_id": "abc", "data": 42}
)
assert result == 'submit abc "42"'
def test_process_preview_none_action(self):
from agent.display import build_tool_preview
result = build_tool_preview("process", {"action": None, "session_id": "abc"})
assert isinstance(result, str)
def test_process_label_non_string_session_id(self):
from agent.display import build_tool_label
result = build_tool_label("process", {"action": "poll", "session_id": 123})
assert isinstance(result, str)