mirror of
https://github.com/tiennm99/DocsGPT.git
synced 2026-10-13 20:14:16 +00:00
fix: multitool call openai issue
This commit is contained in:
1 parent
0a9f18c58f
commit
3cbd284948
4 files changed
+664
-64
No files matched your search
@@ -836,6 +836,35 @@ class LLMHandler(ABC):
|
||||
updated_messages = messages.copy()
|
||||
pending_actions: List[Dict] = []
|
||||
|
||||
# One assistant message carries the WHOLE parallel batch, followed by
|
||||
# one tool message per call — the layout every provider expects, and
|
||||
# the one the resume (``BaseAgent._resume_*``) and history-replay
|
||||
# (``BaseAgent._build_messages``) paths already emit.
|
||||
#
|
||||
# Emitting a separate assistant message per call breaks chained
|
||||
# Responses requests: ``_trim_for_previous_response`` re-sends only
|
||||
# what follows the LAST assistant message, so every output except the
|
||||
# final one was dropped and the provider rejected the request with
|
||||
# "No tool output found for function call <first unpaired call>".
|
||||
batch_assistant: Optional[Dict[str, Any]] = None
|
||||
|
||||
def _declare_call(tool_call_obj: Dict[str, Any]) -> None:
|
||||
"""Add ``tool_call_obj`` to the batch's assistant message,
|
||||
creating (and appending) that message on first use."""
|
||||
nonlocal batch_assistant
|
||||
if batch_assistant is None:
|
||||
batch_assistant = {
|
||||
"role": "assistant",
|
||||
"content": None,
|
||||
"tool_calls": [],
|
||||
}
|
||||
# Reasoning rides once on the batch message; DeepSeek thinking
|
||||
# mode rejects an active-turn assistant message without it.
|
||||
if reasoning_content:
|
||||
batch_assistant["reasoning_content"] = reasoning_content
|
||||
updated_messages.append(batch_assistant)
|
||||
batch_assistant["tool_calls"].append(tool_call_obj)
|
||||
|
||||
for i, call in enumerate(tool_calls):
|
||||
# Check context limit before executing tool call
|
||||
if hasattr(agent, '_check_context_limit') and agent._check_context_limit(updated_messages):
|
||||
@@ -865,6 +894,12 @@ class LLMHandler(ABC):
|
||||
if compression_successful and rebuilt_messages is not None:
|
||||
# Update the messages list with rebuilt compressed version
|
||||
updated_messages = rebuilt_messages
|
||||
# The rebuilt list no longer contains the batch
|
||||
# message we were appending to, so mutating it
|
||||
# would be a silent no-op. Start a fresh one for
|
||||
# the remaining calls; each assistant message
|
||||
# still owns exactly the results that follow it.
|
||||
batch_assistant = None
|
||||
|
||||
# Yield compression success message
|
||||
yield {
|
||||
@@ -969,14 +1004,7 @@ class LLMHandler(ABC):
|
||||
}
|
||||
if getattr(call, "thought_signature", None):
|
||||
tool_call_obj["thought_signature"] = call.thought_signature
|
||||
assistant_msg: Dict[str, Any] = {
|
||||
"role": "assistant",
|
||||
"content": None,
|
||||
"tool_calls": [tool_call_obj],
|
||||
}
|
||||
if reasoning_content:
|
||||
assistant_msg["reasoning_content"] = reasoning_content
|
||||
updated_messages.append(assistant_msg)
|
||||
_declare_call(tool_call_obj)
|
||||
denial_call = ToolCall(
|
||||
id=pause_info["call_id"],
|
||||
name=call.name,
|
||||
@@ -1081,19 +1109,7 @@ class LLMHandler(ABC):
|
||||
if call.thought_signature:
|
||||
tool_call_obj["thought_signature"] = call.thought_signature
|
||||
|
||||
assistant_msg: Dict[str, Any] = {
|
||||
"role": "assistant",
|
||||
"content": None,
|
||||
"tool_calls": [tool_call_obj],
|
||||
}
|
||||
# Each call in a parallel batch becomes its own
|
||||
# assistant message here, so the same per-round
|
||||
# reasoning has to ride on every one — DeepSeek
|
||||
# thinking mode rejects any assistant message in the
|
||||
# active turn that's missing reasoning_content.
|
||||
if reasoning_content:
|
||||
assistant_msg["reasoning_content"] = reasoning_content
|
||||
updated_messages.append(assistant_msg)
|
||||
_declare_call(tool_call_obj)
|
||||
assistant_appended = True
|
||||
|
||||
# The tool result's tool_call_id must match the id put on the
|
||||
@@ -1110,24 +1126,22 @@ class LLMHandler(ABC):
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Error executing tool: {str(e)}", exc_info=True)
|
||||
# The error tool message's tool_call_id must match the assistant
|
||||
# tool_call id that precedes it. When the success path already
|
||||
# appended one (create_tool_message failed afterwards) that id is
|
||||
# the executor-returned ``call_id``; otherwise the except builds
|
||||
# its own assistant message below from ``call.id``.
|
||||
# The error tool message's tool_call_id must match the
|
||||
# tool_call id declared on the batch assistant message. When
|
||||
# the success path already declared this call that id is the
|
||||
# executor-returned ``call_id``; otherwise the except declares
|
||||
# it below from ``call.id``.
|
||||
error_id = call_id if assistant_appended else call.id
|
||||
error_call = ToolCall(
|
||||
id=error_id, name=call.name, arguments=call.arguments
|
||||
)
|
||||
error_response = f"Error executing tool: {str(e)}"
|
||||
# Mirror the success path: a role:"tool" message must follow
|
||||
# an assistant message carrying its tool_calls, or the next
|
||||
# provider completion 400s ("'tool' must be a response to a
|
||||
# preceding message with 'tool_calls'"). Skip it when the
|
||||
# success path already appended one for this call — a
|
||||
# create_tool_message failure after that append would
|
||||
# otherwise duplicate the assistant message and 400 the
|
||||
# same way an orphan tool message does.
|
||||
# Mirror the success path: every tool message must answer a
|
||||
# call declared on the batch assistant message, or the next
|
||||
# provider completion 400s. Skip re-declaring when the success
|
||||
# path already did it for this call — a create_tool_message
|
||||
# failure after that point would otherwise declare the call
|
||||
# twice and 400 the same way an orphan tool message does.
|
||||
if not assistant_appended:
|
||||
args_str = (
|
||||
json.dumps(call.arguments)
|
||||
@@ -1144,14 +1158,7 @@ class LLMHandler(ABC):
|
||||
}
|
||||
if call.thought_signature:
|
||||
tool_call_obj["thought_signature"] = call.thought_signature
|
||||
assistant_msg: Dict[str, Any] = {
|
||||
"role": "assistant",
|
||||
"content": None,
|
||||
"tool_calls": [tool_call_obj],
|
||||
}
|
||||
if reasoning_content:
|
||||
assistant_msg["reasoning_content"] = reasoning_content
|
||||
updated_messages.append(assistant_msg)
|
||||
_declare_call(tool_call_obj)
|
||||
|
||||
error_message = self.create_tool_message(error_call, error_response)
|
||||
updated_messages.append(error_message)
|
||||
|
||||
+122
-11
@@ -158,6 +158,9 @@ class OpenAILLM(BaseLLM):
|
||||
# chain turns when OPENAI_RESPONSES_STORE is enabled.
|
||||
self._reasoning_for_calls = {}
|
||||
self._last_response_id = None
|
||||
# call_ids the most recent response emitted; the chained-request
|
||||
# coverage guard checks the next turn answers all of them.
|
||||
self._last_response_call_ids = set()
|
||||
self._last_reasoning_items = []
|
||||
self._last_usage = None
|
||||
# One-shot guard consumed by ``_prefer_provider_usage``: fresh
|
||||
@@ -222,6 +225,7 @@ class OpenAILLM(BaseLLM):
|
||||
self._reasoning_for_calls = {}
|
||||
self._last_reasoning_items = []
|
||||
self._last_response_id = None
|
||||
self._last_response_call_ids = set()
|
||||
self._imported_response_id = None
|
||||
self._last_finish_reason = None
|
||||
|
||||
@@ -735,20 +739,100 @@ class OpenAILLM(BaseLLM):
|
||||
@staticmethod
|
||||
def _trim_for_previous_response(messages):
|
||||
"""When chaining via ``previous_response_id`` the server already holds
|
||||
the earlier turns, so only system context plus everything after the
|
||||
last completed assistant response needs to be sent again."""
|
||||
the earlier turns, so only system context plus the tool results
|
||||
answering the chained response's calls need to be sent again.
|
||||
|
||||
The cut is the last assistant message. Any ``tool`` message inside the
|
||||
trailing assistant/tool run *before* that cut is carried over too: a
|
||||
batch written as one assistant message per call — instead of one
|
||||
message carrying the whole batch — would otherwise lose every result
|
||||
but the last, and the provider then rejects the request with "No tool
|
||||
output found for function call <first unpaired call>". The assistant
|
||||
messages in that run are not carried: the server already holds those
|
||||
calls, and chained mode accepts bare ``function_call_output`` items.
|
||||
"""
|
||||
last_assistant = -1
|
||||
for i, message in enumerate(messages):
|
||||
if message.get("role") == "assistant":
|
||||
last_assistant = i
|
||||
if last_assistant < 0:
|
||||
return messages
|
||||
carried = []
|
||||
i = last_assistant - 1
|
||||
while i >= 0:
|
||||
role = messages[i].get("role")
|
||||
if role == "tool":
|
||||
carried.append(messages[i])
|
||||
elif not (role == "assistant" and messages[i].get("tool_calls")):
|
||||
break
|
||||
i -= 1
|
||||
carried.reverse()
|
||||
head = [
|
||||
m
|
||||
for m in messages[: last_assistant + 1]
|
||||
if m.get("role") == "system"
|
||||
]
|
||||
return head + messages[last_assistant + 1:]
|
||||
return head + carried + messages[last_assistant + 1:]
|
||||
|
||||
def _build_responses_input(self, messages, previous_response_id):
|
||||
"""Build the Responses ``input`` list, honouring store-mode chaining.
|
||||
|
||||
Returns ``(input_items, previous_response_id)``. The id comes back
|
||||
``None`` when the chained payload would not carry an output for every
|
||||
call the chained response holds — rather than send a request the
|
||||
provider is certain to reject, the full history is sent unchained.
|
||||
|
||||
A chained request answers exactly the calls of the response it chains
|
||||
to, so outputs are checked in BOTH directions against
|
||||
``_last_response_call_ids``: missing ones abandon chaining, extra ones
|
||||
are dropped. The extras matter because the trim's carry loop cannot
|
||||
tell a split batch (whose stray results must be rescued) from a
|
||||
multi-round grouped history (whose earlier rounds the server already
|
||||
has) — both end in the same assistant/tool run. Without the filter,
|
||||
round N re-sends every prior round's outputs back to the last user
|
||||
turn and the payload grows with the tool loop.
|
||||
"""
|
||||
chained = bool(previous_response_id and settings.OPENAI_RESPONSES_STORE)
|
||||
if not chained:
|
||||
return self._to_responses_input(messages, chained=False), None
|
||||
|
||||
input_items = self._to_responses_input(
|
||||
self._trim_for_previous_response(messages), chained=True
|
||||
)
|
||||
expected = set(self._last_response_call_ids or ())
|
||||
if expected:
|
||||
kept, dropped = [], []
|
||||
for item in input_items:
|
||||
if (
|
||||
item.get("type") == "function_call_output"
|
||||
and item.get("call_id") not in expected
|
||||
):
|
||||
dropped.append(item.get("call_id"))
|
||||
else:
|
||||
kept.append(item)
|
||||
if dropped:
|
||||
logging.debug(
|
||||
"Dropping %d already-answered tool output(s) from the "
|
||||
"chained Responses input (call_ids=%s)",
|
||||
len(dropped),
|
||||
sorted(dropped),
|
||||
)
|
||||
input_items = kept
|
||||
sent = {
|
||||
item.get("call_id")
|
||||
for item in input_items
|
||||
if item.get("type") == "function_call_output"
|
||||
}
|
||||
missing = expected - sent
|
||||
if missing:
|
||||
logging.warning(
|
||||
"Chained Responses request would omit tool output(s) for "
|
||||
"%s; sending the full unchained input instead of a "
|
||||
"request the provider would reject",
|
||||
sorted(missing),
|
||||
)
|
||||
return self._to_responses_input(messages, chained=False), None
|
||||
return input_items, previous_response_id
|
||||
|
||||
@staticmethod
|
||||
def _to_responses_tools(tools):
|
||||
@@ -909,10 +993,33 @@ class OpenAILLM(BaseLLM):
|
||||
self._last_usage = result
|
||||
self._last_usage_claimed = False
|
||||
|
||||
@staticmethod
|
||||
def _function_call_ids(response):
|
||||
"""call_ids of every ``function_call`` item in a Responses output.
|
||||
|
||||
Feeds the chained-request coverage guard in
|
||||
``_build_responses_input``: these are the calls the server will expect
|
||||
a ``function_call_output`` for on the next chained turn.
|
||||
"""
|
||||
output = getattr(response, "output", None)
|
||||
if not isinstance(output, list):
|
||||
return set()
|
||||
call_ids = set()
|
||||
for item in output:
|
||||
if isinstance(item, dict):
|
||||
itype, cid = item.get("type"), item.get("call_id")
|
||||
else:
|
||||
itype = getattr(item, "type", None)
|
||||
cid = getattr(item, "call_id", None)
|
||||
if itype == "function_call" and isinstance(cid, str) and cid:
|
||||
call_ids.add(cid)
|
||||
return call_ids
|
||||
|
||||
def _record_responses_metadata(self, response):
|
||||
rid = getattr(response, "id", None)
|
||||
if rid:
|
||||
self._last_response_id = rid
|
||||
self._last_response_call_ids = self._function_call_ids(response)
|
||||
usage = getattr(response, "usage", None)
|
||||
if usage is not None:
|
||||
prompt = int(getattr(usage, "input_tokens", 0) or 0)
|
||||
@@ -1003,13 +1110,15 @@ class OpenAILLM(BaseLLM):
|
||||
previous_response_id = (
|
||||
self._last_response_id or previous_response_id or self._imported_response_id
|
||||
)
|
||||
# Built before the per-turn state is cleared: the coverage guard reads
|
||||
# the call_ids the chained response emitted.
|
||||
input_items, previous_response_id = self._build_responses_input(
|
||||
messages, previous_response_id
|
||||
)
|
||||
self._last_response_id = None
|
||||
self._last_response_call_ids = set()
|
||||
self._last_usage = None
|
||||
self._last_finish_reason = None
|
||||
chained = bool(previous_response_id and settings.OPENAI_RESPONSES_STORE)
|
||||
if chained:
|
||||
messages = self._trim_for_previous_response(messages)
|
||||
input_items = self._to_responses_input(messages, chained=chained)
|
||||
params = self._build_responses_params(
|
||||
model,
|
||||
input_items,
|
||||
@@ -1073,14 +1182,16 @@ class OpenAILLM(BaseLLM):
|
||||
previous_response_id = (
|
||||
self._last_response_id or previous_response_id or self._imported_response_id
|
||||
)
|
||||
# Built before the per-turn state is cleared: the coverage guard reads
|
||||
# the call_ids the chained response emitted.
|
||||
input_items, previous_response_id = self._build_responses_input(
|
||||
messages, previous_response_id
|
||||
)
|
||||
self._last_response_id = None
|
||||
self._last_response_call_ids = set()
|
||||
self._last_usage = None
|
||||
self._last_finish_reason = None
|
||||
self._stream_reached_finish = False
|
||||
chained = bool(previous_response_id and settings.OPENAI_RESPONSES_STORE)
|
||||
if chained:
|
||||
messages = self._trim_for_previous_response(messages)
|
||||
input_items = self._to_responses_input(messages, chained=chained)
|
||||
params = self._build_responses_params(
|
||||
model,
|
||||
input_items,
|
||||
|
||||
@@ -281,6 +281,169 @@ class TestHandleToolCallsIdPairing:
|
||||
assert pending is None
|
||||
|
||||
|
||||
class _MultiCallStubAgent:
|
||||
"""Agent whose executor echoes each call's own id, optionally failing some."""
|
||||
|
||||
def __init__(self, fail_ids=()):
|
||||
self.llm = Mock()
|
||||
self.tool_executor = _StubToolExecutor()
|
||||
self._fail_ids = set(fail_ids)
|
||||
|
||||
def _execute_tool_action(self, tools_dict, call):
|
||||
yield from ()
|
||||
if call.id in self._fail_ids:
|
||||
raise RuntimeError(f"boom:{call.id}")
|
||||
return {"ok": call.id}, call.id
|
||||
|
||||
|
||||
class TestHandleToolCallsParallelBatchLayout:
|
||||
"""A parallel batch must produce the standard provider layout: ONE
|
||||
assistant message carrying every tool_call, followed by one tool message
|
||||
per call.
|
||||
|
||||
Emitting a separate assistant message per call (the previous behaviour)
|
||||
breaks ``_trim_for_previous_response``, which keeps only what follows the
|
||||
LAST assistant message — so every output except the final one is dropped
|
||||
from a chained Responses request and the provider 400s with
|
||||
"No tool output found for function call ...".
|
||||
"""
|
||||
|
||||
def _drain(self, gen):
|
||||
try:
|
||||
while True:
|
||||
next(gen)
|
||||
except StopIteration as e:
|
||||
return e.value
|
||||
|
||||
def _calls(self, n):
|
||||
return [
|
||||
ToolCall(id=f"call_{i}", name="search", arguments={"q": str(i)})
|
||||
for i in range(n)
|
||||
]
|
||||
|
||||
def test_parallel_batch_produces_single_assistant_message(self):
|
||||
handler = ConcreteHandler()
|
||||
calls = self._calls(3)
|
||||
|
||||
updated, pending = self._drain(
|
||||
handler.handle_tool_calls(
|
||||
_MultiCallStubAgent(), calls, tools_dict={}, messages=[]
|
||||
)
|
||||
)
|
||||
|
||||
assistants = [m for m in updated if m["role"] == "assistant"]
|
||||
assert len(assistants) == 1
|
||||
assert [tc["id"] for tc in assistants[0]["tool_calls"]] == [
|
||||
"call_0",
|
||||
"call_1",
|
||||
"call_2",
|
||||
]
|
||||
assert pending is None
|
||||
|
||||
def test_parallel_batch_orders_assistant_before_every_result(self):
|
||||
handler = ConcreteHandler()
|
||||
calls = self._calls(3)
|
||||
|
||||
updated, _ = self._drain(
|
||||
handler.handle_tool_calls(
|
||||
_MultiCallStubAgent(), calls, tools_dict={}, messages=[]
|
||||
)
|
||||
)
|
||||
|
||||
roles = [m["role"] for m in updated]
|
||||
assert roles == ["assistant", "tool", "tool", "tool"]
|
||||
assert [m["tool_call_id"] for m in updated if m["role"] == "tool"] == [
|
||||
"call_0",
|
||||
"call_1",
|
||||
"call_2",
|
||||
]
|
||||
|
||||
def test_every_call_in_the_batch_has_its_output(self):
|
||||
"""The invariant the provider enforces: every tool_call on the
|
||||
assistant message is answered by a tool message."""
|
||||
handler = ConcreteHandler()
|
||||
calls = self._calls(4)
|
||||
|
||||
updated, _ = self._drain(
|
||||
handler.handle_tool_calls(
|
||||
_MultiCallStubAgent(), calls, tools_dict={}, messages=[]
|
||||
)
|
||||
)
|
||||
|
||||
declared = {
|
||||
tc["id"]
|
||||
for m in updated
|
||||
if m["role"] == "assistant"
|
||||
for tc in m["tool_calls"]
|
||||
}
|
||||
answered = {m["tool_call_id"] for m in updated if m["role"] == "tool"}
|
||||
assert declared == answered
|
||||
|
||||
def test_reasoning_content_attached_once_not_per_call(self):
|
||||
handler = ConcreteHandler()
|
||||
calls = self._calls(3)
|
||||
|
||||
updated, _ = self._drain(
|
||||
handler.handle_tool_calls(
|
||||
_MultiCallStubAgent(),
|
||||
calls,
|
||||
tools_dict={},
|
||||
messages=[],
|
||||
reasoning_content="thinking",
|
||||
)
|
||||
)
|
||||
|
||||
assistants = [m for m in updated if m["role"] == "assistant"]
|
||||
assert len(assistants) == 1
|
||||
assert assistants[0]["reasoning_content"] == "thinking"
|
||||
|
||||
def test_failed_call_shares_the_batch_assistant_message(self):
|
||||
"""A tool that raises still contributes its call to the same assistant
|
||||
message and gets an error tool message — pairing must hold."""
|
||||
handler = ConcreteHandler()
|
||||
calls = self._calls(3)
|
||||
|
||||
updated, _ = self._drain(
|
||||
handler.handle_tool_calls(
|
||||
_MultiCallStubAgent(fail_ids={"call_1"}),
|
||||
calls,
|
||||
tools_dict={},
|
||||
messages=[],
|
||||
)
|
||||
)
|
||||
|
||||
assistants = [m for m in updated if m["role"] == "assistant"]
|
||||
assert len(assistants) == 1
|
||||
declared = {tc["id"] for tc in assistants[0]["tool_calls"]}
|
||||
answered = {m["tool_call_id"] for m in updated if m["role"] == "tool"}
|
||||
assert declared == {"call_0", "call_1", "call_2"} == answered
|
||||
|
||||
def test_single_call_batch_unchanged(self):
|
||||
handler = ConcreteHandler()
|
||||
|
||||
updated, _ = self._drain(
|
||||
handler.handle_tool_calls(
|
||||
_MultiCallStubAgent(), self._calls(1), tools_dict={}, messages=[]
|
||||
)
|
||||
)
|
||||
|
||||
assert [m["role"] for m in updated] == ["assistant", "tool"]
|
||||
assert len(updated[0]["tool_calls"]) == 1
|
||||
|
||||
def test_existing_history_is_preserved(self):
|
||||
handler = ConcreteHandler()
|
||||
history = [{"role": "user", "content": "hi"}]
|
||||
|
||||
updated, _ = self._drain(
|
||||
handler.handle_tool_calls(
|
||||
_MultiCallStubAgent(), self._calls(2), tools_dict={}, messages=history
|
||||
)
|
||||
)
|
||||
|
||||
assert updated[0] == {"role": "user", "content": "hi"}
|
||||
assert [m["role"] for m in updated[1:]] == ["assistant", "tool", "tool"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _append_unsupported_attachments
|
||||
# ---------------------------------------------------------------------------
|
||||
@@ -906,18 +1069,22 @@ class TestHandleToolCalls:
|
||||
except StopIteration as e:
|
||||
messages, _pending = e.value
|
||||
|
||||
# Each tool message must immediately follow an assistant message that
|
||||
# advertises its tool_call_id.
|
||||
answered_ids = set()
|
||||
for idx, msg in enumerate(messages):
|
||||
if msg.get("role") == "tool":
|
||||
assert idx > 0
|
||||
prev = messages[idx - 1]
|
||||
assert prev["role"] == "assistant"
|
||||
advertised = {tc["id"] for tc in prev["tool_calls"]}
|
||||
assert msg["tool_call_id"] in advertised
|
||||
# Every tool message must answer a tool_call declared by an assistant
|
||||
# message EARLIER in the array, and every declared call must be
|
||||
# answered. (Adjacency is deliberately not asserted: the standard
|
||||
# provider layout is one assistant message carrying the whole batch
|
||||
# followed by all its results, so only the first result is adjacent.)
|
||||
declared_ids: set = set()
|
||||
answered_ids: set = set()
|
||||
for msg in messages:
|
||||
if msg.get("role") == "assistant":
|
||||
declared_ids.update(tc["id"] for tc in msg.get("tool_calls", []))
|
||||
elif msg.get("role") == "tool":
|
||||
assert msg["tool_call_id"] in declared_ids, (
|
||||
"tool result precedes the assistant message declaring it"
|
||||
)
|
||||
answered_ids.add(msg["tool_call_id"])
|
||||
assert answered_ids == {"c1", "c2", "c3"}
|
||||
assert answered_ids == {"c1", "c2", "c3"} == declared_ids
|
||||
|
||||
def test_thought_signature_preserved(self):
|
||||
handler = ConcreteHandler()
|
||||
|
||||
@@ -1100,6 +1100,321 @@ def test_to_responses_input_chained_keeps_bare_tool_output(monkeypatch):
|
||||
assert [o["call_id"] for o in outputs] == ["call_server_side"]
|
||||
|
||||
|
||||
def _grouped_batch_messages():
|
||||
"""Standard layout: one assistant message, N tool_calls, N tool results."""
|
||||
return [
|
||||
{"role": "system", "content": "sys"},
|
||||
{"role": "user", "content": "q"},
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": None,
|
||||
"tool_calls": [
|
||||
{
|
||||
"id": f"call_{i}",
|
||||
"type": "function",
|
||||
"function": {"name": "read", "arguments": "{}"},
|
||||
}
|
||||
for i in range(3)
|
||||
],
|
||||
},
|
||||
{"role": "tool", "tool_call_id": "call_0", "content": "r0"},
|
||||
{"role": "tool", "tool_call_id": "call_1", "content": "r1"},
|
||||
{"role": "tool", "tool_call_id": "call_2", "content": "r2"},
|
||||
]
|
||||
|
||||
|
||||
def _split_batch_messages():
|
||||
"""Legacy layout: one assistant message PER call, interleaved with its
|
||||
result. ``_trim_for_previous_response`` must still keep every output."""
|
||||
msgs = [
|
||||
{"role": "system", "content": "sys"},
|
||||
{"role": "user", "content": "q"},
|
||||
]
|
||||
for i in range(3):
|
||||
msgs.append({
|
||||
"role": "assistant",
|
||||
"content": None,
|
||||
"tool_calls": [{
|
||||
"id": f"call_{i}",
|
||||
"type": "function",
|
||||
"function": {"name": "read", "arguments": "{}"},
|
||||
}],
|
||||
})
|
||||
msgs.append(
|
||||
{"role": "tool", "tool_call_id": f"call_{i}", "content": f"r{i}"}
|
||||
)
|
||||
return msgs
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_trim_keeps_every_output_of_a_grouped_batch(monkeypatch):
|
||||
"""Chained requests must carry an output for every call the chained
|
||||
response holds."""
|
||||
llm = _make_llm(monkeypatch, _responses_caps(), store_responses=True)
|
||||
trimmed = llm._trim_for_previous_response(_grouped_batch_messages())
|
||||
items = llm._to_responses_input(trimmed, chained=True)
|
||||
outputs = [i["call_id"] for i in items if i.get("type") == "function_call_output"]
|
||||
assert outputs == ["call_0", "call_1", "call_2"]
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_trim_keeps_every_output_of_a_split_batch(monkeypatch):
|
||||
"""Regression: the trim cut at the LAST assistant message, so a batch
|
||||
written as one assistant message per call lost every output but the
|
||||
last — the provider then 400s "No tool output found for function call
|
||||
call_0" (naming the FIRST unpaired call)."""
|
||||
llm = _make_llm(monkeypatch, _responses_caps(), store_responses=True)
|
||||
trimmed = llm._trim_for_previous_response(_split_batch_messages())
|
||||
items = llm._to_responses_input(trimmed, chained=True)
|
||||
outputs = [i["call_id"] for i in items if i.get("type") == "function_call_output"]
|
||||
assert outputs == ["call_0", "call_1", "call_2"]
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_trim_still_drops_prior_completed_turns(monkeypatch):
|
||||
"""The trim must keep doing its job: a completed assistant answer ends
|
||||
the replayed region, so earlier turns are not re-sent."""
|
||||
llm = _make_llm(monkeypatch, _responses_caps(), store_responses=True)
|
||||
messages = [
|
||||
{"role": "system", "content": "sys"},
|
||||
{"role": "user", "content": "old q"},
|
||||
{"role": "assistant", "content": "old answer"},
|
||||
{"role": "user", "content": "new q"},
|
||||
]
|
||||
trimmed = llm._trim_for_previous_response(messages)
|
||||
assert [m.get("content") for m in trimmed] == ["sys", "new q"]
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_chained_input_falls_back_when_an_output_is_missing(monkeypatch):
|
||||
"""Coverage guard: if the outputs being sent do not cover the calls held
|
||||
in the chained response, send a full non-chained input instead of a
|
||||
request the provider is certain to reject."""
|
||||
llm = _make_llm(monkeypatch, _responses_caps(), store_responses=True)
|
||||
llm._last_response_call_ids = {"call_0", "call_1", "call_2"}
|
||||
# Only the last output survives — the exact shape the old trim produced.
|
||||
messages = _grouped_batch_messages()[:3] + [
|
||||
{"role": "tool", "tool_call_id": "call_2", "content": "r2"},
|
||||
]
|
||||
|
||||
items, prev = llm._build_responses_input(messages, "resp_prev")
|
||||
|
||||
assert prev is None, "must drop out of chained mode"
|
||||
# The unchained rebuild is internally consistent: the one call that still
|
||||
# has its output is kept and paired, the two orphans are dropped by the
|
||||
# pairing guard. Either way the provider gets a request it can accept.
|
||||
calls = [i["call_id"] for i in items if i.get("type") == "function_call"]
|
||||
outputs = [i["call_id"] for i in items if i.get("type") == "function_call_output"]
|
||||
assert calls == ["call_2"]
|
||||
assert outputs == ["call_2"]
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_chained_input_stays_chained_when_outputs_are_complete(monkeypatch):
|
||||
llm = _make_llm(monkeypatch, _responses_caps(), store_responses=True)
|
||||
llm._last_response_call_ids = {"call_0", "call_1", "call_2"}
|
||||
|
||||
items, prev = llm._build_responses_input(_grouped_batch_messages(), "resp_prev")
|
||||
|
||||
assert prev == "resp_prev"
|
||||
outputs = [i["call_id"] for i in items if i.get("type") == "function_call_output"]
|
||||
assert outputs == ["call_0", "call_1", "call_2"]
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_chained_input_skips_the_check_when_calls_are_unknown(monkeypatch):
|
||||
"""After a resume/import we may not know which calls the chained response
|
||||
holds; an unknown set must not be treated as a mismatch."""
|
||||
llm = _make_llm(monkeypatch, _responses_caps(), store_responses=True)
|
||||
llm._last_response_call_ids = set()
|
||||
|
||||
_, prev = llm._build_responses_input(
|
||||
[
|
||||
{"role": "system", "content": "sys"},
|
||||
{"role": "tool", "tool_call_id": "call_server_side", "content": "r"},
|
||||
],
|
||||
"resp_prev",
|
||||
)
|
||||
|
||||
assert prev == "resp_prev"
|
||||
|
||||
|
||||
def _multi_round_messages(rounds, calls_per_round=2):
|
||||
"""A grouped history of N *consecutive* tool rounds — no user message in
|
||||
between, which is what an agent tool loop actually produces."""
|
||||
out = [
|
||||
{"role": "system", "content": "sys"},
|
||||
{"role": "user", "content": "q"},
|
||||
]
|
||||
for r in range(rounds):
|
||||
ids = [f"r{r}c{c}" for c in range(calls_per_round)]
|
||||
out.append({
|
||||
"role": "assistant",
|
||||
"content": None,
|
||||
"tool_calls": [
|
||||
{
|
||||
"id": i,
|
||||
"type": "function",
|
||||
"function": {"name": "t", "arguments": "{}"},
|
||||
}
|
||||
for i in ids
|
||||
],
|
||||
})
|
||||
out += [
|
||||
{"role": "tool", "tool_call_id": i, "content": "res"} for i in ids
|
||||
]
|
||||
return out
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.parametrize("rounds", [2, 3, 5])
|
||||
def test_chained_input_does_not_resend_earlier_rounds_outputs(monkeypatch, rounds):
|
||||
"""Only the chained response's OWN calls get answered.
|
||||
|
||||
The trim's carry loop rescues stray results from a split batch, but the
|
||||
trailing assistant/tool run of a multi-round grouped history has the same
|
||||
shape — so the walk also picked up every earlier round's results, which
|
||||
the server already has. Left unfiltered the payload grows with the round
|
||||
count, defeating the point of chaining.
|
||||
"""
|
||||
llm = _make_llm(monkeypatch, _responses_caps(), store_responses=True)
|
||||
last = rounds - 1
|
||||
llm._last_response_call_ids = {f"r{last}c0", f"r{last}c1"}
|
||||
|
||||
items, prev = llm._build_responses_input(
|
||||
_multi_round_messages(rounds), "resp_prev"
|
||||
)
|
||||
|
||||
assert prev == "resp_prev"
|
||||
outputs = [
|
||||
i["call_id"] for i in items if i.get("type") == "function_call_output"
|
||||
]
|
||||
assert outputs == [f"r{last}c0", f"r{last}c1"], (
|
||||
f"re-sent earlier rounds: {outputs}"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_chained_input_still_rescues_a_split_batch(monkeypatch):
|
||||
"""The filter must not undo the split-batch rescue: those outputs all
|
||||
answer calls the chained response emitted, so they stay."""
|
||||
llm = _make_llm(monkeypatch, _responses_caps(), store_responses=True)
|
||||
llm._last_response_call_ids = {"call_0", "call_1", "call_2"}
|
||||
|
||||
items, prev = llm._build_responses_input(_split_batch_messages(), "resp_prev")
|
||||
|
||||
assert prev == "resp_prev"
|
||||
outputs = [
|
||||
i["call_id"] for i in items if i.get("type") == "function_call_output"
|
||||
]
|
||||
assert outputs == ["call_0", "call_1", "call_2"]
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_chained_input_keeps_a_batch_split_by_compression(monkeypatch):
|
||||
"""Mid-batch compression starts a second assistant message for the same
|
||||
round. Both halves answer the chained response's calls, so both stay."""
|
||||
llm = _make_llm(monkeypatch, _responses_caps(), store_responses=True)
|
||||
llm._last_response_call_ids = {"c0", "c1", "c2"}
|
||||
messages = [
|
||||
{"role": "system", "content": "sys"},
|
||||
{"role": "user", "content": "q"},
|
||||
{"role": "assistant", "content": None, "tool_calls": [
|
||||
{"id": "c0", "type": "function",
|
||||
"function": {"name": "t", "arguments": "{}"}},
|
||||
]},
|
||||
{"role": "tool", "tool_call_id": "c0", "content": "r0"},
|
||||
{"role": "assistant", "content": None, "tool_calls": [
|
||||
{"id": "c1", "type": "function",
|
||||
"function": {"name": "t", "arguments": "{}"}},
|
||||
{"id": "c2", "type": "function",
|
||||
"function": {"name": "t", "arguments": "{}"}},
|
||||
]},
|
||||
{"role": "tool", "tool_call_id": "c1", "content": "r1"},
|
||||
{"role": "tool", "tool_call_id": "c2", "content": "r2"},
|
||||
]
|
||||
|
||||
items, _ = llm._build_responses_input(messages, "resp_prev")
|
||||
|
||||
outputs = [
|
||||
i["call_id"] for i in items if i.get("type") == "function_call_output"
|
||||
]
|
||||
assert outputs == ["c0", "c1", "c2"]
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_chained_input_does_not_filter_when_calls_are_unknown(monkeypatch):
|
||||
"""With no known call set (import/resume) an extra output cannot be told
|
||||
from a needed one, so nothing is dropped."""
|
||||
llm = _make_llm(monkeypatch, _responses_caps(), store_responses=True)
|
||||
llm._last_response_call_ids = set()
|
||||
|
||||
items, prev = llm._build_responses_input(_multi_round_messages(3), "resp_prev")
|
||||
|
||||
assert prev == "resp_prev"
|
||||
outputs = [
|
||||
i["call_id"] for i in items if i.get("type") == "function_call_output"
|
||||
]
|
||||
assert len(outputs) == 6
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
@pytest.mark.parametrize("entrypoint", ["_responses_gen", "_responses_gen_stream"])
|
||||
def test_both_request_builders_route_through_the_coverage_guard(
|
||||
monkeypatch, entrypoint
|
||||
):
|
||||
"""Regression: the guard was wired into the non-streaming builder only,
|
||||
so the streaming path — the one production actually uses — still sent
|
||||
unchecked chained requests. Both must go through it."""
|
||||
llm = _make_llm(monkeypatch, _responses_caps(), store_responses=True)
|
||||
llm._last_response_id = "resp_prev"
|
||||
seen = {}
|
||||
|
||||
real = llm._build_responses_input
|
||||
|
||||
def spy(messages, previous_response_id):
|
||||
seen["called"] = True
|
||||
return real(messages, previous_response_id)
|
||||
|
||||
monkeypatch.setattr(llm, "_build_responses_input", spy)
|
||||
monkeypatch.setattr(
|
||||
llm, "_build_responses_params", lambda *a, **k: {"model": "m"}
|
||||
)
|
||||
# The streaming path iterates the response; the non-streaming one reads
|
||||
# attributes off it.
|
||||
llm.client.responses.create = MagicMock(
|
||||
return_value=iter([])
|
||||
if entrypoint == "_responses_gen_stream"
|
||||
else types.SimpleNamespace(id="r", usage=None, output=[])
|
||||
)
|
||||
|
||||
result = getattr(llm, entrypoint)("m", [{"role": "user", "content": "hi"}])
|
||||
if entrypoint == "_responses_gen_stream":
|
||||
list(result) # drain the generator so the body runs
|
||||
|
||||
assert seen.get("called"), f"{entrypoint} bypassed the coverage guard"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_records_call_ids_emitted_by_a_response(monkeypatch):
|
||||
"""The coverage guard needs to know which calls the response emitted."""
|
||||
llm = _make_llm(monkeypatch, _responses_caps(), store_responses=True)
|
||||
response = _ns(
|
||||
id="resp_1",
|
||||
usage=None,
|
||||
output=[
|
||||
_ns(type="function_call", call_id="call_a"),
|
||||
_ns(type="message", call_id=None),
|
||||
_ns(type="function_call", call_id="call_b"),
|
||||
],
|
||||
)
|
||||
|
||||
llm._record_responses_metadata(response)
|
||||
|
||||
assert llm._last_response_id == "resp_1"
|
||||
assert llm._last_response_call_ids == {"call_a", "call_b"}
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_to_responses_input_drops_out_of_order_pair_entirely(monkeypatch):
|
||||
"""An output that precedes its call is malformed history: BOTH sides
|
||||
|
||||
Reference in new issue
Block a user