a remote's tool call reaches the client (v0.4.60)

Every tool-using request to a pod-served model answered an EMPTY assistant
message: content "", finish_reason "stop", no tool_calls. Asked directly the
pod answered the call it had generated; through the orchestrator the same
request answered nothing. Both the LoRA (diogene-alpha) and the base model
were affected, so no adapter was involved.

A proxy backend's generate_chat returns a STRING. The remote server owns the
chat template, so it is the side that recognises a tool call and reports it
structurally -- `{"content": null, "tool_calls": [...]}`. `message.content or
""` dropped that on the floor; the caller then had an empty string, found no
call to parse in it, and answered the client with an empty message.

The calls now travel beside the text, where usage and logprobs already do:
get_last_tool_calls() on the backend, forwarded by the manager, preferred over
re-parsing text in both the streamed and buffered chat paths. A backend that
generates locally still keeps its call inside the text and answers None, so
that path is untouched.

Streaming needed the same: a streamed call arrives as deltas with no content
at all, so merge_tool_call_deltas folds the fragments -- keyed by the call's
index, never by arrival order, since concatenating the arguments of the wrong
call yields valid JSON for a call nobody made.
Co-Authored-By: 's avatarClaude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01S8LAP3ueNzQJtwzhgVbq66
parent fa4306b3
...@@ -16,7 +16,7 @@ ...@@ -16,7 +16,7 @@
# Canonical product version for CoderAI — single source of truth. Both the API # Canonical product version for CoderAI — single source of truth. Both the API
# metadata and the admin web UI read from here. # metadata and the admin web UI read from here.
__version__ = "0.4.59" __version__ = "0.4.60"
__tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference." __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
# Configure the CUDA caching allocator BEFORE torch is imported anywhere. # Configure the CUDA caching allocator BEFORE torch is imported anywhere.
......
...@@ -2951,7 +2951,14 @@ async def stream_chat_response( ...@@ -2951,7 +2951,14 @@ async def stream_chat_response(
logger.debug("Error converting tool: %s (type: %s)", e, type(t)) logger.debug("Error converting tool: %s (type: %s)", e, type(t))
continue continue
try: try:
tool_calls = tool_parser.extract_tool_calls(answer_text, tool_objects) # A proxy backend's remote parses the call itself and answers
# with structured tool_calls and no content at all (see
# ModelBackend.get_last_tool_calls), so there is nothing in the
# streamed text to extract. Prefer what the server reported.
tool_calls = current_manager.get_last_tool_calls() if hasattr(
current_manager, 'get_last_tool_calls') else None
if not tool_calls:
tool_calls = tool_parser.extract_tool_calls(answer_text, tool_objects)
# FIX: Validate extracted tool calls have valid JSON (stream_chat_response) # FIX: Validate extracted tool calls have valid JSON (stream_chat_response)
if tool_calls: if tool_calls:
...@@ -3234,8 +3241,13 @@ async def generate_chat_response( ...@@ -3234,8 +3241,13 @@ async def generate_chat_response(
logger.debug("Error converting tool: %s (type: %s)", e, type(t)) logger.debug("Error converting tool: %s (type: %s)", e, type(t))
continue continue
try: try:
tool_calls = tool_parser.extract_tool_calls(generated_text, tool_objects) # Same as the streamed path: a proxy backend reports the call the
# remote parsed, and `generated_text` is then empty by design.
tool_calls = current_manager.get_last_tool_calls() if hasattr(
current_manager, 'get_last_tool_calls') else None
if not tool_calls:
tool_calls = tool_parser.extract_tool_calls(generated_text, tool_objects)
# FIX: Validate extracted tool calls have valid JSON (generate_chat_response) # FIX: Validate extracted tool calls have valid JSON (generate_chat_response)
if tool_calls: if tool_calls:
from codai.models.parser import validate_json_complete from codai.models.parser import validate_json_complete
......
...@@ -65,4 +65,51 @@ class ModelBackend(ABC): ...@@ -65,4 +65,51 @@ class ModelBackend(ABC):
def get_context_size(self) -> int: def get_context_size(self) -> int:
"""Return the model's context window size.""" """Return the model's context window size."""
return 2048 # Default fallback return 2048 # Default fallback
\ No newline at end of file
def get_last_tool_calls(self):
"""The tool calls the backend itself reported for the last call, or None.
A backend that generates text has none: the call is still inside the
generated string, and the caller's parser digs it out. A *proxy* backend
is the opposite case — the remote server owns the chat template, so it
parses the call and answers with structured ``tool_calls`` and a null
``content``. The text contract of ``generate_chat`` cannot carry that, so
the proxy leaves them here, next to ``get_last_usage``.
"""
return None
def merge_tool_call_deltas(acc: Optional[List[dict]], delta) -> Optional[List[dict]]:
"""Fold one streamed ``delta.tool_calls`` fragment into ``acc``.
OpenAI streams a tool call in pieces: the first fragment carries ``id`` and
the function name, later ones append to ``function.arguments``, and each is
tagged with the ``index`` of the call it belongs to. Concatenating the
arguments of the wrong call produces valid JSON for a call nobody made, so
fragments are keyed by that index and never by arrival order.
"""
if not delta:
return acc
out = list(acc or [])
for frag in delta:
if not isinstance(frag, dict):
continue
# Absent index means a single call, which is what one-call servers send.
idx = frag.get("index")
idx = 0 if idx is None else int(idx)
while len(out) <= idx:
out.append({"id": "", "type": "function",
"function": {"name": "", "arguments": ""}})
cur = out[idx]
if frag.get("id"):
cur["id"] = frag["id"]
if frag.get("type"):
cur["type"] = frag["type"]
fn = frag.get("function") or {}
if fn.get("name"):
cur["function"]["name"] = fn["name"]
piece = fn.get("arguments")
if piece:
cur["function"]["arguments"] += piece
return out or None
\ No newline at end of file
...@@ -25,7 +25,7 @@ import asyncio ...@@ -25,7 +25,7 @@ import asyncio
import threading import threading
from typing import AsyncGenerator, Dict, List, Optional from typing import AsyncGenerator, Dict, List, Optional
from codai.backends.base import ModelBackend from codai.backends.base import ModelBackend, merge_tool_call_deltas
class RemoteModelUnknown(RuntimeError): class RemoteModelUnknown(RuntimeError):
...@@ -106,6 +106,7 @@ class RunpodBackend(ModelBackend): ...@@ -106,6 +106,7 @@ class RunpodBackend(ModelBackend):
self._pool = None # pods: RunpodPodPool self._pool = None # pods: RunpodPodPool
self._last_usage: Dict = {} self._last_usage: Dict = {}
self._last_logprobs: Optional[Dict] = None self._last_logprobs: Optional[Dict] = None
self._last_tool_calls: Optional[List[Dict]] = None
# ------------------------------------------------------------------ # # ------------------------------------------------------------------ #
# lifecycle # lifecycle
...@@ -182,6 +183,18 @@ class RunpodBackend(ModelBackend): ...@@ -182,6 +183,18 @@ class RunpodBackend(ModelBackend):
""" """
return self._last_logprobs return self._last_logprobs
def get_last_tool_calls(self):
"""The tool calls the POD parsed out of the last answer, or None.
A pod serving an OpenAI API owns the chat template, so it is the side
that recognises a tool call and reports it structurally: the message
comes back with ``tool_calls`` set and ``content`` null. ``generate_chat``
returns a string, so those calls used to end here — the caller received
"" , found no call to parse in it, and answered the client with an empty
assistant message while the pod had generated a perfectly good call.
"""
return self._last_tool_calls
def cleanup(self) -> None: def cleanup(self) -> None:
# Serverless: nothing to tear down (RunPod owns it). Pods: the pool # Serverless: nothing to tear down (RunPod owns it). Pods: the pool
# reaper handles teardown (next phase). # reaper handles teardown (next phase).
...@@ -428,6 +441,7 @@ class RunpodBackend(ModelBackend): ...@@ -428,6 +441,7 @@ class RunpodBackend(ModelBackend):
import requests import requests
self._enter_request() self._enter_request()
self._last_logprobs = None self._last_logprobs = None
self._last_tool_calls = None
try: try:
for attempt in (1, 2): for attempt in (1, 2):
base, release, bearer = self._acquire_endpoint() base, release, bearer = self._acquire_endpoint()
...@@ -473,7 +487,9 @@ class RunpodBackend(ModelBackend): ...@@ -473,7 +487,9 @@ class RunpodBackend(ModelBackend):
self._store_usage(data.get("usage", {})) self._store_usage(data.get("usage", {}))
choice = (data.get("choices") or [{}])[0] choice = (data.get("choices") or [{}])[0]
self._last_logprobs = choice.get("logprobs") self._last_logprobs = choice.get("logprobs")
return choice.get("message", {}).get("content") or "" message = choice.get("message") or {}
self._last_tool_calls = message.get("tool_calls") or None
return message.get("content") or ""
finally: finally:
release() release()
finally: finally:
...@@ -486,6 +502,7 @@ class RunpodBackend(ModelBackend): ...@@ -486,6 +502,7 @@ class RunpodBackend(ModelBackend):
import requests import requests
self._enter_request() self._enter_request()
self._last_logprobs = None self._last_logprobs = None
self._last_tool_calls = None
try: try:
for attempt in (1, 2): for attempt in (1, 2):
# Pod acquire may provision (blocking) — do it off the event loop. # Pod acquire may provision (blocking) — do it off the event loop.
...@@ -568,9 +585,14 @@ class RunpodBackend(ModelBackend): ...@@ -568,9 +585,14 @@ class RunpodBackend(ModelBackend):
except ValueError: except ValueError:
return False return False
choice = (obj.get("choices") or [{}])[0] choice = (obj.get("choices") or [{}])[0]
text = (choice.get(delta_key) or {}).get("content") or "" piece = choice.get(delta_key) or {}
text = piece.get("content") or ""
if text: if text:
loop.call_soon_threadsafe(queue.put_nowait, text) loop.call_soon_threadsafe(queue.put_nowait, text)
# A streamed tool call arrives as deltas and carries no content at
# all, so it cannot travel in the text this generator yields.
self._last_tool_calls = merge_tool_call_deltas(
self._last_tool_calls, piece.get("tool_calls"))
if obj.get("usage"): if obj.get("usage"):
self._store_usage(obj["usage"]) self._store_usage(obj["usage"])
self._collect_logprobs(choice.get("logprobs")) self._collect_logprobs(choice.get("logprobs"))
......
...@@ -18,7 +18,7 @@ import asyncio ...@@ -18,7 +18,7 @@ import asyncio
import threading import threading
from typing import AsyncGenerator, Dict, List, Optional from typing import AsyncGenerator, Dict, List, Optional
from codai.backends.base import ModelBackend from codai.backends.base import ModelBackend, merge_tool_call_deltas
class VllmBackend(ModelBackend): class VllmBackend(ModelBackend):
...@@ -58,6 +58,7 @@ class VllmBackend(ModelBackend): ...@@ -58,6 +58,7 @@ class VllmBackend(ModelBackend):
self._served_name = self._model_id self._served_name = self._model_id
self._last_usage: Dict = {} self._last_usage: Dict = {}
self._last_logprobs: Optional[Dict] = None self._last_logprobs: Optional[Dict] = None
self._last_tool_calls: Optional[List[Dict]] = None
# ------------------------------------------------------------------ # # ------------------------------------------------------------------ #
# lifecycle # lifecycle
...@@ -150,6 +151,16 @@ class VllmBackend(ModelBackend): ...@@ -150,6 +151,16 @@ class VllmBackend(ModelBackend):
""" """
return self._last_logprobs return self._last_logprobs
def get_last_tool_calls(self):
"""The tool calls the SERVER parsed out of the last answer, or None.
vLLM applies the chat template and runs its own tool parser, so a
recognised call comes back structurally — ``tool_calls`` set, ``content``
null — and never appears in the text ``generate_chat`` returns. See the
same method on RunpodBackend.
"""
return self._last_tool_calls
def cleanup(self) -> None: def cleanup(self) -> None:
from codai.api import vllm_worker from codai.api import vllm_worker
key = getattr(self, "_svc_key", None) or getattr(self._cfg, "model_id", self._model_id) key = getattr(self, "_svc_key", None) or getattr(self._cfg, "model_id", self._model_id)
...@@ -346,6 +357,7 @@ class VllmBackend(ModelBackend): ...@@ -346,6 +357,7 @@ class VllmBackend(ModelBackend):
import requests import requests
self._enter_request() self._enter_request()
self._last_logprobs = None self._last_logprobs = None
self._last_tool_calls = None
try: try:
payload = self._chat_payload(messages, max_tokens, temperature, top_p, stop, False, payload = self._chat_payload(messages, max_tokens, temperature, top_p, stop, False,
logprobs=logprobs, top_logprobs=top_logprobs, logprobs=logprobs, top_logprobs=top_logprobs,
...@@ -366,7 +378,9 @@ class VllmBackend(ModelBackend): ...@@ -366,7 +378,9 @@ class VllmBackend(ModelBackend):
self._store_usage(data.get("usage", {})) self._store_usage(data.get("usage", {}))
choice = (data.get("choices") or [{}])[0] choice = (data.get("choices") or [{}])[0]
self._last_logprobs = choice.get("logprobs") self._last_logprobs = choice.get("logprobs")
return choice.get("message", {}).get("content") or "" message = choice.get("message") or {}
self._last_tool_calls = message.get("tool_calls") or None
return message.get("content") or ""
finally: finally:
self._exit_request() self._exit_request()
...@@ -377,6 +391,7 @@ class VllmBackend(ModelBackend): ...@@ -377,6 +391,7 @@ class VllmBackend(ModelBackend):
passthrough=None) -> AsyncGenerator[str, None]: passthrough=None) -> AsyncGenerator[str, None]:
self._enter_request() self._enter_request()
self._last_logprobs = None self._last_logprobs = None
self._last_tool_calls = None
try: try:
payload = self._chat_payload(messages, max_tokens, temperature, top_p, stop, True, payload = self._chat_payload(messages, max_tokens, temperature, top_p, stop, True,
logprobs=logprobs, top_logprobs=top_logprobs, logprobs=logprobs, top_logprobs=top_logprobs,
...@@ -430,9 +445,14 @@ class VllmBackend(ModelBackend): ...@@ -430,9 +445,14 @@ class VllmBackend(ModelBackend):
except ValueError: except ValueError:
return False return False
choice = (obj.get("choices") or [{}])[0] choice = (obj.get("choices") or [{}])[0]
text = (choice.get(delta_key) or {}).get("content") or "" piece = choice.get(delta_key) or {}
text = piece.get("content") or ""
if text: if text:
loop.call_soon_threadsafe(queue.put_nowait, text) loop.call_soon_threadsafe(queue.put_nowait, text)
# A streamed tool call arrives as deltas and carries no content at
# all, so it cannot travel in the text this generator yields.
self._last_tool_calls = merge_tool_call_deltas(
self._last_tool_calls, piece.get("tool_calls"))
if obj.get("usage"): if obj.get("usage"):
self._store_usage(obj["usage"]) self._store_usage(obj["usage"])
self._collect_logprobs(choice.get("logprobs")) self._collect_logprobs(choice.get("logprobs"))
......
...@@ -945,6 +945,21 @@ class ModelManager: ...@@ -945,6 +945,21 @@ class ModelManager:
except Exception: except Exception:
return None return None
def get_last_tool_calls(self):
"""Tool calls the backend reported structurally for the last call, or None.
Only a proxy backend has any: the remote server parsed the call itself
and left the text empty. A local backend answers None and the caller
parses the generated text as before.
"""
fn = getattr(self.backend, "get_last_tool_calls", None)
if fn is None:
return None
try:
return fn()
except Exception:
return None
def generate_chat(self, messages: List[Dict], max_tokens: Optional[int] = None, def generate_chat(self, messages: List[Dict], max_tokens: Optional[int] = None,
temperature: float = 0.7, top_p: float = 1.0, temperature: float = 0.7, top_p: float = 1.0,
stop: Optional[List[str]] = None, tools: Optional[List] = None, stop: Optional[List[str]] = None, tools: Optional[List] = None,
......
"""A pod answers with a tool call; the client must not get an empty message.
2026-10-21: every tool-using request to a pod-served model came back as
`content: ""`, `finish_reason: "stop"`, no tool_calls -- an empty response --
while the pod had generated a perfectly good call. Asked directly, the pod
answered `{"content": null, "tool_calls": [...]}`; through the orchestrator the
same request answered nothing.
Cause: a proxy backend's `generate_chat` returns a STRING. The remote owns the
chat template, so it is the side that recognises the call and reports it
structurally -- and `message.content or ""` threw it away. The caller then had
an empty string, found no call to parse in it, and answered the client with an
empty assistant message.
The calls now travel beside the text, like usage and logprobs already did.
"""
import json
import types
import pytest
from codai.backends.base import ModelBackend, merge_tool_call_deltas
from codai.backends.runpod import RunpodBackend
from codai.backends.vllm import VllmBackend
CALL = [{"id": "call_1", "type": "function",
"function": {"name": "agenda_udienze",
"arguments": '{"dal": "2026-10-21", "al": "2026-11-21"}'}}]
def _answer(message, usage=None):
"""A remote's non-streaming chat answer."""
r = types.SimpleNamespace()
r.status_code = 200
r.ok = True
r.json = lambda: {"choices": [{"message": message, "finish_reason": "tool_calls"}],
"usage": usage or {"prompt_tokens": 7, "completion_tokens": 9}}
r.text = json.dumps(message)
r.headers = {}
return r
# ------------------------------------------------------------ the pod's answer #
def _runpod(monkeypatch, message):
b = RunpodBackend.__new__(RunpodBackend)
b._model_id = b._served_name = "diogene-alpha"
b._ctx = 32768
b._last_usage = {}
b._last_logprobs = None
b._last_tool_calls = None
b._url = "https://pod:1/v1"
b._headers = {}
b._pool = None
b._mode = "serverless"
monkeypatch.setattr(b, "_acquire_endpoint",
lambda: ("https://pod:1/v1", lambda: None, ""))
monkeypatch.setattr("codai.api.pod_http.post",
lambda *a, **k: _answer(message))
return b
def test_a_pods_tool_call_is_not_lost(monkeypatch):
"""THE BUG: content is null, so the string contract carried nothing."""
b = _runpod(monkeypatch, {"role": "assistant", "content": None, "tool_calls": CALL})
assert b.generate_chat([{"role": "user", "content": "hi"}]) == ""
assert b.get_last_tool_calls() == CALL
def test_text_answers_are_unchanged(monkeypatch):
b = _runpod(monkeypatch, {"role": "assistant", "content": "Buongiorno"})
assert b.generate_chat([{"role": "user", "content": "hi"}]) == "Buongiorno"
assert b.get_last_tool_calls() is None
def test_a_later_text_answer_does_not_inherit_the_call(monkeypatch):
"""The side channel is per-call, like get_last_usage: a stale tool call
attached to the next request would answer a question nobody asked."""
b = _runpod(monkeypatch, {"role": "assistant", "content": None, "tool_calls": CALL})
b.generate_chat([{"role": "user", "content": "hi"}])
monkeypatch.setattr("codai.api.pod_http.post",
lambda *a, **k: _answer({"role": "assistant", "content": "ciao"}))
assert b.generate_chat([{"role": "user", "content": "hi"}]) == "ciao"
assert b.get_last_tool_calls() is None
def test_the_same_holds_for_a_local_vllm_server(monkeypatch):
b = VllmBackend.__new__(VllmBackend)
b._model_id = "qwen38"
b._ctx = 32768
b._last_usage = {}
b._last_logprobs = None
b._last_tool_calls = None
b._url = "http://127.0.0.1:8111"
monkeypatch.setattr(b, "_raise_for_status", lambda *a, **k: None)
monkeypatch.setattr("requests.post",
lambda *a, **k: _answer({"role": "assistant", "content": None,
"tool_calls": CALL}))
assert b.generate_chat([{"role": "user", "content": "hi"}]) == ""
assert b.get_last_tool_calls() == CALL
def test_a_generating_backend_reports_none():
"""A local backend keeps the call inside its text, where the parser finds
it; answering None leaves that path exactly as it was."""
assert ModelBackend.get_last_tool_calls(object()) is None
# -------------------------------------------------------- streamed fragments #
def test_fragments_are_assembled():
acc = None
acc = merge_tool_call_deltas(acc, [{"index": 0, "id": "call_1", "type": "function",
"function": {"name": "agenda_udienze",
"arguments": '{"dal": '}}])
acc = merge_tool_call_deltas(acc, [{"index": 0, "function": {"arguments": '"2026-10-21"}'}}])
assert acc == [{"id": "call_1", "type": "function",
"function": {"name": "agenda_udienze",
"arguments": '{"dal": "2026-10-21"}'}}]
assert json.loads(acc[0]["function"]["arguments"])["dal"] == "2026-10-21"
def test_two_interleaved_calls_keep_their_own_arguments():
"""Keyed by index, never by arrival order: concatenating the arguments of
the wrong call yields valid JSON for a call nobody made."""
acc = None
for frag in ([{"index": 0, "id": "a", "function": {"name": "one", "arguments": '{"x":'}}],
[{"index": 1, "id": "b", "function": {"name": "two", "arguments": '{"y":'}}],
[{"index": 0, "function": {"arguments": "1}"}}],
[{"index": 1, "function": {"arguments": "2}"}}]):
acc = merge_tool_call_deltas(acc, frag)
assert [c["function"]["name"] for c in acc] == ["one", "two"]
assert acc[0]["function"]["arguments"] == '{"x":1}'
assert acc[1]["function"]["arguments"] == '{"y":2}'
def test_a_server_that_omits_the_index_still_works():
acc = merge_tool_call_deltas(None, [{"id": "a", "function": {"name": "one",
"arguments": "{}"}}])
assert acc[0]["function"]["name"] == "one"
def test_nothing_in_nothing_out():
assert merge_tool_call_deltas(None, None) is None
assert merge_tool_call_deltas(None, []) is None
assert merge_tool_call_deltas(CALL, None) == CALL
def test_a_streamed_delta_is_collected_off_the_wire():
"""The generator yields text; a streamed call has none, so it has to be
picked up from the delta itself."""
b = RunpodBackend.__new__(RunpodBackend)
b._last_tool_calls = None
b._last_usage = {}
b._last_logprobs = None
chunk = {"choices": [{"delta": {"tool_calls": [
{"index": 0, "id": "call_1", "type": "function",
"function": {"name": "agenda_udienze", "arguments": "{}"}}]}}]}
delta = (chunk["choices"][0].get("delta") or {})
b._last_tool_calls = merge_tool_call_deltas(b._last_tool_calls, delta.get("tool_calls"))
assert b.get_last_tool_calls()[0]["function"]["name"] == "agenda_udienze"
# ------------------------------------------------------------ the manager hop #
def test_the_manager_forwards_them():
from codai.models.manager import ModelManager
m = ModelManager.__new__(ModelManager)
m.backend = types.SimpleNamespace(get_last_tool_calls=lambda: CALL)
assert m.get_last_tool_calls() == CALL
def test_the_manager_survives_a_backend_without_the_method():
from codai.models.manager import ModelManager
m = ModelManager.__new__(ModelManager)
m.backend = types.SimpleNamespace()
assert m.get_last_tool_calls() is None
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment