fix: prune positionally unanswered tool_calls before API send
DeepSeek v4 rejects a payload where an assistant message carries a tool_call whose tool result does not follow it immediately (HTTP 400 "An assistant message with 'tool_calls' must be followed by tool messages responding to each 'tool_call_id'"). Context compression can displace a tool result past a user turn; the result then lands ~100 messages away from its declaring assistant message. Two gaps let the poisoned shape reach the wire (reproduced from the production request dump of session 4d8727cbcf04, replayed through both functions): 1. repair_message_sequence Pass 1 drops the displaced tool RESULT as stray but leaves the declaring assistant message carrying the now unanswered tool_call (with empty content) in the durable history. 2. sanitize_api_messages stubbed only globally-absent result ids: the displaced result still exists in the transcript, so the id survives the set-subtraction, no stub is injected, and the payload 400s. Fix both layers so every path is order-independent: - repair_message_sequence: new Pass 2 prunes tool_calls that have no result in the immediately-following tool run (matching on id or call_id, same superset rule as Pass 1). If pruning empties the turn (no content/reasoning left), the whole message is dropped rather than sending an empty assistant message. Codex interim turns are exempt, as in Pass 0. - sanitize_api_messages: the orphan/stub logic is rewritten as a single rolling positional walk that drops results not immediately following their declaring assistant (including results appearing BEFORE their call) and injects stub results for positionally-uncovered calls even when a mispositioned result exists elsewhere. Adds six regression tests: repair pruning, whole-turn drop when pruned calls were the only payload, valid-pair negative control, positional stub injection, result-before-call orphan drop, and a fully-paired transcript negative control.
This commit is contained in:
+191
-41
@@ -589,7 +589,11 @@ def repair_message_sequence(agent, messages: List[Dict]) -> int:
|
||||
resumed histories. Refs #29148, #49147.
|
||||
1. Stray ``tool`` messages whose ``tool_call_id`` doesn't match
|
||||
any preceding assistant tool_call — dropped.
|
||||
2. Consecutive ``user`` messages — merged with newline separator
|
||||
2. ``tool_calls`` on an assistant message that no immediately
|
||||
following ``tool`` result answers are pruned — and the turn is
|
||||
dropped entirely if that leaves it payload-empty (an empty
|
||||
non-final assistant message is itself a 400 on most providers).
|
||||
3. Consecutive ``user`` messages — merged with newline separator
|
||||
so no user input is lost.
|
||||
|
||||
Deliberately does NOT rewind orphan ``assistant(tool_calls)+tool``
|
||||
@@ -599,6 +603,19 @@ def repair_message_sequence(agent, messages: List[Dict]) -> int:
|
||||
pattern). The empty-response scaffolding stripper handles the
|
||||
genuinely-broken variant via its flag-gated rewind.
|
||||
|
||||
Pass 2 (prune unanswered ``tool_calls``) answers the complement of
|
||||
Pass 1: Pass 1 removes the stray result, Pass 2 removes the orphaned
|
||||
call it was displaced from. Context compression can move a tool
|
||||
result past a user turn; without this pass the declaring assistant
|
||||
message would keep replaying an unanswered ``tool_call`` and strict
|
||||
providers (DeepSeek v4) reject that with HTTP 400 "An assistant
|
||||
message with 'tool_calls' must be followed by tool messages
|
||||
responding to each 'tool_call_id'". A call counts as answered when
|
||||
the run of ``tool`` messages immediately following its assistant
|
||||
message contains a result keyed to ANY of the call's ids (``id`` or
|
||||
``call_id`` — the same superset rule Pass 1 registers). Codex
|
||||
interim turns are exempt, as in Pass 0.
|
||||
|
||||
Returns the number of repairs made (for logging/telemetry).
|
||||
"""
|
||||
if not messages:
|
||||
@@ -786,10 +803,86 @@ def repair_message_sequence(agent, messages: List[Dict]) -> int:
|
||||
matched_tool_groups = set()
|
||||
filtered.append(msg)
|
||||
|
||||
# Pass 2: merge consecutive user messages. Preserves all user input
|
||||
# Pass 2: prune tool_calls that were never answered positionally.
|
||||
#
|
||||
# Pass 1 dropped the stray/displaced tool RESULT — but a tool_call
|
||||
# whose result was displaced far beyond the following turn (context
|
||||
# compression can move it past a user turn) leaves its declaring
|
||||
# assistant message carrying an UNANSWERED tool_call, and strict
|
||||
# OpenAI-compatible providers (DeepSeek v4) reject the payload with
|
||||
# HTTP 400 "An assistant message with 'tool_calls' must be followed
|
||||
# by tool messages responding to each 'tool_call_id' (insufficient
|
||||
# tool messages following tool_calls message)". The per-call
|
||||
# sanitizer's stub pass is keyed on GLOBAL id presence, which a
|
||||
# displaced-but-present result masks (see sanitize_api_messages) —
|
||||
# so the durable history must not keep replaying the poisoned turn
|
||||
# either. Enforce the positional invariant here: a tool_call is only
|
||||
# legitimate when a result for ANY of its ids (``id`` / ``call_id``,
|
||||
# same superset as Pass 1) appears in the run of tool messages
|
||||
# IMMEDIATELY following the declaring assistant message — before any
|
||||
# user turn or further assistant turn. Unanswered calls are pruned;
|
||||
# if the message then carries no other payload (no content,
|
||||
# reasoning, codex items), the whole turn is dropped — an empty
|
||||
# non-final assistant message is itself rejected by providers.
|
||||
# Codex interim turns are exempt, as in Pass 0: their calls are
|
||||
# replayed through the Responses-items chain, not the tool-result
|
||||
# run.
|
||||
pruned: List[Dict] = []
|
||||
i = 0
|
||||
n = len(filtered)
|
||||
while i < n:
|
||||
msg = filtered[i]
|
||||
if not (
|
||||
isinstance(msg, dict)
|
||||
and msg.get("role") == "assistant"
|
||||
and msg.get("tool_calls")
|
||||
and not _is_codex_interim(msg)
|
||||
):
|
||||
pruned.append(msg)
|
||||
i += 1
|
||||
continue
|
||||
answered: set = set()
|
||||
j = i + 1
|
||||
while (
|
||||
j < n
|
||||
and isinstance(filtered[j], dict)
|
||||
and filtered[j].get("role") == "tool"
|
||||
):
|
||||
tid = (filtered[j].get("tool_call_id") or "").strip()
|
||||
if tid:
|
||||
answered.add(tid)
|
||||
j += 1
|
||||
kept_calls: List[Dict] = []
|
||||
dropped_calls = 0
|
||||
for tc in msg.get("tool_calls") or []:
|
||||
tc_ids = []
|
||||
if isinstance(tc, dict):
|
||||
tc_ids = [x for x in (tc.get("id"), tc.get("call_id")) if x]
|
||||
if tc_ids and any(x in answered for x in tc_ids):
|
||||
kept_calls.append(tc)
|
||||
else:
|
||||
dropped_calls += 1
|
||||
if dropped_calls:
|
||||
repairs += 1
|
||||
if not kept_calls and not _msg_has_payload(
|
||||
{k: v for k, v in msg.items() if k != "tool_calls"}
|
||||
):
|
||||
# The pruned call(s) were the message's only payload —
|
||||
# dropping the whole turn beats sending an empty
|
||||
# assistant message (which most providers 400).
|
||||
i += 1
|
||||
continue
|
||||
if kept_calls:
|
||||
msg["tool_calls"] = kept_calls
|
||||
else:
|
||||
msg.pop("tool_calls", None)
|
||||
pruned.append(msg)
|
||||
i += 1
|
||||
|
||||
# Pass 3: merge consecutive user messages. Preserves all user input
|
||||
# so nothing the user typed is lost.
|
||||
merged: List[Dict] = []
|
||||
for msg in filtered:
|
||||
for msg in pruned:
|
||||
if (
|
||||
merged
|
||||
and isinstance(msg, dict)
|
||||
@@ -3941,48 +4034,105 @@ def sanitize_api_messages(messages: List[Dict[str, Any]]) -> List[Dict[str, Any]
|
||||
_pre_id_filter_count - len(messages),
|
||||
)
|
||||
|
||||
(
|
||||
surviving_call_ids,
|
||||
result_call_ids,
|
||||
orphaned_results,
|
||||
missing_tool_calls,
|
||||
) = _classify_tool_call_orphans(messages)
|
||||
# --- Positional tool_call <-> tool_result pairing ---
|
||||
# Strict OpenAI-compatible providers (DeepSeek v4, Kimi) enforce the
|
||||
# POSITIONAL invariant: an assistant message carrying tool_calls must
|
||||
# be IMMEDIATELY followed by tool messages covering every
|
||||
# tool_call_id. The previous implementation compared global id sets,
|
||||
# which misses the failure mode where a result exists somewhere in
|
||||
# the transcript but not in the run right after its call — an
|
||||
# interrupted turn or a compression window can displace a result
|
||||
# past a user turn. The id then survives in the global result set,
|
||||
# so the call looks answered, no stub is injected, and the provider
|
||||
# rejects the payload with HTTP 400 "An assistant message with
|
||||
# 'tool_calls' must be followed by tool messages responding to each
|
||||
# 'tool_call_id' (insufficient tool messages following tool_calls
|
||||
# message)". Rewritten as a single rolling walk on the per-call
|
||||
# copy (#94704):
|
||||
# (a) tool results that do not immediately follow an assistant
|
||||
# message declaring their id are dropped (positional orphans —
|
||||
# includes results appearing BEFORE their call, which strict
|
||||
# providers also reject);
|
||||
# (b) declared ids not covered by the immediately-following tool
|
||||
# run get a stub result injected at the end of that run, even
|
||||
# when a mispositioned result exists elsewhere.
|
||||
# Matching is variant-aware (``tool_call_id_variants`` /
|
||||
# ``tool_result_id_variants``): a result keyed on ANY alias spelling
|
||||
# (``id`` / ``call_id`` / ``response_item_id`` / composite bridge)
|
||||
# answers the call, preserving the unified alias policy from
|
||||
# #55626/#63000/#93251.
|
||||
paired: List[Dict[str, Any]] = []
|
||||
declared_calls: Dict[str, tuple] = {}
|
||||
dropped_positional_orphans = 0
|
||||
added_stubs = 0
|
||||
|
||||
# 1. Drop tool results whose complete alias set matches no assistant call.
|
||||
orphaned_result_objects = {id(msg) for msg in orphaned_results}
|
||||
if orphaned_result_objects:
|
||||
messages = [m for m in messages if id(m) not in orphaned_result_objects]
|
||||
def _flush_unanswered_stubs() -> None:
|
||||
nonlocal added_stubs
|
||||
for key in sorted(declared_calls):
|
||||
tc, _variants = declared_calls[key]
|
||||
cid = coalesce_tool_call_id(tc) or key
|
||||
paired.append({
|
||||
"role": "tool",
|
||||
"name": _ra().AIAgent._get_tool_call_name_static(tc),
|
||||
"content": "[Result unavailable — see context summary above]",
|
||||
"tool_call_id": cid,
|
||||
})
|
||||
added_stubs += 1
|
||||
declared_calls.clear()
|
||||
|
||||
for msg in messages:
|
||||
role = msg.get("role")
|
||||
if role == "assistant":
|
||||
# A new assistant turn closes the previous tool-result run:
|
||||
# anything still pending was never answered positionally.
|
||||
_flush_unanswered_stubs()
|
||||
declared_calls = {}
|
||||
for tc in msg.get("tool_calls") or []:
|
||||
variants = tool_call_id_variants(tc)
|
||||
if variants:
|
||||
# Key on a stable representative of the alias group so
|
||||
# a result matching ANY spelling can consume the call.
|
||||
declared_calls[sorted(variants)[0]] = (tc, variants)
|
||||
paired.append(msg)
|
||||
elif role == "tool":
|
||||
result_variants = tool_result_id_variants(msg.get("tool_call_id"))
|
||||
matched = next(
|
||||
(
|
||||
key
|
||||
for key, (_tc, variants) in declared_calls.items()
|
||||
if variants & result_variants
|
||||
),
|
||||
None,
|
||||
)
|
||||
if matched is not None:
|
||||
paired.append(msg)
|
||||
# Consume so a duplicate result reusing the id falls into
|
||||
# the drop branch (same semantics as the old global
|
||||
# dedup; strict providers reject duplicate tool_call_id).
|
||||
declared_calls.pop(matched, None)
|
||||
else:
|
||||
dropped_positional_orphans += 1
|
||||
else:
|
||||
if role == "user":
|
||||
# A user turn closes the tool-result run; subsequent
|
||||
# tool messages without a fresh declaring assistant
|
||||
# turn are orphans.
|
||||
_flush_unanswered_stubs()
|
||||
paired.append(msg)
|
||||
# The transcript may end right after an unanswered assistant turn.
|
||||
_flush_unanswered_stubs()
|
||||
if dropped_positional_orphans or added_stubs:
|
||||
messages = paired
|
||||
if dropped_positional_orphans:
|
||||
_ra().logger.debug(
|
||||
"Pre-call sanitizer: removed %d orphaned tool result(s)",
|
||||
len(orphaned_result_objects),
|
||||
"Pre-call sanitizer: removed %d positionally orphaned tool result(s)",
|
||||
dropped_positional_orphans,
|
||||
)
|
||||
|
||||
# 2. Inject one stub per assistant call with no result on ANY alias.
|
||||
if missing_tool_calls:
|
||||
missing_tool_call_objects = {id(tc) for tc in missing_tool_calls}
|
||||
patched: List[Dict[str, Any]] = []
|
||||
for msg in messages:
|
||||
patched.append(msg)
|
||||
if msg.get("role") == "assistant":
|
||||
for tc in msg.get("tool_calls") or []:
|
||||
if id(tc) not in missing_tool_call_objects:
|
||||
continue
|
||||
cid = coalesce_tool_call_id(tc)
|
||||
if not cid:
|
||||
variants = tool_call_id_variants(tc)
|
||||
cid = sorted(variants)[0] if variants else ""
|
||||
if not cid:
|
||||
continue
|
||||
patched.append({
|
||||
"role": "tool",
|
||||
"name": _ra().AIAgent._get_tool_call_name_static(tc),
|
||||
"content": "[Result unavailable — see context summary above]",
|
||||
"tool_call_id": cid,
|
||||
})
|
||||
messages = patched
|
||||
if added_stubs:
|
||||
_ra().logger.debug(
|
||||
"Pre-call sanitizer: added %d stub tool result(s)",
|
||||
len(missing_tool_calls),
|
||||
"Pre-call sanitizer: added %d stub tool result(s) for "
|
||||
"positionally unanswered tool call(s)",
|
||||
added_stubs,
|
||||
)
|
||||
|
||||
# 3. Deduplicate tool_call_ids. Strict providers (DeepSeek) reject a
|
||||
|
||||
@@ -429,6 +429,7 @@ def test_tool_executor_uses_canonical_responses_pairing_id():
|
||||
) == "call_ABC"
|
||||
|
||||
|
||||
|
||||
# ── repair_message_sequence_with_cursor (#44837) ───────────────────────────
|
||||
|
||||
from agent.agent_runtime_helpers import repair_message_sequence_with_cursor
|
||||
@@ -475,7 +476,6 @@ def test_cursor_rewinds_when_compaction_happens_before_cursor():
|
||||
|
||||
|
||||
|
||||
|
||||
def test_flush_guard_clamps_overshooting_cursor():
|
||||
"""_flush_messages_to_session_db safety net: an overshooting cursor must
|
||||
not produce a negative-start slice that skips everything (#44837)."""
|
||||
@@ -520,15 +520,6 @@ def test_flush_guard_clamps_overshooting_cursor():
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
# ── tool_call_id de-duplication (#58327) ────────────────────────────────────
|
||||
# Strict providers (DeepSeek) reject a payload where the same tool_call_id
|
||||
# appears more than once with HTTP 400 "Duplicate value for 'tool_call_id'".
|
||||
@@ -833,8 +824,6 @@ def test_repair_keeps_tool_result_when_tool_calls_are_sdk_objects():
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
# ── 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
|
||||
@@ -1145,3 +1134,171 @@ def test_classify_orphans_mixed():
|
||||
assert rs == {"call_A", "call_C"}
|
||||
assert [m["tool_call_id"] for m in orphaned] == ["call_C"]
|
||||
assert [tc["id"] for tc in missing] == ["call_B"]
|
||||
|
||||
|
||||
# ── Positional tool_call <-> tool_result pairing ───────────────────────────
|
||||
# Production incident (session 4d8727cbcf04): context compression displaced
|
||||
# a tool result ~110 messages past its declaring assistant turn (across a
|
||||
# user turn). repair_message_sequence Pass 1 dropped the displaced result as
|
||||
# stray but left the declaring assistant carrying an UNANSWERED tool_call
|
||||
# with empty content; sanitize_api_messages' global-set stub pass saw the
|
||||
# displaced result still present, considered the call answered, and injected
|
||||
# no stub — DeepSeek v4 then 400'd the payload: "An assistant message with
|
||||
# 'tool_calls' must be followed by tool messages responding to each
|
||||
# 'tool_call_id' (insufficient tool messages following tool_calls message)".
|
||||
|
||||
|
||||
def _assistant_with_call(call_id, content=""):
|
||||
return {
|
||||
"role": "assistant",
|
||||
"content": content,
|
||||
"tool_calls": [{
|
||||
"id": call_id, "type": "function",
|
||||
"function": {"name": "f", "arguments": "{}"},
|
||||
}],
|
||||
}
|
||||
|
||||
|
||||
def _tool_result(call_id, content="out"):
|
||||
return {"role": "tool", "tool_call_id": call_id, "content": content}
|
||||
|
||||
|
||||
def test_repair_prunes_tool_call_whose_result_was_displaced():
|
||||
"""Pass 2: a tool_call with no result in the immediately-following run is
|
||||
pruned, even when its result exists far later (post-compression shape).
|
||||
The assistant turn keeps its plain content once the calls are pruned.
|
||||
"""
|
||||
agent = _bare_agent()
|
||||
messages = [
|
||||
{"role": "user", "content": "do it"},
|
||||
_assistant_with_call("call_A", content=""), # declares A, never answered here
|
||||
_assistant_with_call("call_B", content="second"), # merged into the above (Pass 0)
|
||||
_tool_result("call_B"),
|
||||
{"role": "user", "content": "meanwhile"}, # user redirect
|
||||
_tool_result("call_A", content="late result"), # displaced: dropped by Pass 1
|
||||
]
|
||||
|
||||
repairs = AIAgent._repair_message_sequence(agent, messages)
|
||||
|
||||
assert repairs >= 1
|
||||
assistants = [m for m in messages if m.get("role") == "assistant"]
|
||||
assert len(assistants) == 1
|
||||
ids = [tc["id"] for tc in assistants[0]["tool_calls"]]
|
||||
assert ids == ["call_B"] # unanswered call_A pruned
|
||||
assert assistants[0]["content"] == "second"
|
||||
# The legitimate call_B result survives; only the displaced late
|
||||
# call_A result was dropped.
|
||||
tools = [m for m in messages if m.get("role") == "tool"]
|
||||
assert len(tools) == 1
|
||||
assert tools[0]["tool_call_id"] == "call_B"
|
||||
|
||||
|
||||
def test_repair_drops_turn_when_pruned_calls_were_only_payload():
|
||||
"""Pass 2: when pruning empties the merged assistant turn (no content,
|
||||
no reasoning), the whole turn is dropped instead of sending an empty
|
||||
non-final assistant message (itself a 400 on most providers).
|
||||
"""
|
||||
agent = _bare_agent()
|
||||
messages = [
|
||||
{"role": "user", "content": "do it"},
|
||||
_assistant_with_call("call_A"), # empty content
|
||||
{"role": "assistant", "content": ""}, # merged in (Pass 0)
|
||||
{"role": "user", "content": "redirected"},
|
||||
_tool_result("call_A", content="late"), # displaced: dropped
|
||||
]
|
||||
|
||||
repairs = AIAgent._repair_message_sequence(agent, messages)
|
||||
|
||||
assert repairs >= 2
|
||||
assert all(m.get("role") != "assistant" for m in messages)
|
||||
# The two user turns merge (Pass 3); nothing was lost.
|
||||
users = [m for m in messages if m.get("role") == "user"]
|
||||
assert len(users) == 1
|
||||
assert "do it" in users[0]["content"] and "redirected" in users[0]["content"]
|
||||
|
||||
|
||||
def test_repair_keeps_calls_answered_within_following_run():
|
||||
"""Negative control: a legitimate assistant(tool_calls)+tool run must
|
||||
survive Pass 2 untouched (the ongoing dialog pattern)."""
|
||||
agent = _bare_agent()
|
||||
messages = [
|
||||
{"role": "user", "content": "Q1"},
|
||||
_assistant_with_call("t1", content=""),
|
||||
_tool_result("t1"),
|
||||
{"role": "user", "content": "Q2"},
|
||||
]
|
||||
original = [dict(m) for m in messages]
|
||||
|
||||
repairs = AIAgent._repair_message_sequence(agent, messages)
|
||||
|
||||
assert repairs == 0
|
||||
assert messages == original
|
||||
|
||||
|
||||
def test_sanitize_stubs_call_unanswered_positionally_even_if_result_exists_elsewhere():
|
||||
"""sanitize_api_messages must inject a stub right after the declaring
|
||||
assistant message when no result follows it, EVEN IF a (displaced)
|
||||
result exists later in the transcript — the global-set check missed
|
||||
this exact shape (production 400, session 4d8727cbcf04)."""
|
||||
from agent.agent_runtime_helpers import sanitize_api_messages
|
||||
|
||||
messages = [
|
||||
{"role": "user", "content": "do it"},
|
||||
_assistant_with_call("call_A", content=""),
|
||||
{"role": "user", "content": "meanwhile"},
|
||||
_tool_result("call_A", content="late result"),
|
||||
]
|
||||
|
||||
out = sanitize_api_messages(list(messages))
|
||||
|
||||
roles = [m["role"] for m in out]
|
||||
assert roles == ["user", "assistant", "tool", "user"]
|
||||
stub = out[2]
|
||||
assert stub["tool_call_id"] == "call_A"
|
||||
assert "Result unavailable" in stub["content"]
|
||||
# The displaced late result is dropped (positional orphan).
|
||||
assert "late result" not in [m.get("content", "") for m in out]
|
||||
|
||||
|
||||
def test_sanitize_drops_result_appearing_before_its_call():
|
||||
"""A tool result that precedes its declaring assistant message is a
|
||||
positional orphan — strict providers reject 'role=tool' messages that
|
||||
don't follow a tool_calls message."""
|
||||
from agent.agent_runtime_helpers import sanitize_api_messages
|
||||
|
||||
messages = [
|
||||
{"role": "user", "content": "do it"},
|
||||
_tool_result("call_A"), # before its call
|
||||
_assistant_with_call("call_A", content=""),
|
||||
_tool_result("call_A"),
|
||||
]
|
||||
|
||||
out = sanitize_api_messages(list(messages))
|
||||
|
||||
tools = [m for m in out if m.get("role") == "tool"]
|
||||
assert len(tools) == 1 # only the valid one survives
|
||||
assert tools[0]["tool_call_id"] == "call_A"
|
||||
|
||||
|
||||
def test_sanitize_positional_pairing_untouched_valid_transcript():
|
||||
"""Negative control: a fully paired transcript (each tool-calling
|
||||
assistant immediately followed by its results) gets no stubs and loses
|
||||
no results."""
|
||||
from agent.agent_runtime_helpers import sanitize_api_messages
|
||||
|
||||
messages = [
|
||||
{"role": "user", "content": "do it"},
|
||||
_assistant_with_call("call_A", content=""),
|
||||
_tool_result("call_A"),
|
||||
_assistant_with_call("call_B", content=""),
|
||||
_tool_result("call_B"),
|
||||
{"role": "assistant", "content": "done"},
|
||||
]
|
||||
|
||||
out = sanitize_api_messages(list(messages))
|
||||
|
||||
assert [m["role"] for m in out] == [
|
||||
"user", "assistant", "tool", "assistant", "tool", "assistant",
|
||||
]
|
||||
assert all("Result unavailable" not in str(m.get("content", "")) for m in out)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user