fix(errors): self-heal empty-content non-final messages before send
Third layer of the empty-stub fix: full self-recovery. A poisoned transcript (empty assistant stub or empty user turn already persisted before the write- time guard, or fed in from a host history) previously 400'd every subsequent request until it scrolled out — needing a manual DB edit + gateway restart. sanitize_api_messages() (the unconditional pre-send chokepoint) now repairs empty non-final messages on the per-call copy by substituting a minimal '[response interrupted]' placeholder, so the session recovers itself IN MEMORY on the very next send. The final message is left untouched (empty final assistant is legal); stored history is never mutated; reasoning-only and tool_call turns are preserved (negative controls). Tests: production-shape repro (tool -> empty assistant -> user), empty-user case, non-destructive guarantee, and negative controls. RED verified by disabling the wire-in.
This commit is contained in:
Aaron Weiker
Teknium
parent
4587d77e0e
commit
df45811198
|
|
@ -2779,6 +2779,117 @@ def repair_tool_call(agent, tool_name: str) -> str | None:
|
|||
|
||||
|
||||
|
||||
# Placeholder substituted for an empty non-final message that would otherwise
|
||||
# make the provider reject the whole request. Kept identical to the stub-
|
||||
# creation placeholder in chat_completion_helpers so a healed transcript reads
|
||||
# consistently whether the empty turn was caught at write time or send time.
|
||||
_INTERRUPTED_PLACEHOLDER = "[response interrupted]"
|
||||
|
||||
|
||||
def _msg_has_payload(msg: Dict[str, Any]) -> bool:
|
||||
"""True if ``msg`` carries anything the API treats as non-empty content.
|
||||
|
||||
Covers string content, non-empty multimodal content lists, tool_calls,
|
||||
tool_call_id linkage (tool results), and reasoning payloads. Mirrors the
|
||||
emptiness checks used by ``AIAgent._is_thinking_only_assistant`` but is
|
||||
role-agnostic so it can vet user/assistant/tool turns uniformly.
|
||||
"""
|
||||
content = msg.get("content")
|
||||
if isinstance(content, str):
|
||||
if content.strip():
|
||||
return True
|
||||
elif isinstance(content, list):
|
||||
for block in content:
|
||||
if isinstance(block, dict):
|
||||
# any typed block (text/image/tool_use/document/...) counts,
|
||||
# as long as a text block is not itself blank
|
||||
if block.get("type") == "text":
|
||||
if isinstance(block.get("text"), str) and block["text"].strip():
|
||||
return True
|
||||
continue
|
||||
return True
|
||||
elif block:
|
||||
return True
|
||||
elif content not in (None, ""):
|
||||
return True
|
||||
# Structural payloads that make an "empty-content" message still valid.
|
||||
if msg.get("tool_calls"):
|
||||
return True
|
||||
if isinstance(msg.get("reasoning_content"), str) and msg["reasoning_content"].strip():
|
||||
return True
|
||||
if msg.get("reasoning") or msg.get("reasoning_details"):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def repair_empty_non_final_messages(
|
||||
messages: List[Dict[str, Any]],
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""Heal empty-content non-final messages before they reach the provider.
|
||||
|
||||
Root-cause context: a stream that dies with 0 recovered characters (peer
|
||||
reset, stall-kill) could persist an assistant turn with ``content=None``
|
||||
and no tool_calls. The Anthropic message schema — and the litellm/Bedrock
|
||||
proxies in front of it — reject ANY request whose transcript contains an
|
||||
empty non-final message:
|
||||
|
||||
"all messages must have non-empty content except for the optional
|
||||
final assistant message" (HTTP 400 INVALID_REQUEST_BODY)
|
||||
|
||||
Once such a message lands mid-transcript it poisons EVERY subsequent turn
|
||||
of that session until it scrolls out of context. The write-time guard in
|
||||
``chat_completion_helpers`` stops NEW stubs, but sessions already carrying
|
||||
one (persisted before the guard, or fed in from a host history) stay stuck
|
||||
and previously needed a manual DB edit + gateway restart to recover.
|
||||
|
||||
This pass is the self-healing counterpart: it runs unconditionally on the
|
||||
per-call ``api_messages`` copy, so a poisoned transcript repairs itself
|
||||
IN MEMORY on the very next send — no restart, no DB surgery. The final
|
||||
message is left untouched (an empty final assistant turn is legal). The
|
||||
stored conversation history is never mutated; only the wire copy is
|
||||
repaired, so the UI/session trace stays faithful.
|
||||
|
||||
Repair strategy is substitution, not deletion: dropping a mid-transcript
|
||||
turn can break role alternation and tool-call pairing, whereas an honest
|
||||
minimal placeholder keeps the sequence intact and reads correctly as an
|
||||
interrupted turn on replay.
|
||||
"""
|
||||
if not messages or len(messages) < 2:
|
||||
return messages
|
||||
|
||||
repaired: List[Dict[str, Any]] = []
|
||||
healed = 0
|
||||
last_idx = len(messages) - 1
|
||||
for idx, msg in enumerate(messages):
|
||||
if (
|
||||
idx != last_idx
|
||||
and isinstance(msg, dict)
|
||||
# tool results are validated by their own orphan/pairing pass; an
|
||||
# empty tool result is a separate (and rarer) concern.
|
||||
and msg.get("role") in ("assistant", "user")
|
||||
and not _msg_has_payload(msg)
|
||||
):
|
||||
# Shallow-copy so stored history / prompt caching stays byte-stable.
|
||||
fixed = dict(msg)
|
||||
fixed["content"] = _INTERRUPTED_PLACEHOLDER
|
||||
repaired.append(fixed)
|
||||
healed += 1
|
||||
else:
|
||||
repaired.append(msg)
|
||||
|
||||
if healed:
|
||||
_ra().logger.warning(
|
||||
"Pre-call sanitizer: healed %d empty non-final message(s) by "
|
||||
"substituting placeholder content — an empty-content turn was in "
|
||||
"the transcript and would 400 the request ('messages must have "
|
||||
"non-empty content' / INVALID_REQUEST_BODY). Self-recovering the "
|
||||
"poisoned transcript in memory; no restart needed.",
|
||||
healed,
|
||||
)
|
||||
return repaired
|
||||
return messages
|
||||
|
||||
|
||||
def sanitize_api_messages(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
||||
"""Fix orphaned tool_call / tool_result pairs before every LLM call.
|
||||
|
||||
|
|
@ -2799,6 +2910,15 @@ def sanitize_api_messages(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]
|
|||
filtered.append(msg)
|
||||
messages = filtered
|
||||
|
||||
# --- Heal empty-content non-final messages (self-recovery) ---
|
||||
# A dead stream can leave an empty assistant stub (or an empty user turn)
|
||||
# mid-transcript; the provider then 400s EVERY subsequent request until it
|
||||
# scrolls out. Repair it here, on the per-call copy, so a poisoned session
|
||||
# recovers itself in memory on the next send — no restart, no DB edit.
|
||||
# Done first so a substituted turn participates normally in the tool-pair
|
||||
# and dedup passes below.
|
||||
messages = repair_empty_non_final_messages(messages)
|
||||
|
||||
# --- Drop empty / malformed tool_calls arrays on assistant messages ---
|
||||
# An assistant message carrying ``tool_calls: []`` (an empty array) — or a
|
||||
# non-list value under the key — is semantically identical to an assistant
|
||||
|
|
|
|||
|
|
@ -770,3 +770,101 @@ def test_sanitize_preserves_populated_tool_calls():
|
|||
out = sanitize_api_messages(list(messages))
|
||||
assistant = [m for m in out if m.get("role") == "assistant"][0]
|
||||
assert [tc["id"] for tc in assistant["tool_calls"]] == ["call_Z"]
|
||||
|
||||
|
||||
# ── Self-recovery: heal empty-content non-final messages ──────────────────
|
||||
# Repro of the production incident: a dead stream persisted an empty-content
|
||||
# assistant stub mid-transcript, and every later request 400'd with
|
||||
# "all messages must have non-empty content except for the optional final
|
||||
# assistant message" (INVALID_REQUEST_BODY). sanitize_api_messages now heals
|
||||
# such turns on the per-call copy so the session recovers itself in memory.
|
||||
|
||||
def test_sanitize_heals_empty_assistant_stub_between_tool_and_user():
|
||||
"""The exact production shape: tool -> EMPTY assistant (finish=length) ->
|
||||
user recovery. The empty assistant would 400 the whole request; it must be
|
||||
substituted with non-empty placeholder content, not left empty."""
|
||||
from agent.agent_runtime_helpers import sanitize_api_messages
|
||||
|
||||
messages = [
|
||||
{"role": "user", "content": "do a thing"},
|
||||
{"role": "assistant", "content": None, "tool_calls": [
|
||||
{"id": "call_A", "type": "function",
|
||||
"function": {"name": "foo", "arguments": "{}"}}]},
|
||||
{"role": "tool", "tool_call_id": "call_A", "content": "result"},
|
||||
{"role": "assistant", "content": None}, # <-- poison stub (non-final)
|
||||
{"role": "user", "content": "[System: previous response was cut off]"},
|
||||
]
|
||||
out = sanitize_api_messages(list(messages))
|
||||
# Every non-final message now has non-empty content.
|
||||
for m in out[:-1]:
|
||||
c = m.get("content")
|
||||
assert m.get("tool_calls") or (isinstance(c, str) and c.strip()) or (
|
||||
isinstance(c, list) and c), f"empty non-final survived: {m}"
|
||||
# The stub specifically became the placeholder.
|
||||
stub = out[3]
|
||||
assert stub["role"] == "assistant"
|
||||
assert stub["content"] == "[response interrupted]"
|
||||
|
||||
|
||||
def test_sanitize_heals_empty_user_message():
|
||||
"""An empty user turn mid-transcript is equally invalid and is healed."""
|
||||
from agent.agent_runtime_helpers import sanitize_api_messages
|
||||
|
||||
messages = [
|
||||
{"role": "assistant", "content": "hello"},
|
||||
{"role": "user", "content": ""}, # <-- empty user (non-final)
|
||||
{"role": "assistant", "content": "still here"},
|
||||
]
|
||||
out = sanitize_api_messages(list(messages))
|
||||
assert out[1]["content"] == "[response interrupted]"
|
||||
|
||||
|
||||
def test_sanitize_allows_empty_final_assistant():
|
||||
"""Negative control: an empty FINAL assistant message is legal per the API
|
||||
('except for the optional final assistant message') and must NOT be
|
||||
touched."""
|
||||
from agent.agent_runtime_helpers import sanitize_api_messages
|
||||
|
||||
messages = [
|
||||
{"role": "user", "content": "hi"},
|
||||
{"role": "assistant", "content": ""}, # final — allowed empty
|
||||
]
|
||||
out = sanitize_api_messages(list(messages))
|
||||
assert out[-1]["content"] == "" # untouched
|
||||
|
||||
|
||||
def test_sanitize_empty_heal_is_non_destructive():
|
||||
"""Healing must not mutate the caller's persisted dicts — only the per-call
|
||||
copy is repaired, so the stored trajectory keeps the true empty turn."""
|
||||
from agent.agent_runtime_helpers import sanitize_api_messages
|
||||
|
||||
original_stub = {"role": "assistant", "content": None}
|
||||
messages = [
|
||||
{"role": "user", "content": "q"},
|
||||
original_stub,
|
||||
{"role": "user", "content": "recovery"},
|
||||
]
|
||||
sanitize_api_messages(list(messages))
|
||||
assert original_stub["content"] is None # untouched in-place
|
||||
|
||||
|
||||
def test_sanitize_preserves_reasoning_only_and_toolcall_turns():
|
||||
"""Negative control: a non-final assistant turn that is 'empty content' but
|
||||
carries reasoning OR tool_calls is valid payload and must NOT be rewritten
|
||||
to the placeholder (that would clobber real model output)."""
|
||||
from agent.agent_runtime_helpers import sanitize_api_messages
|
||||
|
||||
messages = [
|
||||
{"role": "user", "content": "q"},
|
||||
{"role": "assistant", "content": None, "tool_calls": [
|
||||
{"id": "call_R", "type": "function",
|
||||
"function": {"name": "foo", "arguments": "{}"}}]},
|
||||
{"role": "tool", "tool_call_id": "call_R", "content": "r"},
|
||||
{"role": "user", "content": "next"},
|
||||
]
|
||||
out = sanitize_api_messages(list(messages))
|
||||
# tool_call assistant keeps its tool_calls, content not clobbered to text
|
||||
tc_asst = out[1]
|
||||
assert tc_asst.get("tool_calls") and tc_asst["tool_calls"][0]["id"] == "call_R"
|
||||
assert tc_asst["content"] != "[response interrupted]"
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue