I've observed two failure modes in the streaming gemma4 tool-parser, when using it from opencode:
The fix (Path A — atomic-per-tool-call streaming). Replace upstream's tag-counting state machine in _extract_streaming with an emitter that, on every call, computes new_ends = current_text.count("<tool_call|>") - previous_text.count("<tool_call|>"), runs the same tool_call_regex the non-streaming path uses on current_text, and emits one DeltaToolCall per newly-closed tool call (for i in range(prev_end_count, end_count)) — each carrying index=i, freshly-minted id, type="function", function.name, and the complete arguments JSON (parsed via _parse_gemma4_args then json.dumps'd in one shot, no incremental diff, no trailing-char stripping). All emissions for a given delta are packed into a single DeltaMessage(tool_calls=[...]), so multi-boundary deltas are handled by construction; the strict-client trio (id / type / function.name) is present on every emission by construction; and the prev_tool_call_arr / streamed_args_for_tool / tool_call_ids slots are kept consistent so chat_completion/serving.py:820+'s finish-reason coverage check still works. The tradeoff is that clients no longer see character-by-character arguments deltas — each tool call's full args arrives in one SSE chunk when its <tool_call|> close is observed — which is a no-op for our workload (opencode buffers parallel tool calls before dispatch anyway, and read_file args fit in a single TCP segment) and the price of eliminating an entire class of state-machine race conditions. The change is a ~150-line rewrite confined to vllm-patches/gemma4_tool_parser.py (no other files touched), guarded by three new pytest regressions (multi-tool / mega-delta / mid-close-marker chunk splits) that run as the existing Dockerfile build-time smoke check.
"""Regression tests for the datapatcher gemma4 tool-parser patches.
Two failure modes are guarded here:
1. **Strict-client field re-emission** (original patch): vLLM 0.20.2's
upstream ``Gemma4ToolParser`` emitted ``id`` / ``type`` /
``function.name`` only on the *first* streaming chunk for a tool
call. @ai-sdk's ``openai-compatible`` provider (used by opencode)
Zod-validates EVERY chunk and bails out mid-stream with::
AI_InvalidResponseDataError: Expected 'id' to be a string
AI_InvalidResponseDataError: Expected 'function.name' to be a string
Hit ~64% / ~42% of agents in two production runs.
2. **Multi-boundary delta mis-attribution** (Path A patch): upstream's
``_extract_streaming`` was a single-Case dispatch (Case 1/2/3/4 —
at most one branch fires per call). When a single delta brought
end-of-tool-N + start-of-tool-N+1 + end-of-tool-N+1, the new tool's
Case 2 advance was skipped (``start_count > end_count`` guard
failed), and tool N's stripped trailing arg fragments leaked out
under index N+1 in the next delta. Reproduced reliably by
``dp-vllm-toolstream-test`` at c=500/1000 stream as
``args_invalid_json`` (35% / 21% per-request success vs 100%
non-stream on the same model).
Path A fixes this by emitting each tool call as a SINGLE
DeltaToolCall when its ``<tool_call|>`` close is observed --- no
incremental arg diffs, the streaming path uses the same regex as
the non-streaming path so success is by construction.
The tests below cover both failure modes:
* ``test_every_chunk_carries_the_required_strict_fields`` --- single
tool call, every emitted DeltaToolCall has ``id``, ``type``,
``function.name``. In Path A this collapses to a single emission
per tool call so the assertion is trivially satisfied; the test is
kept as a smoke check that the strict-client fields are still
populated.
* ``test_id_resets_between_requests`` --- two requests through the
same parser instance get distinct ids (no state leak).
* ``test_multiple_tools_emit_distinct_indices`` --- THREE parallel
tool calls (the production opencode pattern) each get a unique
index, id, name, and ``json.loads()``-able arguments.
* ``test_mega_delta_with_all_tools_in_one_chunk`` --- the worst-case
upstream regression: the WHOLE tool-call payload arrives in one
delta (which is what continuous-batching scheduler does under
high load). Path A must still emit one DeltaToolCall per tool
with valid args.
* ``test_partial_delta_split_inside_close_marker`` --- a delta
boundary that lands inside a ``<tool_call|>`` token must not
cause a tool call to be skipped or double-emitted.
Run this against the patched parser::
python -m pytest vllm-patches/test_gemma4_id_patch.py -q
It is also wired into the Dockerfile as a build-time smoke check, so an
upstream bump that quietly drops either patch will fail the image
build.
"""
from __future__ import annotations
import importlib.util
import json
import os
import sys
from pathlib import Path
import pytest
def _load_parser_module():
"""Load the patched parser module by path (works against the file in
``vllm-patches/`` *or* an already-installed vllm). We prefer the
in-repo file so the test runs against the version we ship.
"""
here = Path(__file__).resolve().parent
repo_copy = here / "gemma4_tool_parser.py"
# Make sure ``vllm`` is importable so the parser's own imports work.
import vllm # noqa: F401 (raises ImportError if vllm not installed)
spec = importlib.util.spec_from_file_location(
"datapatcher_gemma4_parser_under_test",
repo_copy,
)
module = importlib.util.module_from_spec(spec)
assert spec.loader is not None
spec.loader.exec_module(module)
return module
@pytest.fixture(scope="module")
def parser_cls():
mod = _load_parser_module()
return mod.Gemma4ToolParser
def _make_parser(parser_cls):
"""Construct a Gemma4ToolParser with a stub tokenizer.
``Gemma4ToolParser.__init__`` does ``self.vocab.get(TOOL_CALL_START)``
via the abstract base's ``vocab`` cached_property, which calls
``self.model_tokenizer.get_vocab()`` and raises ``RuntimeError`` if
either ``<|tool_call>`` or ``<tool_call|>`` is missing. So the stub
has to expose a ``get_vocab()`` method returning a dict that contains
those two tokens. Streaming itself doesn't actually use the token
IDs (it pattern-matches on the string forms in ``current_text``), so
the IDs we pick are arbitrary -- we just need them to be non-None.
"""
class _StubTokenizer:
def get_vocab(self):
return {
"<|tool_call>": 100001,
"<tool_call|>": 100002,
}
def encode(self, s, add_special_tokens=False): # pragma: no cover
return list(s.encode())
def decode(self, ids, skip_special_tokens=True): # pragma: no cover
return ""
return parser_cls(_StubTokenizer())
def _stream_chunks(parser, full_text: str, *, chunk_size: int = 8):
"""Feed ``full_text`` through the parser one slice at a time and
collect every emitted DeltaToolCall.
Returns a list of (index, id, type, name, arguments_delta) tuples.
"""
seen: list[tuple[int, str | None, str | None, str | None, str | None]] = []
previous = ""
for i in range(0, len(full_text), chunk_size):
delta_text = full_text[i : i + chunk_size]
current_text = previous + delta_text
delta_msg = parser.extract_tool_calls_streaming(
previous_text=previous,
current_text=current_text,
delta_text=delta_text,
previous_token_ids=[],
current_token_ids=[],
delta_token_ids=[],
request=None,
)
previous = current_text
if delta_msg is None or not getattr(delta_msg, "tool_calls", None):
continue
for tc in delta_msg.tool_calls:
fn = getattr(tc, "function", None) or {}
if not isinstance(fn, dict):
fn = fn.model_dump(exclude_none=True)
seen.append(
(
tc.index,
getattr(tc, "id", None),
getattr(tc, "type", None),
fn.get("name"),
fn.get("arguments"),
)
)
return seen
# A representative gemma4 tool-call transcript: one tool, one string
# arg, one numeric arg.
#
# IMPORTANT: gemma4's tool-call markers are *asymmetric* (note where the
# pipe is) and the body uses ``call:funcname{key:value,...}`` syntax,
# NOT a python-style ``funcname(arg=val)``:
#
# start marker: ``<|tool_call>`` (pipe BEFORE, no pipe AFTER)
# end marker: ``<tool_call|>`` (no pipe BEFORE, pipe AFTER)
# body regex: r"<\|tool_call>call:([\w\-\.]+)\{(.*?)\}<tool_call\|>"
# string delim: ``<|"|>`` (token 52 in the gemma4 vocab)
#
# Earlier versions of this test used "<|tool_call|>" on both sides plus
# python-call-style arguments; the parser's regex never matched, so the
# stream produced zero ``DeltaToolCall`` events and the assertion below
# fired with ``assert []``. See ``vllm-patches/gemma4_tool_parser.py``
# (constants near top, ``tool_call_regex`` in ``__init__``) if the
# format ever changes upstream.
TOOL_CALL_START = "<|tool_call>"
TOOL_CALL_END = "<tool_call|>"
_TRANSCRIPT = (
"Some preamble.\n"
f"{TOOL_CALL_START}call:read"
'{file_path:<|"|>/work/PROMPT.md<|"|>,limit:200}'
f"{TOOL_CALL_END}"
"And some trailing text."
)
def test_every_chunk_carries_the_required_strict_fields(parser_cls):
"""Every emitted DeltaToolCall must carry the trio of fields that
@ai-sdk's Zod parser validates: ``id``, ``type``, ``function.name``.
First-chunk values are the canonical ones we mint in
``_handle_tool_call_middle``; follow-up chunks must re-emit
byte-identical copies.
"""
parser = _make_parser(parser_cls)
chunks = _stream_chunks(parser, _TRANSCRIPT, chunk_size=6)
assert chunks, "expected at least one DeltaToolCall to be emitted"
first_id = chunks[0][1]
first_type = chunks[0][2]
first_name = chunks[0][3]
assert isinstance(first_id, str) and first_id, (
f"first chunk id must be a non-empty string, got {first_id!r}"
)
assert first_type == "function", (
f"first chunk type must be 'function', got {first_type!r}"
)
assert isinstance(first_name, str) and first_name, (
f"first chunk function.name must be a non-empty string, "
f"got {first_name!r}"
)
for idx, (tc_index, tc_id, tc_type, name, args) in enumerate(chunks):
ctx = (f"chunk {idx} (tool_index={tc_index}, name={name!r}, "
f"type={tc_type!r}, args={args!r})")
# id
assert isinstance(tc_id, str), (
f"{ctx} has non-string id {tc_id!r}; "
f"@ai-sdk Zod will reject this"
)
assert tc_id == first_id, (
f"{ctx} id={tc_id!r} differs from first-chunk id "
f"{first_id!r}; tool_call_ids tracking is broken"
)
# type
assert tc_type == "function", (
f"{ctx} type={tc_type!r}; @ai-sdk Zod requires the literal "
f"'function' on every chunk"
)
# function.name
assert isinstance(name, str), (
f"{ctx} has non-string function.name; @ai-sdk Zod will "
f"reject this with 'Expected function.name to be a string'"
)
assert name == first_name, (
f"{ctx} function.name={name!r} differs from first-chunk "
f"name {first_name!r}; prev_tool_call_arr lookup is broken"
)
def test_id_resets_between_requests(parser_cls):
"""A fresh request must mint a fresh id (no leakage of tool_call_ids
from the previous request)."""
parser = _make_parser(parser_cls)
first = _stream_chunks(parser, _TRANSCRIPT, chunk_size=6)
parser._reset_streaming_state()
second = _stream_chunks(parser, _TRANSCRIPT, chunk_size=6)
assert first and second
assert first[0][1] != second[0][1], (
"expected a fresh tool-call id after _reset_streaming_state(), "
f"got the same id twice: {first[0][1]!r}"
)
# --------------------------------------------------------------------------
# Path A regression: multi-boundary deltas (the c=500/1000 stream failure)
# --------------------------------------------------------------------------
# Three parallel `read_file` calls — the exact production opencode shape
# (PROMPT.md / GUIDELINES.md / sample/response_1.md). This is the payload
# that upstream's _extract_streaming mis-attributed under high concurrency.
_PARALLEL_TRANSCRIPT = (
"Sure, reading those files now.\n"
f'{TOOL_CALL_START}call:read_file{{filePath:<|"|>GUIDELINES.md<|"|>}}{TOOL_CALL_END}'
f'{TOOL_CALL_START}call:read_file{{filePath:<|"|>PROMPT.md<|"|>}}{TOOL_CALL_END}'
f'{TOOL_CALL_START}call:read_file{{filePath:<|"|>sample/response_1.md<|"|>}}{TOOL_CALL_END}'
"Done."
)
_EXPECTED_PARALLEL_ARGS = [
{"filePath": "GUIDELINES.md"},
{"filePath": "PROMPT.md"},
{"filePath": "sample/response_1.md"},
]
def _collect_tool_calls_by_index(chunks):
"""Group emitted (index, id, type, name, arguments) tuples by index
and return ``{index: (id, type, name, concatenated_args)}``.
For Path A every index has exactly one entry (we emit atomically),
but the helper concatenates to also work against the legacy
incremental-emit shape for diff-debugging.
"""
by_index: dict[int, list] = {}
for tc_index, tc_id, tc_type, name, args in chunks:
by_index.setdefault(tc_index, []).append((tc_id, tc_type, name, args))
out: dict[int, tuple[str | None, str | None, str | None, str]] = {}
for idx, entries in by_index.items():
# First non-None id / type / name across the entries; concatenate args.
ids = [e[0] for e in entries if e[0]]
types = [e[1] for e in entries if e[1]]
names = [e[2] for e in entries if e[2]]
args_concat = "".join(e[3] or "" for e in entries)
out[idx] = (
ids[0] if ids else None,
types[0] if types else None,
names[0] if names else None,
args_concat,
)
return out
def test_multiple_tools_emit_distinct_indices(parser_cls):
"""Three parallel tool calls must each emit with a unique index, a
unique id, a populated name, and arguments that round-trip through
``json.loads()`` to the expected dict.
This is the production opencode pattern; under upstream's
_extract_streaming it failed at high concurrency because one tool's
arg fragments would leak out under the next tool's index.
"""
parser = _make_parser(parser_cls)
chunks = _stream_chunks(parser, _PARALLEL_TRANSCRIPT, chunk_size=12)
by_index = _collect_tool_calls_by_index(chunks)
assert set(by_index.keys()) == {0, 1, 2}, (
f"expected exactly indices {{0, 1, 2}}, got {sorted(by_index.keys())}"
)
seen_ids: set[str] = set()
for idx in (0, 1, 2):
tc_id, tc_type, name, args_str = by_index[idx]
ctx = f"tool index {idx}"
assert isinstance(tc_id, str) and tc_id, (
f"{ctx}: expected non-empty string id, got {tc_id!r}"
)
assert tc_id not in seen_ids, (
f"{ctx}: duplicate id {tc_id!r} (also used by an earlier index)"
)
seen_ids.add(tc_id)
assert tc_type == "function", f"{ctx}: type={tc_type!r}"
assert name == "read_file", f"{ctx}: name={name!r}"
try:
parsed = json.loads(args_str)
except (json.JSONDecodeError, TypeError) as exc:
raise AssertionError(
f"{ctx}: arguments are not valid JSON: {args_str!r} ({exc})"
) from exc
assert parsed == _EXPECTED_PARALLEL_ARGS[idx], (
f"{ctx}: arguments {parsed!r} != expected "
f"{_EXPECTED_PARALLEL_ARGS[idx]!r}"
)
def test_mega_delta_with_all_tools_in_one_chunk(parser_cls):
"""Worst-case upstream regression: the WHOLE tool-call payload
arrives in one delta (continuous-batching scheduler under load).
Three closes, three opens, all in one call to
``extract_tool_calls_streaming``.
Upstream's single-Case dispatch fires Case 3 once and emits at most
ONE tool call's args under index 0; tools 1 and 2 are dropped or
mis-attributed. Path A must emit all three as a single
``DeltaMessage(tool_calls=[...])``.
"""
parser = _make_parser(parser_cls)
chunks = _stream_chunks(
parser,
_PARALLEL_TRANSCRIPT,
chunk_size=len(_PARALLEL_TRANSCRIPT), # the whole thing in one delta
)
by_index = _collect_tool_calls_by_index(chunks)
assert set(by_index.keys()) == {0, 1, 2}, (
f"expected exactly indices {{0, 1, 2}}, got {sorted(by_index.keys())}"
f"; this is the upstream-failure smoking gun"
)
for idx in (0, 1, 2):
tc_id, tc_type, name, args_str = by_index[idx]
assert isinstance(tc_id, str) and tc_id
assert tc_type == "function"
assert name == "read_file"
assert json.loads(args_str) == _EXPECTED_PARALLEL_ARGS[idx]
def test_partial_delta_split_inside_close_marker(parser_cls):
"""A delta boundary that lands INSIDE a ``<tool_call|>`` token must
not cause a tool call to be skipped or double-emitted.
The buffer in ``_buffer_delta_text`` is supposed to handle this;
this test pins that contract for the multi-tool case so an upstream
bump that changes buffer semantics is caught.
"""
parser = _make_parser(parser_cls)
# Pick a chunk size that's coprime with both the start and end tag
# lengths so we exercise multiple split points across the three
# tool calls.
chunks = _stream_chunks(parser, _PARALLEL_TRANSCRIPT, chunk_size=7)
by_index = _collect_tool_calls_by_index(chunks)
assert set(by_index.keys()) == {0, 1, 2}, (
f"chunk_size=7 split: expected indices {{0, 1, 2}}, got "
f"{sorted(by_index.keys())}"
)
for idx in (0, 1, 2):
_, _, _, args_str = by_index[idx]
assert json.loads(args_str) == _EXPECTED_PARALLEL_ARGS[idx]
if __name__ == "__main__":
sys.exit(pytest.main([__file__, "-q"]))
Your current environment
The output of
vllm docker: vllm-openai:v0.20.2-ubuntu2404python collect_env.pyis not relevant to the issue🐛 Describe the bug
I've observed two failure modes in the streaming gemma4 tool-parser, when using it from opencode:
Strict-client id / type / function.name re-emission (the original patch). The OpenAI streaming spec only requires id, type, and function.name to be set on the first DeltaToolCall chunk for a given tool-call index — subsequent chunks are deduped by index and may omit those fields. Stock vLLM 0.20.2's Gemma4ToolParser follows that letter-of-the-spec behaviour: _handle_tool_call_middle emits the trio on the first chunk, then _emit_argument_diff and _handle_tool_call_end emit follow-up chunks with only index + function.arguments set. But @ai-sdk's openai-compatible provider (used by opencode) Zod-validates every chunk against {id: z.string(), type: z.literal('function'), function: {name: z.string(), arguments: z.string()}} and bails out mid-stream the first time any of those is missing/null, surfacing as AI_InvalidResponseDataError: Expected 'id' to be a string (~64% of agents in run 11726821) and then, once id was patched, the sibling Expected 'function.name' to be a string (~42% of agents in run 11730617). The patch stashes the minted id per-tool-call (tool_call_ids: list[str]) and re-emits the trio on every follow-up DeltaToolCall for that index — a no-op for compliant clients (they dedupe), unblocking for strict ones.
Multi-boundary delta mis-attribution under load (the new Path A patch). Upstream's _extract_streaming is a single-Case dispatch (Case 1/2/3/4 — at most one branch fires per call to extract_tool_calls_streaming). Under high concurrency, vLLM's continuous-batching scheduler produces larger per-step text deltas, and a single delta routinely brings end-of-tool-N + start-of-tool-N+1 (or even multiple complete tool calls) in one shot. When that happens, Case 2's guard start_count > end_count is false (closes caught up), so current_tool_id is never advanced for the new tool; Case 3 then fires _handle_tool_call_end for the old tool only, and _extract_partial_call's current_text.rfind("<|tool_call>") always points at the latest tool — so the old tool's stripped trailing arg fragments (the }/" chars _emit_argument_diff deliberately withholds) leak out under the wrong index in the next delta. dp-vllm-toolstream-test reproduces this cleanly: 100% per-request success at c=10 / c=100 stream, collapsing to 35% at c=500 and 21% at c=1000, all classified as args_invalid_json with the captured bodies showing index 1's content being precisely the missing tail of index 0's args. Path A sidesteps the whole state machine by emitting each tool call as a single DeltaToolCall when its <tool_call|> close is observed, parsed via the same regex the (already-correct, 100%-at-every-N) non-streaming path uses — so the multi-boundary case is handled by construction.
The fix (Path A — atomic-per-tool-call streaming). Replace upstream's tag-counting state machine in _extract_streaming with an emitter that, on every call, computes new_ends = current_text.count("<tool_call|>") - previous_text.count("<tool_call|>"), runs the same tool_call_regex the non-streaming path uses on current_text, and emits one DeltaToolCall per newly-closed tool call (for i in range(prev_end_count, end_count)) — each carrying index=i, freshly-minted id, type="function", function.name, and the complete arguments JSON (parsed via _parse_gemma4_args then json.dumps'd in one shot, no incremental diff, no trailing-char stripping). All emissions for a given delta are packed into a single DeltaMessage(tool_calls=[...]), so multi-boundary deltas are handled by construction; the strict-client trio (id / type / function.name) is present on every emission by construction; and the prev_tool_call_arr / streamed_args_for_tool / tool_call_ids slots are kept consistent so chat_completion/serving.py:820+'s finish-reason coverage check still works. The tradeoff is that clients no longer see character-by-character arguments deltas — each tool call's full args arrives in one SSE chunk when its <tool_call|> close is observed — which is a no-op for our workload (opencode buffers parallel tool calls before dispatch anyway, and read_file args fit in a single TCP segment) and the price of eliminating an entire class of state-machine race conditions. The change is a ~150-line rewrite confined to vllm-patches/gemma4_tool_parser.py (no other files touched), guarded by three new pytest regressions (multi-tool / mega-delta / mid-close-marker chunk splits) that run as the existing Dockerfile build-time smoke check.
Script below is a conceptual repro: