refactor: move the stream tool-call accumulator out of the chat-completions facade

Offsets the facade growth from the carrier capture (code-health FILE_LINES/CC
ratchets): _ToolCallAccumulator moves to chat_completion_helpers_tool_calls.py,
the Copilot reasoning_text fallback folds into StreamCarriers.feed, and the
reasoning_text display rides NormalizedResponse.reasoning instead of a new
branch in agent_runtime_helpers.extract_reasoning.
This commit is contained in:
Teknium committed 2026-10-10 19:22:07 -07:00
1 parent 2b2e4d908d
commit 208fd6e43d
6 files changed
+89 -82

No files matched your search

-3
View File
@@ -1516,9 +1516,6 @@ def extract_reasoning(agent, assistant_message) -> Optional[str]:
parts.append(text)
_add(getattr(assistant_message, "reasoning", None))
_add(getattr(assistant_message, "reasoning_content", None))
if not parts: # Copilot /chat/completions names its readable reasoning ``reasoning_text``
from agent.reasoning_carriers import field
_add(field(assistant_message, "reasoning_text"))
# reasoning_details: [{"type": "reasoning.summary", "summary": "...", ...}, ...]
for detail in getattr(assistant_message, "reasoning_details", None) or []:
if isinstance(detail, dict):
+3 -72
View File
@@ -2731,73 +2731,6 @@ class _BedrockStream:
return _with_stream_emitters(self.agent, self._poll)
class _ToolCallAccumulator:
"""Assemble streamed tool-call deltas into complete ``tool_calls`` entries
(``acc``: slot index -> entry dict). Ollama-compatible endpoints reuse index 0
for every call in a parallel batch, distinguishing them only by id, so a new
id at an already-seen raw index is redirected to a fresh slot."""
def __init__(self):
self.acc: dict = {}
self._notified: set = set()
self._last_id_at_idx: dict = {} # raw_index -> last seen non-empty id
self._active_slot_by_idx: dict = {} # raw_index -> current slot in acc
# Argument deltas are collected per slot and joined once in ``materialize`` —
# ``+=`` per chunk rebuilds the whole string every delta (quadratic on big args).
self._argument_parts: dict[int, list[str]] = {}
def materialize(self) -> dict:
"""Join buffered argument deltas into each entry's ``arguments``; idempotent. Returns ``acc``."""
for idx, parts in self._argument_parts.items():
self.acc[idx]["function"]["arguments"] = "".join(parts)
return self.acc
def feed(self, tc_delta) -> Optional[str]:
"""Merge one delta; return the tool name the first time it is complete."""
raw_idx = getattr(tc_delta, "index", None)
if raw_idx is None:
raw_idx = 0
tc_id = getattr(tc_delta, "id", None)
delta_id = tc_id or ""
if isinstance(tc_id, int): # Poolside sends integer ids
tc_id = str(tc_id)
self._active_slot_by_idx.setdefault(raw_idx, raw_idx)
if delta_id and raw_idx in self._last_id_at_idx and delta_id != self._last_id_at_idx[raw_idx]:
self._active_slot_by_idx[raw_idx] = max(self.acc, default=-1) + 1
if delta_id:
self._last_id_at_idx[raw_idx] = delta_id
idx = self._active_slot_by_idx[raw_idx]
entry = self.acc.setdefault(
idx, {"id": tc_id or "", "type": "function", "function": {"name": "", "arguments": ""}, "extra_content": None},
)
parts = self._argument_parts.setdefault(idx, [])
if tc_id:
entry["id"] = tc_id
tc_function = getattr(tc_delta, "function", None)
if tc_function:
if getattr(tc_function, "name", None):
# Assignment, not +=: names arrive complete and some providers (MiniMax via
# NVIDIA NIM) resend the full name every chunk — += gives "read_fileread_file".
entry["function"]["name"] = tc_function.name
if getattr(tc_function, "arguments", None):
parts.append(tc_function.arguments)
from agent.reasoning_carriers import field
if isinstance(fn_sig := field(tc_function, "thought_signature"), str) and fn_sig:
entry["thought_signature"] = fn_sig
extra = getattr(tc_delta, "extra_content", None)
if extra is None and hasattr(tc_delta, "model_extra"):
extra = (tc_delta.model_extra if isinstance(tc_delta.model_extra, dict) else {}).get("extra_content")
if extra is not None:
entry["extra_content"] = _dump_if_model(extra)
name = entry["function"]["name"]
if name and idx not in self._notified:
self._notified.add(idx)
return name
return None
class _StreamingCall(StreamingWaitMonitor):
"""One streaming request on the chat_completions / anthropic_messages wire.
State shared between the request worker and the poll-loop monitor (heartbeat,
@@ -3111,6 +3044,7 @@ class _StreamingCall(StreamingWaitMonitor):
refusal_parts: list[str] = []
reasoning_details: list = [] # OpenRouter replay data (signatures, encrypted blocks)
pending_text_parts: list[str] = []
from agent.chat_completion_helpers_tool_calls import _ToolCallAccumulator
tool_calls = _ToolCallAccumulator()
tool_calls_acc = tool_calls.acc
from agent.reasoning_carriers import StreamCarriers
@@ -3197,9 +3131,7 @@ class _StreamingCall(StreamingWaitMonitor):
if reasoning_text is None and isinstance(getattr(delta, "model_extra", None), dict):
reasoning_text = delta.model_extra.get("reasoning_content") or delta.model_extra.get("reasoning")
# Copilot /chat/completions names its readable reasoning ``reasoning_text``.
copilot_reasoning = carriers.feed(delta)
if reasoning_text is None:
reasoning_text = copilot_reasoning
reasoning_text = carriers.feed(delta, reasoning_text)
if reasoning_text:
# Summary-part models omit the separator between markdown blocks; re-insert it.
reasoning_text = separate_glued_reasoning_blocks(
@@ -3289,8 +3221,7 @@ class _StreamingCall(StreamingWaitMonitor):
"length" if runaway else finish_reason, model_name, usage_obj, flush_pending=_flush_pending_stream_text,
response_id=response_id, upstream_provider=upstream_provider, reasoning_details=reasoning_details,
refusal_parts=refusal_parts)
if getattr(response, "choices", None):
carriers.apply(response.choices[0].message)
carriers.apply(response)
if runaway:
# Cut, not finished: the length path ends the turn on this mark instead of continuing.
response._runaway_repetition = True
+1 -1
View File
@@ -12,7 +12,7 @@ from __future__ import annotations
from types import SimpleNamespace
from typing import Any
from agent.chat_completion_helpers import _ToolCallAccumulator
from agent.chat_completion_helpers_tool_calls import _ToolCallAccumulator
from agent.message_content import flatten_message_text
from agent.reasoning_summaries import separate_glued_reasoning_blocks
@@ -0,0 +1,73 @@
"""Streamed tool-call delta assembly for the chat-completions stream (and Relay's accumulator)."""
from typing import Optional
from agent.reasoning_carriers import field
class _ToolCallAccumulator:
"""Assemble streamed tool-call deltas into complete ``tool_calls`` entries
(``acc``: slot index -> entry dict). Ollama-compatible endpoints reuse index 0
for every call in a parallel batch, distinguishing them only by id, so a new
id at an already-seen raw index is redirected to a fresh slot."""
def __init__(self):
self.acc: dict = {}
self._notified: set = set()
self._last_id_at_idx: dict = {} # raw_index -> last seen non-empty id
self._active_slot_by_idx: dict = {} # raw_index -> current slot in acc
# Argument deltas are collected per slot and joined once in ``materialize`` —
# ``+=`` per chunk rebuilds the whole string every delta (quadratic on big args).
self._argument_parts: dict[int, list[str]] = {}
def materialize(self) -> dict:
"""Join buffered argument deltas into each entry's ``arguments``; idempotent. Returns ``acc``."""
for idx, parts in self._argument_parts.items():
self.acc[idx]["function"]["arguments"] = "".join(parts)
return self.acc
def feed(self, tc_delta) -> Optional[str]:
"""Merge one delta; return the tool name the first time it is complete."""
raw_idx = getattr(tc_delta, "index", None)
if raw_idx is None:
raw_idx = 0
tc_id = getattr(tc_delta, "id", None)
delta_id = tc_id or ""
if isinstance(tc_id, int): # Poolside sends integer ids
tc_id = str(tc_id)
self._active_slot_by_idx.setdefault(raw_idx, raw_idx)
if delta_id and raw_idx in self._last_id_at_idx and delta_id != self._last_id_at_idx[raw_idx]:
self._active_slot_by_idx[raw_idx] = max(self.acc, default=-1) + 1
if delta_id:
self._last_id_at_idx[raw_idx] = delta_id
idx = self._active_slot_by_idx[raw_idx]
entry = self.acc.setdefault(
idx, {"id": tc_id or "", "type": "function", "function": {"name": "", "arguments": ""}, "extra_content": None},
)
parts = self._argument_parts.setdefault(idx, [])
if tc_id:
entry["id"] = tc_id
tc_function = getattr(tc_delta, "function", None)
if tc_function:
if getattr(tc_function, "name", None):
# Assignment, not +=: names arrive complete and some providers (MiniMax via
# NVIDIA NIM) resend the full name every chunk — += gives "read_fileread_file".
entry["function"]["name"] = tc_function.name
if getattr(tc_function, "arguments", None):
parts.append(tc_function.arguments)
# Copilot's Gemini 3 variant signs the call inside ``function``.
if isinstance(fn_sig := field(tc_function, "thought_signature"), str) and fn_sig:
entry["thought_signature"] = fn_sig
extra = getattr(tc_delta, "extra_content", None)
if extra is None and hasattr(tc_delta, "model_extra"):
extra = (tc_delta.model_extra if isinstance(tc_delta.model_extra, dict) else {}).get("extra_content")
if extra is not None:
from agent.chat_completion_helpers import _dump_if_model
entry["extra_content"] = _dump_if_model(extra)
name = entry["function"]["name"]
if name and idx not in self._notified:
self._notified.add(idx)
return name
return None
+9 -5
View File
@@ -61,18 +61,22 @@ class StreamCarriers:
self.opaque: Optional[str] = None
self.text_parts: list[str] = []
def feed(self, delta: Any) -> Optional[str]:
"""Absorb one delta; returns its ``reasoning_text`` fragment (Copilot's readable reasoning)."""
def feed(self, delta: Any, reasoning: Optional[str]) -> Optional[str]:
"""Absorb one delta; returns ``reasoning``, else its ``reasoning_text`` fragment (Copilot's readable reasoning)."""
self.signature = thought_signature(field(delta, "extra_content")) or self.signature
opaque = field(delta, "reasoning_opaque")
self.opaque = opaque if isinstance(opaque, str) and opaque else self.opaque
text = field(delta, "reasoning_text")
if not (isinstance(text, str) and text):
return None
return reasoning
self.text_parts.append(text)
return text
return text if reasoning is None else reasoning
def apply(self, message: Any) -> None:
def apply(self, response: Any) -> None:
choices = getattr(response, "choices", None)
message = getattr(choices[0], "message", None) if choices else None
if message is None:
return
if self.signature:
message.extra_content = {"google": {"thought_signature": self.signature}}
if self.opaque:
+3 -1
View File
@@ -664,7 +664,9 @@ class ChatCompletionsTransport(ProviderTransport):
return NormalizedResponse(
content=content, tool_calls=tool_calls, finish_reason=finish_reason,
reasoning=getattr(msg, "reasoning", None), usage=usage, provider_data=provider_data or None,
# Copilot /chat/completions names its readable reasoning ``reasoning_text``.
reasoning=getattr(msg, "reasoning", None) or provider_data.get("reasoning_text"), usage=usage,
provider_data=provider_data or None,
)
def _normalize_tool_call(self, tc: Any) -> ToolCall | None: