diff --git a/application/llm/handlers/base.py b/application/llm/handlers/base.py index 44cb558d..22a9fef5 100644 --- a/application/llm/handlers/base.py +++ b/application/llm/handlers/base.py @@ -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 ". + 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) diff --git a/application/llm/openai.py b/application/llm/openai.py index 75555546..48b90726 100644 --- a/application/llm/openai.py +++ b/application/llm/openai.py @@ -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 ". 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, diff --git a/tests/llm/handlers/test_llm_handlers.py b/tests/llm/handlers/test_llm_handlers.py index c0a2b6fd..a480b27e 100644 --- a/tests/llm/handlers/test_llm_handlers.py +++ b/tests/llm/handlers/test_llm_handlers.py @@ -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() diff --git a/tests/llm/test_openai_responses.py b/tests/llm/test_openai_responses.py index 967393a8..7373a4b7 100644 --- a/tests/llm/test_openai_responses.py +++ b/tests/llm/test_openai_responses.py @@ -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