f109e7f0e6* Studio: parse Mistral [TOOL_CALLS] and rehearsal tool-call shapes
Extends the rescue parsers in core/tool_healing.py and
core/inference/tool_call_parser.py to recognise two extra serialisations
local models commonly emit when bypassing native function calling:
* [TOOL_CALLS]name{json_args} (Devstral-Small-2, Mistral-Small-3.x).
* name[ARGS]{json_args} (reasoning-model rehearsal).
Both extractors use a brace-balance scan that honours escapes and
quoted strings so nested JSON args stay intact.
Also pre-strips <think>...</think> and [THINK]...[/THINK] blocks before
matching so calls emitted after a reasoning preamble are recognised
regardless of position.
Streaming gates (TOOL_XML_SIGNALS, llama_cpp.py _TOOL_XML_SIGNALS) and
the SSE strip regex (routes/inference.py _TOOL_XML_RE) gain the new
sentinels so the parser is actually invoked and the raw markup never
leaks to the UI.
* [pre-commit.ci] auto fixes from pre-commit.com hooks
for more information, see https://pre-commit.ci
* Strip unclosed think blocks and catch rehearsal [ARGS] mid-buffer
The pre-existing ``_THINK_TAG_RE`` only matched closed thinking
blocks (``<think>...</think>`` or ``[THINK]...[/THINK]``). During
streaming the model is still inside the open block when the parser
runs, so any tool-shaped markup the model is REHEARSING inside that
block survived the strip and could be executed as a real call.
Switch both copies of the regex (parser + healing) to accept the
trailing block being terminated by end-of-string in addition to
the explicit closer.
The ``_TOOL_XML_SIGNALS`` list on the llama_cpp streaming buffer
included ``[ARGS]`` to catch rehearsal syntax, but the gate used a
``startswith`` check against the buffer head -- rehearsal is shaped
``name[ARGS]{json}``, so the buffer never STARTS with ``[ARGS]``
and the signal had no effect. Add a substring fallback for the
bracket-style signals so the BUFFERING window can still divert the
stream into DRAINING when rehearsal markup arrives mid-buffer.
Adds three regression tests covering rehearsal inside unclosed
``<think>`` / ``[THINK]`` blocks (must yield no calls) and the
positive case after a closed think block (still parsed).
* [pre-commit.ci] auto fixes from pre-commit.com hooks
for more information, see https://pre-commit.ci
* [pre-commit.ci] auto fixes from pre-commit.com hooks
for more information, see https://pre-commit.ci
* Studio: harden bracket-tag tool-call parsing and streaming strip
Address review findings on the Mistral [TOOL_CALLS] / rehearsal [ARGS] paths:
- Accept hyphenated tool names in the bracket parsers and strip patterns.
_MISTRAL_BRACKET_RE and _REHEARSAL_RE used \w+, which dropped or truncated
MCP function names containing dashes (mcp__srv__list-issues). Use [\w-]+ to
match the XML and Gemma parsers.
- Strip a partial bracket marker streamed before its opening brace. The
trailing-unclosed patterns required the {, so a [TOOL_CALLS]web_search or
python[ARGS] split across deltas leaked the raw marker to the UI. Match the
bare marker to end-of-text, mirroring how the bare open tags are stripped.
Closed pairs are unchanged so in-progress markup stays buffered until parsed.
- Strip a truncated bracket tail in the route-level display regex. _TOOL_XML_RE
required a balanced JSON object; a tool call truncated by EOS now strips up
to \Z, like the orphan-opening XML shapes. Complete calls still strip only
their balanced JSON so following prose survives.
Add regression tests for hyphenated names, the streaming partial-marker strip,
and the unclosed-tail route strip.
* Studio: preserve XML parameter indentation in tool_healing
The chat template emits <parameter=k>\nVALUE\n</parameter>; the parameter-start
regex consumed the wrapping newline AND the value's first-line indentation via a
trailing \s*, then str.strip() removed the rest, corrupting code/diff arguments.
Narrow the trailing class to horizontal whitespace and trim exactly one wrapping
newline (_trim_param_value), preserving indentation. Matches SGLang's qwen3_coder
detector and the same fix on the multi-format parser. Add a regression test.
* Studio: tighten Mistral/rehearsal tool-call comments
Compress the comments in the Mistral [TOOL_CALLS] / rehearsal [ARGS] healing shim
and its callers to one or two lines, keeping the bracket-tag stripping rationale,
the thinking-block handling note, and the forge attribution intact.
Comment-only: no code or behavior change (verified with comment_tools.py check
--strip-docstrings; tests green).
* Studio: fix think-strip arg corruption and nested bracket-JSON strip
Review follow-up for the Mistral/rehearsal healing shim:
- The <think>/[THINK] strip ran unconditionally over the whole content before
parsing, so a real tool argument that legitimately contained a <think> /
[THINK] literal was silently corrupted. Don't delete the blocks: compute the
reasoning-block spans and skip any tool-call candidate that STARTS inside one,
across all parse paths (JSON, Gemma, XML, bracket, rehearsal). A rehearsed call
inside reasoning is still ignored; a real call after </think> still parses.
- The bracket-tag display strip used a fixed one-level-nesting regex, so a call
with two-level-nested JSON args either leaked raw markup or, in final mode, let
the catch-all eat the trailing prose. Add a balanced-brace
_strip_bracket_tag_calls pass (any nesting depth) used by strip_tool_call_markup
and the route display strip.
Add regressions: <think>/[THINK] literal inside a real argument, rehearsal-inside-
think with a real call after, and two-level-nested bracket/rehearsal strip keeping
trailing prose.
* Studio: correct think-block comments to match span-skip behavior
The think-strip fix replaced the unconditional think-block strip with a
span-skip (the block is kept and any tool-call candidate starting inside it is
ignored), but two comments still described the old strip-first behavior. Update
the _THINK_TAG_RE comment and the parse_tool_calls_from_text docstring.
* Studio: parse Mistral arrays and call-ids, unify bracket parse/strip, keep it linear
- Parse the canonical Mistral array form (TOOL_CALLS followed by a JSON list of
calls) and emit every call; parse the v11 shape that carries an opaque CALL_ID
token between the name and ARGS (the function name is the token after
TOOL_CALLS, never the call-id); and parse a Mistral call plus a rehearsal call
in one message (the second was dropped yet still stripped from display).
- One shared balanced forward scan (_iter_bracket_spans) backs both the parser
and the strip path, so they no longer diverge. It is linear: each regex is
re-searched only once its cached match falls behind the cursor, replacing the
per-match full-tail re-scan that was O(n^2) (O(n^3) over a stream). A length cap
before the scan is a backstop.
- strip_tool_call_markup preserves think/reasoning blocks verbatim (the parser
skips tool markup inside them), stripping only the visible text around them.
- _in_think uses bisect over the sorted think spans (was a linear scan per
candidate).
- GGUF streaming strip runs the balanced bracket pre-pass before the regex
patterns so nested-arg calls do not leak or eat trailing prose, and the
BUFFERING ARGS detector requires the rehearsal name-ARGS shape.
- Tests: canonical array, array string-args, array strip keeps prose, Mistral
plus rehearsal multi-call, v11 call-id name, think-rehearsal strip
preservation, and bracket-strip linearity.
* Studio: preserve reasoning blocks in the route and streaming strip paths too
Addresses Gemini/Codex review: making strip_tool_call_markup preserve think
blocks left the route display strip and the GGUF streaming strip inconsistent,
so a rehearsed call inside a reasoning block was still deleted from the visible
text on those paths.
- Extract the think-block segmentation into one shared helper (strip_outside_think)
and route all three strip paths through it: strip_tool_call_markup,
_strip_tool_xml_for_display, and the GGUF _strip_tool_markup_streaming closure.
- Add a route-strip regression test that a rehearsal inside a reasoning block is
preserved while a real call outside it is still stripped.
* Studio: fix bracket-tag strip/buffer review findings
Address the live code-review findings on the Mistral bracket-tag / rehearsal
tool-call rescue path:
- tool_healing: a literal think block inside a tool-call argument is no longer
treated as a reasoning block. strip_outside_think now excludes think spans
that sit inside a complete tool-call span, so the call is stripped whole
instead of the split hiding its open/close pair and leaking the raw call.
- tool_healing: the rehearsal trailing-strip pattern requires a following brace
or end-of-text, so prose that merely mentions name[ARGS] is not truncated as
a phantom call. The bracket strip patterns are aligned with the parser
regexes (whitespace, v11 [CALL_ID]/[ARGS] metadata, and the [CALL_ID]
lookbehind).
- routes: strip a truncated canonical Mistral array ([TOOL_CALLS] [{... with no
closing bracket) that the balanced scan cannot remove, align the display
regex with the parser regexes, and apply the same rehearsal-prose guard.
- safetensors loop: mirror the GGUF [ARGS] rehearsal-substring check during
BUFFERING so a rehearsal name does not stream before its [ARGS] arrives.
Adds regression tests for each; existing parser suite stays green.
* Studio: hold split rehearsal tool-name prefix in both streaming loops
A reasoning-model rehearsal call can stream the tool name and its [ARGS] arm in
separate chunks (web_search then [ARGS]{...}). The buffering detector only
recognised the rehearsal once [ARGS] was present, so the bare tool name was
emitted as visible content before the call drained and executed.
Add _is_rehearsal_prefix (mirrored in the safetensors loop and the GGUF loop):
when a no-signal buffer is a bare active-tool name -- or a partial prefix of
NAME[ARGS] -- hold it as a prefix instead of streaming it, so the next chunk's
[ARGS] flips it to a drain. A whitespace in the buffer means prose, not a split
call, so ordinary text still streams.
Adds regression tests for the split rehearsal in both loops and a guard that a
plain non-tool word still streams.
* Studio: route Anthropic tool-call cleanup through the protected display strip
The Anthropic stream, non-stream, and passthrough paths cleaned content with raw
_TOOL_XML_RE.sub instead of _strip_tool_xml_for_display, so a rehearsal call
inside <think> was deleted from the reasoning and a nested [TOOL_CALLS] call
dropped its trailing prose (the OpenAI-compatible paths already use the helper).
Route all four sites (prior-assistant cleanup, streaming content events,
non-stream aggregation, passthrough conversion) through the protected helper, and
add a source-level guard test so raw _TOOL_XML_RE.sub stays confined to the
helper itself.
* Studio: stop split rehearsal tool names leaking once streaming, uncapped, or unrestricted
The split-rehearsal guard (NAME in one chunk, [ARGS]{...} in the next) only held
the name in the initial BUFFERING state. Three gaps remained where the bare tool
name still streamed as visible content before the call drained:
- STREAMING: after prose had already streamed, both loops emitted a trailing
active-tool-name token (and the GGUF/safetensors [ARGS] boundary was not pulled
back over the name). Hold the trailing rehearsal token and release it on the
next chunk, with an end-of-stream flush so a plain answer that merely ends on a
tool-name word is never dropped.
- Buffer cap: a realistic MCP name longer than the 32-char _MAX_BUFFER_CHARS cap
defeated the BUFFERING hold. A rehearsal prefix is self-bounding (it stops
matching once it grows past NAME[ARGS]), so the generic cap no longer applies to
it.
- Unrestricted mode (tools=[]): with no declared tool list, any bare identifier
may be a NAME[ARGS] rehearsal, so the prefix check now recognises one instead of
leaking the name and mis-parsing the call.
Regression tests cover the streaming, long-name, and unrestricted cases plus the
plain-prose paths that must not be held or corrupted.
* [pre-commit.ci] auto fixes from pre-commit.com hooks
for more information, see https://pre-commit.ci
* Studio tools: protect think blocks in safetensors streaming, hold split rehearsal on initial flush, advertise Mistral tools
Pass-3 review follow-ups on the Mistral [TOOL_CALLS] / rehearsal [ARGS] work:
- Safetensors streaming display strip now preserves think / [THINK] reasoning
verbatim (routes through strip_outside_think like the GGUF path). A call
rehearsed inside a reasoning block was stripped mid-stream and then restored by
the final strip, a non-monotonic shrink/grow that corrupted append-by-length
stream consumers and the visible reasoning.
- The first flush out of BUFFERING (safetensors and GGUF) now applies the same
trailing-name hold the STREAMING branch uses, so a split rehearsal (prose plus a
trailing active tool name in one chunk, [ARGS]{...} in the next) no longer leaks
the bare name before the call drains.
- Safetensors capability gate no longer suppresses tools for Mistral [TOOL_CALLS]
templates, which the shared bracket-tag parser now handles end to end. Llama
python_tag stays suppressed (still unparseable).
- Route display strip applies the open-ended / bare-marker tail arms only on the
segment after the last reasoning block (closed-only regex before it), matching
strip_tool_call_markup, so a bare foo[ARGS] before a reasoning block is preserved
while complete calls are still removed in every segment.
Adds regression tests for each and updates the now-stale Mistral capability test.
* [pre-commit.ci] auto fixes from pre-commit.com hooks
for more information, see https://pre-commit.ci
* Fix tool-call think-marker and bracket-wrapper edge cases
Round-1 review follow-ups on the Mistral/rehearsal tool-call healing:
- tool_healing: a reasoning marker that opens INSIDE a tool call's
arguments is argument data, not a reasoning block. Add
_think_spans_outside_tool_markup (start-inside test) and use it in
both parse_tool_calls_from_text and strip_outside_think so a literal
marker in one call's args no longer hides a later call (parse) or
leaks the raw markup (strip) when the greedy match runs past the
call's closer.
- tool_healing: strip the orphan Mistral v11 [/TOOL_CALLS] closer left
behind after the balanced scan removes the call body. Add a route arm
for the same closer in _TOOL_XML_RE / _TOOL_XML_CLOSED_RE.
- safetensors + llama_cpp streaming strip: run the open-ended (EOS
anchored) tail patterns only on the last segment; segments before a
reasoning block use the closed-only patterns, matching the final
strip and the route strip. A bare foo[ARGS] before a reasoning block
is prose, not a truncated call.
- safetensors streaming detector: validate each [ARGS] hit before
draining. A bare foo[ARGS] in prose (no active tool name in front)
no longer drains the rest of the turn; a later real NAME[ARGS] call
is still found and the prose in between is preserved.
Regression tests added for each case across the parser, strip helpers,
and both streaming loops.
* Strip incomplete-XML tool markup with literal think tags; widen render-html detector
Round-2 review follow-ups.
- tool_healing: an UNCLOSED <tool_call> / <function= call that the parser still
executes via allow_incomplete leaked its markup when an argument contained a
literal think marker. _tool_call_markup_spans only covered closed calls, so the
literal was treated as a reasoning block to preserve. Extend it to the
open-ended XML tail forms (shared as _TOOL_OPEN_XML_TAIL_PATS) so a think marker
inside an unclosed call is argument data and the call's markup is stripped. A
complete call's opener stays bounded to its closed span, and a real reasoning
block with no tool call is still preserved.
- safetensors render-html provisional card: _detect_render_html_tool_start was
XML-only, so a Mistral [TOOL_CALLS]render_html or rehearsal render_html[ARGS]
call executed but skipped the early card. Detect the earliest tool-call marker
across every serialization the loop executes and fire when it is render_html.
Regression tests added for both.
* [pre-commit.ci] auto fixes from pre-commit.com hooks
for more information, see https://pre-commit.ci
* Studio tools: gate [ARGS] on active tools and skip think-block render_html rehearsal
Round 3 review fixes for the Mistral / rehearsal tool-call parsing path. Both are
asymmetric-fix bugs where one code path applied a guard the analogous paths did not.
- [ARGS] active-tool gating: the streaming state already validates a rehearsal
NAME[ARGS] against the active tool list before draining, but the BUFFERING
detection and the end-of-stream safety-net checks (safetensors and GGUF) treated
any word[ARGS] substring as a tool boundary. An answer containing a literal
foo[ARGS]{...} in prose, where foo is not an enabled tool, was drained, parsed into
a disabled foo no-op, and forced an extra generation turn. Gate those checks on the
active tool name too (unrestricted mode still accepts any name), so inactive-name
prose is neither drained nor parsed. Adds a shared _has_genuine_tool_signal helper
(safetensors) and _gguf_rehearsal_signal_pos / _gguf_has_genuine_tool_signal (GGUF).
- render_html provisional card vs think blocks: the parser skips tool candidates that
start inside a <think>/[THINK] reasoning block, but the provisional render_html
detector scanned raw content. A render_html rehearsed inside <think> followed by a
real non-render_html call emitted a provisional render_html tool_start (reusing the
later call's id) that the loop never executed. Drop candidates that start inside a
think span and use the first marker of each shape outside the blocks. Also resolve
the [TOOL_CALLS] [{...}] array shape through the parser so a nested "name" argument
key no longer fires a false provisional card ahead of the real top-level tool name.
Adds regression tests for both loops: inactive-name foo[ARGS]{...} is not drained into
a disabled no-op or a retry turn, a think-block render_html rehearsal emits no
provisional card, and the array top-level name is read correctly.
* [pre-commit.ci] auto fixes from pre-commit.com hooks
for more information, see https://pre-commit.ci
* Gate ambiguous bare-rehearsal parse and strip on the active tool list
A bare NAME[ARGS]{json} is a genuine rehearsal call only when NAME is an
active tool; otherwise it is prose. The earlier round gated only detection
(so an inactive foo[ARGS] no longer drained the buffer or forced a retry
turn), but the parse and strip stayed unrestricted, which produced two
regressions:
1. An inactive foo[ARGS]{...} placed immediately before a real
web_search[ARGS]{...} in the same content span made the real call fail
to execute (parse consumed the phantom foo call).
2. An inactive foo[ARGS]{...} in a prose answer had its markup stripped
from the visible text, corrupting the sentence to " is just syntax."
Thread enabled_tool_names through the shared parser/strip so parse and
strip apply the SAME active-tool gate as detection:
- core/tool_healing.py: _iter_bracket_spans skips an inactive rehearsal
span; parse_tool_calls_from_text, _strip_bracket_tag_calls,
_strip_markup_segment and strip_tool_call_markup accept and thread the
gate; apply_tool_strip_patterns keeps an inactive rehearsal match.
- core/inference/tool_call_parser.py: wrappers forward the gate.
- core/inference/safetensors_agentic.py and core/inference/llama_cpp.py:
compute the gate from the active tool list (None when unrestricted, to
keep the legacy strip-all behavior) and thread it into every parse and
streaming/final strip site.
- routes/inference.py: _strip_tool_xml_for_display accepts the gate and
keeps an inactive rehearsal via a capture group on its rehearsal arm, so
the display cleanup does not re-strip the already-correct loop output.
The [TOOL_CALLS] control-token arms still strip unconditionally. Wire
the current turn's active tool names into the GGUF and safetensors
content-display sites.
Tests: parse and strip gate coverage in test_tool_call_parser_strict.py,
test_tool_xml_strip.py and test_safetensors_tool_loop.py; end-to-end GGUF
coverage for the real-call-after-inactive-rehearsal case and a
strengthened assertion that the inactive rehearsal prose survives intact.
* [pre-commit.ci] auto fixes from pre-commit.com hooks
for more information, see https://pre-commit.ci
* Studio: render the reasoning block for safetensors and MLX like GGUF
enable_thinking chat templates (Qwen3/Qwen3.5/GLM) prefill an unclosed <think>
into the generation prompt, so the model emits only the closing </think> then
the answer. The safetensors/MLX chat stream emitted that as plain content, so
the reasoning showed inline with no collapsible thinking block, while GGUF
(which surfaces reasoning via reasoning_content) rendered one. This brings
safetensors and MLX to parity.
- _ResponsesReasoningExtractor gains a reasoning_prefilled mode that starts
inside the reasoning block and splits on the first </think>; default False
keeps GGUF and every existing caller byte-identical. It suppresses a stray
re-emitted <think> and holds partial markers back across chunk boundaries.
- _sf_reasoning_prefill_mode gates the mode on reasoning being enabled for the
request, an enable_thinking or enable_thinking_effort style, and the template
actually using the standard <think>/</think> markers. Models with a bespoke
reasoning channel (e.g. gemma's <|think|>/<|channel>) are excluded so their
answer is never swallowed; gpt-oss (Harmony) and thinking-off requests are
excluded too.
- sf_tool_stream and stream_chunks (the latter also serves MLX) feed text
through the extractor, emitting reasoning_content then content deltas, with a
per-turn reset in the tool loop and a flush before each tool_start; only the
visible delta reaches the monitor reply. The two non-streaming drains split
reasoning_content the same way.
- Tests: extractor prefilled mode (streaming and edge cases), the gate matrix
including the gemma-style exclusion, and a route-replay of the tool-loop
reasoning stream.
* [pre-commit.ci] auto fixes from pre-commit.com hooks
for more information, see https://pre-commit.ci
* studio: skip tool calls rehearsed in prefilled reasoning
Reasoning models (Qwen3.5 enable_thinking) open <think> in the prompt, so the
generated text starts inside the thought and emits only a closing </think> with
no opener. _think_spans_outside_tool_markup only found spans with an explicit
opener, so a NAME[ARGS]{...} or [TOOL_CALLS] call rehearsed in that leading
thought was parsed and executed as a real call.
Add a leading think span (offset 0 through the first close marker) when the
content opens with a bare close, so the rehearsed call is skipped and the
reasoning is preserved by strip_outside_think. Guarded by the existing call-span
check: a literal </think> inside a real call's arguments does not trigger the
span, so a genuine leading call still fires. Tests for both cases.
* [pre-commit.ci] auto fixes from pre-commit.com hooks
for more information, see https://pre-commit.ci
* studio: do not start prefilled reasoning mode when reasoning_effort is none
enable_thinking_effort models (e.g. GLM-5.2) express thinking-off via
reasoning_effort="none" rather than enable_thinking=False, but
_sf_reasoning_prefill_mode only looked at enable_thinking, so such a request
started the extractor in prefilled mode. With thinking off the model never emits
</think>, so the whole answer was captured as reasoning_content and the visible
content/stream came back empty. Thread reasoning_effort through and return False
when it is "none". Tests for none vs a real effort level.
* [pre-commit.ci] auto fixes from pre-commit.com hooks
for more information, see https://pre-commit.ci
* studio: only treat a leading bare </think> as prefilled reasoning when a real call follows
The prefilled-reasoning virtual span fired on any unmatched leading close marker,
so a non-prefilled turn that emits a real call before a stray </think> (for
example "Now web_search[ARGS]{...}</think> answer") had the call swallowed by the
span and dropped. Require that a real tool call also appear after the close (the
actual turn that follows the thought) before adding the span, so a stray close in
a normal answer no longer suppresses a genuine leading call. The rehearse-then-
call case still skips the rehearsal. Test for the stray-close case.
* Studio: trim redundant comments (comment-only, AST-verified)
* studio: keep tool_healing importable on Python 3.9
_balanced_json_span was annotated -> int | None. With no
from __future__ import annotations, that PEP 604 union is evaluated at
import time, so on Python 3.9 (which the package still supports,
requires-python >=3.9, and where external inference servers import this
module standalone) the def raises TypeError and the whole module fails
to import before any parsing runs.
Add from __future__ import annotations so annotations stay lazy strings,
matching the prevailing convention across studio/backend. No behavior
change: the module has no runtime annotation introspection.
* Studio: gate the Anthropic tool-stream display strip on declared tools
The Anthropic streaming and non-streaming tool paths called
_strip_tool_xml_for_display without enabled_tool_names, so with the default
strip-all behavior a final answer that literally contains an inactive-name
NAME[ARGS]{json} (prose, not a call) lost those bytes in the delivered text.
The GGUF and safetensors paths already pass _display_tool_name_gate(tools);
these two sites were missed when that gate was threaded through.
Compute the gate from the declared tools and pass it at both sites (threading
openai_tools into _anthropic_tool_non_streaming and its caller), so an
inactive-name rehearsal survives while an active-name one is still stripped.
Add a regression test.
* Studio: hold a split unrestricted rehearsal prefix at the bracket
In unrestricted tool mode (tools=[]) the rehearsal-prefix regex required
[A after the bracket, so a chunk boundary landing right after NAME[ (e.g.
web_search[ then ARGS]{...}) failed the prefix check and streamed the
partial tool markup web_search[ to the client before the call drained.
Restricted mode already holds this via a startswith check. Make the bracket
and each ARGS letter individually optional so NAME[ is held too, matching
the documented intent. Add a regression test.
* Studio: gate rehearsal detection and history strip on the original tool set
Two display/loop gate fixes so a spent one-shot tool is handled consistently:
- Rehearsal DETECTION (safetensors and GGUF loops) now uses the ORIGINAL tool
list, matching the strip gate, instead of the post-removal active_tools. After a
one-shot tool (render_html) runs it is dropped from active_tools; a repeat
render_html[ARGS]{...} while another tool is still active was stripped from
display yet never detected, so it was not routed to the render_html_repeat no-op
and the turn ended as a blank continuation. Detection now fires for it.
- The GGUF assistant-history sanitiser forwards the enabled-tool-name gate (like
the live-response strip), so a prior turn documenting an inactive foo[ARGS]{...}
shape is preserved in the replayed prompt context instead of being deleted.
Add regression tests for both loops and the history strip.
* Studio: thread the tool-name gate through the remaining rehearsal/history sites
Follow-up to the rehearsal-detection and history-strip gate fixes, covering the
sibling sites that were missed:
- GGUF loop: the rehearsal-prefix and trailing-name hold checks now use the
original tool list (_detect_tools) like the detection path, so a spent one-shot's
split repeat (bare render_html then [ARGS]{...}) is held instead of flushed as
visible text.
- The safetensors and Anthropic assistant-history sanitisers and the Anthropic
non-streaming passthrough now forward the enabled-tool-name gate to
_strip_tool_xml_for_display, matching the GGUF history sanitiser and the live
strips, so a prior turn documenting an inactive foo[ARGS]{...} example is
preserved in the replayed prompt / final text instead of deleted.
Add regression tests.
* [pre-commit.ci] auto fixes from pre-commit.com hooks
for more information, see https://pre-commit.ci
* Tile bracket-call spans per array item and include the v11 closer
Two with_spans fixes for the Mistral bracket parser, both hit through the
client-tool passthrough healers:
- A multi-call [TOOL_CALLS] array carried its whole markup span on the first
call and zero-width spans after, so a consumer that filters promotions by
the declared tool set either re-emitted the full raw array as text next to
the promoted call or silently dropped a filtered call's bytes. The region is
now tiled across the call-producing items (each call's span covers its own
JSON object plus the separator bytes before it; the last span runs to the
region end), so promoted markup strips exactly once and a skipped call's
bytes stay visible.
- The v11 wrapper closer [/TOOL_CALLS] sat outside the reported span and
leaked as stray text after promotion; the region now extends over an
immediately-following closer.
* [pre-commit.ci] auto fixes from pre-commit.com hooks
for more information, see https://pre-commit.ci
* Address review: decouple healer signals from the loop signal set
The passthrough healer buffered on every TOOL_XML_SIGNALS entry, so the bare
[ARGS] rehearsal marker this branch adds for the loops (where it is gated on
active tool names) put legitimate prose like 'Use foo[ARGS] in templates'
into the holding state and stalled the stream until finalization. The healer
can never promote a bare rehearsal call, so it now buffers only on formats
its parser promotes: <tool_call>, <|tool_call>, <function=, [TOOL_CALLS].
* [pre-commit.ci] auto fixes from pre-commit.com hooks
for more information, see https://pre-commit.ci
* Condense comments in the Mistral tool-call rescue to contract essentials
* verify_import_hoist: exempt __future__ imports and same-diff relocations
Two false positives fired on this PR's refactor. A from __future__ import
is a compiler directive whose name never appears as a runtime load, so
HOISTED-IMPORT-UNUSED can never see it used, yet the file requires it for
PEP 604 annotations on Python 3.9. TARGET-CHANGED flagged the deliberate
move of the strip-pattern constants into core.inference.tool_call_parser
as a silent re-point even though the old module-level target was removed
and the new one added in the same diff. Both get narrow exemptions; a
re-point to a pre-existing target is still caught, and the self-test
negative controls all pass unchanged.
* Drain the whole Mistral [TOOL_CALLS] array in streaming passthrough healing
StreamToolCallHealer._drain promoted only the first parsed call per pass and
dropped the rest of the buffer past that one span. For a well-formed Mistral
parallel-tool-call array streamed through client-tool passthrough
([TOOL_CALLS][{...},{...}]), the per-item spans are contiguous, so after the
first call was promoted the residue began with ,{...}] (no leading signal) and
was flushed as raw text: every call after the first was lost.
_drain now walks the contiguous run of parsed calls (adjacent tiled spans =
one array), promoting each declared call and relaying undeclared ones as data,
and stops at the first gap (prose) or incomplete trailing block so separate
blocks still stream incrementally in document order. This mirrors the
non-streaming heal_openai_message / finalize promote-or-flush loop and the
server-side safetensors loop, which already handled multi-call arrays.
Added regression tests: 2-call array in one feed and char-by-char, an
undeclared middle call kept as text, and an array followed by trailing prose.
* Drain comma-less Mistral tool-call arrays and normalize null arguments
The array branch fed the whole body to a single json.loads, which rejects the
comma-less multi-call form the repo's own Mistral/Ollama templates render (the
range loop in ollama_template_mappers.py emits the objects with no separator) and
so dropped every call. Decode elements individually with the existing
comma-tolerant raw_decode helper, now _decode_array_items, which also returns the
objects, so all calls are recovered while the span tiling is unchanged.
Also normalize a non-object array argument such as arguments null to an empty
object, matching the wrapped tool_call path, instead of serializing None to the
string "null" that auto-heal would turn into a bogus query of "null".
* Gate safetensors reasoning prefill on the rendered generation prompt
reasoning_always_on fires on any paired <think></think> in the template,
including markup that only renders PAST assistant history (Kimi-K2-Thinking)
while the generation prompt opens no <think>. Starting the reasoning extractor
in prefilled mode there captured a normal answer entirely as reasoning_content
and returned blank visible content. Prefill only when rendering the generation
prompt actually leaves <think> open (DeepSeek-R1 / QwQ / Qwen3-Thinking);
history-only templates start the extractor in normal mode and parse the model's
own <think>...</think>. Adds a Kimi-shape regression test.
* Keep bare scalar Mistral array arguments raw instead of double-encoding
A scalar string argument in the canonical Mistral [TOOL_CALLS] array
(for example [TOOL_CALLS][{"name":"web_search","arguments":"weather"}])
was run through json.dumps, turning weather into the JSON string
"weather". The downstream argument healer then wrapped that quoted
form, so a single-string tool like web_search searched for the literal
"weather" with quotes. The <tool_call> path already keeps a scalar
argument raw; mirror it here so only a dict is serialized. Add a
regression test asserting both paths yield the same healed arguments.
* Tighten tool-call rescue and reasoning-prefill comments
* [pre-commit.ci] auto fixes from pre-commit.com hooks
for more information, see https://pre-commit.ci
* [pre-commit.ci] auto fixes from pre-commit.com hooks
for more information, see https://pre-commit.ci
---------
Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
2624 lines
94 KiB
Python
2624 lines
94 KiB
Python
# SPDX-License-Identifier: AGPL-3.0-only
|
|
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved. See /studio/LICENSE.AGPL-3.0
|
|
|
|
"""Focused tests for the GGUF llama.cpp agentic tool loop.
|
|
|
|
These tests drive ``LlamaCppBackend.generate_chat_completion_with_tools``
|
|
with fake llama-server SSE streams. They require no model, subprocess, GPU,
|
|
or network access.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
import copy
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
_BACKEND_DIR = str(Path(__file__).resolve().parent.parent)
|
|
if _BACKEND_DIR not in sys.path:
|
|
sys.path.insert(0, _BACKEND_DIR)
|
|
|
|
from core.inference.llama_cpp import (
|
|
_MAX_REPROMPTS,
|
|
_PROVISIONAL_ARGS_MIN_CHARS,
|
|
LlamaCppBackend,
|
|
)
|
|
from state import tool_approvals
|
|
from state.tool_approvals import TOOL_REJECTED_MESSAGE, resolve_tool_decision
|
|
|
|
|
|
def _sse(delta: dict) -> str:
|
|
return "data: " + json.dumps({"choices": [{"index": 0, "delta": delta}]}) + "\n"
|
|
|
|
|
|
def _done() -> str:
|
|
return "data: [DONE]\n"
|
|
|
|
|
|
def _make_backend(monkeypatch, streams: list[list[str]], payloads: list[dict]):
|
|
backend = LlamaCppBackend.__new__(LlamaCppBackend)
|
|
backend._process = object()
|
|
backend._healthy = True
|
|
backend._port = 48847
|
|
backend._api_key = None
|
|
backend._effective_context_length = 4096
|
|
backend._supports_reasoning = False
|
|
backend._reasoning_always_on = False
|
|
backend._reasoning_style = "enable_thinking"
|
|
backend._supports_preserve_thinking = False
|
|
|
|
@contextlib.contextmanager
|
|
def fake_stream_with_retry(
|
|
_client,
|
|
_url,
|
|
payload,
|
|
_cancel_event,
|
|
headers = None,
|
|
first_token_deadline = None,
|
|
):
|
|
payloads.append(copy.deepcopy(payload))
|
|
yield type("FakeResponse", (), {"status_code": 200, "chunks": streams.pop(0)})()
|
|
|
|
def fake_iter_text_cancellable(
|
|
response,
|
|
_cancel_event,
|
|
first_token_deadline = None,
|
|
):
|
|
yield from response.chunks
|
|
|
|
monkeypatch.setattr(backend, "_stream_with_retry", fake_stream_with_retry)
|
|
monkeypatch.setattr(backend, "_iter_text_cancellable", fake_iter_text_cancellable)
|
|
return backend
|
|
|
|
|
|
def _tool_names(payload: dict) -> list[str]:
|
|
return [
|
|
(tool.get("function") or {}).get("name")
|
|
for tool in payload.get("tools", [])
|
|
if (tool.get("function") or {}).get("name")
|
|
]
|
|
|
|
|
|
def _patch_monotonic(monkeypatch, values: list[float]) -> None:
|
|
import core.inference.llama_cpp as llama_cpp_mod
|
|
|
|
it = iter(values)
|
|
last = values[-1]
|
|
|
|
def fake_monotonic() -> float:
|
|
nonlocal last
|
|
try:
|
|
last = next(it)
|
|
except StopIteration:
|
|
pass
|
|
return last
|
|
|
|
monkeypatch.setattr(llama_cpp_mod.time, "monotonic", fake_monotonic)
|
|
|
|
|
|
def _structured_tool_call(tool_name: str, arguments: dict, call_id: str) -> list[str]:
|
|
return [
|
|
_sse(
|
|
{
|
|
"tool_calls": [
|
|
{
|
|
"index": 0,
|
|
"id": call_id,
|
|
"type": "function",
|
|
"function": {
|
|
"name": tool_name,
|
|
"arguments": json.dumps(arguments),
|
|
},
|
|
}
|
|
]
|
|
}
|
|
),
|
|
_done(),
|
|
]
|
|
|
|
|
|
def test_structured_tool_call_after_visible_preface_is_executed(monkeypatch):
|
|
"""llama-server may emit content first and then native delta.tool_calls.
|
|
|
|
Studio must not drop that tool call after it has streamed the preface.
|
|
"""
|
|
|
|
tool_call_id = "call_render_late"
|
|
first_stream = [
|
|
_sse({"content": "Here is the canvas.\n\n"}),
|
|
_sse(
|
|
{
|
|
"tool_calls": [
|
|
{
|
|
"index": 0,
|
|
"id": tool_call_id,
|
|
"type": "function",
|
|
"function": {
|
|
"name": "render_html",
|
|
"arguments": json.dumps(
|
|
{
|
|
"code": "<html><body><div>red</div></body></html>",
|
|
"title": "Simple Red Square",
|
|
}
|
|
),
|
|
},
|
|
}
|
|
]
|
|
}
|
|
),
|
|
_done(),
|
|
]
|
|
second_stream = [
|
|
_sse({"content": "Done."}),
|
|
_done(),
|
|
]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [first_stream, second_stream], payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
calls.append((name, arguments))
|
|
return "Rendered HTML canvas: Simple Red Square."
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
|
|
tools = [
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "render_html",
|
|
"description": "Render HTML.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {"code": {"type": "string"}},
|
|
"required": ["code"],
|
|
},
|
|
},
|
|
}
|
|
]
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "Make a red square."}],
|
|
tools = tools,
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
content_events = [e for e in events if e.get("type") == "content"]
|
|
assert content_events[0]["text"] == "Here is the canvas.\n\n"
|
|
|
|
first_content_index = next(
|
|
i for i, event in enumerate(events) if event.get("type") == "content"
|
|
)
|
|
actual_tool_start_index = next(
|
|
i
|
|
for i, event in enumerate(events)
|
|
if event.get("type") == "tool_start" and event.get("arguments", {}).get("code")
|
|
)
|
|
assert first_content_index < actual_tool_start_index
|
|
|
|
assert calls == [
|
|
(
|
|
"render_html",
|
|
{
|
|
"code": "<html><body><div>red</div></body></html>",
|
|
"title": "Simple Red Square",
|
|
},
|
|
)
|
|
]
|
|
assert any(e.get("type") == "tool_end" and e.get("tool_name") == "render_html" for e in events)
|
|
|
|
# The second llama-server request should include the assistant preface
|
|
# plus the structured tool call, preserving OpenAI-compatible ordering.
|
|
assert len(payloads) == 2
|
|
assistant_messages = [m for m in payloads[1]["messages"] if m.get("role") == "assistant"]
|
|
assert assistant_messages[-1]["content"] == "Here is the canvas.\n\n"
|
|
assert assistant_messages[-1]["tool_calls"][0]["id"] == tool_call_id
|
|
assert assistant_messages[-1]["tool_calls"][0]["function"]["name"] == "render_html"
|
|
|
|
|
|
def test_buffered_reasoning_answer_emits_backend_summary(monkeypatch):
|
|
stream = [
|
|
_sse({"reasoning_content": "I am thinking."}),
|
|
_sse({"reasoning_content": " Still thinking."}),
|
|
_sse({"content": "Final answer."}),
|
|
_done(),
|
|
]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [stream], payloads)
|
|
_patch_monotonic(monkeypatch, [100.0, 110.0, 172.0, 172.0])
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "answer"}],
|
|
tools = [{"type": "function", "function": {"name": "web_search"}}],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
summary_index = next(
|
|
i for i, event in enumerate(events) if event["type"] == "reasoning_summary"
|
|
)
|
|
content_index = next(i for i, event in enumerate(events) if event["type"] == "content")
|
|
assert summary_index < content_index
|
|
assert events[summary_index]["duration_ms"] == 62000
|
|
assert (
|
|
events[content_index]["text"]
|
|
== "<think>I am thinking. Still thinking.</think>Final answer."
|
|
)
|
|
|
|
|
|
def test_consumed_tool_final_pass_emits_latest_reasoning_summary(monkeypatch):
|
|
tool_stream = [
|
|
_sse({"reasoning_content": "Need a render."}),
|
|
_sse(
|
|
{
|
|
"content": '<tool_call>{"name":"render_html","arguments":{"code":"<html>ok</html>"}}</tool_call>'
|
|
}
|
|
),
|
|
_done(),
|
|
]
|
|
final_stream = [
|
|
_sse({"reasoning_content": "Now synthesize."}),
|
|
_sse({"content": "Final from tool."}),
|
|
_done(),
|
|
]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [tool_stream, final_stream], payloads)
|
|
_patch_monotonic(monkeypatch, [200.0, 201.0, 203.0, 300.0, 400.0, 405.0, 405.0])
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
return "Rendered HTML canvas: Done."
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "render then answer"}],
|
|
tools = [{"type": "function", "function": {"name": "render_html"}}],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
summaries = [event for event in events if event["type"] == "reasoning_summary"]
|
|
assert [event["duration_ms"] for event in summaries] == [2000, 5000]
|
|
final_summary_index = events.index(summaries[-1])
|
|
final_content_index = next(
|
|
i
|
|
for i, event in enumerate(events)
|
|
if event.get("type") == "content" and "Final from tool." in event.get("text", "")
|
|
)
|
|
assert final_summary_index < final_content_index
|
|
|
|
|
|
def test_repeat_render_html_nudge_is_not_user_visible_error(monkeypatch):
|
|
"""A repeated render_html call is an internal no-op, not a visible card."""
|
|
|
|
first_stream = [
|
|
_sse(
|
|
{
|
|
"tool_calls": [
|
|
{
|
|
"index": 0,
|
|
"id": "call_first",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "render_html",
|
|
"arguments": json.dumps(
|
|
{
|
|
"code": "<html><body>first</body></html>",
|
|
"title": "First",
|
|
}
|
|
),
|
|
},
|
|
}
|
|
]
|
|
}
|
|
),
|
|
_done(),
|
|
]
|
|
repeat_stream = [
|
|
_sse(
|
|
{
|
|
"tool_calls": [
|
|
{
|
|
"index": 0,
|
|
"id": "call_repeat",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "render_html",
|
|
"arguments": json.dumps(
|
|
{
|
|
"code": "<html><body>repeat</body></html>",
|
|
"title": "Repeat",
|
|
}
|
|
),
|
|
},
|
|
}
|
|
]
|
|
}
|
|
),
|
|
_done(),
|
|
]
|
|
final_stream = [_sse({"content": "Short note."}), _done()]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [first_stream, repeat_stream, final_stream], payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
calls.append((name, arguments))
|
|
return "Rendered HTML canvas: First."
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
|
|
tools = [
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "render_html",
|
|
"description": "Render HTML.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {"code": {"type": "string"}},
|
|
"required": ["code"],
|
|
},
|
|
},
|
|
},
|
|
{"type": "function", "function": {"name": "web_search"}},
|
|
]
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "Make a red square."}],
|
|
tools = tools,
|
|
max_tool_iterations = 2,
|
|
)
|
|
)
|
|
|
|
assert calls == [
|
|
(
|
|
"render_html",
|
|
{"code": "<html><body>first</body></html>", "title": "First"},
|
|
)
|
|
]
|
|
assert _tool_names(payloads[1]) == ["web_search"]
|
|
|
|
actual_tool_starts = [
|
|
event
|
|
for event in events
|
|
if event.get("type") == "tool_start" and event.get("arguments", {}).get("code")
|
|
]
|
|
tool_ends = [
|
|
event
|
|
for event in events
|
|
if event.get("type") == "tool_end" and event.get("tool_name") == "render_html"
|
|
]
|
|
assert len(actual_tool_starts) == 1
|
|
assert len(tool_ends) == 1
|
|
|
|
assert len(payloads) == 3
|
|
render_tool_messages = [
|
|
message
|
|
for message in payloads[2]["messages"]
|
|
if message.get("role") == "tool" and message.get("name") == "render_html"
|
|
]
|
|
assert len(render_tool_messages) == 1
|
|
internal_nudges = [
|
|
message
|
|
for message in payloads[2]["messages"]
|
|
if message.get("role") == "user"
|
|
and "Do not call render_html again" in message.get("content", "")
|
|
]
|
|
assert len(internal_nudges) == 1
|
|
|
|
|
|
def test_render_html_success_drops_tool_schema_before_final_pass(monkeypatch):
|
|
first_stream = [
|
|
_sse(
|
|
{
|
|
"tool_calls": [
|
|
{
|
|
"index": 0,
|
|
"id": "call_first",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "render_html",
|
|
"arguments": json.dumps({"code": "<html>ok</html>"}),
|
|
},
|
|
}
|
|
]
|
|
}
|
|
),
|
|
_done(),
|
|
]
|
|
final_stream = [_sse({"content": "Done."}), _done()]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [first_stream, final_stream], payloads)
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
return "Rendered HTML canvas: Done."
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "Render this."}],
|
|
tools = [{"type": "function", "function": {"name": "render_html"}}],
|
|
max_tool_iterations = 3,
|
|
)
|
|
)
|
|
|
|
assert len(payloads) == 2
|
|
assert "tools" not in payloads[1]
|
|
assert any(event.get("type") == "content" and event.get("text") == "Done." for event in events)
|
|
final_user_messages = [
|
|
m.get("content", "") for m in payloads[1]["messages"] if m.get("role") == "user"
|
|
]
|
|
assert not any("used all available tool calls" in message for message in final_user_messages)
|
|
|
|
|
|
def test_non_consecutive_duplicate_web_search_is_internal_noop(monkeypatch):
|
|
first_search = [
|
|
_sse(
|
|
{
|
|
"tool_calls": [
|
|
{
|
|
"index": 0,
|
|
"id": "call_search_1",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "web_search",
|
|
"arguments": json.dumps({"query": "gpu prices 2026"}),
|
|
},
|
|
}
|
|
]
|
|
}
|
|
),
|
|
_done(),
|
|
]
|
|
python_call = [
|
|
_sse(
|
|
{
|
|
"tool_calls": [
|
|
{
|
|
"index": 0,
|
|
"id": "call_python",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "python",
|
|
"arguments": json.dumps({"code": "print('ok')"}),
|
|
},
|
|
}
|
|
]
|
|
}
|
|
),
|
|
_done(),
|
|
]
|
|
duplicate_search = [
|
|
_sse(
|
|
{
|
|
"tool_calls": [
|
|
{
|
|
"index": 0,
|
|
"id": "call_search_2",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "web_search",
|
|
"arguments": json.dumps({"query": "gpu prices 2026"}),
|
|
},
|
|
}
|
|
]
|
|
}
|
|
),
|
|
_done(),
|
|
]
|
|
final_stream = [_sse({"content": "Final answer from gathered data."}), _done()]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(
|
|
monkeypatch,
|
|
[first_search, python_call, duplicate_search, final_stream],
|
|
payloads,
|
|
)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
calls.append((name, arguments))
|
|
return f"ok:{name}"
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
|
|
tools = [
|
|
{"type": "function", "function": {"name": "web_search"}},
|
|
{"type": "function", "function": {"name": "python"}},
|
|
]
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "search gpus in 2026 prices and use python"}],
|
|
tools = tools,
|
|
max_tool_iterations = 3,
|
|
)
|
|
)
|
|
|
|
assert calls == [
|
|
("web_search", {"query": "gpu prices 2026"}),
|
|
("python", {"code": "print('ok')"}),
|
|
]
|
|
assert [
|
|
event.get("tool_name")
|
|
for event in events
|
|
if event.get("type") == "tool_start" and event.get("tool_name")
|
|
] == ["web_search", "python"]
|
|
assert [
|
|
event.get("tool_name")
|
|
for event in events
|
|
if event.get("type") == "tool_end" and event.get("tool_name")
|
|
] == ["web_search", "python"]
|
|
assert not [
|
|
event
|
|
for event in events
|
|
if event.get("tool_call_id") == "call_search_2"
|
|
and event.get("type") in {"tool_start", "tool_end"}
|
|
]
|
|
assert len(payloads) == 4
|
|
assert _tool_names(payloads[3]) == ["web_search", "python"]
|
|
duplicate_nudges = [
|
|
message
|
|
for message in payloads[3]["messages"]
|
|
if message.get("role") == "user"
|
|
and "already completed successfully" in message.get("content", "")
|
|
]
|
|
assert len(duplicate_nudges) == 1
|
|
|
|
|
|
def test_duplicate_web_search_noop_allows_distinct_followup_tool(monkeypatch):
|
|
first_search = [
|
|
_sse(
|
|
{
|
|
"tool_calls": [
|
|
{
|
|
"index": 0,
|
|
"id": "call_search_1",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "web_search",
|
|
"arguments": json.dumps({"query": "gpu prices 2026"}),
|
|
},
|
|
}
|
|
]
|
|
}
|
|
),
|
|
_done(),
|
|
]
|
|
duplicate_search = [
|
|
_sse(
|
|
{
|
|
"tool_calls": [
|
|
{
|
|
"index": 0,
|
|
"id": "call_search_2",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "web_search",
|
|
"arguments": json.dumps({"query": "gpu prices 2026"}),
|
|
},
|
|
}
|
|
]
|
|
}
|
|
),
|
|
_done(),
|
|
]
|
|
python_call = [
|
|
_sse(
|
|
{
|
|
"tool_calls": [
|
|
{
|
|
"index": 0,
|
|
"id": "call_python",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "python",
|
|
"arguments": json.dumps({"code": "print('ok')"}),
|
|
},
|
|
}
|
|
]
|
|
}
|
|
),
|
|
_done(),
|
|
]
|
|
final_stream = [_sse({"content": "Final answer from gathered data."}), _done()]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(
|
|
monkeypatch,
|
|
[first_search, duplicate_search, python_call, final_stream],
|
|
payloads,
|
|
)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
calls.append((name, arguments))
|
|
return f"ok:{name}"
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
|
|
tools = [
|
|
{"type": "function", "function": {"name": "web_search"}},
|
|
{"type": "function", "function": {"name": "python"}},
|
|
]
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "search gpus in 2026 prices and use python"}],
|
|
tools = tools,
|
|
max_tool_iterations = 4,
|
|
)
|
|
)
|
|
|
|
assert calls == [
|
|
("web_search", {"query": "gpu prices 2026"}),
|
|
("python", {"code": "print('ok')"}),
|
|
]
|
|
assert [
|
|
event.get("tool_name")
|
|
for event in events
|
|
if event.get("type") == "tool_start" and event.get("tool_name")
|
|
] == ["web_search", "python"]
|
|
assert [
|
|
event.get("tool_name")
|
|
for event in events
|
|
if event.get("type") == "tool_end" and event.get("tool_name")
|
|
] == ["web_search", "python"]
|
|
assert not [
|
|
event
|
|
for event in events
|
|
if event.get("tool_call_id") == "call_search_2"
|
|
and event.get("type") in {"tool_start", "tool_end"}
|
|
]
|
|
assert len(payloads) == 4
|
|
assert _tool_names(payloads[2]) == ["web_search", "python"]
|
|
duplicate_nudges = [
|
|
message
|
|
for message in payloads[2]["messages"]
|
|
if message.get("role") == "user"
|
|
and "already completed successfully" in message.get("content", "")
|
|
]
|
|
assert len(duplicate_nudges) == 1
|
|
|
|
|
|
def test_repeated_duplicate_noop_transitions_to_final_pass(monkeypatch):
|
|
first_search = [
|
|
_sse(
|
|
{
|
|
"tool_calls": [
|
|
{
|
|
"index": 0,
|
|
"id": "call_search_1",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "web_search",
|
|
"arguments": json.dumps({"query": "gpu prices 2026"}),
|
|
},
|
|
}
|
|
]
|
|
}
|
|
),
|
|
_done(),
|
|
]
|
|
duplicate_one = [
|
|
_sse(
|
|
{
|
|
"tool_calls": [
|
|
{
|
|
"index": 0,
|
|
"id": "call_search_2",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "web_search",
|
|
"arguments": json.dumps({"query": "gpu prices 2026"}),
|
|
},
|
|
}
|
|
]
|
|
}
|
|
),
|
|
_done(),
|
|
]
|
|
duplicate_two = [
|
|
_sse(
|
|
{
|
|
"tool_calls": [
|
|
{
|
|
"index": 0,
|
|
"id": "call_search_3",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "web_search",
|
|
"arguments": json.dumps({"query": "gpu prices 2026"}),
|
|
},
|
|
}
|
|
]
|
|
}
|
|
),
|
|
_done(),
|
|
]
|
|
final_stream = [_sse({"content": "Final answer from first search."}), _done()]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(
|
|
monkeypatch,
|
|
[first_search, duplicate_one, duplicate_two, final_stream],
|
|
payloads,
|
|
)
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
calls.append((name, arguments))
|
|
return "result"
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "search gpus"}],
|
|
tools = [{"type": "function", "function": {"name": "web_search"}}],
|
|
max_tool_iterations = 10,
|
|
)
|
|
)
|
|
|
|
assert calls == [("web_search", {"query": "gpu prices 2026"})]
|
|
assert [event.get("tool_call_id") for event in events if event.get("type") == "tool_end"] == [
|
|
"call_search_1"
|
|
]
|
|
assert len(payloads) == 4
|
|
assert "tools" not in payloads[-1]
|
|
assert any(
|
|
event.get("type") == "content" and event.get("text") == "Final answer from first search."
|
|
for event in events
|
|
)
|
|
|
|
|
|
def test_same_turn_duplicate_web_search_is_internal_noop(monkeypatch):
|
|
same_turn_duplicates = [
|
|
_sse(
|
|
{
|
|
"tool_calls": [
|
|
{
|
|
"index": 0,
|
|
"id": "call_search_1",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "web_search",
|
|
"arguments": json.dumps({"query": "gpu prices 2026"}),
|
|
},
|
|
},
|
|
{
|
|
"index": 1,
|
|
"id": "call_search_2",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "web_search",
|
|
"arguments": json.dumps({"query": "gpu prices 2026"}),
|
|
},
|
|
},
|
|
]
|
|
}
|
|
),
|
|
_done(),
|
|
]
|
|
final_stream = [_sse({"content": "Final answer."}), _done()]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [same_turn_duplicates, final_stream], payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
calls.append((name, arguments))
|
|
return "search-result"
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "search gpus"}],
|
|
tools = [{"type": "function", "function": {"name": "web_search"}}],
|
|
max_tool_iterations = 2,
|
|
)
|
|
)
|
|
|
|
assert calls == [("web_search", {"query": "gpu prices 2026"})]
|
|
assert [event.get("tool_call_id") for event in events if event.get("type") == "tool_end"] == [
|
|
"call_search_1"
|
|
]
|
|
assert not [
|
|
event
|
|
for event in events
|
|
if event.get("tool_call_id") == "call_search_2"
|
|
and event.get("type") in {"tool_start", "tool_end"}
|
|
]
|
|
|
|
|
|
def test_same_turn_repeated_render_html_does_not_emit_second_provisional_start(monkeypatch):
|
|
same_turn_render_calls = [
|
|
_sse(
|
|
{
|
|
"tool_calls": [
|
|
{
|
|
"index": 0,
|
|
"id": "call_html_1",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "render_html",
|
|
"arguments": json.dumps({"code": "<html>one</html>"}),
|
|
},
|
|
},
|
|
{
|
|
"index": 1,
|
|
"id": "call_html_2",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "render_html",
|
|
"arguments": json.dumps({"code": "<html>two</html>"}),
|
|
},
|
|
},
|
|
]
|
|
}
|
|
),
|
|
_done(),
|
|
]
|
|
final_stream = [_sse({"content": "Final answer."}), _done()]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [same_turn_render_calls, final_stream], payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
calls.append((name, arguments))
|
|
return "Rendered HTML canvas: One."
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "render html"}],
|
|
tools = [{"type": "function", "function": {"name": "render_html"}}],
|
|
max_tool_iterations = 2,
|
|
)
|
|
)
|
|
|
|
assert calls == [("render_html", {"code": "<html>one</html>"})]
|
|
assert [
|
|
event.get("tool_call_id")
|
|
for event in events
|
|
if event.get("type") == "tool_start" and not event.get("arguments")
|
|
] == ["call_html_1"]
|
|
assert not [
|
|
event
|
|
for event in events
|
|
if event.get("tool_call_id") == "call_html_2"
|
|
and event.get("type") in {"tool_start", "tool_end"}
|
|
]
|
|
assert len(payloads) == 2
|
|
assert "tools" not in payloads[1]
|
|
render_nudges = [
|
|
message
|
|
for message in payloads[1]["messages"]
|
|
if message.get("role") == "user"
|
|
and "Do not call render_html again" in message.get("content", "")
|
|
]
|
|
assert len(render_nudges) == 1
|
|
|
|
|
|
def test_disabled_tool_call_is_internal_noop(monkeypatch):
|
|
disabled_python = [
|
|
_sse(
|
|
{
|
|
"tool_calls": [
|
|
{
|
|
"index": 0,
|
|
"id": "call_python_disabled",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "python",
|
|
"arguments": json.dumps({"code": "print(1)"}),
|
|
},
|
|
}
|
|
]
|
|
}
|
|
),
|
|
_done(),
|
|
]
|
|
final_stream = [_sse({"content": "I cannot run Python here."}), _done()]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [disabled_python, final_stream], payloads)
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
raise AssertionError(f"unexpected tool execution: {name} {arguments}")
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "run python"}],
|
|
tools = [{"type": "function", "function": {"name": "web_search"}}],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
assert not [event for event in events if event.get("type") in {"tool_start", "tool_end"}]
|
|
assert len(payloads) == 2
|
|
disabled_nudges = [
|
|
message
|
|
for message in payloads[1]["messages"]
|
|
if message.get("role") == "user" and "not enabled" in message.get("content", "")
|
|
]
|
|
assert len(disabled_nudges) == 1
|
|
|
|
|
|
def test_render_html_success_does_not_reprompt_render_html_intent(monkeypatch):
|
|
"""After render_html succeeds, do not force another render_html call.
|
|
|
|
The post-tool model pass can say it will use render_html again without
|
|
emitting a tool call. That should be accepted as a final model mistake,
|
|
not turned into repeated internal re-prompts after the canvas already
|
|
exists.
|
|
"""
|
|
|
|
first_stream = [
|
|
_sse(
|
|
{
|
|
"tool_calls": [
|
|
{
|
|
"index": 0,
|
|
"id": "call_first",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "render_html",
|
|
"arguments": json.dumps(
|
|
{
|
|
"code": "<html><body>first</body></html>",
|
|
"title": "First",
|
|
}
|
|
),
|
|
},
|
|
}
|
|
]
|
|
}
|
|
),
|
|
_done(),
|
|
]
|
|
post_tool_stream = [
|
|
_sse({"content": "I will now use render_html again."}),
|
|
_done(),
|
|
]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [first_stream, post_tool_stream], payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
calls.append((name, arguments))
|
|
return "Rendered HTML canvas: First."
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
|
|
tools = [
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "render_html",
|
|
"description": "Render HTML.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {"code": {"type": "string"}},
|
|
"required": ["code"],
|
|
},
|
|
},
|
|
}
|
|
]
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "Make a red square."}],
|
|
tools = tools,
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
assert len(payloads) == 2
|
|
assert len(calls) == 1
|
|
assert any(
|
|
event.get("type") == "content" and event.get("text") == "I will now use render_html again."
|
|
for event in events
|
|
)
|
|
|
|
|
|
def test_internal_reprompt_attempts_do_not_duplicate_visible_text(monkeypatch):
|
|
"""No-tool re-prompt attempts should not concatenate into the UI."""
|
|
|
|
# One initial response plus one stream per re-prompt; derive the count from the shared cap.
|
|
streams = [[_sse({"content": "I will use render_html now."}), _done()]]
|
|
streams += [
|
|
[_sse({"content": "Understood. I will use render_html now."}), _done()]
|
|
for _ in range(_MAX_REPROMPTS)
|
|
]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, streams, payloads)
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
raise AssertionError(f"unexpected tool execution: {name} {arguments}")
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
|
|
tools = [
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "render_html",
|
|
"description": "Render HTML.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {"code": {"type": "string"}},
|
|
"required": ["code"],
|
|
},
|
|
},
|
|
}
|
|
]
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "Make a red square."}],
|
|
tools = tools,
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
content_texts = [event.get("text", "") for event in events if event.get("type") == "content"]
|
|
assert content_texts == ["I will use render_html now."]
|
|
assert len(payloads) == _MAX_REPROMPTS + 1
|
|
|
|
|
|
def test_forced_reprompt_plain_final_answer_is_visible(monkeypatch):
|
|
"""A hidden forced re-prompt may fall back to a plain final answer."""
|
|
|
|
streams = [
|
|
[_sse({"content": "I will use render_html now."}), _done()],
|
|
[
|
|
_sse({"content": "No tool is needed. Final answer: use a red square."}),
|
|
_done(),
|
|
],
|
|
]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, streams, payloads)
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
raise AssertionError(f"unexpected tool execution: {name} {arguments}")
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "Make a red square."}],
|
|
tools = [
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "render_html",
|
|
"description": "Render HTML.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {"code": {"type": "string"}},
|
|
"required": ["code"],
|
|
},
|
|
},
|
|
}
|
|
],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
content_texts = [event.get("text", "") for event in events if event.get("type") == "content"]
|
|
assert content_texts == [
|
|
"I will use render_html now.",
|
|
"No tool is needed. Final answer: use a red square.",
|
|
]
|
|
assert len(payloads) == 2
|
|
|
|
|
|
def test_internal_reprompt_disabled_when_auto_heal_disabled(monkeypatch):
|
|
streams = [[_sse({"content": "I will use render_html now."}), _done()]]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, streams, payloads)
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
raise AssertionError(f"unexpected tool execution: {name} {arguments}")
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
|
|
tools = [
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "render_html",
|
|
"description": "Render HTML.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {"code": {"type": "string"}},
|
|
"required": ["code"],
|
|
},
|
|
},
|
|
}
|
|
]
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "Make a red square."}],
|
|
tools = tools,
|
|
max_tool_iterations = 1,
|
|
auto_heal_tool_calls = False,
|
|
)
|
|
)
|
|
|
|
content_texts = [event.get("text", "") for event in events if event.get("type") == "content"]
|
|
assert content_texts == ["I will use render_html now."]
|
|
assert len(payloads) == 1
|
|
|
|
|
|
def test_auto_heal_disabled_parses_well_formed_xml_when_tools_enabled(monkeypatch):
|
|
streams = [
|
|
[
|
|
_sse(
|
|
{
|
|
"content": '<tool_call>{"name":"web_search","arguments":{"query":"x"}}</tool_call>'
|
|
}
|
|
),
|
|
_done(),
|
|
],
|
|
[_sse({"content": "done"}), _done()],
|
|
]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, streams, payloads)
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
calls.append((name, arguments))
|
|
return "result"
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "search"}],
|
|
tools = [{"type": "function", "function": {"name": "web_search"}}],
|
|
auto_heal_tool_calls = False,
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
assert calls == [("web_search", {"query": "x"})]
|
|
assert not any(
|
|
event.get("type") == "content" and "<tool_call>" in event.get("text", "")
|
|
for event in events
|
|
)
|
|
|
|
|
|
def test_textual_mistral_marker_not_leaked_when_inline_with_preface(monkeypatch):
|
|
# Textual Mistral ``[TOOL_CALLS]`` inline with visible preface: the DRAINING flush must use the
|
|
# shared parser patterns (which know ``[TOOL_CALLS]``); the legacy set leaked the marker to clients.
|
|
streams = [
|
|
[_sse({"content": 'Let me search. [TOOL_CALLS]web_search{"query":"cats"}'}), _done()],
|
|
[_sse({"content": "done"}), _done()],
|
|
]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, streams, payloads)
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
calls.append((name, arguments))
|
|
return "result"
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "search"}],
|
|
tools = [{"type": "function", "function": {"name": "web_search"}}],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
assert calls == [("web_search", {"query": "cats"})]
|
|
content_texts = [e.get("text", "") for e in events if e.get("type") == "content"]
|
|
assert all("[TOOL_CALLS]" not in t for t in content_texts), content_texts
|
|
assert any("Let me search." in t for t in content_texts)
|
|
|
|
|
|
def test_textual_llama_python_tag_marker_not_leaked(monkeypatch):
|
|
# Same leak class for the Llama-3 built-in ``<|python_tag|>NAME.call(...)`` form.
|
|
streams = [
|
|
[_sse({"content": '<|python_tag|>web_search.call(query="cats")'}), _done()],
|
|
[_sse({"content": "done"}), _done()],
|
|
]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, streams, payloads)
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
calls.append((name, arguments))
|
|
return "result"
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "search"}],
|
|
tools = [{"type": "function", "function": {"name": "web_search"}}],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
assert calls == [("web_search", {"query": "cats"})]
|
|
content_texts = [e.get("text", "") for e in events if e.get("type") == "content"]
|
|
assert all("<|python_tag|>" not in t for t in content_texts), content_texts
|
|
|
|
|
|
def test_reprompted_tool_call_still_streams_final_answer(monkeypatch):
|
|
"""Suppression ends once a forced re-prompt actually calls a tool."""
|
|
|
|
streams = [
|
|
[_sse({"content": "I will use render_html now."}), _done()],
|
|
[
|
|
_sse(
|
|
{
|
|
"tool_calls": [
|
|
{
|
|
"index": 0,
|
|
"id": "call_forced",
|
|
"type": "function",
|
|
"function": {
|
|
"name": "render_html",
|
|
"arguments": json.dumps(
|
|
{
|
|
"code": "<html><body>forced</body></html>",
|
|
"title": "Forced",
|
|
}
|
|
),
|
|
},
|
|
}
|
|
]
|
|
}
|
|
),
|
|
_done(),
|
|
],
|
|
[_sse({"content": "Final note after tool."}), _done()],
|
|
]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, streams, payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
calls.append((name, arguments))
|
|
return "Rendered HTML canvas: Forced."
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
|
|
tools = [
|
|
{
|
|
"type": "function",
|
|
"function": {
|
|
"name": "render_html",
|
|
"description": "Render HTML.",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {"code": {"type": "string"}},
|
|
"required": ["code"],
|
|
},
|
|
},
|
|
}
|
|
]
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "Make a red square."}],
|
|
tools = tools,
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
assert len(calls) == 1
|
|
content_texts = [event.get("text", "") for event in events if event.get("type") == "content"]
|
|
assert content_texts == ["I will use render_html now.", "Final note after tool."]
|
|
assert len(payloads) == 3
|
|
|
|
|
|
def test_confirm_tool_calls_allow_executes_gguf_tool(monkeypatch):
|
|
streams = [
|
|
_structured_tool_call("python", {"code": "print(1)"}, "call_py"),
|
|
[_sse({"content": "Done."}), _done()],
|
|
]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, streams, payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
calls.append((name, arguments))
|
|
return "OK"
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
monkeypatch.setattr("core.inference.llama_cpp.new_approval_id", lambda: "approval-1")
|
|
monkeypatch.setattr(
|
|
"core.inference.llama_cpp.begin_tool_decision",
|
|
lambda *_a, **_k: object(),
|
|
)
|
|
monkeypatch.setattr("core.inference.llama_cpp.wait_tool_decision", lambda *_a, **_k: "allow")
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "run python"}],
|
|
tools = [{"type": "function", "function": {"name": "python"}}],
|
|
max_tool_iterations = 1,
|
|
confirm_tool_calls = True,
|
|
session_id = "sess",
|
|
)
|
|
)
|
|
|
|
starts = [event for event in events if event.get("type") == "tool_start"]
|
|
assert len(starts) == 1
|
|
assert starts[0]["approval_id"]
|
|
assert starts[0]["awaiting_confirmation"] is True
|
|
assert calls == [("python", {"code": "print(1)"})]
|
|
assert any(event.get("type") == "tool_end" and event.get("result") == "OK" for event in events)
|
|
|
|
|
|
def test_confirm_tool_calls_close_after_prompt_cleans_gguf_slot(monkeypatch):
|
|
approval_id = "approval-close"
|
|
streams = [_structured_tool_call("python", {"code": "print(1)"}, "call_py")]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, streams, payloads)
|
|
|
|
monkeypatch.setattr(
|
|
"core.inference.tools.execute_tool",
|
|
lambda *_a, **_k: (_ for _ in ()).throw(AssertionError("tool should not run")),
|
|
)
|
|
monkeypatch.setattr("core.inference.llama_cpp.new_approval_id", lambda: approval_id)
|
|
|
|
with tool_approvals._lock:
|
|
tool_approvals._pending.clear()
|
|
|
|
gen = backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "run python"}],
|
|
tools = [{"type": "function", "function": {"name": "python"}}],
|
|
max_tool_iterations = 1,
|
|
confirm_tool_calls = True,
|
|
session_id = "sess",
|
|
)
|
|
try:
|
|
assert next(gen)["type"] == "status"
|
|
start = next(gen)
|
|
assert start["type"] == "tool_start"
|
|
assert start["approval_id"] == approval_id
|
|
with tool_approvals._lock:
|
|
assert approval_id in tool_approvals._pending
|
|
finally:
|
|
gen.close()
|
|
|
|
with tool_approvals._lock:
|
|
assert approval_id not in tool_approvals._pending
|
|
assert resolve_tool_decision(approval_id, "allow", session_id = "sess") is False
|
|
|
|
|
|
def test_confirm_tool_calls_skips_gguf_rag_autoinject(monkeypatch):
|
|
streams = [[_sse({"content": "Done."}), _done()]]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, streams, payloads)
|
|
|
|
def fail_autoinject(*_args, **_kwargs):
|
|
raise AssertionError("RAG autoinject must not run before approval")
|
|
|
|
monkeypatch.setattr("core.inference.tools.build_rag_autoinject", fail_autoinject)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "use docs"}],
|
|
tools = [{"type": "function", "function": {"name": "search_knowledge_base"}}],
|
|
max_tool_iterations = 1,
|
|
confirm_tool_calls = True,
|
|
session_id = "sess",
|
|
rag_scope = {"thread_id": "t1"},
|
|
)
|
|
)
|
|
|
|
assert any(event.get("type") == "content" and event.get("text") == "Done." for event in events)
|
|
|
|
|
|
def test_confirm_tool_calls_deny_skips_gguf_tool_and_retry_can_execute(monkeypatch):
|
|
same_call = _structured_tool_call("python", {"code": "print(1)"}, "call_py")
|
|
streams = [
|
|
same_call,
|
|
_structured_tool_call("python", {"code": "print(1)"}, "call_py_retry"),
|
|
[_sse({"content": "Done."}), _done()],
|
|
]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, streams, payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
calls.append((name, arguments))
|
|
return "OK"
|
|
|
|
decisions = iter(["deny", "allow"])
|
|
approvals = iter(["approval-1", "approval-2"])
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
monkeypatch.setattr("core.inference.llama_cpp.new_approval_id", lambda: next(approvals))
|
|
monkeypatch.setattr(
|
|
"core.inference.llama_cpp.begin_tool_decision",
|
|
lambda *_a, **_k: object(),
|
|
)
|
|
monkeypatch.setattr(
|
|
"core.inference.llama_cpp.wait_tool_decision",
|
|
lambda *_a, **_k: next(decisions),
|
|
)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "run python"}],
|
|
tools = [{"type": "function", "function": {"name": "python"}}],
|
|
max_tool_iterations = 2,
|
|
confirm_tool_calls = True,
|
|
session_id = "sess",
|
|
)
|
|
)
|
|
|
|
starts = [event for event in events if event.get("type") == "tool_start"]
|
|
ends = [event for event in events if event.get("type") == "tool_end"]
|
|
assert len(starts) == 2
|
|
assert [event["result"] for event in ends] == [TOOL_REJECTED_MESSAGE, "OK"]
|
|
assert calls == [("python", {"code": "print(1)"})]
|
|
|
|
|
|
def _streamed_structured_tool_call(
|
|
tool_name: str,
|
|
arguments: dict,
|
|
call_id: str,
|
|
frag: int = 24,
|
|
) -> list[str]:
|
|
"""A structured tool call whose arguments arrive token-by-token across many
|
|
deltas (id + name on the first delta), mirroring how llama-server streams a
|
|
large tool-call argument such as a full HTML/code file."""
|
|
args_json = json.dumps(arguments)
|
|
fragments = [args_json[i : i + frag] for i in range(0, len(args_json), frag)] or [""]
|
|
chunks = [
|
|
_sse(
|
|
{
|
|
"tool_calls": [
|
|
{
|
|
"index": 0,
|
|
"id": call_id,
|
|
"type": "function",
|
|
"function": {"name": tool_name, "arguments": fragments[0]},
|
|
}
|
|
]
|
|
}
|
|
)
|
|
]
|
|
for fragment in fragments[1:]:
|
|
chunks.append(_sse({"tool_calls": [{"index": 0, "function": {"arguments": fragment}}]}))
|
|
chunks.append(_done())
|
|
return chunks
|
|
|
|
|
|
def test_large_python_tool_call_emits_early_provisional_start(monkeypatch):
|
|
"""Regression: a large streamed tool-call argument surfaces a provisional
|
|
tool card BEFORE the full arguments finish, so the UI shows progress during
|
|
generation instead of a frozen 'Generating...'. (The bug: only render_html
|
|
surfaced early; python/terminal/etc. were silent until the call completed.)"""
|
|
|
|
big_code = "total = 0\n" + "\n".join(f"total += {i}" for i in range(120))
|
|
args_json = json.dumps({"code": big_code})
|
|
assert len(args_json) > _PROVISIONAL_ARGS_MIN_CHARS
|
|
|
|
first_stream = _streamed_structured_tool_call("python", {"code": big_code}, "call_py_big")
|
|
final_stream = [_sse({"content": "Done."}), _done()]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [first_stream, final_stream], payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
calls.append((name, arguments))
|
|
return "OK"
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "write code"}],
|
|
tools = [{"type": "function", "function": {"name": "python"}}],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
tool_starts = [e for e in events if e.get("type") == "tool_start"]
|
|
provisional = [e for e in tool_starts if not e.get("arguments")]
|
|
real = [e for e in tool_starts if e.get("arguments", {}).get("code")]
|
|
|
|
# Exactly one provisional (empty args) and one real (full args), same id so
|
|
# the frontend reconciles them into a single card.
|
|
assert len(provisional) == 1, tool_starts
|
|
assert provisional[0]["tool_name"] == "python"
|
|
assert provisional[0]["tool_call_id"] == "call_py_big"
|
|
assert provisional[0]["provenance"].get("provisional") is True
|
|
assert len(real) == 1
|
|
assert real[0]["tool_call_id"] == "call_py_big"
|
|
# The provisional card appears before the real (completed) tool_start.
|
|
assert events.index(provisional[0]) < events.index(real[0])
|
|
|
|
assert calls == [("python", {"code": big_code})]
|
|
assert any(e.get("type") == "tool_end" and e.get("tool_name") == "python" for e in events)
|
|
|
|
|
|
def test_small_python_tool_call_has_no_provisional_start(monkeypatch):
|
|
"""A small tool-call argument finishes streaming instantly, so it keeps the
|
|
existing behavior of a single (real) tool_start with no provisional card."""
|
|
|
|
first_stream = _structured_tool_call("python", {"code": "print(1)"}, "call_py_small")
|
|
final_stream = [_sse({"content": "Done."}), _done()]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [first_stream, final_stream], payloads)
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", lambda *_a, **_k: "OK")
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "x"}],
|
|
tools = [{"type": "function", "function": {"name": "python"}}],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
tool_starts = [e for e in events if e.get("type") == "tool_start"]
|
|
assert [e for e in tool_starts if not e.get("arguments")] == []
|
|
assert len([e for e in tool_starts if e.get("arguments", {}).get("code")]) == 1
|
|
|
|
|
|
def _streamed_parallel_tool_calls(specs, frag: int = 24) -> list[str]:
|
|
"""Two or more structured tool calls, each streamed token-by-token across
|
|
deltas, one index fully before the next, mirroring how llama-server streams
|
|
several parallel tool calls whose arguments are large."""
|
|
chunks: list[str] = []
|
|
for index, (tool_name, arguments, call_id) in enumerate(specs):
|
|
args_json = json.dumps(arguments)
|
|
fragments = [args_json[i : i + frag] for i in range(0, len(args_json), frag)] or [""]
|
|
chunks.append(
|
|
_sse(
|
|
{
|
|
"tool_calls": [
|
|
{
|
|
"index": index,
|
|
"id": call_id,
|
|
"type": "function",
|
|
"function": {"name": tool_name, "arguments": fragments[0]},
|
|
}
|
|
]
|
|
}
|
|
)
|
|
)
|
|
for fragment in fragments[1:]:
|
|
chunks.append(
|
|
_sse({"tool_calls": [{"index": index, "function": {"arguments": fragment}}]})
|
|
)
|
|
chunks.append(_done())
|
|
return chunks
|
|
|
|
|
|
def test_parallel_large_tool_calls_each_emit_provisional_start(monkeypatch):
|
|
"""With parallel tool use enabled (the default), every streamed large tool
|
|
call surfaces its own provisional card, not just the first one, so the UI
|
|
shows progress for each call as its arguments stream."""
|
|
|
|
big_code = "total = 0\n" + "\n".join(f"total += {i}" for i in range(120))
|
|
big_cmd = "echo start\n" + "\n".join(f"echo line {i}" for i in range(60))
|
|
assert len(json.dumps({"code": big_code})) > _PROVISIONAL_ARGS_MIN_CHARS
|
|
assert len(json.dumps({"command": big_cmd})) > _PROVISIONAL_ARGS_MIN_CHARS
|
|
|
|
first_stream = _streamed_parallel_tool_calls(
|
|
[
|
|
("python", {"code": big_code}, "call_py"),
|
|
("terminal", {"command": big_cmd}, "call_term"),
|
|
]
|
|
)
|
|
final_stream = [_sse({"content": "Done."}), _done()]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [first_stream, final_stream], payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
calls.append((name, arguments))
|
|
return "OK"
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "do both"}],
|
|
tools = [
|
|
{"type": "function", "function": {"name": "python"}},
|
|
{"type": "function", "function": {"name": "terminal"}},
|
|
],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
provisional = [e for e in events if e.get("type") == "tool_start" and not e.get("arguments")]
|
|
assert sorted(e["tool_call_id"] for e in provisional) == ["call_py", "call_term"]
|
|
assert all(e["provenance"].get("provisional") is True for e in provisional)
|
|
# Both calls actually executed (parallel tool use is enabled by default).
|
|
assert sorted(name for name, _ in calls) == ["python", "terminal"]
|
|
|
|
|
|
def test_parallel_disabled_suppresses_provisional_for_later_calls(monkeypatch):
|
|
"""When parallel tool use is disabled the downstream truncates to the first
|
|
call, so only the first streamed call may surface a provisional; a later
|
|
call must not get a card that could never reconcile or be closed."""
|
|
|
|
big_code = "total = 0\n" + "\n".join(f"total += {i}" for i in range(120))
|
|
big_cmd = "echo start\n" + "\n".join(f"echo line {i}" for i in range(60))
|
|
|
|
first_stream = _streamed_parallel_tool_calls(
|
|
[
|
|
("python", {"code": big_code}, "call_py"),
|
|
("terminal", {"command": big_cmd}, "call_term"),
|
|
]
|
|
)
|
|
final_stream = [_sse({"content": "Done."}), _done()]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [first_stream, final_stream], payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
calls.append((name, arguments))
|
|
return "OK"
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "do both"}],
|
|
tools = [
|
|
{"type": "function", "function": {"name": "python"}},
|
|
{"type": "function", "function": {"name": "terminal"}},
|
|
],
|
|
max_tool_iterations = 1,
|
|
disable_parallel_tool_use = True,
|
|
)
|
|
)
|
|
|
|
provisional = [e for e in events if e.get("type") == "tool_start" and not e.get("arguments")]
|
|
assert [e["tool_call_id"] for e in provisional] == ["call_py"]
|
|
# Only the first call executes when parallel use is disabled.
|
|
assert calls == [("python", {"code": big_code})]
|
|
# The lone provisional is closed exactly once (no dangling card).
|
|
closing = [
|
|
e for e in events if e.get("type") == "tool_end" and e.get("tool_call_id") == "call_py"
|
|
]
|
|
assert len(closing) == 1
|
|
|
|
|
|
def test_connect_error_during_tool_call_closes_provisional_card(monkeypatch):
|
|
"""If llama-server drops mid tool-call after a provisional card is shown, the
|
|
loop must close that card before surfacing the error so the UI never leaves a
|
|
tool spinning forever."""
|
|
import httpx
|
|
|
|
big_code = "total = 0\n" + "\n".join(f"total += {i}" for i in range(120))
|
|
fragments = _streamed_structured_tool_call("python", {"code": big_code}, "call_py_err")
|
|
# Drop the trailing [DONE]; raise a connection error after the fragments
|
|
# stream (and after the provisional card has been emitted).
|
|
fragments = fragments[:-1]
|
|
|
|
def raising_stream():
|
|
for chunk in fragments:
|
|
yield chunk
|
|
raise httpx.ConnectError("connection lost mid stream")
|
|
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [raising_stream()], payloads)
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", lambda *_a, **_k: "OK")
|
|
|
|
collected: list[dict] = []
|
|
raised = False
|
|
gen = backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "write code"}],
|
|
tools = [{"type": "function", "function": {"name": "python"}}],
|
|
max_tool_iterations = 1,
|
|
)
|
|
try:
|
|
for event in gen:
|
|
collected.append(event)
|
|
except RuntimeError as exc:
|
|
raised = True
|
|
assert "Lost connection" in str(exc)
|
|
|
|
assert raised
|
|
provisional = [e for e in collected if e.get("type") == "tool_start" and not e.get("arguments")]
|
|
assert len(provisional) == 1
|
|
assert provisional[0]["tool_call_id"] == "call_py_err"
|
|
# The provisional card is closed before the error propagates.
|
|
closing = [
|
|
e
|
|
for e in collected
|
|
if e.get("type") == "tool_end" and e.get("tool_call_id") == "call_py_err"
|
|
]
|
|
assert len(closing) == 1
|
|
# The closing card is marked as an error, not an empty success, so the UI
|
|
# renders it as failed.
|
|
assert "Error" in (closing[0].get("result") or "")
|
|
|
|
|
|
def test_empty_tool_call_id_does_not_emit_provisional_card(monkeypatch):
|
|
"""llama.cpp can stream a tool call whose id is an empty string. A provisional
|
|
card keyed by "" cannot reconcile with the real tool_start (the frontend mints
|
|
its own id per event), so it must not be emitted -- otherwise the empty card
|
|
would dangle. The real call must still execute normally."""
|
|
|
|
big_code = "total = 0\n" + "\n".join(f"total += {i}" for i in range(120))
|
|
assert len(json.dumps({"code": big_code})) > _PROVISIONAL_ARGS_MIN_CHARS
|
|
|
|
# Same large streamed call as the provisional test, but with an empty id.
|
|
first_stream = _streamed_structured_tool_call("python", {"code": big_code}, "")
|
|
final_stream = [_sse({"content": "Done."}), _done()]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [first_stream, final_stream], payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
calls.append((name, arguments))
|
|
return "OK"
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "write code"}],
|
|
tools = [{"type": "function", "function": {"name": "python"}}],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
# No provisional card (empty-args tool_start) was surfaced for the empty id.
|
|
provisional = [e for e in events if e.get("type") == "tool_start" and not e.get("arguments")]
|
|
assert provisional == []
|
|
# The real call still executes despite the missing id.
|
|
assert calls == [("python", {"code": big_code})]
|
|
|
|
|
|
def _streamed_content(text: str, frag: int = 4) -> list[str]:
|
|
"""Stream content token-by-token like llama-server; ``frag`` sets the chunk size."""
|
|
chunks = [_sse({"content": text[i : i + frag]}) for i in range(0, len(text), frag)]
|
|
chunks.append(_done())
|
|
return chunks
|
|
|
|
|
|
def test_bare_json_tool_call_streamed_is_not_leaked_and_executes(monkeypatch):
|
|
"""A wrapper-less bare-JSON call must be held while incomplete, drained silently, and executed with nothing leaking."""
|
|
|
|
bare_call = '{"name": "web_search", "parameters": {"query": "weather in Sydney"}}'
|
|
first_stream = _streamed_content(bare_call)
|
|
final_stream = [_sse({"content": "It is sunny in Sydney."}), _done()]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [first_stream, final_stream], payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
calls.append((name, arguments))
|
|
return "Weather: sunny, 22C."
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "weather in Sydney?"}],
|
|
tools = [{"type": "function", "function": {"name": "web_search"}}],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
# The tool ran with the parsed arguments.
|
|
assert calls == [("web_search", {"query": "weather in Sydney"})]
|
|
assert any(
|
|
event.get("type") == "tool_end" and event.get("tool_name") == "web_search"
|
|
for event in events
|
|
)
|
|
|
|
# The bare JSON never leaked to the user-visible stream.
|
|
content_texts = [e.get("text", "") for e in events if e.get("type") == "content"]
|
|
assert all('"name"' not in t for t in content_texts), content_texts
|
|
assert all("web_search" not in t for t in content_texts), content_texts
|
|
# The post-tool synthesis is still streamed.
|
|
assert any("sunny in Sydney" in t for t in content_texts), content_texts
|
|
|
|
|
|
def test_ordinary_json_with_name_key_is_shown_not_treated_as_tool_call(monkeypatch):
|
|
"""Markerless JSON with a non-enabled name is the answer, not a phantom call."""
|
|
|
|
answer = '{"name": "Alice", "parameters": {"age": 30}}'
|
|
first_stream = _streamed_content(answer)
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [first_stream], payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
monkeypatch.setattr(
|
|
"core.inference.tools.execute_tool",
|
|
lambda n, a, **_k: (calls.append((n, a)) or "x"),
|
|
)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "give me a person record"}],
|
|
tools = [{"type": "function", "function": {"name": "web_search"}}],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
assert calls == [], calls
|
|
content_texts = [e.get("text", "") for e in events if e.get("type") == "content"]
|
|
assert any("Alice" in t for t in content_texts), content_texts
|
|
|
|
|
|
def test_incomplete_bare_json_truncation_is_not_leaked(monkeypatch):
|
|
"""If generation is cut off mid bare-JSON object (no closing brace), the held
|
|
fragment must be stripped at stream end rather than dumped to the user."""
|
|
|
|
truncated = '{"name": "web_search", "parameters": {"query": "weather in S'
|
|
stream = _streamed_content(truncated)
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [stream], payloads)
|
|
|
|
monkeypatch.setattr(
|
|
"core.inference.tools.execute_tool",
|
|
lambda *_a, **_k: (_ for _ in ()).throw(AssertionError("no complete call")),
|
|
)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "weather?"}],
|
|
tools = [{"type": "function", "function": {"name": "web_search"}}],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
content_texts = [e.get("text", "") for e in events if e.get("type") == "content"]
|
|
assert all('{"name"' not in t for t in content_texts), content_texts
|
|
|
|
|
|
def test_gguf_truncated_ordinary_json_with_name_key_is_shown_not_suppressed(monkeypatch):
|
|
"""A truncated markerless object whose "name" is NOT an enabled tool (a person
|
|
record cut off mid-stream, ``{"name":"Alice","age":``) must still be shown. The
|
|
end-of-stream ``_is_bare_tc`` heuristic routed any ``{...,"name",...}`` fragment
|
|
to DRAINING (dropped); it is now gated on the enabled tool names so only a real
|
|
truncated tool call is suppressed, ordinary JSON streams through."""
|
|
|
|
truncated = '{"name": "Alice", "age": 30, "bio": "loves '
|
|
stream = _streamed_content(truncated)
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [stream], payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
monkeypatch.setattr(
|
|
"core.inference.tools.execute_tool",
|
|
lambda n, a, **_k: (calls.append((n, a)) or "x"),
|
|
)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "start a person record"}],
|
|
tools = [{"type": "function", "function": {"name": "web_search"}}],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
assert calls == [], calls
|
|
content_texts = [e.get("text", "") for e in events if e.get("type") == "content"]
|
|
assert any("Alice" in t for t in content_texts), content_texts
|
|
|
|
|
|
def test_gguf_truncated_disabled_name_json_is_preserved_when_tools_active(monkeypatch):
|
|
"""A truncated JSON answer with a non-enabled name must still be shown (resolvers are gated on enabled names)."""
|
|
|
|
truncated = '{"name": "Alice", "parameters": {"age": 30'
|
|
stream = _streamed_content(truncated)
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [stream], payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
monkeypatch.setattr(
|
|
"core.inference.tools.execute_tool",
|
|
lambda n, a, **_k: (calls.append((n, a)) or "x"),
|
|
)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "give json"}],
|
|
tools = [{"type": "function", "function": {"name": "web_search"}}],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
assert calls == [], calls
|
|
content_texts = [e.get("text", "") for e in events if e.get("type") == "content"]
|
|
assert any("Alice" in t for t in content_texts), content_texts
|
|
|
|
|
|
def test_gguf_truncated_enabled_name_json_is_still_suppressed(monkeypatch):
|
|
"""Counterpart guard: a truncated ENABLED-tool bare call (``web_search``) cut off
|
|
mid-JSON still must NOT leak -- the gate only spares disabled / non-tool names."""
|
|
|
|
truncated = '{"name": "web_search", "parameters": {"query": "weather in S'
|
|
stream = _streamed_content(truncated)
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [stream], payloads)
|
|
|
|
monkeypatch.setattr(
|
|
"core.inference.tools.execute_tool",
|
|
lambda *_a, **_k: (_ for _ in ()).throw(AssertionError("no complete call")),
|
|
)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "weather?"}],
|
|
tools = [{"type": "function", "function": {"name": "web_search"}}],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
content_texts = [e.get("text", "") for e in events if e.get("type") == "content"]
|
|
assert all("web_search" not in t for t in content_texts), content_texts
|
|
assert all('{"name"' not in t for t in content_texts), content_texts
|
|
|
|
|
|
def test_gguf_oversized_disabled_name_json_is_preserved(monkeypatch):
|
|
"""An oversized still-open JSON answer with a non-enabled name streams as content, not a phantom drain."""
|
|
|
|
cap = 16384
|
|
big = "A" * (cap + 5000)
|
|
answer = '{"name":"Alice","parameters":{"bio":"' + big # never closes
|
|
first_stream = [_sse({"content": answer[i : i + 2000]}) for i in range(0, len(answer), 2000)]
|
|
first_stream.append(_done())
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [first_stream], payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
monkeypatch.setattr(
|
|
"core.inference.tools.execute_tool",
|
|
lambda n, a, **_k: (calls.append((n, a)) or "x"),
|
|
)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "long json"}],
|
|
tools = [{"type": "function", "function": {"name": "web_search"}}],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
assert calls == [], calls
|
|
content_texts = [e.get("text", "") for e in events if e.get("type") == "content"]
|
|
assert any("Alice" in t for t in content_texts), content_texts[:1]
|
|
|
|
|
|
def test_gemma_wrapperless_call_streamed_is_not_leaked_and_executes(monkeypatch):
|
|
"""Gemma 4 GGUF (skip_special_tokens) streams a wrapper-less ``call:NAME{..}``
|
|
with no XML signal. Like bare JSON, the BUFFERING scan must recognise it via
|
|
_GEMMA_BARE_TC_RE, drain it silently, and execute the tool -- never leaking
|
|
the ``call:`` markup to the user-visible stream."""
|
|
|
|
gemma_call = 'call:web_search{query:"weather in Sydney"}'
|
|
first_stream = _streamed_content(gemma_call)
|
|
final_stream = [_sse({"content": "It is sunny in Sydney."}), _done()]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [first_stream, final_stream], payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
calls.append((name, arguments))
|
|
return "Weather: sunny, 22C."
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "weather in Sydney?"}],
|
|
tools = [{"type": "function", "function": {"name": "web_search"}}],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
assert calls == [("web_search", {"query": "weather in Sydney"})]
|
|
content_texts = [e.get("text", "") for e in events if e.get("type") == "content"]
|
|
assert all("call:" not in t for t in content_texts), content_texts
|
|
assert any("sunny in Sydney" in t for t in content_texts), content_texts
|
|
|
|
|
|
def _usage_done(usage: dict, finish_reason: str = "stop") -> str:
|
|
"""A terminal SSE chunk carrying llama-server's ``usage`` block, the way the
|
|
real server reports it on the final chunk of a completion."""
|
|
return (
|
|
"data: "
|
|
+ json.dumps(
|
|
{
|
|
"choices": [{"index": 0, "delta": {}, "finish_reason": finish_reason}],
|
|
"usage": usage,
|
|
}
|
|
)
|
|
+ "\n"
|
|
)
|
|
|
|
|
|
def test_metadata_event_preserves_prompt_tokens_details(monkeypatch):
|
|
"""The tool loop's metadata event must carry llama-server's
|
|
``prompt_tokens_details`` (KV-cache hits) through ``_build_metadata_event``,
|
|
so the route reports real ``cached_tokens`` instead of always 0 (#6570).
|
|
|
|
This drives the *real* generator; the route-level test feeds a pre-built
|
|
metadata event and so never exercises this code.
|
|
"""
|
|
stream = [
|
|
_sse({"content": "The answer is 42."}),
|
|
_usage_done(
|
|
{
|
|
"prompt_tokens": 20,
|
|
"completion_tokens": 4,
|
|
"prompt_tokens_details": {"cached_tokens": 16},
|
|
}
|
|
),
|
|
_done(),
|
|
]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [stream], payloads)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "hi"}],
|
|
tools = [],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
metadata = [e for e in events if e.get("type") == "metadata"]
|
|
assert metadata, "expected a metadata event"
|
|
usage = metadata[-1]["usage"]
|
|
assert usage["prompt_tokens_details"] == {"cached_tokens": 16}
|
|
assert usage["prompt_tokens"] == 20
|
|
assert usage["completion_tokens"] == 4
|
|
|
|
|
|
def test_metadata_event_omits_prompt_tokens_details_when_absent(monkeypatch):
|
|
"""No KV-cache block from the server -> the key isn't fabricated, so the
|
|
route falls back to its 0-default instead of reading a bogus value."""
|
|
stream = [
|
|
_sse({"content": "hi"}),
|
|
_usage_done({"prompt_tokens": 5, "completion_tokens": 2}),
|
|
_done(),
|
|
]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [stream], payloads)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "hi"}],
|
|
tools = [],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
metadata = [e for e in events if e.get("type") == "metadata"]
|
|
assert metadata, "expected a metadata event"
|
|
assert "prompt_tokens_details" not in metadata[-1]["usage"]
|
|
|
|
|
|
def test_gguf_rehearsal_name_split_before_args_is_not_leaked(monkeypatch):
|
|
"""Finding 6: a rehearsal call whose name (``web_search``) and ``[ARGS]{...}``
|
|
arrive in separate content deltas must hold the bare name in the buffer until
|
|
``[ARGS]`` flips it to a drain. Without _is_rehearsal_prefix the GGUF path
|
|
streams the tool name as visible content before the call executes."""
|
|
|
|
first_stream = [
|
|
_sse({"content": "web_search"}),
|
|
_sse({"content": '[ARGS]{"query":"cats"}'}),
|
|
_done(),
|
|
]
|
|
final_stream = [_sse({"content": "Found cats."}), _done()]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [first_stream, final_stream], payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
|
|
def fake_execute_tool(name, arguments, **_kwargs):
|
|
calls.append((name, arguments))
|
|
return "result"
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", fake_execute_tool)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "search cats"}],
|
|
tools = [{"type": "function", "function": {"name": "web_search"}}],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
assert calls == [("web_search", {"query": "cats"})], calls
|
|
content_texts = [e.get("text", "") for e in events if e.get("type") == "content"]
|
|
assert all("web_search" not in t for t in content_texts), content_texts
|
|
assert all("[ARGS]" not in t for t in content_texts), content_texts
|
|
|
|
|
|
def test_gguf_initial_buffer_flush_holds_split_rehearsal_name(monkeypatch):
|
|
"""The first flush out of BUFFERING (prose plus a trailing active-tool-name in
|
|
the first delta, ``[ARGS]{...}`` in the next) must apply the same trailing-name
|
|
hold the STREAMING branch uses. The first delta has spaces so it is not a
|
|
rehearsal prefix and falls to the initial flush, which previously emitted the
|
|
bare name before the call drained."""
|
|
|
|
first_stream = [
|
|
_sse({"content": "I will use web_search"}),
|
|
_sse({"content": '[ARGS]{"query":"cats"}'}),
|
|
_done(),
|
|
]
|
|
final_stream = [_sse({"content": "Found cats."}), _done()]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [first_stream, final_stream], payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
monkeypatch.setattr(
|
|
"core.inference.tools.execute_tool",
|
|
lambda name, arguments, **_k: (calls.append((name, arguments)) or "result"),
|
|
)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "search cats"}],
|
|
tools = [{"type": "function", "function": {"name": "web_search"}}],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
assert calls == [("web_search", {"query": "cats"})], calls
|
|
content_texts = [e.get("text", "") for e in events if e.get("type") == "content"]
|
|
assert all("web_search" not in t for t in content_texts), content_texts
|
|
assert all("[ARGS]" not in t for t in content_texts), content_texts
|
|
|
|
|
|
def test_gguf_rehearsal_name_after_prose_in_streaming_is_not_leaked(monkeypatch):
|
|
"""Finding 9: the BUFFERING guard only covers a rehearsal at the turn start.
|
|
When prose has already streamed (STREAMING state) and the model then emits the
|
|
tool name and ``[ARGS]{...}`` in later deltas, the bare name must still be held,
|
|
not flushed as visible content before the call drains."""
|
|
|
|
first_stream = [
|
|
_sse({"content": "Let me think. "}),
|
|
_sse({"content": "I will search "}),
|
|
_sse({"content": "web_search"}),
|
|
_sse({"content": '[ARGS]{"query":"cats"}'}),
|
|
_done(),
|
|
]
|
|
final_stream = [_sse({"content": "Found cats."}), _done()]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [first_stream, final_stream], payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
monkeypatch.setattr(
|
|
"core.inference.tools.execute_tool",
|
|
lambda name, arguments, **_k: (calls.append((name, arguments)) or "result"),
|
|
)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "search cats"}],
|
|
tools = [{"type": "function", "function": {"name": "web_search"}}],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
assert calls == [("web_search", {"query": "cats"})], calls
|
|
content_texts = [e.get("text", "") for e in events if e.get("type") == "content"]
|
|
assert all("web_search" not in t for t in content_texts), content_texts
|
|
|
|
|
|
def test_gguf_plain_answer_ending_with_tool_name_word_is_preserved(monkeypatch):
|
|
"""End-of-stream flush: a plain answer that ENDS on a tool-name word with no
|
|
``[ARGS]`` following is real prose and must not be dropped by the streaming
|
|
rehearsal hold."""
|
|
|
|
first_stream = [
|
|
_sse({"content": "I think "}),
|
|
_sse({"content": "you should "}),
|
|
_sse({"content": "web_search"}),
|
|
_done(),
|
|
]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [first_stream], payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
monkeypatch.setattr(
|
|
"core.inference.tools.execute_tool",
|
|
lambda name, arguments, **_k: (calls.append((name, arguments)) or "result"),
|
|
)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "advise"}],
|
|
tools = [{"type": "function", "function": {"name": "web_search"}}],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
assert calls == [], calls
|
|
content_texts = [e.get("text", "") for e in events if e.get("type") == "content"]
|
|
assert any(t.rstrip().endswith("web_search") for t in content_texts), content_texts
|
|
|
|
|
|
def test_gguf_long_tool_name_split_rehearsal_is_not_capped_and_executes(monkeypatch):
|
|
"""Finding 11: a realistic MCP name longer than the 32-char buffer cap split as
|
|
NAME then [ARGS]{...} must still be held (a rehearsal prefix is self-bounding),
|
|
so the name does not leak and the call executes."""
|
|
name = "mcp__github__create_pull_request"
|
|
assert len(name) >= 32, len(name)
|
|
|
|
first_stream = [
|
|
_sse({"content": name}),
|
|
_sse({"content": '[ARGS]{"x":1}'}),
|
|
_done(),
|
|
]
|
|
final_stream = [_sse({"content": "done"}), _done()]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [first_stream, final_stream], payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
monkeypatch.setattr(
|
|
"core.inference.tools.execute_tool",
|
|
lambda n, a, **_k: (calls.append((n, a)) or "result"),
|
|
)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "go"}],
|
|
tools = [{"type": "function", "function": {"name": name}}],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
assert calls == [(name, {"x": 1})], calls
|
|
content_texts = [e.get("text", "") for e in events if e.get("type") == "content"]
|
|
assert not any(name in t for t in content_texts), content_texts
|
|
|
|
|
|
def test_gguf_streaming_keeps_bare_args_before_think_block(monkeypatch):
|
|
"""F4: the GGUF streaming strip must run its open-ended ``[ARGS]`` tail cleanup
|
|
only on the LAST segment. A bare ``foo[ARGS]`` (no JSON body, ``foo`` not a tool)
|
|
before a <think> block is prose, not a truncated call, so the final visible text
|
|
must keep it verbatim instead of dropping ``foo[ARGS]`` and corrupting the
|
|
sentence."""
|
|
|
|
first_stream = [
|
|
_sse({"content": "Please pass foo[ARGS] "}),
|
|
_sse({"content": "<think>pause</think> "}),
|
|
_sse({"content": "to the template."}),
|
|
_done(),
|
|
]
|
|
backend = _make_backend(monkeypatch, [first_stream], [])
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
monkeypatch.setattr(
|
|
"core.inference.tools.execute_tool",
|
|
lambda name, arguments, **_k: (calls.append((name, arguments)) or "result"),
|
|
)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "x"}],
|
|
tools = [{"type": "function", "function": {"name": "web_search"}}],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
assert calls == [], calls
|
|
content_texts = [e.get("text", "") for e in events if e.get("type") == "content"]
|
|
assert content_texts, events
|
|
assert content_texts[-1] == "Please pass foo[ARGS] <think>pause</think> to the template."
|
|
|
|
|
|
def test_gguf_inactive_name_args_in_prose_is_not_drained(monkeypatch):
|
|
"""BUG A: an inactive-name ``foo[ARGS]{...}`` in a prose answer must not be treated
|
|
as a tool call. The BUFFERING and end-of-stream safety-net ``[ARGS]`` checks gate on
|
|
active tool names (like the safetensors loop and the mid-stream path), so ``foo``
|
|
(``web_search`` is the only enabled tool) is neither drained/parsed into a disabled
|
|
no-op nor forced into another generation turn."""
|
|
first_stream = [
|
|
_sse({"content": 'foo[ARGS]{"x":1} is just syntax.'}),
|
|
_done(),
|
|
]
|
|
backend = _make_backend(monkeypatch, [first_stream], [])
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
monkeypatch.setattr(
|
|
"core.inference.tools.execute_tool",
|
|
lambda name, arguments, **_k: (calls.append((name, arguments)) or "result"),
|
|
)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "x"}],
|
|
tools = [{"type": "function", "function": {"name": "web_search"}}],
|
|
max_tool_iterations = 2,
|
|
)
|
|
)
|
|
|
|
# No tool executed for the inactive name; a spurious no-op re-prompt would exhaust the
|
|
# single supplied stream and error.
|
|
assert calls == [], calls
|
|
assert not any(e.get("type") in ("tool_start", "tool_end") for e in events), events
|
|
content_texts = [e.get("text", "") for e in events if e.get("type") == "content"]
|
|
# The inactive ``foo[ARGS]{...}`` is prose: the name-gated strip keeps the whole sentence.
|
|
assert any('foo[ARGS]{"x":1} is just syntax.' in t for t in content_texts), content_texts
|
|
|
|
|
|
def test_gguf_inactive_rehearsal_before_active_call_executes_and_keeps_prose(monkeypatch):
|
|
"""BUG X (#5704): an inactive ``foo[ARGS]{...}`` before a real ``web_search[ARGS]{...}``
|
|
in one delta must NOT swallow the real call; web_search executes while the inactive
|
|
rehearsal stays visible as prose."""
|
|
first_stream = [
|
|
_sse({"content": 'foo[ARGS]{"a":1} web_search[ARGS]{"query":"cats"}'}),
|
|
_done(),
|
|
]
|
|
final_stream = [_sse({"content": "Found cats."}), _done()]
|
|
backend = _make_backend(monkeypatch, [first_stream, final_stream], [])
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
monkeypatch.setattr(
|
|
"core.inference.tools.execute_tool",
|
|
lambda name, arguments, **_k: (calls.append((name, arguments)) or "result"),
|
|
)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "search cats"}],
|
|
tools = [{"type": "function", "function": {"name": "web_search"}}],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
# The real call runs; ``foo`` is not executed as a phantom disabled call.
|
|
assert calls == [("web_search", {"query": "cats"})], calls
|
|
content_texts = [e.get("text", "") for e in events if e.get("type") == "content"]
|
|
# The inactive rehearsal is preserved as prose; the active one is stripped.
|
|
assert any('foo[ARGS]{"a":1}' in t for t in content_texts), content_texts
|
|
assert all("web_search[ARGS]" not in t for t in content_texts), content_texts
|
|
|
|
|
|
def test_gguf_rehearsal_detection_recognises_spent_one_shot_with_original_tools():
|
|
# Rehearsal detection is fed the ORIGINAL tool list, so a spent one-shot's re-emitted
|
|
# repeat is still detected (matching the strip gate) instead of blanking the turn.
|
|
from core.inference.llama_cpp import _gguf_has_genuine_tool_signal
|
|
from core.inference.tool_call_parser import TOOL_XML_SIGNALS
|
|
|
|
repeat = 'render_html[ARGS]{"code":"<html>x</html>"}'
|
|
active_only = [{"type": "function", "function": {"name": "web_search"}}]
|
|
original = active_only + [{"type": "function", "function": {"name": "render_html"}}]
|
|
assert not _gguf_has_genuine_tool_signal(repeat, TOOL_XML_SIGNALS, active_only)
|
|
assert _gguf_has_genuine_tool_signal(repeat, TOOL_XML_SIGNALS, original)
|
|
|
|
|
|
def test_gguf_rehearsal_prefix_and_tail_hold_recognise_spent_one_shot():
|
|
# The BUFFERING prefix check and STREAMING/flush tail-holds use the ORIGINAL tool list,
|
|
# so a spent one-shot's split repeat is held rather than leaked as visible text.
|
|
from core.inference.llama_cpp import _held_rehearsal_tail_len, _is_rehearsal_prefix
|
|
|
|
active_only = [{"type": "function", "function": {"name": "web_search"}}]
|
|
original = active_only + [{"type": "function", "function": {"name": "render_html"}}]
|
|
assert not _is_rehearsal_prefix("render_html", active_only)
|
|
assert _is_rehearsal_prefix("render_html", original)
|
|
assert _held_rehearsal_tail_len("answer render_html", active_only) == 0
|
|
assert _held_rehearsal_tail_len("answer render_html", original) == len("render_html")
|
|
|
|
|
|
def test_gguf_oversized_bare_json_not_leaked_and_executes(monkeypatch):
|
|
"""An oversized bare-JSON call drains rather than streams, and still executes via the safety net."""
|
|
|
|
cap = 16384
|
|
big = "A" * (cap + 5000)
|
|
full = '{"name":"python","parameters":{"code":"' + big + '"}}'
|
|
first_stream = [_sse({"content": full[i : i + 2000]}) for i in range(0, len(full), 2000)]
|
|
first_stream.append(_done())
|
|
final_stream = [_sse({"content": "done"}), _done()]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [first_stream, final_stream], payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
monkeypatch.setattr(
|
|
"core.inference.tools.execute_tool",
|
|
lambda name, arguments, **_k: (calls.append((name, arguments)) or "OK"),
|
|
)
|
|
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "run"}],
|
|
tools = [{"type": "function", "function": {"name": "python"}}],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
content_texts = [e.get("text", "") for e in events if e.get("type") == "content"]
|
|
assert not any(t.lstrip().startswith('{"name') for t in content_texts), content_texts[:1]
|
|
assert calls and calls[0][0] == "python"
|
|
assert len(calls[0][1].get("code", "")) > cap
|
|
|
|
|
|
def test_gguf_bare_json_call_not_replayed_in_next_turn_content(monkeypatch):
|
|
"""After a bare-JSON call executes, the kept assistant message must not carry the raw call as content."""
|
|
|
|
import copy
|
|
|
|
first_stream = [
|
|
_sse({"content": '{"name":"web_search","parameters":{"query":"cats"}}'}),
|
|
_done(),
|
|
]
|
|
final_stream = [_sse({"content": "Found."}), _done()]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [first_stream, final_stream], payloads)
|
|
|
|
monkeypatch.setattr("core.inference.tools.execute_tool", lambda *_a, **_k: "RESULT")
|
|
|
|
list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "cats"}],
|
|
tools = [{"type": "function", "function": {"name": "web_search"}}],
|
|
max_tool_iterations = 2,
|
|
)
|
|
)
|
|
|
|
assert len(payloads) >= 2
|
|
asst = [m for m in payloads[1]["messages"] if m.get("role") == "assistant"]
|
|
assert asst and not any('"name"' in (m.get("content") or "") for m in asst), asst
|
|
|
|
|
|
def test_gguf_textual_fallback_caps_distinct_tool_calls_per_turn(monkeypatch):
|
|
"""A single textual-fallback turn that parses many DISTINCT tool calls must be
|
|
capped at _MAX_TOOL_CALLS_PER_TURN (structured delta.tool_calls are grammar
|
|
bounded by llama-server; text parsed from content is not). Mirrors the
|
|
safetensors loop so one runaway turn cannot fan out into dozens of executions."""
|
|
from core.inference.llama_cpp import _MAX_TOOL_CALLS_PER_TURN
|
|
|
|
n = _MAX_TOOL_CALLS_PER_TURN + 4
|
|
blocks = "".join(
|
|
'<tool_call>{"name":"t%d","arguments":{"i":%d}}</tool_call>' % (i, i) for i in range(n)
|
|
)
|
|
first_stream = [_sse({"content": blocks}), _done()]
|
|
final_stream = [_sse({"content": "done"}), _done()]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [first_stream, final_stream], payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
monkeypatch.setattr(
|
|
"core.inference.tools.execute_tool",
|
|
lambda name, arguments, **_k: (calls.append((name, arguments)) or "OK"),
|
|
)
|
|
|
|
list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "go"}],
|
|
tools = [{"type": "function", "function": {"name": f"t{i}"}} for i in range(n)],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
assert len(calls) == _MAX_TOOL_CALLS_PER_TURN, [c[0] for c in calls]
|
|
# The cap keeps the first calls in order (no reordering / drop of leading ones).
|
|
assert [c[0] for c in calls] == [f"t{i}" for i in range(_MAX_TOOL_CALLS_PER_TURN)]
|
|
|
|
|
|
def test_gguf_textual_fallback_collapses_duplicate_tool_calls(monkeypatch):
|
|
"""Exact-duplicate textual calls in one turn collapse to a single execution."""
|
|
blocks = '<tool_call>{"name":"web_search","arguments":{"query":"cats"}}</tool_call>' * 5
|
|
first_stream = [_sse({"content": blocks}), _done()]
|
|
final_stream = [_sse({"content": "done"}), _done()]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, [first_stream, final_stream], payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
monkeypatch.setattr(
|
|
"core.inference.tools.execute_tool",
|
|
lambda name, arguments, **_k: (calls.append((name, arguments)) or "OK"),
|
|
)
|
|
|
|
list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "cats"}],
|
|
tools = [{"type": "function", "function": {"name": "web_search"}}],
|
|
max_tool_iterations = 1,
|
|
)
|
|
)
|
|
|
|
assert len(calls) == 1, [c[0] for c in calls]
|
|
|
|
|
|
def test_gguf_drain_truncated_enabled_name_json_preserved_when_auto_heal_disabled(monkeypatch):
|
|
"""Auto-Heal OFF keeps a truncated enabled-name fragment visible; ON suppresses it (strip gated on auto_heal_tool_calls)."""
|
|
|
|
trunc = '{"name":"web_search","parameters":{"query":"weather'
|
|
|
|
def _run(auto_heal):
|
|
stream = [_sse({"content": trunc}), _done()]
|
|
backend = _make_backend(monkeypatch, [stream], [])
|
|
calls: list[tuple[str, dict]] = []
|
|
monkeypatch.setattr(
|
|
"core.inference.tools.execute_tool",
|
|
lambda name, arguments, **_k: (calls.append((name, arguments)) or "result"),
|
|
)
|
|
events = list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "x"}],
|
|
tools = [{"type": "function", "function": {"name": "web_search"}}],
|
|
max_tool_iterations = 1,
|
|
auto_heal_tool_calls = auto_heal,
|
|
)
|
|
)
|
|
contents = "".join(e.get("text", "") for e in events if e.get("type") == "content")
|
|
return calls, contents
|
|
|
|
calls_off, contents_off = _run(False)
|
|
assert calls_off == [], calls_off
|
|
assert "web_search" in contents_off, contents_off
|
|
|
|
calls_on, contents_on = _run(True)
|
|
assert calls_on == [], calls_on
|
|
assert "web_search" not in contents_on, contents_on
|
|
|
|
|
|
def test_gguf_valid_tool_calls_respect_max_tool_iterations(monkeypatch):
|
|
"""Re-prompt slots must not extend the tool budget: stop after ``max_tool_iterations`` executed rounds."""
|
|
# More tool-call streams than the budget: if re-prompt slots leaked into the budget (the bug) the
|
|
# loop would run 2+3=5 rounds; honouring it stops after 2, then a tool-less final-answer pass.
|
|
streams = [
|
|
_structured_tool_call("web_search", {"query": f"q{i}"}, f"call_{i}") for i in range(6)
|
|
]
|
|
payloads: list[dict] = []
|
|
backend = _make_backend(monkeypatch, streams, payloads)
|
|
|
|
calls: list[tuple[str, dict]] = []
|
|
monkeypatch.setattr(
|
|
"core.inference.tools.execute_tool",
|
|
lambda name, arguments, **_k: (calls.append((name, arguments)) or "result"),
|
|
)
|
|
|
|
list(
|
|
backend.generate_chat_completion_with_tools(
|
|
messages = [{"role": "user", "content": "search repeatedly"}],
|
|
tools = [{"type": "function", "function": {"name": "web_search"}}],
|
|
max_tool_iterations = 2,
|
|
)
|
|
)
|
|
|
|
# Exactly two executed tool rounds, then one final-answer pass.
|
|
assert len(calls) == 2, calls
|
|
assert len(payloads) == 3, len(payloads)
|
|
# The final pass is the budget-exhausted nudge and carries no tools.
|
|
assert _tool_names(payloads[2]) == [], _tool_names(payloads[2])
|
|
assert any(
|
|
m.get("role") == "user" and "used all available tool calls" in m.get("content", "")
|
|
for m in payloads[2]["messages"]
|
|
), payloads[2]["messages"]
|