Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
185 changes: 129 additions & 56 deletions cecli/coders/base_coder.py
Original file line number Diff line number Diff line change
Expand Up @@ -3440,11 +3440,17 @@ async def add_assistant_reply_to_cur_messages(self):
to be `None` when `tool_calls` are present.
"""
msg = dict(role="assistant")
response = (
self.partial_response_chunks[0]
if not self.stream
else litellm.stream_chunk_builder(self.partial_response_chunks)
)

# Prefer the response already produced by consolidate_chunks(): it carries
# the provider-specific fields (e.g. reasoning_items) that we preserved
# across all chunks, which a fresh litellm.stream_chunk_builder() pass
# alone would drop or truncate.
if self.partial_response_consolidated:
response = self.partial_response_consolidated[0]
elif not self.stream:
response = self.partial_response_chunks[0]
else:
response = litellm.stream_chunk_builder(self.partial_response_chunks)

try:
# Use response_dict as a regular dictionary
Expand Down Expand Up @@ -3963,63 +3969,51 @@ def consolidate_chunks(self):
if getattr(last_chunk, "usage", None):
response.usage = last_chunk.usage

# Collect provider-specific fields from chunks to preserve them
# We need to track both by ID (primary) and index (fallback) since
# early chunks might not have IDs established yet
provider_specific_fields_by_id = {}
provider_specific_fields_by_index = {}

# Collect message-level provider-specific fields (e.g. `reasoning_items`
# for reasoning models) from ALL chunks. litellm's stream_chunk_builder()
# merges these with last-wins semantics for list fields, silently dropping
# every reasoning item except the final one. Reasoning models depend on the
# full ordered item list being present in the assistant message so that
# exact-prefix prompt caching keeps working across turns, so we collect the
# fields ourselves and concatenate list-valued entries.
message_provider_specific_fields = {}
for chunk in self.partial_response_chunks:
try:
if chunk.choices and chunk.choices[0].delta and chunk.choices[0].delta.tool_calls:
for tool_call in chunk.choices[0].delta.tool_calls:
if (
hasattr(tool_call, "provider_specific_fields")
and tool_call.provider_specific_fields
):
# Ensure provider_specific_fields is a dictionary
psf = tool_call.provider_specific_fields
if not isinstance(psf, dict):
continue

# Try to use ID first
if hasattr(tool_call, "id") and tool_call.id:
tool_id = tool_call.id
if tool_id not in provider_specific_fields_by_id:
provider_specific_fields_by_id[tool_id] = {}
# Merge provider-specific fields for this tool ID
provider_specific_fields_by_id[tool_id].update(psf)
# Also track by index as fallback
elif hasattr(tool_call, "index"):
tool_index = tool_call.index
if tool_index not in provider_specific_fields_by_index:
provider_specific_fields_by_index[tool_index] = {}
provider_specific_fields_by_index[tool_index].update(psf)
if chunk.choices and chunk.choices[0].delta:
psf = getattr(chunk.choices[0].delta, "provider_specific_fields", None)
if psf and isinstance(psf, dict):
for key, value in psf.items():
if isinstance(value, list):
message_provider_specific_fields.setdefault(key, []).extend(value)
elif value is not None:
message_provider_specific_fields[key] = value
except (AttributeError, IndexError):
continue

if message_provider_specific_fields:
message_psf = getattr(response.choices[0].message, "provider_specific_fields", None)
if not isinstance(message_psf, dict):
message_psf = {}
message_psf.update(message_provider_specific_fields)
response.choices[0].message.provider_specific_fields = message_psf

try:
if response.choices[0].message.tool_calls:
for i, tool_call in enumerate(response.choices[0].message.tool_calls):
# Add provider-specific fields if we collected any for this tool
tool_id = tool_call.id

# Try ID first
if tool_id in provider_specific_fields_by_id:
# Add provider-specific fields directly to the tool call object
tool_call.provider_specific_fields = provider_specific_fields_by_id[tool_id]
# Fall back to index
elif i in provider_specific_fields_by_index:
# Add provider-specific fields directly to the tool call object
tool_call.provider_specific_fields = provider_specific_fields_by_index[i]

# Only append to partial_response_tool_calls if it's empty
if len(self.partial_response_tool_calls) == 0:
self.partial_response_tool_calls.append(tool_call)

self.partial_response_function_call = (
response.choices[0].message.tool_calls[0].function
)
message_tool_calls = response.choices[0].message.tool_calls
if message_tool_calls and len(message_tool_calls):
if self.stream:
built_tool_calls = self._build_tool_calls_from_chunks()
if built_tool_calls:
response.choices[0].message.tool_calls = built_tool_calls
self.partial_response_tool_calls = built_tool_calls
else:
# Fall back to litellm's merged list, keeping every call
self.partial_response_tool_calls = list(message_tool_calls)
else:
# Non-streaming: the single response chunk already carries the
# full tool_calls list
self.partial_response_tool_calls = list(message_tool_calls)

self.partial_response_function_call = self.partial_response_tool_calls[0].function
except AttributeError as e:
func_err = e

Expand Down Expand Up @@ -4075,6 +4069,85 @@ def consolidate_chunks(self):
self.partial_response_consolidated = (response, func_err, content_err)
return response, func_err, content_err

def _build_tool_calls_from_chunks(self):
"""Rebuild tool calls from the raw streaming chunks, keyed by delta index.

Streaming deltas for parallel tool calls arrive interleaved and may start
at any index (not necessarily 0). Indexing into a dict by the delta's
tool-call ``index`` before converting it back to a list ensures every
parallel call is preserved, correctly ordered, and keeps its
provider-specific fields (e.g. thought signatures) attached.
"""
from litellm.types.utils import ChatCompletionMessageToolCall, Function

tool_calls_dict = {}

for chunk in self.partial_response_chunks:
try:
if not (chunk.choices and chunk.choices[0].delta):
continue

delta = chunk.choices[0].delta
for tool_call in delta.tool_calls or []:
if tool_call is None:
continue

if nested.getter(tool_call, "function") is None:
continue

index = nested.getter(tool_call, "index")
if index is None:
index = len(tool_calls_dict)

entry = tool_calls_dict.setdefault(
index,
{
"id": None,
"name": None,
"type": "function",
"arguments": [],
"provider_specific_fields": {},
},
)

entry["id"] = nested.getter(tool_call, "id") or entry["id"]
entry["type"] = nested.getter(tool_call, "type") or entry["type"]
entry["name"] = nested.getter(tool_call, "function.name") or entry["name"]

arguments = nested.getter(tool_call, "function.arguments")
if arguments:
entry["arguments"].append(arguments)

psf = nested.getter(tool_call, "provider_specific_fields")
if not psf:
psf = nested.getter(tool_call, "function.provider_specific_fields")
if psf and isinstance(psf, dict):
entry["provider_specific_fields"].update(psf)
except (AttributeError, IndexError):
continue

tool_calls = []
for index in sorted(tool_calls_dict.keys()):
data = tool_calls_dict[index]
if not (data["id"] and data["name"]):
continue

function = Function(
arguments="".join(data["arguments"]) or "{}",
name=data["name"],
)
params = {
"id": data["id"],
"function": function,
"type": data["type"] or "function",
}
if data["provider_specific_fields"]:
params["provider_specific_fields"] = data["provider_specific_fields"]

tool_calls.append(ChatCompletionMessageToolCall(**params))

return tool_calls

def stream_wrapper(self, content, final):
if not hasattr(self, "_streaming_buffer_length"):
self._streaming_buffer_length = 0
Expand Down
6 changes: 5 additions & 1 deletion cecli/helpers/io_proxy.py
Original file line number Diff line number Diff line change
Expand Up @@ -51,6 +51,9 @@ def __init__(self, target: T, coder: Any) -> None:
super().__setattr__("_coder", weakref.ref(coder))
# Per-coder task storage: {coder_uuid: {attr_name: asyncio.Task}}
super().__setattr__("_per_coder", {coder_uuid: {}})
# Last tool `type` emitted via tool_output — lives on the proxy,
# never on the shared target (like coder_uuid)
super().__setattr__("_last_type", None)

# Register a per-coder input queue (TUI mode only)
# Allows the TUI to push input directly to this coder's queue,
Expand All @@ -76,6 +79,7 @@ def tool_output(self, *messages: Any, **kwargs: Any) -> Any:
"""Forward tool_output with coder_uuid injected."""
if "coder_uuid" not in kwargs:
kwargs["coder_uuid"] = self._coder_uuid
self._last_type = kwargs.get("type")
return self._target.tool_output(*messages, **kwargs)

def tool_error(self, message: str = "", strip: bool = True, **kwargs: Any) -> Any:
Expand Down Expand Up @@ -265,7 +269,7 @@ def __getattr__(self, name: str) -> Any:

def __setattr__(self, name: str, value: Any) -> None:
# Proxy-internal attributes — store on proxy instance only
if name in ("_target", "_coder_uuid", "_coder", "_per_coder"):
if name in ("_target", "_coder_uuid", "_coder", "_per_coder", "_last_type"):
super().__setattr__(name, value)
# Per-coder task attributes — isolate per-coder so coders don't
# compete for the same promise on the shared InputOutput instance
Expand Down
52 changes: 51 additions & 1 deletion cecli/helpers/requests.py
Original file line number Diff line number Diff line change
Expand Up @@ -172,7 +172,7 @@ def add_continue_for_no_prefill(model, messages, tools):

if not model.info.get("supports_assistant_prefill", False):
# Only add "Continue" if the last message is not a user message
if not messages or messages[-1].get("role") != "user":
if not messages or messages[-1].get("role") not in ("user", "tool"):
# Add a user message with content "Continue" to the messages list
append_message = True

Expand Down Expand Up @@ -214,6 +214,7 @@ def prevent_consecutive_assistant_messages(messages):


def model_request_parser(model, messages, tools):
messages = _copy_messages_for_request(messages)
messages = thought_signature(model, messages)
messages = remove_empty_tool_calls(messages)
messages = concatenate_user_messages(messages)
Expand All @@ -222,3 +223,52 @@ def model_request_parser(model, messages, tools):
messages = add_continue_for_no_prefill(model, messages, tools)
messages = prevent_consecutive_assistant_messages(messages)
return messages


def _copy_psf_container(container):
"""Shallow-copy a container dict and give its provider_specific_fields a fresh dict."""
container = dict(container)

psf = container.get("provider_specific_fields")
if isinstance(psf, dict):
container["provider_specific_fields"] = dict(psf)

return container


def _copy_messages_for_request(messages):
"""Return a shallow copy of the message list that request formatting can safely mutate.

The formatters in this module (``thought_signature``, ``add_reasoning_content``,
``add_continue_for_no_prefill``, ``ensure_alternating_roles``, ...) mutate message
dicts in place. The messages passed in come straight from the conversation store via
``BaseMessage.to_dict()``, which shares the nested ``provider_specific_fields`` dicts
with the stored history, so a bare top-level copy would corrupt the stored messages —
e.g. ``reasoning_items`` (relied on for exact-prefix prompt caching) would be popped
out of ``provider_specific_fields`` and lost from the store.

Shallow-copy the top-level message dict plus the nested ``provider_specific_fields``
dicts (message-level and per tool-call) — the only structures the formatters mutate —
instead of deep-copying the whole message.
"""
copied = []
for msg in messages:
msg = dict(msg)

psf = msg.get("provider_specific_fields")
if isinstance(psf, dict):
msg["provider_specific_fields"] = dict(psf)

tool_calls = msg.get("tool_calls")
if isinstance(tool_calls, list):
msg["tool_calls"] = [
_copy_psf_container(call) if isinstance(call, dict) else call for call in tool_calls
]

function_call = msg.get("function_call")
if isinstance(function_call, dict):
msg["function_call"] = _copy_psf_container(function_call)

copied.append(msg)

return copied
Loading
Loading