fix: stop infinite loop when assistant content is a block list

strip_think_blocks() ran re.sub() directly on content that could be a
list of blocks (Anthropic via OpenRouter returns assistant content as
[{type:text,...},{type:thinking,...}]). A list reaching re.sub raised
'TypeError: expected string or bytes-like object, got list', which the
outer conversation loop swallowed and retried forever — the observed
infinite 'preparing terminal...' loop that re-emitted the same
assistant text every iteration.

The live-turn path normalized list content to a string, but
_interim_assistant_visible_text reads a *stored* history message whose
content was persisted as a list and passes it straight into the shared
strip_think_blocks helper. Fix at the shared choke point: coerce
list/dict content to visible text (dropping reasoning blocks, which is
the function's job) before any regex runs, so every caller is safe.
This commit is contained in:
teknium1 2026-07-17 15:23:23 -07:00 committed by Teknium
parent e7f208fd74
commit 296494db0e
2 changed files with 67 additions and 0 deletions

View file

@ -672,6 +672,39 @@ def strip_think_blocks(agent, content: str) -> str:
"""
if not content:
return ""
# Coerce non-string content to text before any regex runs. Providers
# that return assistant ``content`` as a list of blocks (Anthropic via
# OpenRouter emits ``[{"type":"text",...}, {"type":"thinking",...}]``) or
# as a dict flow into this shared helper from several callers — most
# notably ``_interim_assistant_visible_text`` reading a *stored* history
# message whose content was persisted as a list. A raw list/dict reaching
# ``re.sub`` below raises ``TypeError: expected string or bytes-like
# object, got 'list'``, which the outer conversation loop swallows and
# retries forever (observed as an infinite "preparing terminal…" loop on
# Anthropic models via OpenRouter). Flatten here so every caller is safe.
if not isinstance(content, str):
if isinstance(content, list):
_parts: list[str] = []
for _part in content:
if isinstance(_part, str):
_parts.append(_part)
elif isinstance(_part, dict):
_ptype = str(_part.get("type") or "").strip().lower()
# Drop reasoning/thinking blocks outright — this function's
# whole job is to strip them, and their text lives under
# different keys ("thinking", "reasoning") per provider.
if _ptype in {"thinking", "reasoning", "redacted_thinking"}:
continue
_text = _part.get("text")
if isinstance(_text, str) and _text:
_parts.append(_text)
content = "".join(_parts)
elif isinstance(content, dict):
content = str(content.get("text") or content.get("content") or "")
else:
content = str(content)
if not content:
return ""
# 1. Closed tag pairs — case-insensitive for all variants so
# mixed-case tags (<THINK>, <Thinking>) don't slip through to
# the unterminated-tag pass and take trailing content with them.

View file

@ -445,6 +445,40 @@ class TestStripThinkBlocks:
def test_none_returns_empty(self, agent):
assert agent._strip_think_blocks(None) == ""
def test_list_content_flattened_no_crash(self, agent):
"""Anthropic-via-OpenRouter returns content as a block list.
A raw list reaching ``re.sub`` raised ``TypeError: expected string
or bytes-like object, got 'list'``, which the outer conversation
loop swallowed and retried forever (infinite "preparing terminal…"
loop). ``strip_think_blocks`` must flatten list content to visible
text and drop reasoning blocks.
"""
result = agent._strip_think_blocks(
[
{"type": "text", "text": "visible answer"},
{"type": "thinking", "thinking": "internal reasoning"},
]
)
assert isinstance(result, str)
assert "visible answer" in result
assert "internal reasoning" not in result
def test_dict_content_flattened_no_crash(self, agent):
"""Some servers return content as a single dict block."""
result = agent._strip_think_blocks({"type": "text", "text": "hello world"})
assert isinstance(result, str)
assert "hello world" in result
def test_list_of_only_thinking_returns_empty(self, agent):
"""A list carrying only reasoning blocks yields no visible text."""
assert (
agent._strip_think_blocks([{"type": "thinking", "thinking": "x"}]) == ""
)
def test_empty_list_returns_empty(self, agent):
assert agent._strip_think_blocks([]) == ""
def test_no_blocks_unchanged(self, agent):
assert agent._strip_think_blocks("hello world") == "hello world"