mirror of
https://github.com/CopilotKit/CopilotKit.git
synced 2026-09-14 16:26:20 +08:00
fix(sdk-python): strip orphan OpenAI Responses function_call content blocks (#6781)
## What does this PR do? `after_model` strips intercepted frontend `tool_calls` off the last AIMessage, but under `responses/v1` output the equivalent `function_call` **content blocks** stay behind in `message.content`. If the run is cancelled mid-turn, that partial turn persists without a `ToolMessage`. On replay, `langchain-openai` serializes the orphaned block into the Responses API `input` with no matching `function_call_output`, and OpenAI rejects the request: ``` 400 - "No tool output found for function call call_..." ``` Because the poisoned message replays with the thread history, the thread then fails **permanently** — every subsequent turn 400s, not just the one that raced the cancellation. The existing sanitizer (`_fix_messages_for_bedrock`) only strips unanswered Anthropic `tool_use` blocks. This PR extends the same three strip sites to OpenAI Responses `function_call` blocks, keyed on `call_id` (`block["id"]` on those is the item id, not the call id): 1. blocks whose `call_id` is missing from `msg.tool_calls` 2. all of them when `tool_calls` is empty (e.g. after_model intercepted everything) 3. the ones belonging to unanswered (non-adjacent) `tool_calls` This is a re-land of the fix from #4473 (closed unmerged, tracked as OSS-71); the defect is still present on current `main`. ## Side-effect analysis — orphan-handoff contract The sanitizer runs request-scoped inside `wrap_model_call`/`awrap_model_call`, in the same pass that already strips Anthropic `tool_use` blocks on `main`; it writes nothing to the checkpoint and does not change `after_agent`'s restore behavior: - the intercepted frontend call is still restored with its synthetic `forwarded_to_frontend` result before the sanitizer runs, so an **answered** `function_call` block survives the wrap (`test_next_model_call_keeps_answered_function_call_blocks_on_restore`); - only blocks with no matching answered tool call are stripped — the poisoned-replay shape is covered end-to-end through `wrap_model_call`/`awrap_model_call` (`test_replayed_cancelled_turn_strips_orphan_function_call_blocks`, sync + async). ## Related PRs and Issues Fixes #6676 Re-lands #4473 ## Checklist - [x] Tests pass: `pytest tests/test_copilotkit_lg_middleware.py` → 91 passed (4 regression tests: the two unit shapes plus the two request-scoped wrap paths above, sync + async) - [x] Full local suite: `pytest tests/` → 232 passed, 11 skipped (the 4 `test_intercepted_tool_call_events.py` failures reproduce on pristine `main` in the same local environment — dependency drift, not from this branch) - [x] `ruff format --check` clean on the changed files; no new `ruff check` findings - [x] `__tests__` / devSK: N/A (Python SDK)
This commit is contained in:
Ben Taylor
@@ -649,6 +649,14 @@ class CopilotKitMiddleware(AgentMiddleware[StateSchema, Any]):
|
||||
the duplicate toolResult IDs. We keep the real result (non-interrupted)
|
||||
over the placeholder, falling back to the last occurrence if both look
|
||||
real.
|
||||
|
||||
Content-block syncing covers both provider shapes: Anthropic `tool_use`
|
||||
blocks (keyed on block["id"]) and OpenAI Responses `function_call`
|
||||
blocks (keyed on block["call_id"]; block["id"] on those is the item id,
|
||||
not the call id). Orphaned `function_call` blocks make langchain-openai
|
||||
re-emit them as Responses input items with no matching
|
||||
function_call_output, which OpenAI rejects with
|
||||
"No tool output found for function call call_...".
|
||||
"""
|
||||
# 4. Deduplicate ToolMessages by tool_call_id before all other processing.
|
||||
# patch_orphan_tool_calls adds "…was interrupted before completion."
|
||||
@@ -706,26 +714,36 @@ class CopilotKitMiddleware(AgentMiddleware[StateSchema, Any]):
|
||||
|
||||
tool_calls = getattr(msg, "tool_calls", None) or []
|
||||
|
||||
# 1. Sync content with tool_calls: remove tool_use content blocks
|
||||
# that aren't in msg.tool_calls (e.g. stripped by after_model
|
||||
# but content blocks left behind in checkpoint).
|
||||
# 1. Sync content with tool_calls: remove tool_use (Anthropic) and
|
||||
# function_call (OpenAI Responses) content blocks that aren't in
|
||||
# msg.tool_calls (e.g. stripped by after_model but content blocks
|
||||
# left behind in checkpoint).
|
||||
def _orphan_tool_block(block, tc_ids):
|
||||
if not isinstance(block, dict):
|
||||
return False
|
||||
btype = block.get("type")
|
||||
if btype == "tool_use":
|
||||
return block.get("id") not in tc_ids
|
||||
if btype == "function_call":
|
||||
return block.get("call_id") not in tc_ids
|
||||
return False
|
||||
|
||||
if tool_calls and isinstance(msg.content, list):
|
||||
tc_ids = {tc.get("id") for tc in tool_calls}
|
||||
msg.content = [
|
||||
block
|
||||
for block in msg.content
|
||||
if not (
|
||||
isinstance(block, dict)
|
||||
and block.get("type") == "tool_use"
|
||||
and block.get("id") not in tc_ids
|
||||
)
|
||||
if not _orphan_tool_block(block, tc_ids)
|
||||
]
|
||||
elif not tool_calls and isinstance(msg.content, list):
|
||||
# No tool_calls at all — strip ALL tool_use content blocks
|
||||
# No tool_calls at all — strip ALL tool_use / function_call blocks
|
||||
msg.content = [
|
||||
block
|
||||
for block in msg.content
|
||||
if not (isinstance(block, dict) and block.get("type") == "tool_use")
|
||||
if not (
|
||||
isinstance(block, dict)
|
||||
and block.get("type") in ("tool_use", "function_call")
|
||||
)
|
||||
]
|
||||
|
||||
if not tool_calls:
|
||||
@@ -753,15 +771,23 @@ class CopilotKitMiddleware(AgentMiddleware[StateSchema, Any]):
|
||||
tc for tc in tool_calls if tc.get("id") in adjacent_tc_ids
|
||||
]
|
||||
|
||||
# Also strip matching content blocks
|
||||
# Also strip matching content blocks (tool_use + function_call)
|
||||
if isinstance(msg.content, list):
|
||||
msg.content = [
|
||||
block
|
||||
for block in msg.content
|
||||
if not (
|
||||
isinstance(block, dict)
|
||||
and block.get("type") == "tool_use"
|
||||
and block.get("id") in unanswered_ids
|
||||
and (
|
||||
(
|
||||
block.get("type") == "tool_use"
|
||||
and block.get("id") in unanswered_ids
|
||||
)
|
||||
or (
|
||||
block.get("type") == "function_call"
|
||||
and block.get("call_id") in unanswered_ids
|
||||
)
|
||||
)
|
||||
)
|
||||
]
|
||||
|
||||
@@ -1117,10 +1143,10 @@ class CopilotKitMiddleware(AgentMiddleware[StateSchema, Any]):
|
||||
|
||||
for message in messages:
|
||||
if isinstance(message, AIMessage) and message.id == original_message_id:
|
||||
restored_tool_calls = (
|
||||
copilotkit_state.get("original_tool_calls")
|
||||
or [*(message.tool_calls or []), *intercepted_tool_calls]
|
||||
)
|
||||
restored_tool_calls = copilotkit_state.get("original_tool_calls") or [
|
||||
*(message.tool_calls or []),
|
||||
*intercepted_tool_calls,
|
||||
]
|
||||
updated_messages.append(
|
||||
self._copy_ai_message_with_tool_calls(
|
||||
message,
|
||||
|
||||
@@ -1188,6 +1188,210 @@ def test_bedrock_fix_repairs_string_args_to_dicts():
|
||||
assert repaired.tool_calls[0]["args"] == {"q": "hello"}
|
||||
|
||||
|
||||
def test_bedrock_fix_strips_orphan_function_call_content_blocks_no_tool_calls():
|
||||
"""When tool_calls is empty (e.g. after_model intercepted the frontend tool),
|
||||
any leftover OpenAI Responses 'function_call' content block must be stripped
|
||||
so langchain-openai doesn't re-emit it as an unanswered Responses input
|
||||
item ("No tool output found for function call call_..." 400 from OpenAI).
|
||||
"""
|
||||
ai = AIMessage(
|
||||
content=[
|
||||
{"type": "text", "text": "calling tool"},
|
||||
{
|
||||
"type": "function_call",
|
||||
"call_id": "call_orphan",
|
||||
"name": "frontend_action",
|
||||
"arguments": "{}",
|
||||
"id": "fc_item_1",
|
||||
},
|
||||
],
|
||||
id="ai-1",
|
||||
)
|
||||
messages: list[Any] = [HumanMessage("hi"), ai]
|
||||
|
||||
CopilotKitMiddleware._fix_messages_for_bedrock(messages)
|
||||
|
||||
repaired = next(m for m in messages if isinstance(m, AIMessage))
|
||||
assert isinstance(repaired.content, list)
|
||||
types_left = [b.get("type") for b in repaired.content if isinstance(b, dict)]
|
||||
assert "function_call" not in types_left
|
||||
assert "text" in types_left
|
||||
|
||||
|
||||
def test_bedrock_fix_strips_orphan_function_call_blocks_when_tool_calls_partial():
|
||||
"""function_call content blocks whose call_id isn't in tool_calls must
|
||||
be removed alongside their tool_calls peers."""
|
||||
ai = AIMessage(
|
||||
content=[
|
||||
{
|
||||
"type": "function_call",
|
||||
"call_id": "call_kept",
|
||||
"name": "search",
|
||||
"arguments": "{}",
|
||||
},
|
||||
{
|
||||
"type": "function_call",
|
||||
"call_id": "call_orphan",
|
||||
"name": "frontend_action",
|
||||
"arguments": "{}",
|
||||
},
|
||||
],
|
||||
tool_calls=[{"id": "call_kept", "name": "search", "args": {}}],
|
||||
id="ai-1",
|
||||
)
|
||||
answered = ToolMessage(content="ok", tool_call_id="call_kept")
|
||||
messages: list[Any] = [HumanMessage("hi"), ai, answered]
|
||||
|
||||
CopilotKitMiddleware._fix_messages_for_bedrock(messages)
|
||||
|
||||
repaired = next(m for m in messages if isinstance(m, AIMessage))
|
||||
call_ids = [
|
||||
b.get("call_id")
|
||||
for b in repaired.content
|
||||
if isinstance(b, dict) and b.get("type") == "function_call"
|
||||
]
|
||||
assert call_ids == ["call_kept"]
|
||||
|
||||
|
||||
def _function_call_block(call_id: str, name: str, arguments: str) -> dict[str, Any]:
|
||||
return {
|
||||
"type": "function_call",
|
||||
"call_id": call_id,
|
||||
"name": name,
|
||||
"arguments": arguments,
|
||||
"id": f"fc_{call_id}",
|
||||
}
|
||||
|
||||
|
||||
def _function_call_ids(message: AIMessage) -> list[str]:
|
||||
return [
|
||||
b.get("call_id")
|
||||
for b in message.content
|
||||
if isinstance(b, dict) and b.get("type") == "function_call"
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("use_async", [False, True])
|
||||
def test_next_model_call_keeps_answered_function_call_blocks_on_restore(use_async):
|
||||
"""The restore path must not over-strip: when the intercepted frontend
|
||||
call is restored with its synthetic result (and the backend call has its
|
||||
real result), the equivalent `function_call` content blocks stay in the
|
||||
request-scoped history — they are answered, so langchain-openai can pair
|
||||
each Responses input item with its function_call_output.
|
||||
"""
|
||||
middleware = CopilotKitMiddleware()
|
||||
fe_tool = {"function": {"name": "navigate"}}
|
||||
backend_call = {"id": "1", "name": "backend_search", "args": {"q": "hi"}}
|
||||
frontend_call = {"id": "2", "name": "navigate", "args": {"path": "/x"}}
|
||||
initial_state = {
|
||||
"messages": [
|
||||
HumanMessage("hi"),
|
||||
AIMessage(
|
||||
content=[
|
||||
{"type": "text", "text": "calling tools"},
|
||||
_function_call_block("1", "backend_search", '{"q": "hi"}'),
|
||||
_function_call_block("2", "navigate", '{"path": "/x"}'),
|
||||
],
|
||||
tool_calls=[backend_call, frontend_call],
|
||||
id="ai-1",
|
||||
),
|
||||
],
|
||||
"copilotkit": {"actions": [fe_tool]},
|
||||
}
|
||||
|
||||
after_model = middleware.after_model(initial_state, MagicMock(name="runtime"))
|
||||
assert after_model is not None
|
||||
# after_model strips tool_calls but leaves the content blocks behind
|
||||
stripped_ai = after_model["messages"][-1]
|
||||
assert [tc["id"] for tc in stripped_ai.tool_calls] == ["1"]
|
||||
assert _function_call_ids(stripped_ai) == ["1", "2"]
|
||||
|
||||
messages = [
|
||||
*after_model["messages"],
|
||||
ToolMessage(content='{"hits": 1}', tool_call_id="1"),
|
||||
]
|
||||
state = {
|
||||
"messages": messages,
|
||||
"copilotkit": {**initial_state["copilotkit"], **after_model["copilotkit"]},
|
||||
}
|
||||
request = _make_request(state=state, messages=messages)
|
||||
|
||||
if use_async:
|
||||
received: dict[str, ModelRequest] = {}
|
||||
|
||||
async def handler(req: ModelRequest):
|
||||
received["req"] = req
|
||||
return "ok"
|
||||
|
||||
asyncio.run(middleware.awrap_model_call(request, handler))
|
||||
seen = received["req"]
|
||||
else:
|
||||
seen, _ = _run_wrap(middleware, request)
|
||||
|
||||
seen_ai = next(
|
||||
m for m in seen.messages if isinstance(m, AIMessage) and m.id == "ai-1"
|
||||
)
|
||||
assert [tc["id"] for tc in seen_ai.tool_calls] == ["1", "2"]
|
||||
assert _function_call_ids(seen_ai) == ["1", "2"]
|
||||
tool_messages = [m for m in seen.messages if isinstance(m, ToolMessage)]
|
||||
assert [m.tool_call_id for m in tool_messages] == ["1", "2"]
|
||||
assert json.loads(tool_messages[1].content) == {"status": "forwarded_to_frontend"}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("use_async", [False, True])
|
||||
def test_replayed_cancelled_turn_strips_orphan_function_call_blocks(use_async):
|
||||
"""Regression for #6676: a run cancelled mid-turn persists the assistant
|
||||
turn with its Responses `function_call` content blocks but without any
|
||||
ToolMessage — the backend tool never ran and the intercept state was
|
||||
never written. On replay, the model must receive a history with no
|
||||
orphaned `function_call` blocks: langchain-openai would otherwise
|
||||
serialize them as Responses input items with no matching
|
||||
function_call_output, and OpenAI rejects every subsequent turn with
|
||||
400 "No tool output found for function call call_...".
|
||||
"""
|
||||
middleware = CopilotKitMiddleware()
|
||||
poisoned_ai = AIMessage(
|
||||
content=[
|
||||
{"type": "text", "text": "calling tools"},
|
||||
_function_call_block("1", "backend_search", '{"q": "hi"}'),
|
||||
_function_call_block("2", "navigate", '{"path": "/x"}'),
|
||||
],
|
||||
# after_model already stripped the frontend call; the run was
|
||||
# cancelled before the backend tool produced a ToolMessage.
|
||||
tool_calls=[{"id": "1", "name": "backend_search", "args": {"q": "hi"}}],
|
||||
id="ai-1",
|
||||
)
|
||||
messages: list[Any] = [HumanMessage("hi"), poisoned_ai]
|
||||
state = {
|
||||
"messages": messages,
|
||||
"copilotkit": {"actions": [{"function": {"name": "navigate"}}]},
|
||||
}
|
||||
request = _make_request(state=state, messages=messages)
|
||||
|
||||
if use_async:
|
||||
received: dict[str, ModelRequest] = {}
|
||||
|
||||
async def handler(req: ModelRequest):
|
||||
received["req"] = req
|
||||
return "ok"
|
||||
|
||||
asyncio.run(middleware.awrap_model_call(request, handler))
|
||||
seen = received["req"]
|
||||
else:
|
||||
seen, _ = _run_wrap(middleware, request)
|
||||
|
||||
seen_ai = next(
|
||||
m for m in seen.messages if isinstance(m, AIMessage) and m.id == "ai-1"
|
||||
)
|
||||
assert _function_call_ids(seen_ai) == []
|
||||
types_left = [b.get("type") for b in seen_ai.content if isinstance(b, dict)]
|
||||
assert "text" in types_left
|
||||
# the unanswered backend call is stripped as well — every tool call the
|
||||
# model sees must have an answer
|
||||
assert seen_ai.tool_calls == []
|
||||
assert not any(isinstance(m, ToolMessage) for m in seen.messages)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Checkpoint roundtrip — orphan-handoff contract
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user