unsloth/studio/backend/tests/test_openai_tool_passthrough.py
Daniel Han c608649552
feat(studio): run chats in parallel in the Chat tab (#7455)
* feat(studio): run chats in parallel in the Chat tab

New Chat used to cancel whatever the current conversation was generating.
It now leaves it running, like switching to the Train or Export tab: the
sidebar shows which chats are still going, and Stop is per conversation.

Plain `unsloth studio` launched llama-server with one decode slot, so the
admission queue serialised every chat regardless of what the UI did. Both
entry points now default to the same slot count as `unsloth studio run`.

A model swap still ends every running chat, since they all decode on one
llama-server. /load and /unload now refuse with 409 and name those chats
unless the caller passes force_cancel_active, and the UI asks first.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* fix(studio): scope the composer tool badge to its own conversation

The green "Running Python: ..." badge above the composer read a single
global store value, so one chat's tool call showed above every other
chat's composer, including a brand-new empty one. Its elapsed counter
also restarted at 0 on every thread switch, and a run ending anywhere
cleared the badge everywhere.

Key the status by thread and store the moment it started, so each
conversation shows only its own tool call and the counter resumes rather
than restarts. Also adds a test that every conversation gets its own
tool sandbox directory, which parallel tool calls depend on.

* Fix stalled tool calls while awaiting approval for PR #7455

Three problems, all from the approval prompt behaving as though only one
chat could ever run.

Arguments were not streamed for a gated call, so the chat stayed blank for
as long as the model took to write the payload, which for a large file is
minutes. Nothing runs before the decision either way, and the code is what
is being approved, so python and terminal now stream their card while
gated. render_html stays suppressed: its card renders the payload.

The status read "Running ..." with a climbing timer while the call had not
started. It now reports that it is waiting for approval, then switches to
running once allowed.

The admission lease was held across the wait, so four unanswered prompts
held all four decode slots and no other chat could start while llama-server
sat idle. A parked run keeps its lease but no longer counts against
capacity.

Measured with four prompts left open: every gated call streamed its code,
none reported running, and a fresh chat answered in 0.4s where it
previously waited 290s and never did.

* Fix duplicated and truncated tool cards for PR #7455

A gated tool call rendered two cards: the provisional one that streams the
arguments, plus a second one keyed by the approval id. Only the second ever
got its tool_end, so the first spun "Running" for the rest of the chat.
Reuse the open part when the approval prompt arrives.

The terminal card also showed nothing but a 60-char trigger label, so a long
heredoc read as no progress at all. It now renders the command the same way
the Python card renders its script, and neither is capped at 10k chars.

Both cells moved inside the collapsible, so one chevron hides the code with
the output and Copy / Download exist only while the card is open. A card
parked on the prompt says so instead of counting up "Running".

* Fix review findings on the parallel-chat gate for PR #7455

Backend:
- /unload rechecks active generations under the lifecycle gate, like /load,
  and lets its 409 through the catch-all instead of rewriting it as a 500.
- /load gates only once _load_model_impl has decided this is a real reload,
  so an Apply on the already-loaded model no longer refuses, and the retry it
  asks for no longer cancels every chat before returning already_loaded.
- The direct /v1/responses stream registers in the cancel registry, so a
  non-forced unload can no longer tear llama-server down under it.
- run_server defaults to the same slot count as the CLI. colab.py calls it
  without the argument, so Colab was still serialising every chat.

Frontend:
- Cancelling a backgrounded chat aborts its own request rather than only
  posting a cancel id, which is the only thing that ends an external-provider
  or audio run.
- The model-swap dialog counts local runs only, and falls back to the backend
  when this tab's map is empty, so a reload or a second tab still gets asked.
- Context usage and the diffusion canvas are scoped to the chat that produced
  them; a compare row reads activity from its member threads.

Tests:
- The extracted-source cancel harnesses supply the active-generations module,
  which the tracked-cancel class now depends on.

* Fix the swap confirmation scope and cancel timing for PR #7455

A forced load cancelled every chat before the model identifier, GPU selection,
training coexistence and download checks had run, so a load that then failed
those checks stopped the chats and replaced nothing. The refusal still happens
early, but the destructive cancel now sits immediately before the teardown it
is paying for, and rechecks under the gate like /unload does.

The swap dialog only reconciled with the backend when this tab looked idle, so
one local chat was enough to hide a second tab's runs. Confirming then sent
force_cancel_active, which cancels every backend run, including the ones the
dialog never mentioned. The backend snapshot is now merged in every time, so
the dialog names what will actually stop. External-provider runs are never
registered there, so the union stays local-only.

Also drops the active-generations docstring claim about restoring sidebar
spinners, which nothing consumes.

* Defer destructive cancels and track every local stream for PR #7455

/unload cancelled the running chats before it had resolved that it unloads
anything. A stale model_path, which a second tab produces routinely, killed
every chat and then no-opped, leaving the resident model up. It now refuses
early and cancels only at each teardown, matching /load.

The swap dialog also stopped every chat locally the moment the user confirmed,
which threw away the two-phase backend behaviour: a load that then failed
identifier resolution, GPU validation or the training guard had already
truncated the replies. The backend now owns the cancel.

Three local streams decoded on llama-server without registering, so a
non-forced unload counted zero generations and tore the server down mid
response: /v1/completions streaming, and the plain and server-tool Anthropic
streams, the first of which is the default /v1/messages path. Note this makes
a non-forced load return 409 during those runs rather than draining quietly,
the same trade the /v1/responses fix made.

The safetensors tool loop still announced a gated call as running while it
waited on a human; only the GGUF loop had been fixed. A source-level parity
test now pins both.

Also drops stopAllChatThreads, which has no callers left.

* Studio: close three load/unload gate races found in review

Re-check the in-flight load guard after the stop-running-chats confirm.
The confirm always GETs active-generations before its zero-running
early-out, so the guard no longer sits atomically ahead of the
reservation and two picks in that window both reached performLoad over
the same refs. ejectModel had the same shape and gets the same re-check.

Reject a sidecar swap immediately before the forced cancel in both load
branches. The previous check was back at the top of preflight, so an
install reserving during identifier resolution, the tier probe, the
training guard or the download check made the post-drain recheck 409 a
load whose chats had already been stopped.

Enter the Anthropic passthrough's cancel tracker inside its body
generator. It was entered eagerly and returned through
_sse_streaming_response, which sets no unstarted_cleanup, so a response
whose body never started left the run registered forever and 409'd every
later non-forced load and unload.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Trim comments across the files this PR touches

Tightens the comments and doc blocks in the backend, CLI, tests and frontend
files changed by this PR: collapses multi-line explanations to a single line
where they still read clearly, and drops the ones the code already says.

No code changes, verified by an AST comparison against the previous commit.

* Studio: defer the destructive cancel and close two gate gaps

Move the forced cancel behind every check that can still reject a swap.
The drain now runs first with the runs it is about to cancel discounted,
so it waits only for inference the cancel cannot end, then the sidecar
check decides, then the cancel fires, then a second drain lets those runs
unwind before teardown. A sidecar install reserving during the drain no
longer 409s a load whose chats have already been stopped.

Track the non-streaming /v1/completions proxy. It was the last local
decode path missing from active_generations, so an unload, which runs no
drain, tore llama-server down under it and force_cancel_active could not
signal it. It now uses the same tracked cancel event and dedicated client
as the OpenAI pass-through.

Skip the client's preliminary unload while chats are generating and let
/load evict at its own post-preflight point instead. Forwarding
force_cancel_active there truncated replies before identifier
resolution, the GPU and training guards and the download check had run.

Keep per-thread context usage so returning to a chat whose background run
finished restores its bar instead of leaving it blank until the next turn.

Make the running-flag clear run-specific. Every run without a resolved
thread id shares the "__default" key, so concurrent compare panes could
clear each other's flag and strand a live stop handle.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Studio: register the embeddings proxy with the swap gate

/v1/embeddings proxied straight through the pooled client with no tracked
cancel event, so it never appeared in active_generations. /unload runs no
idle drain, so a concurrent non-forced unload counted zero generations and
killed llama-server mid-request, and force_cancel_active had no event to
signal. Mirrors the completions proxy: tracked event, dedicated unpooled
client closed by a cancel/disconnect watcher, unregister in a nested
finally so a close failure cannot leave a phantom generation behind.

* Trim comments on the newest changes in this PR

Comments only, no code changes: shorten the ones added by the load-gate
ordering, embeddings and per-thread usage work down to the same density as
the rest of the diff.

* Studio: register the legacy generate stream with the swap gate

/generate/stream built a cancel event but never entered the tracker, so it
was invisible to active_generations. Being in the keep-warm middleware's
inference suffixes only covers /load, which drains; /unload does not, so a
non-forced unload passed the 409 gate and then blocked on the standard
backend's generation lock, and a forced swap had no event to signal.
Registered inside the body generator under a nested finally so a teardown
failure cannot skip the unregister.

The AST contract test asserted the cleanup finally by overwriting its flag
per Try node, so a nested try made the last one win. Accumulate instead,
which is what the existence claim meant.

* Studio: three more swap-gate gaps found in review

Register /audio/generate with the gate. TTS holds the model for the whole
request and /unload runs no drain, so unregistered a non-forced swap counted
zero generations and tore the model down mid-generation; the orchestrator
path only waits 15s for the generation lock, which real TTS exceeds. No
cancel keys: no backend takes a cancel_event for audio, so the event has no
observer and a forced swap still cannot interrupt audio already in flight.

Thread the tracked cancel event into the /v1/responses admission wait. It
was the only admission caller passing None, so a queued run could not be
reached by cancel_all() and a plain /inference/cancel could not stop it at
all. Same omission fixed at the upstream send there and on /v1/completions.

Let an unforced unload of a stale model path reach the no-op check. Before
this PR that request returned 200 and did nothing; the new gate refused it
with 409 for a request that reaches no teardown branch. Gate both refusal
passes on the disjunction of the route's own teardown conditions, including
not is_loaded, so a mid-load GGUF still refuses.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Studio: register the remaining non-streaming decode paths

stream defaults to false on all three of these, so they are the ordinary
shape of their routes, and each holds a local backend for the whole
request. /unload runs no idle drain, so with no registry entry a non-forced
swap counted zero generations and tore the backend down mid-request instead
of returning 409, and a forced one had no event to signal.

Non-streaming /v1/messages: all three helpers ran with an empty registry,
since only the streaming siblings were tracked. Registered at the call site
because the pass-through takes no cancel_event of its own, and with no
cancel keys, matching those siblings.

Non-streaming standard chat and audio-input chat: the trackers in this route
sit inside their `if payload.stream:` arms, so neither else branch was
covered. The GGUF sibling already registers its own non-streaming branch.

Each exit is in a finally on the branch's existing try, so the except arms
are covered too: a leaked entry 409s every later swap until restart.

* Studio: tighten the swap-gate comments

Comment-only pass over the newest swap-gate registrations: collapse the multi-line rationales in /unload, the legacy generate stream, audio generation and the non-streaming chat branches, and the matching test preambles, to the shortest form that still carries the reason. No code changes.

* Studio: stop the reselect dialog promising a stop that never happens

Picking an external provider leaves the local model resident and stops the
status poll mirroring it, so reselecting that model showed the stop-chats
dialog, and /load then answered already_loaded ahead of its cancel hook.
Confirmed with the live backend: the same pick with force_cancel_active set
still returned already_loaded and the chat kept streaming. Not stopping
those chats is right, since the load never interrupts them, so remove the
prompt rather than honour it. Blanket-skipping is unsafe, because the same
id and variant with one sampling setting changed is a real reload and 409s,
so the branch only fires when a status fetch confirms the resident
checkpoint and variant match, and then adopts it without calling /load.

Redact native model paths from the active-generations response. Registering
/generate/stream recorded backend.active_model_name verbatim, which is an
absolute path for a native local model, and this route is the only place
that serialises it. Redacting at the response covers every tracker rather
than the one that surfaced it.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Studio: keep hydrated context usage in the per-thread map

The history loader restores a saved conversation's usage through
setContextUsage only, and it runs once per mount, so switching away and
back left the bar blank for a hydrated chat even after the per-thread map
landed. setContextUsage now writes the value through to the visible
thread's own entry and clears that entry when passed null, which covers
both hydration call sites and any future writer.

* Studio: unblock load cancellation and share unresolved thread keys

Run the two stop-loading fast paths ahead of the unload route's pre-gate
refusal. _unload_may_evict returns True for exactly the model being
cancelled, so the refusal was blocking the branch that cancels a load which
has replaced nothing and can interrupt no chat. The client made that
unrecoverable: cancelLoading sends the unload without force, drops the
result, and its abort never reaches /load, which takes no signal, so the
load ran on and could later cancel those chats and swap the model. Nothing
else is exempted; an unload that would tear down a serving model matches
neither fast path and still 409s. The comment claiming the client lets that
409 surface is corrected, since it discards it.

Hold every owner behind a shared thread key. Runs with no resolved thread id
share "__default" (concurrent compare panes, since startCompare clears
activeThreadId), so a single owner slot let a second run replace the first's
token and then delete the shared entry while it was still generating, and
the server-cancel map lost the older handle the same way. Both now hold a
list, the running and local flags survive until the last owner clears, and
stopChatThread stops every handle under the key.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Studio: carry a confirmed swap into the sidecar install, key restored usage by thread

Picking a model that needs a newer transformers while chats generate raised the
"stop N chats" prompt, but the answer never reached the install that runs before
the load: /install-latest-transformers refused on those same chats and took no
force flag, so Retry hit the same 409 and nothing in the flow stopped them.

Carry force_cancel_active through the consent dialog into the installer. Only
the pre-gate fast path is skipped: the recheck under the lifecycle gate still
has to pass, so an unconfirmed caller is refused as before. The cancel runs last
inside the gate, after every check that can still reject the install, and the
drain behind it is bounded since it holds the gate and the sidecar reservation.

Also key restored context usage by the thread the loader read. history.load()
captures remoteId before two awaited round trips, so a switch inside that window
filed one thread's usage under another and setActiveThreadId kept re-applying it.

Preserve sibling owners when a run key is cleared without an owner: the image
rejection gate now uses its own token, and the reducer leaves owned runs alone.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Studio: bound the post-cancel drains, and make cancellation reach the paths that ignored it

A forced swap cancels the chats it interrupts, then waits for them to unwind.
That wait had no deadline while holding the lifecycle gate, and TTS on the
subprocess backend observes no cancel event at all, so one audio generation
could pin every load, unload and new request for its whole duration. Bound both
post-cancel drains. Pre-cancel drains stay unbounded: the swap can still be
refused there, so shortening them would weaken what they protect.

/unload had the opposite problem and no drain at all, cancelling and tearing
down on the next line, which turned a clean stream end into a dropped
connection. Give it the same bounded wait, gated on the cancel having cancelled
something so an idle Eject pays nothing.

Make the cancel actually land where it can. GGUF TTS now takes a cancel_event
and a watcher closes its client to break the blocking POST. The Anthropic
non-streaming pass-through did the same thing the completions and embeddings
paths used to: register with the gate, then run both POSTs on the pooled client
that cannot be closed. It now uses a per-request client like they do.

Also: park and unpark the admission queue the reservation actually holds, since
queues are keyed by base_url and a reload mints a new port; key tool output by
remoteId on both sides, so the first turn of a New Chat stops writing under one
key and reading another; and give tool status a run owner, so a finishing run
cannot blank the badge a concurrent one is still showing.

Clamp --parallel to 1 on a llama-server without --kv-unified. The new default of
4 would otherwise split -c four ways on such a build, quartering the context
window for a feature it cannot serve.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Studio: scope a chat's Stop to its own generation, and clear the way before a confirmed install

Safetensors generation is serialized on _gen_lock and the worker has a single
cancel event, so a chat still queued on that lock owns no generation. Its Stop
handler called reset_generation_state() anyway, which set the shared event and
ended whichever conversation was actually running. Parallel chats is what makes
that reachable.

_generate_inner now records its cancel_event as the current holder once it takes
the lock, and reset_generation_state drops a reset from anyone else. Every route
call site passes its own request event. A reset with no event stays global, so
unload and model switch cannot leave a generation alive, and a reset while
nothing runs still resets, so an error path before generation is not a no-op.
The other two backends take the argument too, or the standard one raises
TypeError on every cancel.

The sidecar install had the mirror of the /load ordering problem: it cancelled
the chats first and drained second, so an unrelated counted request the cancel
cannot reach (a count_tokens, say) was still there for the recheck, which then
refused an install that had already stopped every chat for nothing. Drain the
unreachable remainder first, discounting the registered chats, then cancel.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Studio: close the windows the previous round's fixes left open

Three follow-ups, two of them holes in the fixes just before them.

The worker claim went in after _send_cmd, so the command was already running
unclaimed and a queued chat's Stop in that window still reset it. Claim first,
with the send inside the same try, so a failed send releases it too.

Tool status kept one entry per key with an owner. That stops a foreign clear but
not an overwrite: under the shared unresolved-thread key the second run replaced
the first's entry, and its own clear then removed the only one while the first
tool was still running. Keep per-run entries and render the newest.

/unload gated its drain on having cancelled something, so a request that passed
the keep-warm middleware but had not reached its tracker yet was invisible to it
and the teardown landed on an already-admitted request. Drain on the middleware
count instead, which covers that window as well as the cancelled runs, then
re-cancel whatever registered while waiting. Bounded, not a refusal: an unload is
deliberate, and on expiry it proceeds exactly as before.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Trim the parallel-chats comments to their reasons

Compress the multi-line rationales added by this branch into shorter forms and drop
restatements of the code below them. The reasons behind the drain bounds, the deferred
cancel, the per-request generation ownership and the thread-scoped tool and usage keys
are kept, just said in fewer lines.

* Studio: own the worker per generation, and make a resumed chat requeue for its slot

Ownership was a single lock holder, so dispatched runs (compare mode bypasses
_gen_lock by design) never claimed it and the guard fell straight through to the
global reset: a Stop on one of them ended its siblings. Track the generations
actually running instead, claimed before the send and released in the same
finally on both paths. A reset still proceeds when nothing is running, so an
error path ahead of generation is not swallowed.

park() hands the freed slot to a waiter, so a chat resuming from a tool approval
could take it back while that waiter was still decoding, putting two holders on
a one-slot server and sending the resumed tool loop past the admission limit.
unpark_async waits for room; the plain unpark stays for a holder tearing down,
which will not decode again.

Audio only observed its cancel event on a forced swap. An explicit Stop just
aborts the fetch, and this route has no cancel id, so llama-server ran on to the
request timeout after the chat reported it stopped. Watch the disconnect.

Also read tool status by remoteId, matching the key the adapter writes and the
fix already made for tool output, and stop an unresolved run from writing its
usage into whichever conversation the user moved to.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Studio: let only the generation the worker is running speak for it, and hold a slot for a resuming chat

The ownership list recorded admission, but the subprocess runs generations one
at a time, so a dispatched request queued behind another counted as an owner and
its Stop signalled the shared cancel event, ending the request that was actually
running. Keep admission for release bookkeeping and gate ownership on execution
instead, promoted when the worker first answers that request. Nothing executing
still permits a reset, so an error path ahead of generation is not swallowed.

The worker has one cancel event and no per-request cancellation, so this decides
who may pull the lever rather than making the lever per-request.

A resuming chat also polled for a slot it could never see: release() grants to
the next waiter under the same lock, so later arrivals overtook an approved chat
indefinitely. A pending unpark now reserves the next slot and they queue behind
it.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Studio: cover the prefill window, and keep a first turn's tool output readable

Gating worker ownership on execution left the interval between the send and the
first response uncovered: nothing is executing then, and the empty case admitted
anyone, so a queued chat's Stop still ended the one in prefill. Split the empty
case. Nothing claimed at all still permits a reset, so an error path ahead of
generation is not swallowed; claimed but unanswered resolves to the oldest
claim, which is what a FIFO command queue is working on.

Putting both sides of the tool-output scope on remoteId left the first turn of a
New Chat writing under the unresolved scope for its whole life while the readers
recomputed the moment the autosave assigned an id, so the card blanked mid-run.
The readers now fall back to the unresolved scope, which only an unpersisted
first turn can occupy.

* Studio: order the parked approvals, and tie a worker claim to its enqueue

The reservation added for admission fairness was a bare count, so every approved
holder counted against every other: park two chats, approve both, and once the
last decoder released, nothing could ever satisfy the check again. That is a
deadlock where the problem it fixed was only unfairness. Make it a FIFO ticket
so a pending unpark blocks the ones behind it and no others.

_owns_worker reads claim order to decide which request the worker is prefilling,
which only holds if claiming and enqueuing cannot interleave. Hold one lock
across both on the dispatched and the locked path.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Studio: follow a first turn's run when its thread gets an id, and key the denoising canvas per chat

A run started before its thread existed filed every handle under "__default". Nothing
moved them once autosave assigned the real id, so the sidebar row showed no spinner and
Stop could not reach the generation, which kept holding a slot.

adoptDefaultThreadRun re-keys the run maps onto the real id from the thread adapter's
initialize(), where the id first exists; anything already filed under that id wins, since
that is a later run. The adapter captures its key once at run start, so it now resolves
the live key per use through runKeyForOwner, looking its own serverCancel up in the owner
map. Without that the migrated entries are stranded and the spinner never clears.

The denoising canvas was one global slot, so two diffusion chats overwrote each other and
the ownership tag then hid the visible preview until that thread emitted again. It is now
activeDiffusionCanvasByThreadId, written and cleared per thread, and the frame no longer
carries a threadId of its own. The bubble reads threadListItem.remoteId, dropping the dead
threadListItem.id arm: the writer tags unstable_threadId, which is exactly remoteId.

Two existing backend tests needed the same treatment. _bare_orchestrator skips __init__, so
it now sets the claim bookkeeping the worker ownership check reads. The Anthropic
passthrough gate test anchored on comment prose that a rewrap had broken; it anchors on the
code instead.

* Studio: hand the worker over cleanly between generations, and stop unresolved runs sharing each other's state

Worker ownership moved off the consumer and onto the dispatcher. Consumers read their
mailbox whenever they get around to it, so a request whose gen_done had been routed still
owned the worker while the next one ran, and a late Stop for it cancelled that one. The
dispatcher is the only place responses arrive in the order the worker produced them: it
now retires a request at its terminal response and promotes the next one, and answering a
request makes it the sole executor, since the subprocess runs one generation at a time.

reserve()'s immediate path ignored the unpark tickets that _grant_waiters_locked already
honours, so a request arriving between a slot freeing and an approved chat's next poll
took it, repeatedly. It applies the same reservation now.

Three places let concurrent first turns share state through the "__default" key. Nothing
links a run filed there to the id its thread later receives, so rather than guess, each
now declines when the key is ambiguous: adoption only re-keys a lone run, the composer
badge only claims a lone status, and the tool-output fallback only applies to a thread
that is still running. That leaves two concurrent first turns where they were before
adoption existed instead of handing one thread the other's handles.

A first turn's usage was never filed, because its key stayed null for the whole run while
autosave moved activeThreadId to the real id, so the context bar went blank after the
first reply. It resolves the adopted key like the cleanup handles do.

Cancelling a forced load left the UI with no model: the previous one stays resident until
/load's teardown, and the cancel path cleared the checkpoint without rolling back. It now
resyncs from the backend, which is right whether or not the load got that far.

The sidecar install drain is weighted 1:4 rather than halved, total unchanged. Only the
second half benefits from patience, and cutting it short refused installs whose chats had
already been stopped for nothing.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Studio: give a first turn its real thread id before the run starts

A first turn filed every run handle under a shared unresolved key because
assistant-ui binds unstable_threadId before the thread is persisted. Two of them
overlapping there is unresolvable afterwards, and the last round's migration could
only decline rather than guess, which left neither sidebar row showing its run.

The id is available earlier than I claimed. append() already tracks
threadListItem.initialize() by the user message id, and createPersistedRunAdapter
already awaits that promise before invoking the adapter, so the thread is persisted
by the time the run begins. It was only being discarded: the tracked promise resolved
to void. It now resolves to the assigned id, and the wrapper hands it to the adapter
when assistant-ui had none. An id that is already set is never replaced, since that
would move a running chat's handles out from under the row watching them. The
existing unresolved-key guards stay as a safety net but should no longer carry weight.

The sidebar counted running thread ids rather than rows, so one compare conversation
read as two chats. It folds ids into rows through the same threadIds the row spinner
uses, and still counts a running id that matches no row.

_TrackedCancel always registered kind="chat", so an embeddings or raw completions
request appeared in the model-swap prompt as an unnamed conversation and confirming
cancelled it while calling it a chat. The non-conversation routes now pass their own
kind, and the prompt says "requests" whenever the snapshot is not all chats.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Studio: withhold the shared worker cancel from a request the worker has left

Moving ownership to the dispatcher fixed reset_generation_state, but the token loop
signals the shared worker event directly and did not carry the same rule. A dispatched
consumer runs with mark_started off and can still be draining tokens buffered before
its gen_done was routed, so stopping it there ended whichever request the worker had
started next.

It now signals only when _owns_worker agrees, the same predicate reset_generation_state
uses. The local drain and return are unconditional, since those touch nothing but this
stream. The remaining _cancel_generation callers are deliberately global: subprocess
shutdown, the pre-load kill and unload_model.

* Studio: add the AGPL-3.0 header to the first-turn identity test

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Studio: stop the dispatcher and a _gen_lock stream fighting over the response queue

Nothing stopped the dispatcher starting under a _gen_lock generation, so once compare
was opened while an ordinary chat was still streaming, both consumed _resp_queue and
whichever response the dispatcher took without a mailbox was dropped, gen_done included.
That chat truncated or hung. This PR is what makes it reachable, since navigating into
compare no longer ends the chat behind it.

Delaying the dispatcher would serialise compare behind whatever chat happens to be
streaming, so the direct readers get a mailbox instead. _direct_reader returns a reader,
a cancel drain and a release, and files the mailbox under _direct_mailboxes rather than
_mailboxes, which means "compare requests are in flight" to the unload and distributed
paths and must not count an ordinary chat.

Both directions close. The dispatcher finds the direct reader's mailbox instead of
dropping. And this reader can already be blocked on the queue when a compare request's
dispatcher starts, so a response that is not ours goes to its own mailbox rather than
being consumed, which would have corrupted the chat and hung the pane. All three
_gen_lock readers use it, and the cancel drain goes through it too.

The sidebar's return target still picked a raw pane id while the count grouped by row,
and /chat addresses compare with `compare`, not `thread`. It resolves through the same
items now, so a running compare row returns to its pair.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Studio: keep worker ownership honest across audio, API traffic and a replaced worker

The audio-input send got a mailbox last round but stayed unclaimed, so a compare request
queued behind it looked like the oldest owner and stopping that queued request signalled
the shared event into the audio chat. It claims under the send lock and releases in the
finally, like _generate_inner.

Ownership is keyed on cancel-event identity with nothing tying it to a worker generation,
so a consumer still blocked on its mailbox when the process was replaced stayed recorded
as the executor, and a generation on the fresh worker could not be stopped.
_shutdown_subprocess clears that state once the process is confirmed dead, mailboxes
included: nothing routes to them again, and a stale one reads as compare activity to the
unload path. Not on the survived-SIGKILL path, which keeps its handle on purpose.

The four public /v1/messages trackers were registering as chats. The distinction is a
Studio thread, not the protocol, and those branches already say "No thread_id: public API
surface" while the Studio path passes payload.thread_id separately. They carry their own
kind now, so the swap prompt stops calling an external request a chat.

The swap confirmation still counted raw pane ids, so a compare conversation asked to stop
two chats and listed its title twice. It folds panes onto pairId and lowers the count by
what it collapsed, leaving a first turn the backend can count but not name.

Deep Research set runningByThreadId but registered no server-cancel handle, and that map
is how Stop, archive and delete reach a thread that is no longer active. Leaving the
outgoing thread running is this PR's doing, so the run was left unreachable while its
supervisor kept working against a conversation the user could delete.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Studio: tighten the parallel-chats comments

* Studio: replay a Deep Research stop that arrived before the run existed

The handle is registered before createResearchRun resolves because the thread can be
stopped while that request is in flight, but it had no id to act on and dropped the stop.
The supervisor then followed a run the user had already stopped, archived or deleted.

It latches instead: a stop with no id yet sets a flag, and the adapter replays it against
the id the moment creation returns rather than starting to follow.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Studio: fix worker ownership on a raced reroute, and the stop-chats prompt

Four review findings on the parallel-chats work, all reproduced first.

- _direct_reader hands a foreign response to its own mailbox, but skipped the
  ownership move the dispatcher makes. A _gen_lock reader already blocked on
  resp_queue can beat the compare dispatcher to that request's first response,
  and the compare consumer opts out of marking, so nothing promoted it: the
  direct request stayed the recorded executor, its late reset cancelled the
  compare generation, and the compare chat's own Stop was ignored.
- A chat stopped while queued on _gen_lock was still claimed and sent once the
  lock freed. Cancellation is only checked on a token, so a long prefill, or a
  generation reaching gen_done without one, occupied the worker after Stop.
  Same hole in the audio-input path, which shares the lock.
- The stop-chats prompt counted generation handles, not conversations. One chat
  holds several while a tool continuation registers its next leg before the
  previous unwinds, so it offered to stop two chats and listed one title.
- Ejecting a model confirms through that dialog, which told the user
  "Unloading the model reloads the model" and offered "Stop and reload".
  Confirming calls /unload and leaves nothing loaded.

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Studio: name the TTS run's thread so the stop prompt counts it once

The audio branch registers its run locally under the thread key but sent no
thread_id, so the backend tracker filed the same generation under no thread.
The stop-chats prompt then had a named local run and an unnamed backend one and,
since e8e7594 started adding unnamed entries to the named ones, counted a single
TTS chat as two requests. The backend already reads payload.thread_id, so
sending it lines both registries up on the same run.

* [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>
2026-07-28 04:40:38 -07:00

7227 lines
270 KiB
Python

# SPDX-License-Identifier: AGPL-3.0-only
# Copyright 2026-present the Unsloth AI Inc. team. All rights reserved.
"""Tests for the OpenAI /v1/chat/completions client-side tool pass-through."""
import os
import sys
import asyncio
import json
import threading
import time
from types import SimpleNamespace
_backend = os.path.join(os.path.dirname(__file__), "..")
sys.path.insert(0, _backend)
import httpx
import pytest
from fastapi import HTTPException
from pydantic import ValidationError
from models.inference import (
ChatCompletionRequest,
ChatMessage,
CompletionChoice,
CompletionMessage,
ResponsesRequest,
)
from core.inference.anthropic_compat import (
anthropic_tool_choice_to_openai,
)
from core.inference.api_monitor import ApiMonitor
from core.inference.llama_admission import (
ADMISSION_KEEPALIVE_INTERVAL_ENV,
ADMISSION_MAX_QUEUE_ENV,
ADMISSION_QUEUE_TIMEOUT_ENV,
LlamaAdmissionCancelled,
LlamaAdmissionConfig,
get_llama_admission_queue,
reset_llama_admission_queues,
)
from routes.inference import (
_aclose_stream_resources,
_build_chat_request,
_build_openai_passthrough_body,
_build_passthrough_payload,
_clamp_finish_reason,
_cmpl_stream_event_out,
_coalesce_consecutive_user_turns,
_drop_empty_assistant_sentinels,
_effective_max_tokens,
_effective_openai_max_tokens,
_effective_openai_max_tokens_from_values,
_extract_content_parts,
_friendly_error,
_friendly_upstream_error,
_merge_user_content,
_monitor_openai_chunk,
_monitor_openai_sse_event,
_normalize_openai_passthrough_sse_line,
_openai_compat_stream_stall_timeout,
_openai_llama_admission_capacity,
_openai_messages_for_gguf_chat,
_openai_passthrough_sse_line_terminal_state,
_openai_passthrough_upstream_headers,
_openai_passthrough_non_streaming,
_openai_passthrough_stream,
_responses_stream,
_openai_stream_error_sse,
_openai_stream_usage_chunk,
_openai_admission_wait_stream_chunks,
_wait_for_openai_admission_non_streaming,
_proxy_to_external_provider,
_SameTaskStreamingResponse,
_OPENAI_COMPAT_STREAM_STALL_TIMEOUT_ENV,
_set_or_prepend_system_message,
openai_completions,
openai_embeddings,
openai_chat_completions,
)
from state.tool_policy import reset_tool_policy, set_tool_policy
@pytest.fixture(autouse = True)
def _reset_admission_queues():
reset_llama_admission_queues()
yield
reset_llama_admission_queues()
def test_aclose_stream_resources_attempts_remaining_closes_after_cancel():
class Closeable:
def __init__(self, *, cancel = False):
self.cancel = cancel
self.closed = False
async def aclose(self):
self.closed = True
if self.cancel:
raise asyncio.CancelledError()
async def _run():
iterator = Closeable(cancel = True)
resp = Closeable()
client = Closeable()
with pytest.raises(asyncio.CancelledError):
await _aclose_stream_resources(iterator = iterator, resp = resp, client = client)
assert iterator.closed
assert resp.closed
assert client.closed
asyncio.run(_run())
class TestFriendlyUpstreamError:
def test_grammar_parse_failure_gets_actionable_message(self):
raw = '{"error":{"code":400,"message":"Failed to initialize samplers: failed to parse grammar","type":"invalid_request_error"}}'
msg = _friendly_upstream_error(raw)
assert "failed to parse grammar" not in msg # raw body is not surfaced verbatim
assert "tool-calling grammar" in msg and "Update Unsloth" in msg
def test_failed_to_initialize_samplers_alone_matches(self):
assert "tool-calling grammar" in _friendly_upstream_error("Failed to initialize samplers")
def test_unrelated_error_passes_through(self):
assert _friendly_upstream_error("out of memory") == "llama-server error: out of memory"
def test_openai_passthrough_error_rewrites_grammar_failure(self):
# OpenAI-compatible agents (opencode/openclaw/hermes/pi via /v1/chat/completions)
# get the same actionable message as the Anthropic passthrough, not the raw body.
from routes.inference import _openai_passthrough_error
exc = _openai_passthrough_error(
400, '{"error":{"message":"Failed to initialize samplers: failed to parse grammar"}}'
)
assert "tool-calling grammar" in exc.detail
# An unrelated upstream error still passes through verbatim.
assert "llama-server error:" in _openai_passthrough_error(500, "disk full").detail
# =====================================================================
# ChatMessage — tool role, tool_calls, optional content
# =====================================================================
class TestChatMessageToolRoles:
def test_tool_role_with_tool_call_id(self):
msg = ChatMessage(
role = "tool",
tool_call_id = "call_abc123",
content = '{"temperature": 72}',
)
assert msg.role == "tool"
assert msg.tool_call_id == "call_abc123"
assert msg.content == '{"temperature": 72}'
def test_tool_role_with_name(self):
msg = ChatMessage(
role = "tool",
tool_call_id = "call_abc123",
name = "get_weather",
content = '{"temperature": 72}',
)
assert msg.name == "get_weather"
def test_assistant_with_tool_calls_no_content(self):
msg = ChatMessage(
role = "assistant",
content = None,
tool_calls = [
{
"id": "call_1",
"type": "function",
"function": {
"name": "get_weather",
"arguments": '{"city": "Paris"}',
},
}
],
)
assert msg.role == "assistant"
assert msg.content is None
assert msg.tool_calls is not None
assert len(msg.tool_calls) == 1
assert msg.tool_calls[0]["function"]["name"] == "get_weather"
def test_assistant_with_content_and_tool_calls(self):
msg = ChatMessage(
role = "assistant",
content = "Let me check the weather.",
tool_calls = [
{
"id": "call_1",
"type": "function",
"function": {"name": "get_weather", "arguments": "{}"},
}
],
)
assert msg.content == "Let me check the weather."
assert msg.tool_calls[0]["id"] == "call_1"
def test_plain_user_message_still_works(self):
msg = ChatMessage(role = "user", content = "Hello")
assert msg.role == "user"
assert msg.tool_call_id is None
assert msg.tool_calls is None
assert msg.name is None
def test_invalid_role_rejected(self):
with pytest.raises(ValidationError):
ChatMessage(role = "function", content = "x")
def test_content_absent_on_assistant_tool_call_defaults_to_none(self):
# Assistant messages carrying only tool_calls are the one documented
# case where `content=None` is permitted.
msg = ChatMessage(
role = "assistant",
tool_calls = [
{
"id": "call_1",
"type": "function",
"function": {"name": "f", "arguments": "{}"},
}
],
)
assert msg.content is None
def test_tool_role_missing_tool_call_id_left_for_request_validator(self):
# Per-message: missing tool_call_id is now allowed at this layer.
# ChatCompletionRequest's walkback fills it from the prior assistant
# tool_calls; see test_inference_model_validation.py for resolution
# coverage.
msg = ChatMessage(role = "tool", content = '{"temperature": 72}')
assert msg.tool_call_id is None
assert msg.content == '{"temperature": 72}'
def test_tool_role_empty_tool_call_id_left_for_request_validator(self):
msg = ChatMessage(
role = "tool",
tool_call_id = "",
content = '{"temperature": 72}',
)
# Empty-string is treated the same as missing by the walkback.
assert msg.tool_call_id in (None, "")
# ── Role-aware content requirements ────────────────────────────
@pytest.mark.parametrize("role", ["user", "system"])
def test_empty_string_content_allowed(self, role):
msg = ChatMessage(role = role, content = "")
assert msg.content == ""
def test_user_missing_content_rejected(self):
with pytest.raises(ValidationError):
ChatMessage(role = "user")
def test_user_empty_list_content_rejected(self):
with pytest.raises(ValidationError):
ChatMessage(role = "user", content = [])
def test_tool_empty_content_accepted(self):
# Empty tool output (mkdir, git add, ...) is routine in agentic loops;
# OpenAI and llama-server both accept it, so Unsloth must not 400.
msg = ChatMessage(role = "tool", tool_call_id = "call_1", content = "")
assert msg.content == ""
def test_assistant_without_content_or_tool_calls_tolerated(self):
# Stop-button leaves an empty assistant turn; tolerate for replay.
msg = ChatMessage(role = "assistant")
assert msg.content is None
assert msg.tool_calls is None
def test_assistant_empty_string_content_normalised_to_none(self):
msg = ChatMessage(role = "assistant", content = "")
assert msg.content is None
def test_assistant_empty_list_content_normalised_to_none(self):
msg = ChatMessage(role = "assistant", content = [])
assert msg.content is None
# ── Role-constrained tool-call metadata ────────────────────────
def test_tool_calls_on_user_rejected(self):
with pytest.raises(ValidationError) as exc_info:
ChatMessage(
role = "user",
content = "Hi",
tool_calls = [
{
"id": "c1",
"type": "function",
"function": {"name": "f", "arguments": "{}"},
}
],
)
assert "tool_calls" in str(exc_info.value)
def test_tool_call_id_on_user_rejected(self):
with pytest.raises(ValidationError) as exc_info:
ChatMessage(role = "user", content = "Hi", tool_call_id = "call_1")
assert "tool_call_id" in str(exc_info.value)
def test_name_on_user_rejected(self):
with pytest.raises(ValidationError) as exc_info:
ChatMessage(role = "user", content = "Hi", name = "get_weather")
assert "name" in str(exc_info.value)
# =====================================================================
# ChatCompletionRequest — standard OpenAI tool fields
# =====================================================================
class TestChatCompletionRequestToolFields:
def _make(self, **kwargs):
base = {"messages": [{"role": "user", "content": "Hi"}]}
base.update(kwargs)
return ChatCompletionRequest(**base)
def test_tools_parses(self):
req = self._make(
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Return the weather in a city",
"parameters": {
"type": "object",
"properties": {"city": {"type": "string"}},
"required": ["city"],
},
},
}
],
)
assert req.tools is not None
assert len(req.tools) == 1
assert req.tools[0]["function"]["name"] == "get_weather"
def test_image_base64_allows_empty_user_text(self):
req = ChatCompletionRequest(
messages = [{"role": "user", "content": ""}],
image_base64 = "aW1hZ2U=",
)
assert req.messages[0].content == ""
assert req.image_base64 == "aW1hZ2U="
def test_tool_choice_string_auto(self):
assert self._make(tool_choice = "auto").tool_choice == "auto"
def test_tool_choice_string_required(self):
assert self._make(tool_choice = "required").tool_choice == "required"
def test_tool_choice_string_none(self):
assert self._make(tool_choice = "none").tool_choice == "none"
def test_tool_choice_named_function(self):
tc = {"type": "function", "function": {"name": "get_weather"}}
assert self._make(tool_choice = tc).tool_choice == tc
def test_stop_string(self):
assert self._make(stop = "\nUser:").stop == "\nUser:"
def test_stop_list(self):
assert self._make(stop = ["\nUser:", "\nAssistant:"]).stop == ["\nUser:", "\nAssistant:"]
def test_tools_default_none(self):
req = self._make()
assert req.tools is None
assert req.tool_choice is None
assert req.stop is None
def test_extra_fields_accepted(self):
# `frequency_penalty` and `response_format` are not yet explicitly
# declared but must survive Pydantic parsing now that extra="allow" is
# set. `seed` is declared and should land on the typed field instead.
req = self._make(
frequency_penalty = 0.5,
seed = 42,
response_format = {"type": "json_object"},
)
assert req.seed == 42
# Extras land in model_extra
assert req.model_extra is not None
assert req.model_extra.get("frequency_penalty") == 0.5
assert "seed" not in req.model_extra
assert req.model_extra.get("response_format") == {"type": "json_object"}
def test_unsloth_extensions_still_work(self):
req = self._make(
enable_tools = True,
enabled_tools = ["web_search", "python"],
session_id = "abc",
)
assert req.enable_tools is True
assert req.enabled_tools == ["web_search", "python"]
assert req.session_id == "abc"
def test_stream_defaults_false_matching_openai_spec(self):
# OpenAI defaults `stream` to false. Unsloth used to default true,
# breaking naive curl/.NET clients (#5047) that omit it. Pin the fix.
req = self._make()
assert req.stream is False
def test_post_without_stream_field_decodes_to_stream_false_over_http(self, monkeypatch):
# Wire-level guard: a POST body omitting `stream` must deserialise to
# stream=False and return application/json, never text/event-stream.
# Mounts the real router to catch middleware/aliasing regressions;
# backends are bypassed via provider_type + a stubbed proxy.
from fastapi import FastAPI
from fastapi.responses import JSONResponse
from fastapi.testclient import TestClient
import routes.inference as inference_route
from auth.authentication import get_current_subject
captured = {}
async def _fake_proxy(payload, request, current_subject):
assert current_subject == "test-user"
captured["stream"] = payload.stream
return JSONResponse({"choices": [], "object": "chat.completion"})
monkeypatch.setattr(inference_route, "_proxy_to_external_provider", _fake_proxy)
app = FastAPI()
app.include_router(inference_route.router)
app.dependency_overrides[get_current_subject] = lambda: "test-user"
client = TestClient(app)
resp = client.post(
"/chat/completions",
json = {
"messages": [{"role": "user", "content": "hi"}],
"provider_type": "openai",
},
)
assert resp.status_code == 200
assert resp.headers["content-type"].startswith("application/json")
assert "text/event-stream" not in resp.headers["content-type"]
assert captured["stream"] is False
def _v1_client(
self,
monkeypatch,
llama_backend,
inference_backend = None,
):
from fastapi import FastAPI
from fastapi.testclient import TestClient
import routes.inference as inference_route
from auth.authentication import get_current_subject
from utils.api_errors import install_api_error_handlers
monkeypatch.setattr(inference_route, "get_llama_cpp_backend", lambda: llama_backend)
if inference_backend is not None:
monkeypatch.setattr(inference_route, "get_inference_backend", lambda: inference_backend)
app = FastAPI()
app.include_router(inference_route.router, prefix = "/v1")
install_api_error_handlers(app)
app.dependency_overrides[get_current_subject] = lambda: "test-user"
return TestClient(app)
def _assert_unsupported_param(self, response, param):
assert response.status_code == 400
body = response.json()
assert body["error"]["param"] == param
assert body["error"]["code"] == "unsupported_parameter"
def _assert_unsupported_n(self, response):
self._assert_unsupported_param(response, "n")
def test_n_allows_openai_chat_completion_range(self):
req = self._make(n = 128)
assert req.n == 128
with pytest.raises(ValidationError):
self._make(n = 129)
def test_n_rejected_for_external_provider_path(self, monkeypatch):
class _UnusedBackend:
is_loaded = False
client = self._v1_client(monkeypatch, _UnusedBackend())
resp = client.post(
"/v1/chat/completions",
json = {
"messages": [{"role": "user", "content": "hi"}],
"provider_type": "openai",
"n": 2,
},
)
self._assert_unsupported_n(resp)
def test_confirm_tool_calls_rejected_for_provider_tools(self, monkeypatch):
class _UnusedBackend:
is_loaded = False
client = self._v1_client(monkeypatch, _UnusedBackend())
resp = client.post(
"/v1/chat/completions",
json = {
"messages": [{"role": "user", "content": "hi"}],
"provider_type": "openai",
"external_model": "gpt-4.1",
"enable_tools": True,
"enabled_tools": ["web_search"],
"confirm_tool_calls": True,
},
)
assert resp.status_code == 400
body = resp.json()
assert body["error"]["param"] == "confirm_tool_calls"
assert "only supported for local streaming tools" in body["error"]["message"]
def test_logprobs_rejected_until_supported(self, monkeypatch):
class _UnusedBackend:
is_loaded = False
client = self._v1_client(monkeypatch, _UnusedBackend())
resp = client.post(
"/v1/chat/completions",
json = {
"messages": [{"role": "user", "content": "hi"}],
"provider_type": "openai",
"logprobs": True,
},
)
self._assert_unsupported_param(resp, "logprobs")
def test_top_logprobs_rejected_until_supported(self, monkeypatch):
class _UnusedBackend:
is_loaded = False
client = self._v1_client(monkeypatch, _UnusedBackend())
resp = client.post(
"/v1/chat/completions",
json = {
"messages": [{"role": "user", "content": "hi"}],
"provider_type": "openai",
"top_logprobs": 3,
},
)
self._assert_unsupported_param(resp, "top_logprobs")
def test_n_rejected_for_gguf_streaming_path(self, monkeypatch):
class _GGUFBackend:
is_loaded = True
model_identifier = "test-gguf"
supports_tools = False
is_vision = False
_is_audio = False
context_length = 4096
client = self._v1_client(monkeypatch, _GGUFBackend())
resp = client.post(
"/v1/chat/completions",
json = {
"messages": [{"role": "user", "content": "hi"}],
"stream": True,
"n": 2,
},
)
self._assert_unsupported_n(resp)
def test_n_rejected_for_gguf_tools_passthrough_path(self, monkeypatch):
import routes.inference as inference_route
class _GGUFBackend:
is_loaded = True
model_identifier = "test-gguf"
supports_tools = True
is_vision = False
_is_audio = False
context_length = 4096
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inference_route, "api_monitor", monitor)
client = self._v1_client(monkeypatch, _GGUFBackend())
resp = client.post(
"/v1/chat/completions",
json = {
"messages": [{"role": "user", "content": "hi"}],
"tools": [
{
"type": "function",
"function": {
"name": "lookup",
"parameters": {"type": "object"},
},
}
],
"n": 2,
},
)
self._assert_unsupported_n(resp)
[entry] = monitor.snapshot()
assert entry["status"] == "error"
assert "n > 1 is not supported" in entry["error"]
assert monitor.active_count() == 0
def test_client_tools_rejected_when_gguf_template_has_no_tool_support(self, monkeypatch):
import routes.inference as inference_route
class _GGUFBackend:
is_loaded = True
model_identifier = "test-gguf"
supports_tools = False
is_vision = False
_is_audio = False
context_length = 4096
def generate_chat_completion(self, **_kwargs):
raise AssertionError("client tools must not fall through to the standard GGUF path")
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inference_route, "api_monitor", monitor)
client = self._v1_client(monkeypatch, _GGUFBackend())
resp = client.post(
"/v1/chat/completions",
json = {
"messages": [{"role": "user", "content": "hi"}],
"tools": [
{
"type": "function",
"function": {
"name": "lookup",
"parameters": {"type": "object"},
},
}
],
},
)
self._assert_unsupported_param(resp, "tools")
assert "does not advertise tools" in resp.json()["error"]["message"]
[entry] = monitor.snapshot()
assert entry["status"] == "error"
assert "does not advertise tools" in entry["error"]
assert monitor.active_count() == 0
def test_client_tools_use_passthrough_capability_when_tool_loop_is_disabled(self, monkeypatch):
import routes.inference as inference_route
captured = {}
class _GGUFBackend:
is_loaded = True
model_identifier = "test-gguf"
supports_tools = False
supports_tool_passthrough = True
is_vision = False
_is_audio = False
context_length = 4096
base_url = "http://llama.passthrough-capability.test"
_request_reasoning_kwargs = lambda *_args, **_kwargs: None
def generate_chat_completion(self, **_kwargs):
raise AssertionError("client tools must use passthrough")
def generate_chat_completion_with_tools(self, **_kwargs):
raise AssertionError("Unsloth tool loop must stay disabled")
async def fake_passthrough(llama_backend, payload, model_name, **kwargs):
captured["body"] = inference_route._build_openai_passthrough_body(
payload,
backend_ctx = llama_backend.context_length,
llama_backend = llama_backend,
)
inference_route.api_monitor.finish(kwargs.get("monitor_id"))
return inference_route.JSONResponse({"ok": True, "model": model_name})
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inference_route, "api_monitor", monitor)
monkeypatch.setattr(
inference_route,
"_openai_passthrough_non_streaming",
fake_passthrough,
)
client = self._v1_client(monkeypatch, _GGUFBackend())
resp = client.post(
"/v1/chat/completions",
json = {
"messages": [{"role": "user", "content": "use client tool"}],
"tools": [
{
"type": "function",
"function": {
"name": "lookup",
"parameters": {"type": "object"},
},
}
],
},
)
assert resp.status_code == 200
assert resp.json()["ok"] is True
assert captured["body"]["tools"][0]["function"]["name"] == "lookup"
[entry] = monitor.snapshot()
assert entry["status"] == "completed"
assert monitor.active_count() == 0
def test_permission_mode_does_not_reject_client_tool_passthrough(self, monkeypatch):
# A non-streaming client-tool passthrough (client tools, no Unsloth tool
# loop) that also carries permission_mode "ask"/"auto" must reach the
# provider passthrough, not the confirm-without-stream guard: the
# validator leaves confirm_tool_calls unset for passthrough, and a bare
# permission_mode only gates Unsloth's own local tool loop. An explicit
# confirm_tool_calls=True still forces the local-confirm rejection.
# The pre-switch guard only runs when an automatic load may run, so force
# that predicate on to exercise it against a resident passthrough backend.
import routes.inference as inference_route
class _GGUFBackend:
is_loaded = True
model_identifier = "test-gguf"
supports_tools = False
supports_tool_passthrough = True
is_vision = False
_is_audio = False
context_length = 4096
base_url = "http://llama.permission-passthrough.test"
_request_reasoning_kwargs = lambda *_args, **_kwargs: None
def generate_chat_completion(self, **_kwargs):
raise AssertionError("client tools must use passthrough")
def generate_chat_completion_with_tools(self, **_kwargs):
raise AssertionError("Unsloth tool loop must stay disabled")
async def fake_passthrough(llama_backend, payload, model_name, **kwargs):
inference_route.api_monitor.finish(kwargs.get("monitor_id"))
return inference_route.JSONResponse({"ok": True, "model": model_name})
client_tools = [
{
"type": "function",
"function": {"name": "lookup", "parameters": {"type": "object"}},
}
]
def _setup(policy = None):
reset_tool_policy()
if policy is not None:
set_tool_policy(policy)
monkeypatch.setattr(inference_route, "_automatic_model_load_may_run", lambda: True)
monkeypatch.setattr(inference_route, "api_monitor", ApiMonitor(max_entries = 3))
monkeypatch.setattr(
inference_route, "_openai_passthrough_non_streaming", fake_passthrough
)
return self._v1_client(monkeypatch, _GGUFBackend())
# A process --enable-tools policy must not turn a client-tool passthrough
# into an Unsloth local loop, so a policy of None or True both keep the
# passthrough (the guard mirrors _explicit_studio_tool_loop_requested).
for policy in (None, True):
for mode in ("ask", "auto"):
client = _setup(policy)
resp = client.post(
"/v1/chat/completions",
json = {
"messages": [{"role": "user", "content": "use client tool"}],
"tools": client_tools,
"permission_mode": mode,
"stream": False,
},
)
assert resp.status_code == 200, resp.text
assert resp.json()["ok"] is True
# A JSON-schema response_format is guided-decoding passthrough, not a local
# tool loop, so a --enable-tools policy must not 400 a non-streaming ask/auto
# structured-output request under the confirm guard.
for mode in ("ask", "auto"):
client = _setup(True)
resp = client.post(
"/v1/chat/completions",
json = {
"messages": [{"role": "user", "content": "give me json"}],
"response_format": {
"type": "json_schema",
"json_schema": {"name": "s", "schema": {"type": "object"}},
},
"permission_mode": mode,
"stream": False,
},
)
assert resp.status_code == 200, resp.text
assert resp.json()["ok"] is True
# An explicit confirm_tool_calls=True with client tools and no stream is
# still a confirm-without-stream request and must be rejected up front.
client = _setup()
resp = client.post(
"/v1/chat/completions",
json = {
"messages": [{"role": "user", "content": "use client tool"}],
"tools": client_tools,
"confirm_tool_calls": True,
"stream": False,
},
)
assert resp.status_code == 400
assert "requires stream=true" in resp.json()["error"]["message"]
def test_permission_mode_policy_forced_local_loop_rejected_before_switch(self, monkeypatch):
# A process --enable-tools policy forces Unsloth's own tool loop on even
# when the request omits enable_tools and carries no client tools. A
# non-streaming ask/auto request is then confirm-gated with no stream to
# prompt on, so it must 400 at the pre-switch guard -- before
# _maybe_auto_switch_model runs -- rather than evicting the resident model
# and 400ing only at the per-backend check.
import routes.inference as inference_route
class _GGUFBackend:
is_loaded = True
model_identifier = "test-gguf"
supports_tools = True
supports_tool_passthrough = True
is_vision = False
_is_audio = False
context_length = 4096
base_url = "http://llama.policy-forced.test"
_request_reasoning_kwargs = lambda *_args, **_kwargs: None
switch_calls = []
async def _no_switch(*_args, **_kwargs):
switch_calls.append(1)
def _setup():
reset_tool_policy()
set_tool_policy(True)
monkeypatch.setattr(inference_route, "_automatic_model_load_may_run", lambda: True)
monkeypatch.setattr(inference_route, "api_monitor", ApiMonitor(max_entries = 3))
monkeypatch.setattr(inference_route, "_maybe_auto_switch_model", _no_switch)
return self._v1_client(monkeypatch, _GGUFBackend())
try:
for mode in ("ask", "auto"):
switch_calls.clear()
client = _setup()
resp = client.post(
"/v1/chat/completions",
json = {
"messages": [{"role": "user", "content": "hi"}],
"permission_mode": mode,
"stream": False,
},
)
assert resp.status_code == 400, resp.text
assert "requires stream=true" in resp.json()["error"]["message"]
assert switch_calls == [], "guard must reject before the auto-switch"
finally:
reset_tool_policy()
def test_enable_tools_on_non_tool_backend_keeps_client_tools_on_passthrough(self, monkeypatch):
# DiffusionGemma forces supports_tools off while passthrough stays
# available (#6851): enable_tools=True must not steal client tools
# from the passthrough into an Unsloth tool loop that cannot run.
import routes.inference as inference_route
captured = {}
class _GGUFBackend:
is_loaded = True
model_identifier = "test-gguf"
supports_tools = False
supports_tool_passthrough = True
is_vision = False
_is_audio = False
context_length = 4096
base_url = "http://llama.passthrough-capability.test"
_request_reasoning_kwargs = lambda *_args, **_kwargs: None
def generate_chat_completion(self, **_kwargs):
raise AssertionError("client tools must use passthrough")
def generate_chat_completion_with_tools(self, **_kwargs):
raise AssertionError("Unsloth tool loop cannot run on a non-tool backend")
async def fake_passthrough(llama_backend, payload, model_name, **kwargs):
captured["body"] = inference_route._build_openai_passthrough_body(
payload,
backend_ctx = llama_backend.context_length,
llama_backend = llama_backend,
)
inference_route.api_monitor.finish(kwargs.get("monitor_id"))
return inference_route.JSONResponse({"ok": True, "model": model_name})
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inference_route, "api_monitor", monitor)
monkeypatch.setattr(
inference_route,
"_openai_passthrough_non_streaming",
fake_passthrough,
)
client = self._v1_client(monkeypatch, _GGUFBackend())
resp = client.post(
"/v1/chat/completions",
json = {
"messages": [{"role": "user", "content": "use client tool"}],
"enable_tools": True,
"tools": [
{
"type": "function",
"function": {
"name": "lookup",
"parameters": {"type": "object"},
},
}
],
},
)
assert resp.status_code == 200
assert resp.json()["ok"] is True
assert captured["body"]["tools"][0]["function"]["name"] == "lookup"
[entry] = monitor.snapshot()
assert entry["status"] == "completed"
assert monitor.active_count() == 0
def test_tool_choice_none_allows_tool_catalog_without_tool_template(self, monkeypatch):
import routes.inference as inference_route
class _GGUFBackend:
is_loaded = True
model_identifier = "test-gguf"
supports_tools = False
is_vision = False
_is_audio = False
context_length = 4096
def generate_chat_completion(self, **kwargs):
assert kwargs["max_tokens"] is None
yield "plain response"
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inference_route, "api_monitor", monitor)
client = self._v1_client(monkeypatch, _GGUFBackend())
resp = client.post(
"/v1/chat/completions",
json = {
"messages": [{"role": "user", "content": "hi"}],
"tools": [
{
"type": "function",
"function": {
"name": "lookup",
"parameters": {"type": "object"},
},
}
],
"tool_choice": "none",
},
)
assert resp.status_code == 200
assert resp.json()["choices"][0]["message"]["content"] == "plain response"
[entry] = monitor.snapshot()
assert entry["status"] == "completed"
assert entry["reply"] == "plain response"
assert monitor.active_count() == 0
def test_tool_call_history_rejected_when_gguf_template_has_no_tool_support(self, monkeypatch):
import routes.inference as inference_route
class _GGUFBackend:
is_loaded = True
model_identifier = "test-gguf"
supports_tools = False
is_vision = False
_is_audio = False
context_length = 4096
def generate_chat_completion(self, **_kwargs):
raise AssertionError(
"tool-call history must not fall through to the standard GGUF path"
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inference_route, "api_monitor", monitor)
client = self._v1_client(monkeypatch, _GGUFBackend())
resp = client.post(
"/v1/chat/completions",
json = {
"messages": [
{"role": "user", "content": "use a tool"},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {"name": "lookup", "arguments": "{}"},
}
],
},
{"role": "tool", "tool_call_id": "call_1", "content": "{}"},
],
},
)
self._assert_unsupported_param(resp, "messages")
assert "does not advertise tools" in resp.json()["error"]["message"]
[entry] = monitor.snapshot()
assert entry["status"] == "error"
assert "does not advertise tools" in entry["error"]
assert monitor.active_count() == 0
def test_n_rejected_for_non_gguf_path(self, monkeypatch):
class _NoGGUFBackend:
is_loaded = False
supports_tools = False
class _InferenceBackend:
active_model_name = "test-model"
models = {"test-model": {}}
client = self._v1_client(monkeypatch, _NoGGUFBackend(), _InferenceBackend())
resp = client.post(
"/v1/chat/completions",
json = {
"messages": [{"role": "user", "content": "hi"}],
"n": 2,
},
)
self._assert_unsupported_n(resp)
def test_confirm_tool_calls_requires_streaming_for_safetensors_tools(self, monkeypatch):
import routes.inference as inference_route
class _NoGGUFBackend:
is_loaded = False
supports_tools = False
class _InferenceBackend:
active_model_name = "test-model"
models = {"test-model": {"chat_template_info": {"template": "chatml"}}}
def generate_chat_completion_with_tools(self, **kwargs):
raise AssertionError("tool loop should be rejected before starting")
def generate_chat_completion(self, **kwargs):
raise AssertionError("plain path should not be used")
monkeypatch.setattr(
inference_route,
"_detect_safetensors_features",
lambda backend, chat_template, tools = None: {"supports_tools": True},
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inference_route, "api_monitor", monitor)
client = self._v1_client(monkeypatch, _NoGGUFBackend(), _InferenceBackend())
resp = client.post(
"/v1/chat/completions",
json = {
"messages": [{"role": "user", "content": "hi"}],
"enable_tools": True,
"enabled_tools": ["web_search"],
"confirm_tool_calls": True,
"stream": False,
},
)
assert resp.status_code == 400
body = resp.json()
assert body["error"]["param"] == "confirm_tool_calls"
assert "requires stream=true" in body["error"]["message"]
[entry] = monitor.snapshot()
assert entry["status"] == "error"
assert "confirm_tool_calls requires stream=true" in entry["error"]
assert monitor.active_count() == 0
def test_multiturn_tool_loop_messages(self):
req = ChatCompletionRequest(
messages = [
{"role": "user", "content": "What's the weather in Paris?"},
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {
"name": "get_weather",
"arguments": '{"city": "Paris"}',
},
}
],
},
{
"role": "tool",
"tool_call_id": "call_1",
"content": '{"temperature": 14, "unit": "celsius"}',
},
],
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"parameters": {"type": "object"},
},
}
],
)
assert len(req.messages) == 3
assert req.messages[1].role == "assistant"
assert req.messages[1].content is None
assert req.messages[1].tool_calls[0]["id"] == "call_1"
assert req.messages[2].role == "tool"
assert req.messages[2].tool_call_id == "call_1"
# =====================================================================
# anthropic_tool_choice_to_openai — pure translation helper
# =====================================================================
class TestAnthropicToolChoiceToOpenAI:
def test_auto(self):
assert anthropic_tool_choice_to_openai({"type": "auto"}) == "auto"
def test_any_becomes_required(self):
assert anthropic_tool_choice_to_openai({"type": "any"}) == "required"
def test_none(self):
assert anthropic_tool_choice_to_openai({"type": "none"}) == "none"
def test_tool_named(self):
result = anthropic_tool_choice_to_openai({"type": "tool", "name": "get_weather"})
assert result == {"type": "function", "function": {"name": "get_weather"}}
def test_tool_missing_name_returns_none(self):
assert anthropic_tool_choice_to_openai({"type": "tool"}) is None
def test_none_input_returns_none(self):
assert anthropic_tool_choice_to_openai(None) is None
def test_unrecognized_shape_returns_none(self):
assert anthropic_tool_choice_to_openai({"type": "wibble"}) is None
assert anthropic_tool_choice_to_openai("auto") is None
assert anthropic_tool_choice_to_openai(42) is None
# =====================================================================
# _build_passthrough_payload — tool_choice propagation
# =====================================================================
class TestBuildPassthroughPayloadToolChoice:
def _args(self):
return dict(
openai_messages = [{"role": "user", "content": "Hi"}],
openai_tools = [
{
"type": "function",
"function": {"name": "f", "parameters": {"type": "object"}},
}
],
temperature = 0.6,
top_p = 0.95,
top_k = 20,
max_tokens = 128,
stream = False,
)
def test_default_tool_choice_is_auto(self):
body = _build_passthrough_payload(**self._args())
assert body["tool_choice"] == "auto"
def test_override_tool_choice_required(self):
body = _build_passthrough_payload(**self._args(), tool_choice = "required")
assert body["tool_choice"] == "required"
def test_override_tool_choice_none(self):
body = _build_passthrough_payload(**self._args(), tool_choice = "none")
assert body["tool_choice"] == "none"
def test_override_tool_choice_named_function(self):
tc = {"type": "function", "function": {"name": "f"}}
body = _build_passthrough_payload(**self._args(), tool_choice = tc)
assert body["tool_choice"] == tc
def test_llama_incompatible_tool_constraints_are_omitted(self):
args = self._args()
schema = args["openai_tools"][0]["function"]["parameters"]
schema["properties"] = {
"declarationKey": {"type": "string", "pattern": r"\S"},
"exactKey": {"type": "string", "pattern": r"^[A-Z]+$"},
"nested": {
"type": "array",
"items": {
"anyOf": [
{"type": "string", "pattern": "token"},
{"type": "string", "pattern": "^fixed$"},
],
"default": {"pattern": "annotation data"},
},
},
"largeScript": {"type": "string", "minLength": 1, "maxLength": 65536},
"boundedScript": {"type": "string", "maxLength": 2000},
}
body = _build_passthrough_payload(**args)
forwarded = body["tools"][0]["function"]["parameters"]["properties"]
assert forwarded["declarationKey"] == {"type": "string"}
assert forwarded["exactKey"]["pattern"] == r"^[A-Z]+$"
nested = forwarded["nested"]["items"]
assert nested["anyOf"][0] == {"type": "string"}
assert nested["anyOf"][1]["pattern"] == "^fixed$"
assert nested["default"] == {"pattern": "annotation data"}
assert forwarded["largeScript"] == {"type": "string", "minLength": 1}
assert forwarded["boundedScript"]["maxLength"] == 2000
assert schema["properties"]["declarationKey"]["pattern"] == r"\S"
assert schema["properties"]["nested"]["items"]["anyOf"][0]["pattern"] == "token"
assert schema["properties"]["largeScript"]["maxLength"] == 65536
def test_stream_omits_usage_options_when_client_did_not_request_them(self):
args = self._args()
args["stream"] = True
body = _build_passthrough_payload(**args)
assert "stream_options" not in body
def test_stream_forwards_include_usage_when_client_requests_it(self):
args = self._args()
args["stream"] = True
body = _build_passthrough_payload(
**args,
stream_options = {"include_usage": True},
)
assert body.get("stream_options") == {"include_usage": True}
def test_stream_forwards_include_usage_false_when_client_requests_it(self):
args = self._args()
args["stream"] = True
body = _build_passthrough_payload(
**args,
stream_options = {"include_usage": False},
)
assert body.get("stream_options") == {"include_usage": False}
def test_response_format_without_tools_omits_tool_fields(self):
args = self._args()
args["openai_tools"] = None
body = _build_passthrough_payload(
**args,
response_format = {"type": "json_object"},
)
assert body["response_format"] == {"type": "json_object"}
assert "tools" not in body
assert "tool_choice" not in body
def test_repetition_penalty_renamed(self):
body = _build_passthrough_payload(**self._args(), repetition_penalty = 1.1)
assert body.get("repeat_penalty") == 1.1
assert "repetition_penalty" not in body
def test_omitted_passthrough_max_tokens_uses_backend_context(self):
args = self._args()
args["max_tokens"] = None
body = _build_passthrough_payload(**args, backend_ctx = 4096)
assert body["max_tokens"] == 4096
def test_passthrough_body_merges_system_and_developer_messages(self):
payload = ChatCompletionRequest(
model = "default",
messages = [
{"role": "system", "content": "original system"},
{"role": "developer", "content": "developer rules"},
{"role": "user", "content": "hi"},
],
tools = self._args()["openai_tools"],
)
body = _build_openai_passthrough_body(payload, backend_ctx = 4096)
assert body["messages"] == [
{"role": "system", "content": "original system\n\ndeveloper rules"},
{"role": "user", "content": "hi"},
]
class TestOpenAIPassthroughSSETerminalState:
def test_done_sentinel(self):
assert _openai_passthrough_sse_line_terminal_state("data: [DONE]") == "done"
def test_finish_reason_with_space(self):
line = 'data: {"choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}'
assert _openai_passthrough_sse_line_terminal_state(line) == "finish"
def test_finish_reason_without_space(self):
line = 'data:{"choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}]}'
assert _openai_passthrough_sse_line_terminal_state(line) == "finish"
def test_usage_chunk(self):
line = 'data: {"choices":[],"usage":{"prompt_tokens":1,"completion_tokens":2}}'
assert _openai_passthrough_sse_line_terminal_state(line) == "usage"
def test_error_chunk(self):
line = 'data: {"error":{"message":"boom"}}'
assert _openai_passthrough_sse_line_terminal_state(line) == "error"
def test_cap_parallel_tool_calls_accepts_no_space_after_data_colon(self):
line = (
'data:{"choices":[{"delta":{"tool_calls":['
'{"index":0,"function":{"name":"a"}},'
'{"index":1,"function":{"name":"b"}}]}}]}'
)
capped = _normalize_openai_passthrough_sse_line(line, cap_parallel_tool_calls = True)
data = json.loads(capped[len("data:") :].lstrip())
assert data["choices"][0]["delta"]["tool_calls"] == [
{"index": 0, "function": {"name": "a"}}
]
def test_plain_content_line_is_returned_identically(self):
# The relay dispatches terminal classification on `out_line is raw_line`,
# so the no-mutation path must return the identical string object.
line = 'data: {"choices":[{"index":0,"delta":{"content":"hello"},"finish_reason":null}]}'
assert _normalize_openai_passthrough_sse_line(line) is line
assert _normalize_openai_passthrough_sse_line(line, cap_parallel_tool_calls = True) is line
def test_reasoning_key_inside_content_text_keeps_line_identical(self):
# Fast-path substring gate fires, but the parse finds nothing to change:
# the original object must come back so the relay stays byte-identical.
line = (
'data: {"choices":[{"index":0,"delta":{"content":'
'"mentions \\"reasoning_content\\" in text"},"finish_reason":null}]}'
)
assert _normalize_openai_passthrough_sse_line(line) is line
def test_reasoning_only_delta_gets_empty_content(self):
line = (
'data: {"choices":[{"index":0,'
'"delta":{"reasoning_content":"thinking"},'
'"finish_reason":null}]}'
)
normalized = _normalize_openai_passthrough_sse_line(line)
data = json.loads(normalized[len("data:") :].lstrip())
delta = data["choices"][0]["delta"]
assert delta["reasoning_content"] == "thinking"
assert delta["content"] == ""
def test_reasoning_normalization_preserves_done_sentinel(self):
assert _normalize_openai_passthrough_sse_line("data: [DONE]") == "data: [DONE]"
# =====================================================================
# Passthrough reasoning kwargs — enable_thinking / reasoning_effort /
# preserve_thinking must reach llama-server via chat_template_kwargs,
# gated on template capabilities like the non-passthrough paths.
# =====================================================================
def _reasoning_backend(
supports_reasoning = True,
reasoning_style = "enable_thinking",
reasoning_always_on = False,
supports_preserve_thinking = False,
):
"""Bare LlamaCppBackend with just the reasoning capability flags set,
so _build_openai_passthrough_body exercises the real
_request_reasoning_kwargs gating."""
from core.inference.llama_cpp import LlamaCppBackend
backend = LlamaCppBackend.__new__(LlamaCppBackend)
backend._supports_reasoning = supports_reasoning
backend._reasoning_style = reasoning_style
backend._reasoning_always_on = reasoning_always_on
backend._supports_preserve_thinking = supports_preserve_thinking
return backend
class TestPassthroughReasoningKwargs:
def _payload(self, **fields):
return ChatCompletionRequest(
model = "default",
messages = [{"role": "user", "content": "hi"}],
**fields,
)
def test_enable_thinking_forwarded(self):
body = _build_openai_passthrough_body(
self._payload(enable_thinking = False),
backend_ctx = 4096,
llama_backend = _reasoning_backend(),
)
assert body["chat_template_kwargs"] == {"enable_thinking": False}
def test_preserve_thinking_forwarded_when_template_supports_it(self):
body = _build_openai_passthrough_body(
self._payload(enable_thinking = True, preserve_thinking = True),
backend_ctx = 4096,
llama_backend = _reasoning_backend(supports_preserve_thinking = True),
)
assert body["chat_template_kwargs"] == {
"enable_thinking": True,
"preserve_thinking": True,
}
def test_preserve_thinking_dropped_when_template_lacks_it(self):
body = _build_openai_passthrough_body(
self._payload(preserve_thinking = True),
backend_ctx = 4096,
llama_backend = _reasoning_backend(supports_preserve_thinking = False),
)
assert "chat_template_kwargs" not in body
def test_reasoning_effort_forwarded_for_effort_style_models(self):
body = _build_openai_passthrough_body(
self._payload(reasoning_effort = "high"),
backend_ctx = 4096,
llama_backend = _reasoning_backend(reasoning_style = "reasoning_effort"),
)
assert body["chat_template_kwargs"] == {"reasoning_effort": "high"}
def test_reasoning_effort_none_forwarded_for_effort_style_models(self):
body = _build_openai_passthrough_body(
self._payload(enable_thinking = False, reasoning_effort = "none"),
backend_ctx = 4096,
llama_backend = _reasoning_backend(reasoning_style = "reasoning_effort"),
)
assert body["chat_template_kwargs"] == {"reasoning_effort": "none"}
def test_reasoning_effort_minimal_maps_to_low_for_effort_style_models(self):
body = _build_openai_passthrough_body(
self._payload(enable_thinking = True, reasoning_effort = "minimal"),
backend_ctx = 4096,
llama_backend = _reasoning_backend(reasoning_style = "reasoning_effort"),
)
assert body["chat_template_kwargs"] == {"reasoning_effort": "low"}
def test_enable_thinking_maps_to_effort_for_effort_style_models(self):
body = _build_openai_passthrough_body(
self._payload(enable_thinking = False),
backend_ctx = 4096,
llama_backend = _reasoning_backend(reasoning_style = "reasoning_effort"),
)
assert body["chat_template_kwargs"] == {"reasoning_effort": "low"}
def test_always_on_reasoning_skips_thinking_kwargs(self):
body = _build_openai_passthrough_body(
self._payload(enable_thinking = False),
backend_ctx = 4096,
llama_backend = _reasoning_backend(reasoning_always_on = True),
)
assert "chat_template_kwargs" not in body
def test_no_reasoning_fields_omits_chat_template_kwargs(self):
body = _build_openai_passthrough_body(
self._payload(),
backend_ctx = 4096,
llama_backend = _reasoning_backend(supports_preserve_thinking = True),
)
assert "chat_template_kwargs" not in body
# =====================================================================
# OpenAI API compatibility helpers — verified spec edge cases
# =====================================================================
class TestOpenAICompatibilityHelpers:
def test_max_completion_tokens_wins_over_deprecated_max_tokens(self):
payload = SimpleNamespace(max_tokens = 128, max_completion_tokens = 64)
assert _effective_max_tokens(payload) == 64
def test_openai_compat_max_tokens_returns_none_when_omitted(self):
payload = SimpleNamespace(max_tokens = None, max_completion_tokens = None)
assert _effective_openai_max_tokens(payload) is None
@pytest.mark.parametrize(
("payload", "expected"),
[
(SimpleNamespace(max_tokens = 8192, max_completion_tokens = None), 8192),
(SimpleNamespace(max_tokens = 8192, max_completion_tokens = 256), 256),
],
)
def test_openai_compat_explicit_values_pass_through(self, payload, expected):
assert _effective_openai_max_tokens(payload) == expected
@pytest.mark.parametrize(
("payload", "param"),
[
(SimpleNamespace(max_tokens = "128", max_completion_tokens = None), "max_tokens"),
(SimpleNamespace(max_tokens = True, max_completion_tokens = None), "max_tokens"),
(SimpleNamespace(max_tokens = 12.5, max_completion_tokens = None), "max_tokens"),
(
SimpleNamespace(max_tokens = None, max_completion_tokens = "128"),
"max_completion_tokens",
),
],
)
def test_openai_compat_max_tokens_rejects_non_integer_explicit_values(self, payload, param):
with pytest.raises(HTTPException) as exc:
_effective_openai_max_tokens(payload)
assert exc.value.status_code == 400
assert exc.value.detail["error"]["param"] == param
assert exc.value.detail["error"]["code"] == "invalid_type"
def test_openai_compat_max_tokens_zero_is_valid_and_negative_rejected(self):
# Legacy completions spec: max_tokens has minimum 0, so 0 must pass
# through; only negatives are invalid_value.
assert _effective_openai_max_tokens_from_values(0) == 0
with pytest.raises(HTTPException) as exc:
_effective_openai_max_tokens_from_values(-1)
assert exc.value.status_code == 400
assert exc.value.detail["error"]["code"] == "invalid_value"
assert exc.value.detail["error"]["param"] == "max_tokens"
def test_chat_reasoning_chunk_carries_empty_content(self):
from routes.inference import _chat_reasoning_chunk
line = _chat_reasoning_chunk("chatcmpl-test", 123, "gguf", "thinking...")
chunk = json.loads(line[len("data: ") :])
delta = chunk["choices"][0]["delta"]
assert delta["reasoning_content"] == "thinking..."
assert delta["content"] == ""
def test_passthrough_upstream_headers_include_backend_auth(self):
headers = _openai_passthrough_upstream_headers(
llama_backend = SimpleNamespace(_auth_headers = {"Authorization": "Bearer secret"}),
)
assert headers["Authorization"] == "Bearer secret"
assert headers["Connection"] == "close"
def test_openai_admission_capacity_prefers_backend_effective_slots(self):
request = SimpleNamespace(
app = SimpleNamespace(state = SimpleNamespace(llama_parallel_slots = 1))
)
backend = SimpleNamespace(effective_parallel_slots = 3)
assert _openai_llama_admission_capacity(request, backend) == 3
@pytest.mark.parametrize("backend_value", [None, 0, -1, "not-an-int"])
def test_openai_admission_capacity_falls_back_to_app_state(self, backend_value):
request = SimpleNamespace(
app = SimpleNamespace(state = SimpleNamespace(llama_parallel_slots = 2))
)
backend = SimpleNamespace(effective_parallel_slots = backend_value)
assert _openai_llama_admission_capacity(request, backend) == 2
def test_openai_admission_capacity_falls_back_to_one_without_request(self):
assert _openai_llama_admission_capacity(None, SimpleNamespace()) == 1
def test_openai_admission_non_streaming_exits_invalidated_waiter(self):
async def _run():
queue = get_llama_admission_queue("http://llama.invalidated.test")
blocker = queue.reserve(capacity = 1, config = LlamaAdmissionConfig()).lease_nowait()
assert blocker is not None
reservation = queue.reserve(capacity = 1, config = LlamaAdmissionConfig())
assert reservation._waiter is not None
reservation._waiter.future.cancel()
with pytest.raises(LlamaAdmissionCancelled):
await asyncio.wait_for(
_wait_for_openai_admission_non_streaming(
reservation,
LlamaAdmissionConfig(),
request = None,
cancel_event = None,
),
timeout = 0.1,
)
blocker.release()
snapshot = queue.snapshot()
assert snapshot.active == 0
assert snapshot.queued == 0
asyncio.run(_run())
def test_openai_admission_stream_exits_invalidated_waiter(self):
async def _run():
queue = get_llama_admission_queue("http://llama.invalidated.stream.test")
blocker = queue.reserve(capacity = 1, config = LlamaAdmissionConfig()).lease_nowait()
assert blocker is not None
reservation = queue.reserve(capacity = 1, config = LlamaAdmissionConfig())
assert reservation._waiter is not None
reservation._waiter.future.cancel()
chunks = _openai_admission_wait_stream_chunks(
reservation,
LlamaAdmissionConfig(),
request = None,
cancel_event = None,
)
with pytest.raises(LlamaAdmissionCancelled):
await asyncio.wait_for(chunks.__anext__(), timeout = 0.1)
blocker.release()
snapshot = queue.snapshot()
assert snapshot.active == 0
assert snapshot.queued == 0
asyncio.run(_run())
def test_openai_compat_stream_stall_timeout_uses_default(self, monkeypatch):
monkeypatch.delenv(_OPENAI_COMPAT_STREAM_STALL_TIMEOUT_ENV, raising = False)
assert _openai_compat_stream_stall_timeout() == 120.0
def test_openai_compat_stream_stall_timeout_uses_env_override(self, monkeypatch):
monkeypatch.setenv(_OPENAI_COMPAT_STREAM_STALL_TIMEOUT_ENV, "4.5")
assert _openai_compat_stream_stall_timeout() == 4.5
@pytest.mark.parametrize("raw_value", ["", "not-a-float"])
def test_openai_compat_stream_stall_timeout_invalid_env_uses_default(
self, monkeypatch, raw_value
):
monkeypatch.setenv(_OPENAI_COMPAT_STREAM_STALL_TIMEOUT_ENV, raw_value)
assert _openai_compat_stream_stall_timeout() == 120.0
@pytest.mark.parametrize("raw_value", ["0", "-1"])
def test_openai_compat_stream_stall_timeout_non_positive_env_disables(
self, monkeypatch, raw_value
):
monkeypatch.setenv(_OPENAI_COMPAT_STREAM_STALL_TIMEOUT_ENV, raw_value)
assert _openai_compat_stream_stall_timeout() is None
def test_openai_stream_error_sse_closes_with_done(self):
error = {"error": {"message": "boom"}}
assert _openai_stream_error_sse(error) == (
'data: {"error": {"message": "boom"}}\n\ndata: [DONE]\n\n'
)
@pytest.mark.parametrize(
"finish_reason",
["stop", "length", "tool_calls", "content_filter", "function_call"],
)
def test_clamp_finish_reason_preserves_openai_finish_reasons(self, finish_reason):
assert _clamp_finish_reason(finish_reason) == finish_reason
def test_clamp_finish_reason_defaults_unknown_to_stop(self):
assert _clamp_finish_reason(None) == "stop"
assert _clamp_finish_reason("unexpected") == "stop"
def test_non_streaming_completion_choice_accepts_tool_calls_finish_reason(self):
choice = CompletionChoice(
index = 0,
message = CompletionMessage(content = ""),
finish_reason = "tool_calls",
)
assert choice.finish_reason == "tool_calls"
def test_stream_usage_chunk_requires_include_usage(self):
usage = {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5}
payload = SimpleNamespace(stream_options = None)
assert (
_openai_stream_usage_chunk(payload, "chatcmpl-test", 123, "model", usage, None) is None
)
payload.stream_options = {"include_usage": True}
line = _openai_stream_usage_chunk(payload, "chatcmpl-test", 123, "model", usage, None)
assert line is not None
assert '"choices":[]' in line
assert '"usage"' in line
def test_stream_usage_chunk_coerces_nullable_counts(self):
payload = SimpleNamespace(stream_options = {"include_usage": True})
line = _openai_stream_usage_chunk(
payload,
"chatcmpl-test",
123,
"model",
{"prompt_tokens": None, "completion_tokens": 7, "total_tokens": None},
None,
)
assert line is not None
parsed = json.loads(line.removeprefix("data: "))
usage = parsed["usage"]
assert usage["prompt_tokens"] == 0
assert usage["completion_tokens"] == 7
assert usage["total_tokens"] == 7
def test_completion_stream_monitor_reads_usage_before_client_strip(self, monkeypatch):
import routes.inference as inf_mod
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monitor_id = monitor.start(
endpoint = "/v1/completions",
method = "POST",
model = "m",
prompt = "hi",
context_length = 100,
)
event = (
b'data: {"id":"chatcmpl-test","choices":[{"text":"done","finish_reason":"stop"}],'
b'"usage":{"prompt_tokens":4,"completion_tokens":6,"total_tokens":10}}\n'
)
_monitor_openai_sse_event(monitor_id, event, context_length = 100)
out = _cmpl_stream_event_out(event, include_usage = False)
assert out is not None
assert b'"usage"' not in out
[entry] = monitor.snapshot()
assert entry["reply"] == "done"
assert entry["prompt_tokens"] == 4
assert entry["completion_tokens"] == 6
assert entry["total_tokens"] == 10
assert entry["context_usage"] == 0.1
def test_developer_message_preserves_existing_system_prompt(self):
payload = ChatCompletionRequest(
messages = [
{"role": "system", "content": "original system"},
{"role": "developer", "content": "developer rules"},
{"role": "user", "content": "hi"},
]
)
for message in payload.messages:
if message.role == "developer":
message.role = "system"
system_prompt, chat_messages, image_b64 = _extract_content_parts(payload.messages)
assert system_prompt == "original system\n\ndeveloper rules"
assert chat_messages == [{"role": "user", "content": "hi"}]
assert image_b64 is None
# =====================================================================
# _friendly_error — httpx transport failures
# =====================================================================
class TestFriendlyErrorHttpx:
def _req(self):
return httpx.Request("POST", "http://127.0.0.1:65535/v1/chat/completions")
def test_connect_error_mapped(self):
exc = httpx.ConnectError("All connection attempts failed", request = self._req())
assert "Lost connection" in _friendly_error(exc)
def test_read_error_mapped(self):
exc = httpx.ReadError("EOF", request = self._req())
assert "Lost connection" in _friendly_error(exc)
def test_remote_protocol_error_mapped(self):
exc = httpx.RemoteProtocolError("peer closed", request = self._req())
assert "Lost connection" in _friendly_error(exc)
def test_read_timeout_mapped(self):
exc = httpx.ReadTimeout("timed out", request = self._req())
assert "first token within 20 minutes" in _friendly_error(exc)
def test_non_httpx_unchanged(self):
# Non-httpx exceptions still fall through to the substring heuristics
# — a context-size message must still produce "Message too long".
ctx_msg = "request (4096 tokens) exceeds the available context size (2048 tokens)"
assert "Message too long" in _friendly_error(ValueError(ctx_msg))
def test_generic_exception_returns_generic_message(self):
assert _friendly_error(RuntimeError("unrelated")) == "An internal error occurred"
from routes.inference import ( # noqa: E402
_drop_empty_assistant_sentinels,
_openai_messages_for_gguf_chat,
_openai_messages_for_passthrough,
)
class TestDropEmptyAssistantSentinels:
def test_drops_empty_assistant_between_real_turns(self):
msgs = [
{"role": "user", "content": "hi"},
{"role": "assistant", "content": ""},
{"role": "user", "content": "again"},
]
out = _drop_empty_assistant_sentinels(msgs)
assert out == [{"role": "user", "content": "hi"}, {"role": "user", "content": "again"}]
def test_drops_assistant_with_no_content_key(self):
# exclude_none=True strips the content key entirely; filter must catch it.
msgs = [
{"role": "user", "content": "hi"},
{"role": "assistant"},
{"role": "user", "content": "ok"},
]
out = _drop_empty_assistant_sentinels(msgs)
assert out == [{"role": "user", "content": "hi"}, {"role": "user", "content": "ok"}]
def test_preserves_assistant_with_text(self):
msgs = [
{"role": "user", "content": "hi"},
{"role": "assistant", "content": "hello back"},
]
out = _drop_empty_assistant_sentinels(msgs)
assert out == msgs
def test_preserves_assistant_with_tool_calls_only(self):
msgs = [
{"role": "user", "content": "weather?"},
{
"role": "assistant",
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {"name": "get_weather", "arguments": "{}"},
},
],
},
{
"role": "tool",
"tool_call_id": "call_1",
"content": '{"t": 72}',
},
]
out = _drop_empty_assistant_sentinels(msgs)
assert out == msgs
def test_preserves_user_and_system_with_empty_content(self):
# Filter scoped to role="assistant" only.
msgs = [
{"role": "system", "content": ""},
{"role": "user", "content": ""},
]
out = _drop_empty_assistant_sentinels(msgs)
assert out == msgs
def test_openai_messages_for_passthrough_drops_sentinel(self):
"""End-to-end: Stop-sentinel must not reach the wire."""
req = ChatCompletionRequest(
model = "default",
messages = [
ChatMessage(role = "user", content = "hi"),
ChatMessage(role = "assistant", content = ""),
ChatMessage(role = "user", content = "again"),
],
)
out = _openai_messages_for_passthrough(req)
roles = [m["role"] for m in out]
assert roles == ["user", "user"]
for m in out:
assert m.get("content"), m
class TestGgufVisionMessages:
_PNG_B64 = (
"iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAIAAACQd1PeAAAADUlEQVR42mNk"
"+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg=="
)
def test_preserves_multiturn_image_parts_on_original_turns(self):
req = ChatCompletionRequest(
model = "default",
image_base64 = self._PNG_B64,
messages = [
{
"role": "user",
"content": [
{"type": "text", "text": "describe image one"},
{
"type": "image_url",
"image_url": {
"url": f"data:image/png;base64,{self._PNG_B64}",
},
},
],
},
{"role": "assistant", "content": "first answer"},
{
"role": "user",
"content": [
{"type": "text", "text": "describe image two"},
{
"type": "image_url",
"image_url": {
"url": f"data:image/png;base64,{self._PNG_B64}",
},
},
],
},
],
)
messages, has_image = _openai_messages_for_gguf_chat(req, is_vision = True)
assert has_image is True
assert messages[0]["content"][0] == {"type": "text", "text": "describe image one"}
assert messages[0]["content"][1]["type"] == "image_url"
assert len(messages[0]["content"]) == 2
assert messages[2]["content"][0] == {"type": "text", "text": "describe image two"}
assert messages[2]["content"][1]["type"] == "image_url"
assert len(messages[2]["content"]) == 2
assert isinstance(messages[1]["content"], str)
# Legacy top-level image_base64 must be ignored when a message-level
# image exists; otherwise turn 2 ends up with two image parts.
for msg in messages:
content = msg.get("content")
if isinstance(content, list):
image_parts = [p for p in content if p.get("type") == "image_url"]
assert len(image_parts) == 1, msg
def test_legacy_image_base64_is_injected_when_messages_are_text_only(self):
req = ChatCompletionRequest(
model = "default",
image_base64 = self._PNG_B64,
messages = [{"role": "user", "content": "describe this image"}],
)
messages, has_image = _openai_messages_for_gguf_chat(req, is_vision = True)
assert has_image is True
assert messages[0]["content"][0] == {"type": "text", "text": "describe this image"}
assert messages[0]["content"][1]["type"] == "image_url"
assert messages[0]["content"][1]["image_url"]["url"].startswith("data:image/png;base64,")
def test_rejects_image_parts_for_text_only_gguf(self):
req = ChatCompletionRequest(
model = "default",
messages = [
{
"role": "user",
"content": [
{"type": "text", "text": "look"},
{
"type": "image_url",
"image_url": {
"url": f"data:image/png;base64,{self._PNG_B64}",
},
},
],
},
],
)
with pytest.raises(HTTPException) as exc_info:
_openai_messages_for_gguf_chat(req, is_vision = False)
assert "does not support vision" in str(exc_info.value)
def test_tool_nudge_system_update_preserves_image_parts(self):
messages = [
{"role": "system", "content": "Base instructions."},
{
"role": "user",
"content": [
{"type": "text", "text": "describe this"},
{
"type": "image_url",
"image_url": {
"url": f"data:image/png;base64,{self._PNG_B64}",
},
},
],
},
]
updated = _set_or_prepend_system_message(
messages, "Base instructions.\n\nUse tools when appropriate."
)
assert updated[0] == {
"role": "system",
"content": "Base instructions.\n\nUse tools when appropriate.",
}
assert updated[1]["content"][1]["type"] == "image_url"
assert messages[1]["content"][1]["type"] == "image_url"
def test_tool_nudge_system_update_handles_none_messages(self):
assert _set_or_prepend_system_message(None, "") == []
assert _set_or_prepend_system_message(None, "Use tools.") == [
{"role": "system", "content": "Use tools."}
]
def test_tool_nudge_system_update_dedupes_non_leading_system(self):
messages = [
{"role": "user", "content": "earlier"},
{"role": "system", "content": "Mid instructions."},
{"role": "user", "content": "now"},
]
updated = _set_or_prepend_system_message(messages, "Mid instructions.\n\nUse tools.")
assert [m["role"] for m in updated] == ["system", "user", "user"]
assert updated[0]["content"] == "Mid instructions.\n\nUse tools."
class TestGgufVisionToolRouting:
class _Request:
state = SimpleNamespace()
url = SimpleNamespace(path = "/v1/chat/completions")
method = "POST"
async def is_disconnected(self):
return False
@staticmethod
def _drive(coro):
return asyncio.run(coro)
@staticmethod
def _consume_response(response):
async def _consume():
chunks = []
async for chunk in response.body_iterator:
chunks.append(chunk)
return chunks
return TestGgufVisionToolRouting._drive(_consume())
@staticmethod
def _sse_payloads(chunks):
payloads = []
for chunk in chunks:
if isinstance(chunk, bytes):
chunk = chunk.decode()
for line in str(chunk).splitlines():
if not line.startswith("data: "):
continue
data = line.removeprefix("data: ")
if data == "[DONE]":
continue
try:
payloads.append(json.loads(data))
except json.JSONDecodeError:
pass
return payloads
def _run_gguf_case(
self,
monkeypatch,
*,
generate = None,
tool_generate = None,
payload_kwargs = None,
backend_kwargs = None,
):
import routes.inference as inf_mod
reset_tool_policy()
def _plain(**_kwargs):
raise AssertionError("plain GGUF path should not be used")
backend_data = {
"is_loaded": True,
"is_vision": False,
"supports_tools": tool_generate is not None,
"supports_reasoning": True,
"reasoning_always_on": True,
"_is_audio": False,
"model_identifier": "test-gguf",
"context_length": 4096,
"generate_chat_completion": generate or _plain,
}
if tool_generate is not None:
backend_data["generate_chat_completion_with_tools"] = tool_generate
if backend_kwargs:
backend_data.update(backend_kwargs)
backend = SimpleNamespace(**backend_data)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
request_data = {
"model": "default",
"messages": [{"role": "user", "content": "hi"}],
}
if payload_kwargs:
request_data.update(payload_kwargs)
payload = ChatCompletionRequest(**request_data)
response = self._drive(
openai_chat_completions(payload, request = self._Request(), current_subject = "test")
)
result = SimpleNamespace(response = response, monitor = monitor, backend = backend)
if request_data.get("stream"):
result.chunks = self._consume_response(response)
result.payloads = self._sse_payloads(result.chunks)
else:
result.body = json.loads(response.body)
return result
def test_image_request_with_enabled_tools_enters_gguf_tool_loop(self, monkeypatch):
import routes.inference as inf_mod
reset_tool_policy()
captured = {}
def _plain(**kwargs):
raise AssertionError("plain GGUF path should not be used")
def _tools(**kwargs):
captured["kwargs"] = kwargs
yield {"type": "content", "text": "done"}
backend = SimpleNamespace(
is_loaded = True,
is_vision = True,
supports_tools = True,
model_identifier = "gemma-4-12b-it-GGUF",
context_length = 4096,
generate_chat_completion = _plain,
generate_chat_completion_with_tools = _tools,
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
payload = ChatCompletionRequest(
model = "default",
enable_tools = True,
enabled_tools = ["web_search"],
stream = True,
messages = [
{
"role": "user",
"content": [
{"type": "text", "text": "What is in this image?"},
{
"type": "image_url",
"image_url": {
"url": (f"data:image/png;base64,{TestGgufVisionMessages._PNG_B64}"),
},
},
],
},
],
)
response = self._drive(
openai_chat_completions(payload, request = self._Request(), current_subject = "test")
)
self._consume_response(response)
assert "kwargs" in captured
assert captured["kwargs"]["tools"]
tool_messages = captured["kwargs"]["messages"]
assert tool_messages[0]["role"] == "system"
assert tool_messages[1]["role"] == "user"
assert tool_messages[1]["content"][1]["type"] == "image_url"
def test_parallel_tool_calls_false_reaches_gguf_tool_loop(self, monkeypatch):
import routes.inference as inf_mod
reset_tool_policy()
captured = {}
def _plain(**kwargs):
raise AssertionError("plain GGUF path should not be used")
def _tools(**kwargs):
captured["kwargs"] = kwargs
yield {"type": "content", "text": "done"}
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = True,
model_identifier = "test-gguf",
context_length = 4096,
generate_chat_completion = _plain,
generate_chat_completion_with_tools = _tools,
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
payload = ChatCompletionRequest(
model = "default",
enable_tools = True,
enabled_tools = ["web_search"],
parallel_tool_calls = False,
stream = True,
messages = [{"role": "user", "content": "search once"}],
)
response = self._drive(
openai_chat_completions(payload, request = self._Request(), current_subject = "test")
)
self._consume_response(response)
assert captured["kwargs"]["disable_parallel_tool_use"] is True
def test_confirm_tool_calls_requires_streaming_for_gguf_tools(self, monkeypatch):
import routes.inference as inf_mod
def _plain(**kwargs):
raise AssertionError("plain GGUF path should not be used")
def _tools(**kwargs):
raise AssertionError("tool loop should be rejected before starting")
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = True,
model_identifier = "test-gguf",
context_length = 4096,
generate_chat_completion = _plain,
generate_chat_completion_with_tools = _tools,
)
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
payload = ChatCompletionRequest(
model = "default",
enable_tools = True,
enabled_tools = ["web_search"],
confirm_tool_calls = True,
stream = False,
messages = [{"role": "user", "content": "search once"}],
)
with pytest.raises(HTTPException) as exc:
self._drive(
openai_chat_completions(
payload,
request = self._Request(),
current_subject = "test",
)
)
assert exc.value.status_code == 400
assert "requires stream=true" in exc.value.detail["error"]["message"]
[entry] = monitor.snapshot()
assert entry["status"] == "error"
assert "confirm_tool_calls requires stream=true" in entry["error"]
assert monitor.active_count() == 0
def test_standard_gguf_stream_splits_reasoning_content(self, monkeypatch):
def _generate(**_kwargs):
yield "<thi"
yield "<think>plan"
yield "<think>plan</think>vis"
yield "<think>plan</think>visible"
yield {
"type": "metadata",
"usage": {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5},
"finish_reason": "stop",
}
result = self._run_gguf_case(
monkeypatch,
generate = _generate,
payload_kwargs = {"stream": True},
)
deltas = [p["choices"][0].get("delta", {}) for p in result.payloads if p.get("choices")]
assert "".join(d.get("reasoning_content", "") for d in deltas) == "plan"
assert "".join(d.get("content", "") for d in deltas) == "visible"
assert all("<think>" not in d.get("content", "") for d in deltas)
assert all("content" in d for d in deltas if "reasoning_content" in d)
[entry] = result.monitor.snapshot()
assert entry["reply"] == "visible"
def test_standard_gguf_stream_queued_request_sends_keepalive_before_generation(
self, monkeypatch
):
async def _run():
import routes.inference as inf_mod
class Request(self._Request):
app = SimpleNamespace(state = SimpleNamespace(llama_parallel_slots = 1))
def _generate(**_kwargs):
raise AssertionError("standard GGUF generation must not start while queued")
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = False,
supports_reasoning = True,
reasoning_always_on = True,
_is_audio = False,
model_identifier = "test-gguf",
context_length = 4096,
base_url = "http://llama.standard.test",
effective_parallel_slots = 1,
generate_chat_completion = _generate,
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setenv(ADMISSION_KEEPALIVE_INTERVAL_ENV, "0.01")
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
queue = get_llama_admission_queue("http://llama.standard.test")
blocker = queue.reserve(capacity = 1, config = LlamaAdmissionConfig()).lease_nowait()
assert blocker is not None
payload = ChatCompletionRequest(
model = "default",
messages = [{"role": "user", "content": "hi"}],
stream = True,
)
response = await openai_chat_completions(
payload,
request = Request(),
current_subject = "test",
)
iterator = response.body_iterator
try:
chunk = await asyncio.wait_for(iterator.__anext__(), timeout = 0.2)
assert chunk == ": keep-alive\n\n"
snapshot = queue.snapshot()
assert snapshot.active == 1
assert snapshot.queued == 1
finally:
aclose = getattr(iterator, "aclose", None)
if aclose is not None:
await aclose()
blocker.release()
snapshot = queue.snapshot()
assert snapshot.active == 0
assert snapshot.queued == 0
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_standard_gguf_stream_close_after_first_chunk_cleans_tracker(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
cancel_id = "standard-stream-close-cleanup"
def _generate(**_kwargs):
yield "visible"
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = False,
supports_reasoning = True,
reasoning_always_on = True,
_is_audio = False,
model_identifier = "test-gguf",
context_length = 4096,
base_url = "http://llama.standard.test",
effective_parallel_slots = 1,
generate_chat_completion = _generate,
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
payload = ChatCompletionRequest(
model = "default",
messages = [{"role": "user", "content": "hi"}],
stream = True,
cancel_id = cancel_id,
)
response = await openai_chat_completions(
payload,
request = self._Request(),
current_subject = "test",
)
iterator = response.body_iterator
assert cancel_id in inf_mod._CANCEL_REGISTRY
await asyncio.wait_for(iterator.__anext__(), timeout = 0.2)
aclose = getattr(iterator, "aclose", None)
assert aclose is not None
await aclose()
assert cancel_id not in inf_mod._CANCEL_REGISTRY
assert get_llama_admission_queue("http://llama.standard.test").snapshot().active == 0
asyncio.run(_run())
def test_standard_gguf_stream_task_cancel_after_first_chunk_finalizes_monitor(
self, monkeypatch
):
async def _run():
import routes.inference as inf_mod
started = threading.Event()
released = threading.Event()
def _generate(**kwargs):
cancel_event = kwargs["cancel_event"]
started.set()
while not cancel_event.is_set():
time.sleep(0.005)
released.set()
yield from ()
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = False,
supports_reasoning = True,
reasoning_always_on = True,
_is_audio = False,
model_identifier = "test-gguf",
context_length = 4096,
base_url = "http://llama.standard.test",
effective_parallel_slots = 1,
generate_chat_completion = _generate,
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
payload = ChatCompletionRequest(
model = "default",
messages = [{"role": "user", "content": "hi"}],
stream = True,
)
response = await openai_chat_completions(
payload,
request = self._Request(),
current_subject = "test",
)
iterator = response.body_iterator
assert await asyncio.wait_for(iterator.__anext__(), timeout = 0.2)
pending = asyncio.create_task(iterator.__anext__())
assert await asyncio.to_thread(started.wait, 1.0)
await asyncio.sleep(0)
pending.cancel()
with pytest.raises(asyncio.CancelledError):
await asyncio.wait_for(pending, timeout = 1.0)
assert released.is_set()
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert monitor.active_count() == 0
assert get_llama_admission_queue("http://llama.standard.test").snapshot().active == 0
asyncio.run(_run())
def test_gguf_tool_stream_queued_request_sends_keepalive_before_generation(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class Request(self._Request):
app = SimpleNamespace(state = SimpleNamespace(llama_parallel_slots = 1))
async def fake_select_tools(*_args, **_kwargs):
return [
{
"type": "function",
"function": {
"name": "lookup",
"parameters": {"type": "object", "properties": {}},
},
}
]
def _generate(**_kwargs):
raise AssertionError("GGUF tool loop must not start while queued")
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = True,
supports_reasoning = True,
reasoning_always_on = True,
_is_audio = False,
model_identifier = "test-gguf",
context_length = 4096,
base_url = "http://llama.tool.test",
effective_parallel_slots = 1,
generate_chat_completion = lambda **_kwargs: "unused",
generate_chat_completion_with_tools = _generate,
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setenv(ADMISSION_KEEPALIVE_INTERVAL_ENV, "0.01")
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
monkeypatch.setattr(inf_mod, "_select_request_tools", fake_select_tools)
queue = get_llama_admission_queue("http://llama.tool.test")
blocker = queue.reserve(capacity = 1, config = LlamaAdmissionConfig()).lease_nowait()
assert blocker is not None
payload = ChatCompletionRequest(
model = "default",
messages = [{"role": "user", "content": "hi"}],
enable_tools = True,
stream = True,
)
response = await openai_chat_completions(
payload,
request = Request(),
current_subject = "test",
)
iterator = response.body_iterator
try:
chunk = await asyncio.wait_for(iterator.__anext__(), timeout = 0.2)
assert chunk == ": keep-alive\n\n"
snapshot = queue.snapshot()
assert snapshot.active == 1
assert snapshot.queued == 1
finally:
aclose = getattr(iterator, "aclose", None)
if aclose is not None:
await aclose()
blocker.release()
snapshot = queue.snapshot()
assert snapshot.active == 0
assert snapshot.queued == 0
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_gguf_tool_stream_task_cancel_after_first_chunk_finalizes_monitor(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
async def fake_select_tools(*_args, **_kwargs):
return [
{
"type": "function",
"function": {
"name": "lookup",
"parameters": {"type": "object", "properties": {}},
},
}
]
started = threading.Event()
released = threading.Event()
def _tools(**kwargs):
cancel_event = kwargs["cancel_event"]
started.set()
while not cancel_event.is_set():
time.sleep(0.005)
released.set()
yield from ()
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = True,
supports_reasoning = True,
reasoning_always_on = True,
_is_audio = False,
model_identifier = "test-gguf",
context_length = 4096,
base_url = "http://llama.tool.test",
effective_parallel_slots = 1,
generate_chat_completion = lambda **_kwargs: "unused",
generate_chat_completion_with_tools = _tools,
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
monkeypatch.setattr(inf_mod, "_select_request_tools", fake_select_tools)
payload = ChatCompletionRequest(
model = "default",
messages = [{"role": "user", "content": "hi"}],
enable_tools = True,
stream = True,
)
response = await openai_chat_completions(
payload,
request = self._Request(),
current_subject = "test",
)
iterator = response.body_iterator
assert await asyncio.wait_for(iterator.__anext__(), timeout = 0.2)
pending = asyncio.create_task(iterator.__anext__())
assert await asyncio.to_thread(started.wait, 1.0)
await asyncio.sleep(0)
pending.cancel()
with pytest.raises(asyncio.CancelledError):
await asyncio.wait_for(pending, timeout = 1.0)
assert released.is_set()
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert monitor.active_count() == 0
assert get_llama_admission_queue("http://llama.tool.test").snapshot().active == 0
asyncio.run(_run())
def test_global_enable_tools_does_not_preempt_response_format_passthrough(self, monkeypatch):
import routes.inference as inf_mod
reset_tool_policy()
set_tool_policy(True)
captured = {}
def _plain(**_kwargs):
raise AssertionError("plain GGUF path should not be used")
def _tools(**_kwargs):
raise AssertionError("Unsloth tool loop should not steal response_format")
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = True,
_is_audio = False,
model_identifier = "test-gguf",
context_length = 4096,
base_url = "http://llama.policy.test",
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
generate_chat_completion = _plain,
generate_chat_completion_with_tools = _tools,
)
async def fake_passthrough(llama_backend, payload, model_name, **_kwargs):
captured["body"] = inf_mod._build_openai_passthrough_body(
payload,
backend_ctx = llama_backend.context_length,
llama_backend = llama_backend,
)
return inf_mod.JSONResponse({"ok": True, "model": model_name})
try:
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
monkeypatch.setattr(
inf_mod,
"_openai_passthrough_non_streaming",
fake_passthrough,
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
payload = ChatCompletionRequest(
model = "default",
messages = [{"role": "user", "content": "json"}],
response_format = {"type": "json_object"},
)
response = self._drive(
openai_chat_completions(
payload,
request = self._Request(),
current_subject = "test",
)
)
assert json.loads(response.body)["ok"] is True
assert captured["body"]["response_format"] == {"type": "json_object"}
assert "tools" not in captured["body"]
assert "tool_choice" not in captured["body"]
finally:
reset_tool_policy()
def test_global_enable_tools_does_not_replace_client_tools_passthrough(self, monkeypatch):
import routes.inference as inf_mod
reset_tool_policy()
set_tool_policy(True)
captured = {}
client_tools = [
{
"type": "function",
"function": {
"name": "client_lookup",
"parameters": {"type": "object", "properties": {}},
},
}
]
def _plain(**_kwargs):
raise AssertionError("plain GGUF path should not be used")
def _tools(**_kwargs):
raise AssertionError("Unsloth tool loop should not replace client tools")
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = True,
_is_audio = False,
model_identifier = "test-gguf",
context_length = 4096,
base_url = "http://llama.policy.test",
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
generate_chat_completion = _plain,
generate_chat_completion_with_tools = _tools,
)
async def fake_passthrough(llama_backend, payload, model_name, **_kwargs):
captured["body"] = inf_mod._build_openai_passthrough_body(
payload,
backend_ctx = llama_backend.context_length,
llama_backend = llama_backend,
)
return inf_mod.JSONResponse({"ok": True, "model": model_name})
try:
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
monkeypatch.setattr(
inf_mod,
"_openai_passthrough_non_streaming",
fake_passthrough,
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
payload = ChatCompletionRequest(
model = "default",
messages = [{"role": "user", "content": "use client tool"}],
tools = client_tools,
)
response = self._drive(
openai_chat_completions(
payload,
request = self._Request(),
current_subject = "test",
)
)
assert json.loads(response.body)["ok"] is True
assert captured["body"]["tools"] == client_tools
assert captured["body"]["tool_choice"] == "auto"
finally:
reset_tool_policy()
def test_global_enable_tools_honors_client_tool_choice_none(self, monkeypatch):
import routes.inference as inf_mod
reset_tool_policy()
set_tool_policy(True)
client_tools = [
{
"type": "function",
"function": {
"name": "client_lookup",
"parameters": {"type": "object", "properties": {}},
},
}
]
def _plain(**kwargs):
assert kwargs["max_tokens"] is None
yield "plain response"
def _tools(**_kwargs):
raise AssertionError("tool_choice='none' must not start Unsloth's tool loop")
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = True,
_is_audio = False,
model_identifier = "test-gguf",
context_length = 4096,
base_url = "http://llama.policy.test",
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
generate_chat_completion = _plain,
generate_chat_completion_with_tools = _tools,
)
try:
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
payload = ChatCompletionRequest(
model = "default",
messages = [{"role": "user", "content": "do not use tools"}],
tools = client_tools,
tool_choice = "none",
)
response = self._drive(
openai_chat_completions(
payload,
request = self._Request(),
current_subject = "test",
)
)
assert json.loads(response.body)["choices"][0]["message"]["content"] == "plain response"
[entry] = monitor.snapshot()
assert entry["status"] == "completed"
assert entry["reply"] == "plain response"
assert monitor.active_count() == 0
finally:
reset_tool_policy()
def test_enabled_tools_without_enable_tools_keeps_response_format_passthrough(
self, monkeypatch
):
import routes.inference as inf_mod
reset_tool_policy()
captured = {}
def _plain(**_kwargs):
raise AssertionError("plain GGUF path should not be used")
def _tools(**_kwargs):
raise AssertionError("enabled_tools alone must not start Unsloth's tool loop")
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = True,
_is_audio = False,
model_identifier = "test-gguf",
context_length = 4096,
base_url = "http://llama.enabled-tools.test",
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
generate_chat_completion = _plain,
generate_chat_completion_with_tools = _tools,
)
async def fake_passthrough(llama_backend, payload, model_name, **_kwargs):
captured["body"] = inf_mod._build_openai_passthrough_body(
payload,
backend_ctx = llama_backend.context_length,
llama_backend = llama_backend,
)
return inf_mod.JSONResponse({"ok": True, "model": model_name})
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
monkeypatch.setattr(inf_mod, "_openai_passthrough_non_streaming", fake_passthrough)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
payload = ChatCompletionRequest(
model = "default",
messages = [{"role": "user", "content": "json"}],
enabled_tools = ["web_search"],
response_format = {"type": "json_object"},
)
response = self._drive(
openai_chat_completions(
payload,
request = self._Request(),
current_subject = "test",
)
)
assert json.loads(response.body)["ok"] is True
assert captured["body"]["response_format"] == {"type": "json_object"}
def test_enabled_tools_without_enable_tools_keeps_client_tools_passthrough(self, monkeypatch):
import routes.inference as inf_mod
reset_tool_policy()
captured = {}
client_tools = [
{
"type": "function",
"function": {
"name": "client_lookup",
"parameters": {"type": "object", "properties": {}},
},
}
]
def _plain(**_kwargs):
raise AssertionError("plain GGUF path should not be used")
def _tools(**_kwargs):
raise AssertionError("enabled_tools alone must not start Unsloth's tool loop")
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = True,
_is_audio = False,
model_identifier = "test-gguf",
context_length = 4096,
base_url = "http://llama.enabled-tools.test",
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
generate_chat_completion = _plain,
generate_chat_completion_with_tools = _tools,
)
async def fake_passthrough(llama_backend, payload, model_name, **_kwargs):
captured["body"] = inf_mod._build_openai_passthrough_body(
payload,
backend_ctx = llama_backend.context_length,
llama_backend = llama_backend,
)
return inf_mod.JSONResponse({"ok": True, "model": model_name})
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
monkeypatch.setattr(inf_mod, "_openai_passthrough_non_streaming", fake_passthrough)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
payload = ChatCompletionRequest(
model = "default",
messages = [{"role": "user", "content": "use client tool"}],
enabled_tools = ["web_search"],
tools = client_tools,
)
response = self._drive(
openai_chat_completions(
payload,
request = self._Request(),
current_subject = "test",
)
)
assert json.loads(response.body)["ok"] is True
assert captured["body"]["tools"] == client_tools
assert captured["body"]["tool_choice"] == "auto"
def test_reasoning_capable_gguf_stream_splits_reasoning_by_default(self, monkeypatch):
def _generate(**_kwargs):
yield "<think>plan</think>visible"
yield {
"type": "metadata",
"usage": {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5},
"finish_reason": "stop",
}
result = self._run_gguf_case(
monkeypatch,
generate = _generate,
payload_kwargs = {"stream": True},
backend_kwargs = {"reasoning_always_on": False},
)
deltas = [p["choices"][0].get("delta", {}) for p in result.payloads if p.get("choices")]
assert "".join(d.get("reasoning_content", "") for d in deltas) == "plan"
assert "".join(d.get("content", "") for d in deltas) == "visible"
[entry] = result.monitor.snapshot()
assert entry["reply"] == "visible"
def test_reasoning_capable_gguf_stream_sanitizes_think_tags_when_disabled(self, monkeypatch):
def _generate(**_kwargs):
yield "<think>leaked</think>visible"
yield {
"type": "metadata",
"usage": {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5},
"finish_reason": "stop",
}
result = self._run_gguf_case(
monkeypatch,
generate = _generate,
payload_kwargs = {"stream": True, "enable_thinking": False},
backend_kwargs = {"reasoning_always_on": False},
)
deltas = [p["choices"][0].get("delta", {}) for p in result.payloads if p.get("choices")]
assert "".join(d.get("reasoning_content", "") for d in deltas) == "leaked"
assert "".join(d.get("content", "") for d in deltas) == "visible"
assert all("<think>" not in d.get("content", "") for d in deltas)
[entry] = result.monitor.snapshot()
assert entry["reply"] == "visible"
def test_gguf_tool_stream_splits_reasoning_and_strips_gemma_tool_marker(self, monkeypatch):
def _tools(**_kwargs):
yield {
"type": "content",
"text": '<think>plan</think>visible <|tool_call>call:terminal{command:"ls"}<tool_call|>',
}
yield {
"type": "metadata",
"usage": {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5},
"finish_reason": "stop",
}
result = self._run_gguf_case(
monkeypatch,
tool_generate = _tools,
payload_kwargs = {
"stream": True,
"enable_tools": True,
"enabled_tools": ["terminal"],
"messages": [{"role": "user", "content": "list files"}],
},
)
deltas = [p["choices"][0].get("delta", {}) for p in result.payloads if p.get("choices")]
assert "".join(d.get("reasoning_content", "") for d in deltas) == "plan"
combined_content = "".join(d.get("content", "") for d in deltas)
assert combined_content == "visible "
assert "<|tool_call>" not in combined_content
[entry] = result.monitor.snapshot()
assert entry["reply"] == "visible "
def test_gguf_tool_stream_flushes_held_text_before_status_reset(self, monkeypatch):
def _tools(**_kwargs):
yield {"type": "content", "text": "answer <"}
yield {"type": "status", "text": ""}
yield {
"type": "metadata",
"usage": {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5},
"finish_reason": "stop",
}
result = self._run_gguf_case(
monkeypatch,
tool_generate = _tools,
payload_kwargs = {
"stream": True,
"enable_tools": True,
"enabled_tools": ["terminal"],
"messages": [{"role": "user", "content": "say literal"}],
},
)
deltas = [p["choices"][0].get("delta", {}) for p in result.payloads if p.get("choices")]
combined_content = "".join(d.get("content", "") for d in deltas)
assert combined_content == "answer <"
[entry] = result.monitor.snapshot()
assert entry["reply"] == "answer <"
def test_non_streaming_gguf_splits_reasoning_content(self, monkeypatch):
def _generate(**_kwargs):
yield "<think>plan</think>visible"
yield {
"type": "metadata",
"usage": {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5},
"finish_reason": "stop",
}
result = self._run_gguf_case(monkeypatch, generate = _generate)
body = result.body
message = body["choices"][0]["message"]
assert message["content"] == "visible"
assert message["reasoning_content"] == "plan"
[entry] = result.monitor.snapshot()
assert entry["reply"] == "visible"
def test_standard_gguf_non_streaming_admission_timeout_before_generation(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class Request(self._Request):
app = SimpleNamespace(state = SimpleNamespace(llama_parallel_slots = 1))
def _generate(**_kwargs):
raise AssertionError("standard GGUF generation must not start while queued")
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = False,
_is_audio = False,
model_identifier = "test-gguf",
context_length = 4096,
base_url = "http://llama.standard.test",
effective_parallel_slots = 1,
generate_chat_completion = _generate,
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setenv(ADMISSION_QUEUE_TIMEOUT_ENV, "0.01")
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
queue = get_llama_admission_queue("http://llama.standard.test")
blocker = queue.reserve(capacity = 1, config = LlamaAdmissionConfig()).lease_nowait()
assert blocker is not None
payload = ChatCompletionRequest(
model = "default",
messages = [{"role": "user", "content": "hi"}],
)
try:
with pytest.raises(HTTPException) as exc:
await openai_chat_completions(
payload,
request = Request(),
current_subject = "test",
)
assert exc.value.status_code == 503
finally:
blocker.release()
snapshot = queue.snapshot()
assert snapshot.active == 0
assert snapshot.queued == 0
asyncio.run(_run())
def test_standard_gguf_non_streaming_cancel_id_stops_queued_request_before_generation(
self, monkeypatch
):
async def _run():
import routes.inference as inf_mod
class Request(self._Request):
app = SimpleNamespace(state = SimpleNamespace(llama_parallel_slots = 1))
def _generate(**_kwargs):
raise AssertionError("standard GGUF generation must not start after cancel_id")
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = False,
_is_audio = False,
model_identifier = "test-gguf",
context_length = 4096,
base_url = "http://llama.standard.test",
effective_parallel_slots = 1,
generate_chat_completion = _generate,
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
queue = get_llama_admission_queue("http://llama.standard.test")
blocker = queue.reserve(capacity = 1, config = LlamaAdmissionConfig()).lease_nowait()
assert blocker is not None
cancel_id = "standard-nonstream-admission-cancel"
payload = ChatCompletionRequest(
model = "default",
messages = [{"role": "user", "content": "hi"}],
cancel_id = cancel_id,
)
task = asyncio.create_task(
openai_chat_completions(
payload,
request = Request(),
current_subject = "test",
)
)
try:
for _ in range(50):
if cancel_id in inf_mod._CANCEL_REGISTRY:
break
await asyncio.sleep(0.01)
assert cancel_id in inf_mod._CANCEL_REGISTRY
assert inf_mod._cancel_by_cancel_id_or_stash(cancel_id) == 1
with pytest.raises(HTTPException) as exc:
await asyncio.wait_for(task, timeout = 0.5)
assert exc.value.status_code == 499
finally:
if not task.done():
task.cancel()
with pytest.raises(asyncio.CancelledError):
await task
blocker.release()
assert cancel_id not in inf_mod._CANCEL_REGISTRY
snapshot = queue.snapshot()
assert snapshot.active == 0
assert snapshot.queued == 0
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_standard_gguf_non_streaming_admission_task_cancel_cleans_tracker_and_slot(
self, monkeypatch
):
async def _run():
import routes.inference as inf_mod
cancel_id = "standard-nonstream-task-cancel"
async def fake_wait(*_args, **_kwargs):
raise asyncio.CancelledError()
def _generate(**_kwargs):
raise AssertionError("standard GGUF generation must not start after task cancel")
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = False,
_is_audio = False,
model_identifier = "test-gguf",
context_length = 4096,
base_url = "http://llama.standard.test",
effective_parallel_slots = 1,
generate_chat_completion = _generate,
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
monkeypatch.setattr(
inf_mod,
"_wait_for_openai_admission_non_streaming",
fake_wait,
)
payload = ChatCompletionRequest(
model = "default",
messages = [{"role": "user", "content": "hi"}],
cancel_id = cancel_id,
)
with pytest.raises(asyncio.CancelledError):
await openai_chat_completions(
payload,
request = self._Request(),
current_subject = "test",
)
assert cancel_id not in inf_mod._CANCEL_REGISTRY
assert get_llama_admission_queue("http://llama.standard.test").snapshot().active == 0
asyncio.run(_run())
def test_gguf_tool_non_streaming_admission_timeout_before_generation(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class Request(self._Request):
app = SimpleNamespace(state = SimpleNamespace(llama_parallel_slots = 1))
async def fake_select_tools(*_args, **_kwargs):
return [
{
"type": "function",
"function": {
"name": "lookup",
"parameters": {"type": "object", "properties": {}},
},
}
]
def _generate(**_kwargs):
raise AssertionError("GGUF tool loop must not start while queued")
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = True,
_is_audio = False,
model_identifier = "test-gguf",
context_length = 4096,
base_url = "http://llama.tool.test",
effective_parallel_slots = 1,
generate_chat_completion = lambda **_kwargs: "unused",
generate_chat_completion_with_tools = _generate,
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setenv(ADMISSION_QUEUE_TIMEOUT_ENV, "0.01")
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
monkeypatch.setattr(inf_mod, "_select_request_tools", fake_select_tools)
queue = get_llama_admission_queue("http://llama.tool.test")
blocker = queue.reserve(capacity = 1, config = LlamaAdmissionConfig()).lease_nowait()
assert blocker is not None
payload = ChatCompletionRequest(
model = "default",
messages = [{"role": "user", "content": "hi"}],
enable_tools = True,
)
try:
with pytest.raises(HTTPException) as exc:
await openai_chat_completions(
payload,
request = Request(),
current_subject = "test",
)
assert exc.value.status_code == 503
finally:
blocker.release()
snapshot = queue.snapshot()
assert snapshot.active == 0
assert snapshot.queued == 0
asyncio.run(_run())
def test_gguf_tool_non_streaming_cancel_drains_worker_before_releasing_slot(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
async def fake_select_tools(*_args, **_kwargs):
return [
{
"type": "function",
"function": {
"name": "lookup",
"parameters": {"type": "object", "properties": {}},
},
}
]
started = threading.Event()
released = threading.Event()
def _tools(**kwargs):
cancel_event = kwargs["cancel_event"]
started.set()
while not cancel_event.is_set():
time.sleep(0.005)
released.set()
yield from ()
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = True,
_is_audio = False,
model_identifier = "test-gguf",
context_length = 4096,
base_url = "http://llama.tool.test",
effective_parallel_slots = 1,
generate_chat_completion = lambda **_kwargs: "unused",
generate_chat_completion_with_tools = _tools,
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
monkeypatch.setattr(inf_mod, "_select_request_tools", fake_select_tools)
payload = ChatCompletionRequest(
model = "default",
messages = [{"role": "user", "content": "hi"}],
enable_tools = True,
)
task = asyncio.create_task(
openai_chat_completions(
payload,
request = self._Request(),
current_subject = "test",
)
)
assert await asyncio.to_thread(started.wait, 1.0)
task.cancel()
with pytest.raises(asyncio.CancelledError):
await asyncio.wait_for(task, timeout = 1.0)
assert released.is_set()
assert get_llama_admission_queue("http://llama.tool.test").snapshot().active == 0
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_non_streaming_gguf_n_records_all_monitor_replies(self, monkeypatch):
import routes.inference as inf_mod
calls = {"count": 0}
def _generate(**_kwargs):
calls["count"] += 1
text = f"reply {calls['count']}"
yield text
yield {
"type": "metadata",
"usage": {
"prompt_tokens": 3,
"completion_tokens": calls["count"],
"total_tokens": 3 + calls["count"],
},
}
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = False,
_is_audio = False,
model_identifier = "test-gguf",
context_length = 4096,
generate_chat_completion = _generate,
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
payload = ChatCompletionRequest(
model = "default",
n = 2,
messages = [{"role": "user", "content": "two please"}],
)
response = self._drive(
openai_chat_completions(
payload,
request = self._Request(),
current_subject = "test",
)
)
body = json.loads(response.body)
assert [c["message"]["content"] for c in body["choices"]] == ["reply 1", "reply 2"]
[entry] = monitor.snapshot()
assert entry["reply"] == "Choice 1:\nreply 1\n\nChoice 2:\nreply 2"
assert entry["completion_tokens"] == 3
assert monitor.active_count() == 0
def test_non_streaming_gguf_cancel_drains_worker(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
started = threading.Event()
released = threading.Event()
def _generate(**kwargs):
cancel_event = kwargs["cancel_event"]
started.set()
while not cancel_event.is_set():
time.sleep(0.005)
released.set()
yield from ()
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = False,
_is_audio = False,
model_identifier = "test-gguf",
context_length = 4096,
generate_chat_completion = _generate,
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
payload = ChatCompletionRequest(
model = "default",
messages = [{"role": "user", "content": "hi"}],
)
task = asyncio.create_task(
openai_chat_completions(
payload,
request = self._Request(),
current_subject = "test",
)
)
assert await asyncio.to_thread(started.wait, 1.0)
task.cancel()
with pytest.raises(asyncio.CancelledError):
await asyncio.wait_for(task, timeout = 1.0)
assert released.is_set()
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_standard_gguf_merges_system_and_developer_messages(self, monkeypatch):
import routes.inference as inf_mod
captured = {}
def _generate(**kwargs):
captured["messages"] = kwargs["messages"]
yield "done"
yield {
"type": "metadata",
"usage": {"prompt_tokens": 3, "completion_tokens": 1, "total_tokens": 4},
"finish_reason": "stop",
}
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = False,
model_identifier = "test-gguf",
context_length = 4096,
generate_chat_completion = _generate,
)
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
payload = ChatCompletionRequest(
model = "default",
messages = [
{"role": "system", "content": "original system"},
{"role": "developer", "content": "developer rules"},
{"role": "user", "content": "hi"},
],
)
self._drive(
openai_chat_completions(payload, request = self._Request(), current_subject = "test")
)
assert captured["messages"] == [
{"role": "system", "content": "original system\n\ndeveloper rules"},
{"role": "user", "content": "hi"},
]
@pytest.mark.parametrize(
("seed", "expected"),
[
(41, [41, 42, 43]),
(-1, [-1, -1, -1]),
],
)
def test_gguf_n_choices_vary_explicit_non_negative_seed(self, monkeypatch, seed, expected):
import routes.inference as inf_mod
seen_seeds = []
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
def _generate(**kwargs):
seen_seeds.append(kwargs.get("seed"))
yield f"choice-{len(seen_seeds)}"
yield {
"type": "metadata",
"usage": {
"prompt_tokens": 5,
"completion_tokens": 7,
"total_tokens": 12,
},
"finish_reason": "stop",
}
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = False,
model_identifier = "test-gguf",
context_length = 4096,
generate_chat_completion = _generate,
)
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
payload = ChatCompletionRequest(
model = "default",
messages = [{"role": "user", "content": "hi"}],
n = 3,
seed = seed,
)
response = self._drive(
openai_chat_completions(payload, request = self._Request(), current_subject = "test")
)
body = json.loads(response.body)
assert seen_seeds == expected
assert [choice["index"] for choice in body["choices"]] == [0, 1, 2]
assert body["usage"]["prompt_tokens"] == 5
assert body["usage"]["completion_tokens"] == 21
[entry] = monitor.snapshot()
assert entry["prompt_tokens"] == 5
assert entry["completion_tokens"] == 21
assert entry["total_tokens"] == 26
class TestApiMonitorProviderAndCompletionStreams:
class _Request:
state = SimpleNamespace()
url = SimpleNamespace(path = "/v1/chat/completions")
method = "POST"
async def is_disconnected(self):
return False
async def _run_passthrough_stream(
self,
monkeypatch,
lines,
stream_options = None,
):
import routes.inference as inf_mod
class Request:
async def is_disconnected(self):
return False
async def fake_send(*_args, **_kwargs):
return httpx.Response(200, content = b"")
async def fake_items(*_args, **_kwargs):
for line in lines:
yield line
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "_send_stream_with_preheader_cancel", fake_send)
monkeypatch.setattr(inf_mod, "_aiter_llama_stream_items", fake_items)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
stream = True,
stream_options = stream_options,
tools = [
{
"type": "function",
"function": {
"name": "lookup",
"parameters": {"type": "object", "properties": {}},
},
}
],
)
response = await _openai_passthrough_stream(
Request(),
threading.Event(),
SimpleNamespace(
base_url = "http://llama.test",
context_length = 4096,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"gguf",
"chatcmpl-test",
monitor_id = monitor_id,
)
chunks = [chunk async for chunk in response.body_iterator]
return SimpleNamespace(chunks = chunks, body = "".join(chunks), monitor = monitor)
def test_passthrough_stream_preheader_dispatched_with_timeout(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
gate = asyncio.Event()
async def fake_send(*_args, **_kwargs):
await gate.wait()
return httpx.Response(200, content = b"")
class Request:
async def is_disconnected(self):
return False
monitor = ApiMonitor(max_entries = 3)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "_send_stream_with_preheader_cancel", fake_send)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
stream = True,
)
response = await asyncio.wait_for(
_openai_passthrough_stream(
Request(),
threading.Event(),
SimpleNamespace(
base_url = "http://llama.test",
context_length = 4096,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"chatcmpl-test",
"chatcmpl-test",
monitor_id = monitor_id,
),
timeout = 5.0,
)
assert isinstance(response, _SameTaskStreamingResponse)
gate.set()
chunks = [
chunk.decode() if isinstance(chunk, bytes) else chunk
async for chunk in response.body_iterator
]
assert "data: [DONE]\n\n" in "".join(chunks)
asyncio.run(_run())
def test_passthrough_stream_forwards_backend_auth_headers(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
captured_headers = {}
async def fake_send(_client, req, *_args, **_kwargs):
captured_headers.update(dict(req.headers))
return httpx.Response(200, content = b"")
class Request:
async def is_disconnected(self):
return False
monitor = ApiMonitor(max_entries = 3)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "_send_stream_with_preheader_cancel", fake_send)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
stream = True,
tools = [
{
"type": "function",
"function": {
"name": "lookup",
"parameters": {"type": "object", "properties": {}},
},
}
],
)
response = await _openai_passthrough_stream(
Request(),
threading.Event(),
SimpleNamespace(
base_url = "http://llama.test",
context_length = 4096,
_auth_headers = {"Authorization": "Bearer secret"},
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"chatcmpl-test",
"chatcmpl-test",
monitor_id = monitor_id,
)
chunks = [
chunk.decode() if isinstance(chunk, bytes) else chunk
async for chunk in response.body_iterator
]
assert "data: [DONE]\n\n" in "".join(chunks)
assert captured_headers["authorization"] == "Bearer secret"
assert captured_headers["connection"] == "close"
asyncio.run(_run())
def test_passthrough_stream_keepalive_while_upstream_headers_are_pending(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
gate = asyncio.Event()
async def fake_send(*_args, **_kwargs):
await gate.wait()
return httpx.Response(200, content = b"")
class Request:
async def is_disconnected(self):
return False
monitor = ApiMonitor(max_entries = 3)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "_send_stream_with_preheader_cancel", fake_send)
monkeypatch.setattr(
inf_mod,
"_OPENAI_PASSTHROUGH_PENDING_RESPONSE_KEEPALIVE_S",
0.01,
)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
stream = True,
)
response = await asyncio.wait_for(
_openai_passthrough_stream(
Request(),
threading.Event(),
SimpleNamespace(
base_url = "http://llama.test",
context_length = 4096,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"chatcmpl-test",
"chatcmpl-test",
monitor_id = monitor_id,
),
timeout = 0.2,
)
first = await asyncio.wait_for(response.body_iterator.__anext__(), timeout = 0.2)
assert first == ": keep-alive\n\n"
gate.set()
chunks = [
chunk.decode() if isinstance(chunk, bytes) else chunk
async for chunk in response.body_iterator
]
body = "".join(chunks)
assert "data: [DONE]\n\n" in body
asyncio.run(_run())
def test_passthrough_stream_preheader_non_200_in_window(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
async def fake_send(*_args, **_kwargs):
return httpx.Response(400, content = b'{"error":"bad"}')
class Request:
async def is_disconnected(self):
return False
monitor = ApiMonitor(max_entries = 3)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "_send_stream_with_preheader_cancel", fake_send)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
stream = True,
)
with pytest.raises(HTTPException) as exc:
await _openai_passthrough_stream(
Request(),
threading.Event(),
SimpleNamespace(
base_url = "http://llama.test",
context_length = 4096,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"chatcmpl-test",
"chatcmpl-test",
monitor_id = monitor_id,
)
assert exc.value.status_code == 400
asyncio.run(_run())
def test_passthrough_stream_preheader_request_error_in_window(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
async def fake_send(*_args, **_kwargs):
raise httpx.ConnectError("connectivity issue")
class Request:
async def is_disconnected(self):
return False
monitor = ApiMonitor(max_entries = 3)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "_send_stream_with_preheader_cancel", fake_send)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
stream = True,
)
with pytest.raises(HTTPException) as exc:
await _openai_passthrough_stream(
Request(),
threading.Event(),
SimpleNamespace(
base_url = "http://llama.test",
context_length = 4096,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"chatcmpl-test",
"chatcmpl-test",
monitor_id = monitor_id,
)
assert exc.value.status_code == 502
asyncio.run(_run())
def test_passthrough_stream_preheader_delayed_non_200_returns_sse_error(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
gate = asyncio.Event()
async def fake_send(*_args, **_kwargs):
await gate.wait()
return httpx.Response(400, content = b'{"error":"bad"}')
class Request:
async def is_disconnected(self):
return False
monitor = ApiMonitor(max_entries = 3)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "_send_stream_with_preheader_cancel", fake_send)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
stream = True,
)
response = await asyncio.wait_for(
_openai_passthrough_stream(
Request(),
threading.Event(),
SimpleNamespace(
base_url = "http://llama.test",
context_length = 4096,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"chatcmpl-test",
"chatcmpl-test",
monitor_id = monitor_id,
),
timeout = 5.0,
)
assert isinstance(response, _SameTaskStreamingResponse)
gate.set()
chunks = [
chunk.decode() if isinstance(chunk, bytes) else chunk
async for chunk in response.body_iterator
]
body = "".join(chunks)
assert "data:" in body
assert '"error"' in body
assert "data: [DONE]" in body
[entry] = monitor.snapshot()
assert entry["status"] == "error"
assert "bad" in entry["error"]
asyncio.run(_run())
def test_passthrough_stream_preheader_delayed_context_error_keeps_error_envelope(
self, monkeypatch
):
async def _run():
import routes.inference as inf_mod
gate = asyncio.Event()
ctx_msg = "request (4096 tokens) exceeds the available context size (2048 tokens)"
async def fake_send(*_args, **_kwargs):
await gate.wait()
return httpx.Response(400, content = ctx_msg.encode("utf-8"))
class Request:
async def is_disconnected(self):
return False
monitor = ApiMonitor(max_entries = 3)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "_send_stream_with_preheader_cancel", fake_send)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
stream = True,
)
response = await asyncio.wait_for(
_openai_passthrough_stream(
Request(),
threading.Event(),
SimpleNamespace(
base_url = "http://llama.test",
context_length = 2048,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"chatcmpl-test",
"chatcmpl-test",
monitor_id = monitor_id,
),
timeout = 5.0,
)
assert isinstance(response, _SameTaskStreamingResponse)
gate.set()
chunks = [
chunk.decode() if isinstance(chunk, bytes) else chunk
async for chunk in response.body_iterator
]
body = "".join(chunks)
events = [
line.removeprefix("data: ")
for line in body.splitlines()
if line.startswith("data: ")
]
assert events[-1] == "[DONE]"
payload = json.loads(events[0])
assert payload["error"]["code"] == "context_length_exceeded"
assert payload["error"]["param"] == "messages"
assert isinstance(payload["error"], dict)
asyncio.run(_run())
def test_passthrough_stream_preheader_delayed_context_error_retries_truncation(
self, monkeypatch
):
async def _run():
import routes.inference as inf_mod
gate = asyncio.Event()
calls = []
err_body = json.dumps(
{
"error": {
"message": "request (10000 tokens) exceeds the available context size (2048 tokens)",
"n_prompt_tokens": 10000,
"n_ctx": 2048,
}
}
).encode("utf-8")
async def fake_send(_client, req, *_args, **_kwargs):
calls.append(json.loads(req.content.decode("utf-8")))
if len(calls) == 1:
await gate.wait()
return httpx.Response(400, content = err_body)
return httpx.Response(200, content = b"")
class Request:
async def is_disconnected(self):
return False
monitor = ApiMonitor(max_entries = 3)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "_send_stream_with_preheader_cancel", fake_send)
messages = [
ChatMessage(role = "system", content = "system"),
*[
ChatMessage(role = "user", content = f"turn {idx} " + ("x" * 1000))
for idx in range(8)
],
]
payload = ChatCompletionRequest(
model = "default",
messages = messages,
stream = True,
context_overflow = "truncate_middle",
)
response = await asyncio.wait_for(
_openai_passthrough_stream(
Request(),
threading.Event(),
SimpleNamespace(
base_url = "http://llama.test",
context_length = 2048,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"chatcmpl-test",
"chatcmpl-test",
monitor_id = monitor_id,
),
timeout = 5.0,
)
assert isinstance(response, _SameTaskStreamingResponse)
gate.set()
chunks = [
chunk.decode() if isinstance(chunk, bytes) else chunk
async for chunk in response.body_iterator
]
assert "data: [DONE]\n\n" in "".join(chunks)
assert len(calls) == 2
assert len(calls[1]["messages"]) < len(calls[0]["messages"])
[entry] = monitor.snapshot()
assert entry["status"] == "completed"
asyncio.run(_run())
def test_passthrough_stream_preheader_immediate_context_retry_adopts_delayed_response(
self, monkeypatch
):
async def _run():
import routes.inference as inf_mod
gate = asyncio.Event()
calls = []
err_body = json.dumps(
{
"error": {
"message": "request (10000 tokens) exceeds the available context size (2048 tokens)",
"n_prompt_tokens": 10000,
"n_ctx": 2048,
}
}
).encode("utf-8")
ok_lines = [
'data: {"id":"chatcmpl-test","object":"chat.completion.chunk","created":1,'
'"model":"gguf","choices":[{"index":0,"delta":{"content":"OK"},'
'"finish_reason":null}]}',
'data: {"id":"chatcmpl-test","object":"chat.completion.chunk","created":1,'
'"model":"gguf","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}',
"data: [DONE]",
]
async def fake_send(_client, req, *_args, **_kwargs):
calls.append(json.loads(req.content.decode("utf-8")))
if len(calls) == 1:
return httpx.Response(400, content = err_body)
await gate.wait()
return httpx.Response(200, content = b"")
async def fake_items(*_args, **_kwargs):
for line in ok_lines:
yield line
class Request:
async def is_disconnected(self):
return False
monitor = ApiMonitor(max_entries = 3)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "_send_stream_with_preheader_cancel", fake_send)
monkeypatch.setattr(inf_mod, "_aiter_llama_stream_items", fake_items)
messages = [
ChatMessage(role = "system", content = "system"),
*[
ChatMessage(role = "user", content = f"turn {idx} " + ("x" * 1000))
for idx in range(8)
],
]
payload = ChatCompletionRequest(
model = "default",
messages = messages,
stream = True,
context_overflow = "truncate_middle",
)
response = await asyncio.wait_for(
_openai_passthrough_stream(
Request(),
threading.Event(),
SimpleNamespace(
base_url = "http://llama.test",
context_length = 2048,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"chatcmpl-test",
"chatcmpl-test",
monitor_id = monitor_id,
),
timeout = 0.2,
)
assert isinstance(response, _SameTaskStreamingResponse)
gate.set()
chunks = [
chunk.decode() if isinstance(chunk, bytes) else chunk
async for chunk in response.body_iterator
]
body = "".join(chunks)
assert "OK" in body
assert "context_length_exceeded" not in body
assert len(calls) == 2
assert len(calls[1]["messages"]) < len(calls[0]["messages"])
[entry] = monitor.snapshot()
assert entry["status"] == "completed"
asyncio.run(_run())
def test_passthrough_stream_preheader_delayed_request_error_cleans_up(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
gate = asyncio.Event()
cancel_id = "delayed-request-error-cancel"
async def fake_send(*_args, **_kwargs):
await gate.wait()
raise httpx.ConnectError("delayed connectivity issue")
class Request:
async def is_disconnected(self):
return False
monitor = ApiMonitor(max_entries = 3)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "_send_stream_with_preheader_cancel", fake_send)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
stream = True,
cancel_id = cancel_id,
)
response = await asyncio.wait_for(
_openai_passthrough_stream(
Request(),
threading.Event(),
SimpleNamespace(
base_url = "http://llama.test",
context_length = 4096,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"chatcmpl-test",
"chatcmpl-test",
monitor_id = monitor_id,
),
timeout = 5.0,
)
assert isinstance(response, _SameTaskStreamingResponse)
assert cancel_id in inf_mod._CANCEL_REGISTRY
gate.set()
chunks = [
chunk.decode() if isinstance(chunk, bytes) else chunk
async for chunk in response.body_iterator
]
body = "".join(chunks)
assert "data:" in body
assert '"error"' in body
[entry] = monitor.snapshot()
assert entry["status"] == "error"
assert "Lost connection" in entry["error"]
assert cancel_id not in inf_mod._CANCEL_REGISTRY
asyncio.run(_run())
def test_passthrough_stream_preheader_cancel_cleans_pending_send(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
entered = asyncio.Event()
cancelled = asyncio.Event()
cancel_id = "preheader-cancel-cleanup"
async def fake_send(*_args, **_kwargs):
entered.set()
try:
await asyncio.Event().wait()
except asyncio.CancelledError:
cancelled.set()
raise
class Request:
async def is_disconnected(self):
return False
monitor = ApiMonitor(max_entries = 3)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "_send_stream_with_preheader_cancel", fake_send)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
stream = True,
cancel_id = cancel_id,
)
task = asyncio.create_task(
_openai_passthrough_stream(
Request(),
threading.Event(),
SimpleNamespace(
base_url = "http://llama.test",
context_length = 4096,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"chatcmpl-test",
"chatcmpl-test",
monitor_id = monitor_id,
)
)
await asyncio.wait_for(entered.wait(), timeout = 5.0)
assert cancel_id in inf_mod._CANCEL_REGISTRY
task.cancel()
with pytest.raises(asyncio.CancelledError):
await task
await asyncio.wait_for(cancelled.wait(), timeout = 5.0)
assert cancel_id not in inf_mod._CANCEL_REGISTRY
asyncio.run(_run())
def test_passthrough_stream_unstarted_cleanup_closes_completed_send_response(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
gate = asyncio.Event()
returned = asyncio.Event()
cancel_id = "unstarted-completed-send-cleanup"
class Stream(httpx.AsyncByteStream):
async def __aiter__(self):
if False:
yield b""
stream = Stream()
upstream_response = httpx.Response(200, stream = stream)
async def fake_send(*_args, **_kwargs):
await gate.wait()
returned.set()
return upstream_response
class Request:
async def is_disconnected(self):
return False
monitor = ApiMonitor(max_entries = 3)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "_send_stream_with_preheader_cancel", fake_send)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
stream = True,
cancel_id = cancel_id,
)
response = await asyncio.wait_for(
_openai_passthrough_stream(
Request(),
threading.Event(),
SimpleNamespace(
base_url = "http://llama.test",
context_length = 4096,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"chatcmpl-test",
"chatcmpl-test",
monitor_id = monitor_id,
),
timeout = 5.0,
)
assert isinstance(response, _SameTaskStreamingResponse)
assert cancel_id in inf_mod._CANCEL_REGISTRY
gate.set()
await asyncio.wait_for(returned.wait(), timeout = 5.0)
await asyncio.sleep(0)
await response._unstarted_cleanup()
assert upstream_response.is_closed
assert cancel_id not in inf_mod._CANCEL_REGISTRY
asyncio.run(_run())
def test_external_non_streaming_json_updates_monitor(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class DummyExternalClient:
def __init__(self, **_kwargs):
pass
async def stream_chat_completion(self, **kwargs):
assert kwargs["stream"] is False
yield json.dumps(
{
"choices": [{"message": {"content": "provider [DONE] reply"}}],
"usage": {
"prompt_tokens": 3,
"completion_tokens": 4,
"total_tokens": 7,
},
}
)
async def close(self):
pass
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "ExternalProviderClient", DummyExternalClient)
payload = ChatCompletionRequest(
model = "default",
external_model = "gpt-test",
provider_type = "openai",
provider_base_url = "https://api.openai.com/v1",
messages = [ChatMessage(role = "user", content = "hi")],
)
response = await _proxy_to_external_provider(payload, self._Request())
chunks = []
async for chunk in response.body_iterator:
chunks.append(chunk)
assert chunks[-1] == "data: [DONE]\n\n"
[entry] = monitor.snapshot()
assert entry["status"] == "completed"
assert entry["reply"] == "provider [DONE] reply"
assert entry["prompt_tokens"] == 3
assert entry["completion_tokens"] == 4
assert entry["total_tokens"] == 7
asyncio.run(_run())
def test_external_stream_cancel_finalizes_monitor(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class DummyExternalClient:
def __init__(self, **_kwargs):
pass
async def stream_chat_completion(self, **_kwargs):
yield 'data: {"choices":[{"delta":{"content":"hello"}}]}'
await asyncio.sleep(3600)
async def close(self):
pass
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "ExternalProviderClient", DummyExternalClient)
payload = ChatCompletionRequest(
model = "default",
external_model = "gpt-test",
provider_type = "openai",
provider_base_url = "https://api.openai.com/v1",
messages = [ChatMessage(role = "user", content = "hi")],
stream = True,
)
response = await _proxy_to_external_provider(payload, self._Request())
iterator = response.body_iterator
first = await anext(iterator)
assert "hello" in first
pending = asyncio.create_task(anext(iterator))
await asyncio.sleep(0)
pending.cancel()
with pytest.raises(asyncio.CancelledError):
await pending
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert entry["reply"] == "hello"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_completions_preheader_cancel_finalizes_monitor(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class Request:
state = SimpleNamespace()
url = SimpleNamespace(path = "/v1/completions")
method = "POST"
async def json(self):
return {"prompt": "hi", "stream": True}
async def is_disconnected(self):
return False
async def fake_send(*_args, **_kwargs):
return None
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod,
"get_llama_cpp_backend",
lambda: SimpleNamespace(
is_loaded = True,
base_url = "http://llama.test",
context_length = 4096,
model_identifier = "gguf",
),
)
monkeypatch.setattr(inf_mod, "_send_stream_with_preheader_cancel", fake_send)
response = await openai_completions(Request(), current_subject = "test")
chunks = []
async for chunk in response.body_iterator:
chunks.append(chunk)
assert chunks == []
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_completions_stream_cancel_finalizes_monitor(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class Request:
state = SimpleNamespace()
url = SimpleNamespace(path = "/v1/completions")
method = "POST"
async def json(self):
return {"prompt": "hi", "stream": True}
async def is_disconnected(self):
return False
async def fake_send(*_args, **_kwargs):
return httpx.Response(200, content = b"")
async def fake_items(*_args, **_kwargs):
yield b'data: {"choices":[{"text":"hello"}]}\n\n'
await asyncio.sleep(3600)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod,
"get_llama_cpp_backend",
lambda: SimpleNamespace(
is_loaded = True,
base_url = "http://llama.test",
context_length = 4096,
model_identifier = "gguf",
),
)
monkeypatch.setattr(inf_mod, "_send_stream_with_preheader_cancel", fake_send)
monkeypatch.setattr(inf_mod, "_aiter_llama_stream_items", fake_items)
response = await openai_completions(Request(), current_subject = "test")
iterator = response.body_iterator
first = await anext(iterator)
assert b"hello" in first
pending = asyncio.create_task(anext(iterator))
await asyncio.sleep(0)
pending.cancel()
with pytest.raises(asyncio.CancelledError):
await pending
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert entry["reply"] == "hello"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_completions_non_streaming_post_error_finalizes_monitor(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class Request:
state = SimpleNamespace()
url = SimpleNamespace(path = "/v1/completions")
method = "POST"
async def json(self):
return {"prompt": "hi", "stream": False}
async def is_disconnected(self):
return False
class FailingAsyncClient:
async def __aenter__(self):
return self
async def __aexit__(self, *_args):
return False
async def aclose(self):
return None
async def post(self, *_args, **_kwargs):
raise httpx.ConnectError("llama down")
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
# Per-request client so a forced swap can close it mid-call; the pooled one is shared.
monkeypatch.setattr(
inf_mod,
"_cancelable_nonstreaming_client",
lambda: FailingAsyncClient(),
)
monkeypatch.setattr(
inf_mod,
"get_llama_cpp_backend",
lambda: SimpleNamespace(
is_loaded = True,
base_url = "http://llama.test",
context_length = 4096,
model_identifier = "gguf",
),
)
with pytest.raises(httpx.ConnectError):
await openai_completions(Request(), current_subject = "test")
[entry] = monitor.snapshot()
assert entry["status"] == "error"
assert "Lost connection to the model server" in entry["error"]
assert monitor.active_count() == 0
asyncio.run(_run())
def test_completions_omitted_max_tokens_falls_back_to_context(self, monkeypatch):
# With no env knobs set, an omitted max_tokens must forward the
# backend's context length, exactly as on main.
async def _run():
import routes.inference as inf_mod
class Request:
state = SimpleNamespace()
url = SimpleNamespace(path = "/v1/completions")
method = "POST"
async def json(self):
return {"prompt": "hi", "stream": False}
async def is_disconnected(self):
return False
captured = []
class CapturingClient:
async def aclose(self):
return None
async def post(self, _url, *, json, **_kwargs):
captured.append(dict(json))
return httpx.Response(
200,
json = {
"id": "cmpl-test",
"choices": [{"text": "ok"}],
"usage": {
"prompt_tokens": 1,
"completion_tokens": 1,
"total_tokens": 2,
},
},
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod, "_cancelable_nonstreaming_client", lambda: CapturingClient()
)
monkeypatch.setattr(
inf_mod,
"get_llama_cpp_backend",
lambda: SimpleNamespace(
is_loaded = True,
base_url = "http://llama.test",
context_length = 4096,
model_identifier = "gguf",
),
)
await openai_completions(Request(), current_subject = "test")
assert captured[0]["max_tokens"] == 4096
assert monitor.active_count() == 0
asyncio.run(_run())
def test_completions_forwards_spec_valid_zero_max_tokens(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class Request:
state = SimpleNamespace()
url = SimpleNamespace(path = "/v1/completions")
method = "POST"
async def json(self):
return {"prompt": "hi", "stream": False, "max_tokens": 0}
async def is_disconnected(self):
return False
captured = []
class CapturingClient:
async def aclose(self):
return None
async def post(self, _url, *, json, **_kwargs):
captured.append(dict(json))
return httpx.Response(
200,
json = {
"id": "cmpl-test",
"choices": [{"text": "", "finish_reason": "length"}],
"usage": {
"prompt_tokens": 1,
"completion_tokens": 0,
"total_tokens": 1,
},
},
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod, "_cancelable_nonstreaming_client", lambda: CapturingClient()
)
monkeypatch.setattr(
inf_mod,
"get_llama_cpp_backend",
lambda: SimpleNamespace(
is_loaded = True,
base_url = "http://llama.test",
context_length = 4096,
model_identifier = "gguf",
),
)
await openai_completions(Request(), current_subject = "test")
assert captured[0]["max_tokens"] == 0
assert monitor.active_count() == 0
asyncio.run(_run())
def test_completions_rejects_non_integer_max_tokens_before_forwarding(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class Request:
state = SimpleNamespace()
url = SimpleNamespace(path = "/v1/completions")
method = "POST"
async def json(self):
return {"prompt": "hi", "stream": False, "max_tokens": "128"}
class UnusedClient:
async def post(self, *_args, **_kwargs):
raise AssertionError("invalid max_tokens must not reach llama-server")
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "nonstreaming_client", lambda: UnusedClient())
monkeypatch.setattr(inf_mod, "_cancelable_nonstreaming_client", lambda: UnusedClient())
monkeypatch.setattr(
inf_mod,
"get_llama_cpp_backend",
lambda: SimpleNamespace(
is_loaded = True,
base_url = "http://llama.test",
context_length = 4096,
model_identifier = "gguf",
),
)
with pytest.raises(HTTPException) as exc:
await openai_completions(Request(), current_subject = "test")
assert exc.value.status_code == 400
assert exc.value.detail["error"]["param"] == "max_tokens"
assert exc.value.detail["error"]["code"] == "invalid_type"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_monitor_openai_chunk_records_all_choice_replies(self, monkeypatch):
import routes.inference as inf_mod
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monitor_id = monitor.start(
endpoint = "/v1/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
_monitor_openai_chunk(
monitor_id,
{
"choices": [
{"text": "first"},
{"text": "second"},
],
"usage": {
"prompt_tokens": 2,
"completion_tokens": 5,
"total_tokens": 7,
},
},
4096,
)
entry = monitor.get(monitor_id)
assert entry["reply"] == "Choice 1:\nfirst\n\nChoice 2:\nsecond"
assert entry["prompt_tokens"] == 2
assert entry["completion_tokens"] == 5
assert entry["context_length"] == 4096
def test_monitor_openai_chunk_records_tool_call_reply(self, monkeypatch):
import routes.inference as inf_mod
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
_monitor_openai_chunk(
monitor_id,
{
"choices": [
{
"message": {
"tool_calls": [
{
"type": "function",
"function": {
"name": "lookup",
"arguments": '{"query":"weather"}',
},
}
]
}
}
]
},
4096,
)
entry = monitor.get(monitor_id)
assert entry["reply"] == 'Tool call: lookup({"query":"weather"})'
def test_embeddings_request_is_counted_active_and_completed(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class Request:
state = SimpleNamespace()
url = SimpleNamespace(path = "/v1/embeddings")
method = "POST"
async def json(self):
return {"input": ["alpha", "beta"], "model": "embed"}
async def is_disconnected(self):
return False
class FakeAsyncClient:
async def __aenter__(self):
return self
async def __aexit__(self, *_args):
return False
async def aclose(self):
return None
async def post(self, *_args, **_kwargs):
assert monitor.active_count() == 1
return httpx.Response(
200,
json = {
"data": [{"embedding": [0.1]}],
"usage": {"prompt_tokens": 4, "total_tokens": 4},
},
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
# Per-request client so a forced swap can close it mid-call; the pooled one is shared.
monkeypatch.setattr(
inf_mod,
"_cancelable_nonstreaming_client",
lambda: FakeAsyncClient(),
)
monkeypatch.setattr(
inf_mod,
"get_llama_cpp_backend",
lambda: SimpleNamespace(
is_loaded = True,
base_url = "http://llama.test",
context_length = 4096,
model_identifier = "gguf",
),
)
response = await openai_embeddings(Request(), current_subject = "test")
assert response.status_code == 200
[entry] = monitor.snapshot()
assert entry["endpoint"] == "/v1/embeddings"
assert entry["status"] == "completed"
assert entry["prompt_preview"] == "alpha\nbeta"
assert entry["prompt_tokens"] == 4
assert entry["total_tokens"] == 4
assert monitor.active_count() == 0
asyncio.run(_run())
def test_passthrough_stream_task_cancel_finalizes_monitor(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class Request:
async def is_disconnected(self):
return False
async def fake_send(*_args, **_kwargs):
return httpx.Response(200, content = b"")
async def fake_items(*_args, **_kwargs):
yield 'data: {"choices":[{"delta":{"content":"hello"}}]}'
await asyncio.sleep(3600)
cancel_id = "passthrough-stream-delete-cancel"
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "_send_stream_with_preheader_cancel", fake_send)
monkeypatch.setattr(inf_mod, "_aiter_llama_stream_items", fake_items)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
stream = True,
cancel_id = cancel_id,
tools = [
{
"type": "function",
"function": {
"name": "lookup",
"parameters": {"type": "object", "properties": {}},
},
}
],
)
response = await _openai_passthrough_stream(
Request(),
threading.Event(),
SimpleNamespace(
base_url = "http://llama.test",
context_length = 4096,
_auth_headers = {"Authorization": "Bearer secret"},
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"gguf",
"chatcmpl-test",
monitor_id = monitor_id,
)
assert isinstance(response, _SameTaskStreamingResponse)
iterator = response.body_iterator
first = await anext(iterator)
assert "hello" in first
assert cancel_id in inf_mod._CANCEL_REGISTRY
pending = asyncio.create_task(anext(iterator))
await asyncio.sleep(0)
pending.cancel()
with pytest.raises(asyncio.CancelledError):
await pending
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert entry["reply"] == "hello"
assert monitor.active_count() == 0
assert cancel_id not in inf_mod._CANCEL_REGISTRY
asyncio.run(_run())
def test_passthrough_stream_immediate_task_cancel_releases_admission_and_tracker(
self, monkeypatch
):
async def _run():
import routes.inference as inf_mod
async def fake_cancel_check(*_args, **_kwargs):
raise asyncio.CancelledError()
cancel_id = "passthrough-stream-immediate-task-cancel"
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod,
"_raise_if_openai_admission_cancelled",
fake_cancel_check,
)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
stream = True,
cancel_id = cancel_id,
)
backend = SimpleNamespace(
base_url = "http://llama.test",
context_length = 4096,
effective_parallel_slots = 1,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
with pytest.raises(asyncio.CancelledError):
await _openai_passthrough_stream(
self._Request(),
threading.Event(),
backend,
payload,
"gguf",
"chatcmpl-test",
monitor_id = monitor_id,
)
assert cancel_id not in inf_mod._CANCEL_REGISTRY
assert get_llama_admission_queue("http://llama.test").snapshot().active == 0
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_passthrough_stream_queued_cancel_before_inner_first_chunk_runs_cleanup(
self, monkeypatch
):
async def _run():
import routes.inference as inf_mod
class Request(self._Request):
app = SimpleNamespace(state = SimpleNamespace(llama_parallel_slots = 1))
body_holder = {}
cleanup_called = threading.Event()
async def fake_admitted(*_args, admission_lease, tracker, **_kwargs):
async def cleanup():
admission_lease.release()
tracker.__exit__(None, None, None)
cleanup_called.set()
class BlockingBody:
def __init__(self):
self.started = threading.Event()
self.closed = False
def __aiter__(self):
return self
async def __anext__(self):
self.started.set()
await asyncio.sleep(3600)
raise StopAsyncIteration
async def aclose(self):
self.closed = True
await cleanup()
body = BlockingBody()
body_holder["body"] = body
return _SameTaskStreamingResponse(
body,
media_type = "text/event-stream",
unstarted_cleanup = cleanup,
)
monkeypatch.setenv(ADMISSION_KEEPALIVE_INTERVAL_ENV, "0.01")
monkeypatch.setattr(
inf_mod,
"_openai_passthrough_stream_admitted",
fake_admitted,
)
queue = get_llama_admission_queue("http://llama.test")
blocker = queue.reserve(capacity = 1, config = LlamaAdmissionConfig()).lease_nowait()
assert blocker is not None
cancel_id = "queued-inner-unstarted-cleanup"
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
stream = True,
cancel_id = cancel_id,
)
response = await _openai_passthrough_stream(
Request(),
threading.Event(),
SimpleNamespace(
base_url = "http://llama.test",
context_length = 4096,
effective_parallel_slots = 1,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"gguf",
"chatcmpl-test",
)
iterator = response.body_iterator
try:
chunk = await asyncio.wait_for(iterator.__anext__(), timeout = 0.2)
assert chunk == ": keep-alive\n\n"
assert cancel_id in inf_mod._CANCEL_REGISTRY
blocker.release()
pending = asyncio.create_task(iterator.__anext__())
for _ in range(100):
if "body" in body_holder:
break
await asyncio.sleep(0.01)
body = body_holder["body"]
assert await asyncio.to_thread(body.started.wait, 1.0)
pending.cancel()
with pytest.raises(asyncio.CancelledError):
await asyncio.wait_for(pending, timeout = 1.0)
finally:
aclose = getattr(iterator, "aclose", None)
if aclose is not None:
await aclose()
blocker.release()
assert body_holder["body"].closed
assert cleanup_called.is_set()
assert cancel_id not in inf_mod._CANCEL_REGISTRY
assert queue.snapshot().active == 0
asyncio.run(_run())
def test_passthrough_stream_queued_cancel_after_inner_first_chunk_finalizes_monitor(
self, monkeypatch
):
async def _run():
import routes.inference as inf_mod
class Request(self._Request):
app = SimpleNamespace(state = SimpleNamespace(llama_parallel_slots = 1))
async def fake_admitted(
*_args,
monitor_id = None,
admission_lease,
tracker,
**_kwargs,
):
async def cleanup():
admission_lease.release()
tracker.__exit__(None, None, None)
async def body():
try:
yield 'data: {"choices":[{"delta":{"content":"hello"}}]}\n\n'
await asyncio.sleep(3600)
except asyncio.CancelledError:
inf_mod.api_monitor.finish(monitor_id, "cancelled")
raise
finally:
await cleanup()
return _SameTaskStreamingResponse(
body(),
media_type = "text/event-stream",
unstarted_cleanup = cleanup,
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setenv(ADMISSION_KEEPALIVE_INTERVAL_ENV, "0.01")
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod,
"_openai_passthrough_stream_admitted",
fake_admitted,
)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
queue = get_llama_admission_queue("http://llama.test")
blocker = queue.reserve(capacity = 1, config = LlamaAdmissionConfig()).lease_nowait()
assert blocker is not None
cancel_id = "queued-inner-cancel-monitor"
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
stream = True,
cancel_id = cancel_id,
)
response = await _openai_passthrough_stream(
Request(),
threading.Event(),
SimpleNamespace(
base_url = "http://llama.test",
context_length = 4096,
effective_parallel_slots = 1,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"gguf",
"chatcmpl-test",
monitor_id = monitor_id,
)
iterator = response.body_iterator
try:
chunk = await asyncio.wait_for(iterator.__anext__(), timeout = 0.2)
assert chunk == ": keep-alive\n\n"
blocker.release()
first = await asyncio.wait_for(iterator.__anext__(), timeout = 0.2)
assert "hello" in first
pending = asyncio.create_task(iterator.__anext__())
await asyncio.sleep(0)
pending.cancel()
with pytest.raises(asyncio.CancelledError):
await asyncio.wait_for(pending, timeout = 1.0)
finally:
aclose = getattr(iterator, "aclose", None)
if aclose is not None:
await aclose()
blocker.release()
assert cancel_id not in inf_mod._CANCEL_REGISTRY
assert queue.snapshot().active == 0
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_passthrough_stream_synthesizes_missing_finish_reason(self, monkeypatch):
async def _run():
result = await self._run_passthrough_stream(
monkeypatch,
[
(
'data: {"id":"upstream","created":123,"model":"gguf",'
'"choices":[{"index":0,"delta":{"content":"hello"}}]}'
),
"data: [DONE]",
],
)
body = result.body
assert '"finish_reason":"stop"' in body.replace(" ", "")
assert "data: [DONE]" in body
assert result.monitor.active_count() == 0
asyncio.run(_run())
def test_passthrough_stream_synthesizes_tool_call_finish_reason(self, monkeypatch):
async def _run():
result = await self._run_passthrough_stream(
monkeypatch,
[
(
'data: {"id":"upstream","created":123,"model":"gguf",'
'"choices":[{"index":0,"delta":{"tool_calls":[{"index":0,'
'"id":"call_1","type":"function","function":{"name":"lookup",'
'"arguments":"{}"}}]}}]}'
),
"data: [DONE]",
],
)
compact = result.body.replace(" ", "")
assert '"finish_reason":"tool_calls"' in compact
assert '"finish_reason":"stop"' not in compact
assert "data: [DONE]" in result.body
assert result.monitor.active_count() == 0
asyncio.run(_run())
def test_passthrough_stream_error_done_skips_synthetic_finish_reason(self, monkeypatch):
async def _run():
result = await self._run_passthrough_stream(
monkeypatch,
[
'data: {"error":{"message":"boom","type":"server_error"}}',
"data: [DONE]",
],
)
compact = result.body.replace(" ", "")
assert '"error":{"message":"boom","type":"server_error"}' in compact
assert '"finish_reason"' not in compact
assert "data: [DONE]" in result.body
[entry] = result.monitor.snapshot()
assert entry["status"] == "error"
assert entry["error"] == "boom"
assert result.monitor.active_count() == 0
asyncio.run(_run())
def test_passthrough_stream_error_eof_skips_synthetic_finish_reason(self, monkeypatch):
async def _run():
result = await self._run_passthrough_stream(
monkeypatch,
['data: {"error":{"message":"boom","type":"server_error"}}'],
)
compact = result.body.replace(" ", "")
assert '"error":{"message":"boom","type":"server_error"}' in compact
assert '"finish_reason"' not in compact
assert "data: [DONE]" not in result.body
[entry] = result.monitor.snapshot()
assert entry["status"] == "error"
assert entry["error"] == "boom"
assert result.monitor.active_count() == 0
asyncio.run(_run())
def test_passthrough_usage_done_are_separate_sse_events(self, monkeypatch):
async def _run():
result = await self._run_passthrough_stream(
monkeypatch,
[
'data: {"id":"chatcmpl-test","object":"chat.completion.chunk","created":1,"model":"m","choices":[{"index":0,"delta":{"role":"assistant"},"finish_reason":null}]}',
'data: {"id":"chatcmpl-test","object":"chat.completion.chunk","created":1,"model":"m","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}',
'data: {"id":"chatcmpl-test","object":"chat.completion.chunk","created":1,"model":"m","choices":[],"usage":{"prompt_tokens":1,"completion_tokens":1,"total_tokens":2}}',
],
stream_options = {"include_usage": True},
)
assert (
'"usage":{"prompt_tokens":1,"completion_tokens":1,"total_tokens":2' in result.body
)
assert "data: [DONE]" in result.body
assert "}\n\ndata: [DONE]\n\n" in result.body
assert "}\ndata: [DONE]\n\n" not in result.body
asyncio.run(_run())
def test_passthrough_stream_queued_request_sends_keepalive_before_upstream(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class Request:
app = SimpleNamespace(state = SimpleNamespace(llama_parallel_slots = 1))
url = SimpleNamespace(path = "/v1/chat/completions")
async def is_disconnected(self):
return False
async def fail_admitted(*_args, **_kwargs):
raise AssertionError("upstream must not start while request is queued")
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setenv(ADMISSION_KEEPALIVE_INTERVAL_ENV, "0.01")
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "_openai_passthrough_stream_admitted", fail_admitted)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
queue = get_llama_admission_queue("http://llama.test")
blocker = queue.reserve(capacity = 1, config = LlamaAdmissionConfig()).lease_nowait()
assert blocker is not None
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
stream = True,
)
response = await _openai_passthrough_stream(
Request(),
threading.Event(),
SimpleNamespace(
base_url = "http://llama.test",
effective_parallel_slots = 1,
context_length = 4096,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"gguf",
"chatcmpl-test",
monitor_id = monitor_id,
)
iterator = response.body_iterator
try:
chunk = await asyncio.wait_for(iterator.__anext__(), timeout = 0.2)
assert chunk == ": keep-alive\n\n"
snapshot = queue.snapshot()
assert snapshot.active == 1
assert snapshot.queued == 1
finally:
aclose = getattr(iterator, "aclose", None)
if aclose is not None:
await aclose()
blocker.release()
snapshot = queue.snapshot()
assert snapshot.active == 0
assert snapshot.queued == 0
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_passthrough_non_streaming_admission_timeout_before_upstream(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class Request:
app = SimpleNamespace(state = SimpleNamespace(llama_parallel_slots = 1))
url = SimpleNamespace(path = "/v1/chat/completions")
async def is_disconnected(self):
return False
async def fail_upstream(*_args, **_kwargs):
raise AssertionError("upstream must not start while request is queued")
monkeypatch.setenv(ADMISSION_QUEUE_TIMEOUT_ENV, "0.01")
monkeypatch.setattr(
inf_mod,
"_openai_passthrough_non_streaming_upstream",
fail_upstream,
)
queue = get_llama_admission_queue("http://llama.test")
blocker = queue.reserve(capacity = 1, config = LlamaAdmissionConfig()).lease_nowait()
assert blocker is not None
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
)
try:
with pytest.raises(HTTPException) as exc:
await _openai_passthrough_non_streaming(
SimpleNamespace(
base_url = "http://llama.test",
effective_parallel_slots = 1,
context_length = 4096,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"gguf",
request = Request(),
cancel_event = threading.Event(),
)
assert exc.value.status_code == 503
finally:
blocker.release()
snapshot = queue.snapshot()
assert snapshot.active == 0
assert snapshot.queued == 0
asyncio.run(_run())
def test_passthrough_non_streaming_admission_queue_full_before_upstream(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class Request:
app = SimpleNamespace(state = SimpleNamespace(llama_parallel_slots = 1))
url = SimpleNamespace(path = "/v1/chat/completions")
async def is_disconnected(self):
return False
async def fail_upstream(*_args, **_kwargs):
raise AssertionError("upstream must not start when admission queue is full")
monkeypatch.setenv(ADMISSION_MAX_QUEUE_ENV, "1")
monkeypatch.setattr(
inf_mod,
"_openai_passthrough_non_streaming_upstream",
fail_upstream,
)
queue = get_llama_admission_queue("http://llama.test")
blocker = queue.reserve(
capacity = 1,
config = LlamaAdmissionConfig(max_queue = 1),
).lease_nowait()
queued = queue.reserve(capacity = 1, config = LlamaAdmissionConfig(max_queue = 1))
assert blocker is not None
assert queued.lease_nowait() is None
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
)
try:
with pytest.raises(HTTPException) as exc:
await _openai_passthrough_non_streaming(
SimpleNamespace(
base_url = "http://llama.test",
effective_parallel_slots = 1,
context_length = 4096,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"gguf",
request = Request(),
cancel_event = threading.Event(),
)
assert exc.value.status_code == 429
finally:
queued.cancel()
blocker.release()
snapshot = queue.snapshot()
assert snapshot.active == 0
assert snapshot.queued == 0
asyncio.run(_run())
def test_passthrough_non_streaming_immediate_cancel_stops_before_upstream(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
async def fail_upstream(*_args, **_kwargs):
raise AssertionError("upstream must not start after client cancellation")
monkeypatch.setattr(
inf_mod,
"_openai_passthrough_non_streaming_upstream",
fail_upstream,
)
cancel_event = threading.Event()
cancel_event.set()
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
)
with pytest.raises(HTTPException) as exc:
await _openai_passthrough_non_streaming(
SimpleNamespace(
base_url = "http://llama.test",
effective_parallel_slots = 1,
context_length = 4096,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"gguf",
monitor_id = monitor_id,
cancel_event = cancel_event,
)
assert exc.value.status_code == 499
assert get_llama_admission_queue("http://llama.test").snapshot().active == 0
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_passthrough_non_streaming_admission_task_cancel_finalizes_monitor(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
async def fake_wait(*_args, **_kwargs):
raise asyncio.CancelledError()
async def fail_upstream(*_args, **_kwargs):
raise AssertionError("upstream must not start after admission task cancel")
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod,
"_wait_for_openai_admission_non_streaming",
fake_wait,
)
monkeypatch.setattr(
inf_mod,
"_openai_passthrough_non_streaming_upstream",
fail_upstream,
)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
)
with pytest.raises(asyncio.CancelledError):
await _openai_passthrough_non_streaming(
SimpleNamespace(
base_url = "http://llama.test",
effective_parallel_slots = 1,
context_length = 4096,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"gguf",
monitor_id = monitor_id,
cancel_event = threading.Event(),
)
assert get_llama_admission_queue("http://llama.test").snapshot().active == 0
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_passthrough_non_streaming_cancel_finalizes_monitor(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class CancellingAsyncClient:
async def __aenter__(self):
return self
async def __aexit__(self, *_args):
return False
async def post(self, *_args, **_kwargs):
raise asyncio.CancelledError()
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod,
"nonstreaming_client",
lambda: CancellingAsyncClient(),
)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
tools = [
{
"type": "function",
"function": {
"name": "lookup",
"parameters": {"type": "object", "properties": {}},
},
}
],
)
with pytest.raises(asyncio.CancelledError):
await _openai_passthrough_non_streaming(
SimpleNamespace(
base_url = "http://llama.test",
context_length = 4096,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"gguf",
monitor_id = monitor_id,
)
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_passthrough_non_streaming_cancel_closes_blocked_upstream_post(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class HangingCancelableClient:
def __init__(self):
self.started = asyncio.Event()
self.closed = asyncio.Event()
async def post(self, *_args, **_kwargs):
self.started.set()
await self.closed.wait()
raise httpx.ReadError("client closed")
async def aclose(self):
self.closed.set()
class Request:
async def is_disconnected(self):
return False
client = HangingCancelableClient()
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod,
"_cancelable_nonstreaming_client",
lambda: client,
)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
cancel_event = threading.Event()
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
tools = [
{
"type": "function",
"function": {
"name": "lookup",
"parameters": {"type": "object", "properties": {}},
},
}
],
)
task = asyncio.create_task(
_openai_passthrough_non_streaming(
SimpleNamespace(
base_url = "http://llama.test",
context_length = 4096,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"gguf",
monitor_id = monitor_id,
request = Request(),
cancel_event = cancel_event,
)
)
await asyncio.wait_for(client.started.wait(), 0.2)
cancel_event.set()
with pytest.raises(asyncio.CancelledError):
await asyncio.wait_for(task, 0.5)
assert client.closed.is_set()
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_passthrough_non_streaming_route_registers_cancel_id(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class HangingCancelableClient:
def __init__(self):
self.started = asyncio.Event()
self.closed = asyncio.Event()
async def post(self, *_args, **_kwargs):
self.started.set()
await self.closed.wait()
raise httpx.ReadError("client closed")
async def aclose(self):
self.closed.set()
class Request:
state = SimpleNamespace()
url = SimpleNamespace(path = "/v1/chat/completions")
method = "POST"
async def is_disconnected(self):
return False
cancel_id = "passthrough-nonstream-cancel-id"
client = HangingCancelableClient()
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "_cancelable_nonstreaming_client", lambda: client)
def _plain(**_kwargs):
raise AssertionError("plain GGUF path should not be used")
monkeypatch.setattr(
inf_mod,
"get_llama_cpp_backend",
lambda: SimpleNamespace(
is_loaded = True,
is_vision = False,
supports_tools = True,
_is_audio = False,
model_identifier = "test-gguf",
context_length = 4096,
base_url = "http://llama.test",
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
generate_chat_completion = _plain,
),
)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
cancel_id = cancel_id,
tools = [
{
"type": "function",
"function": {
"name": "lookup",
"parameters": {"type": "object", "properties": {}},
},
}
],
)
task = asyncio.create_task(
openai_chat_completions(
payload,
request = Request(),
current_subject = "test",
)
)
await asyncio.wait_for(client.started.wait(), 0.2)
assert cancel_id in inf_mod._CANCEL_REGISTRY
assert inf_mod._cancel_by_cancel_id_or_stash(cancel_id) == 1
with pytest.raises(asyncio.CancelledError):
await asyncio.wait_for(task, 0.5)
assert client.closed.is_set()
assert cancel_id not in inf_mod._CANCEL_REGISTRY
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_passthrough_non_streaming_disconnect_closes_blocked_upstream_post(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class HangingCancelableClient:
def __init__(self):
self.started = asyncio.Event()
self.closed = asyncio.Event()
async def post(self, *_args, **_kwargs):
self.started.set()
await self.closed.wait()
raise httpx.ReadError("client closed")
async def aclose(self):
self.closed.set()
class Request:
def __init__(self):
self.disconnected = False
async def is_disconnected(self):
return self.disconnected
client = HangingCancelableClient()
request = Request()
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod,
"_cancelable_nonstreaming_client",
lambda: client,
)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
cancel_event = threading.Event()
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
tools = [
{
"type": "function",
"function": {
"name": "lookup",
"parameters": {"type": "object", "properties": {}},
},
}
],
)
task = asyncio.create_task(
_openai_passthrough_non_streaming(
SimpleNamespace(
base_url = "http://llama.test",
context_length = 4096,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"gguf",
monitor_id = monitor_id,
request = request,
cancel_event = cancel_event,
)
)
await asyncio.wait_for(client.started.wait(), 0.2)
request.disconnected = True
with pytest.raises(asyncio.CancelledError):
await asyncio.wait_for(task, 0.5)
assert client.closed.is_set()
assert cancel_event.is_set()
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_passthrough_non_streaming_forwards_backend_auth_headers(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
captured = {}
class FakeNonStreamingClient:
async def post(self, *_args, **kwargs):
captured["headers"] = kwargs.get("headers")
return httpx.Response(
200,
json = {
"id": "chatcmpl-test",
"object": "chat.completion",
"created": 123,
"model": "gguf",
"choices": [
{
"index": 0,
"message": {"role": "assistant", "content": "OK"},
"finish_reason": "stop",
}
],
},
)
monitor = ApiMonitor(max_entries = 3)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod,
"nonstreaming_client",
lambda: FakeNonStreamingClient(),
)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
tools = [
{
"type": "function",
"function": {
"name": "lookup",
"parameters": {"type": "object", "properties": {}},
},
}
],
)
response = await _openai_passthrough_non_streaming(
SimpleNamespace(
base_url = "http://llama.test",
context_length = 4096,
_auth_headers = {"Authorization": "Bearer secret"},
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"gguf",
monitor_id = monitor_id,
)
assert json.loads(response.body)["choices"][0]["message"]["content"] == "OK"
assert captured["headers"]["Authorization"] == "Bearer secret"
assert captured["headers"]["Connection"] == "close"
asyncio.run(_run())
def test_passthrough_non_streaming_forces_upstream_stream_false(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
captured = {}
class FakeNonStreamingClient:
async def __aenter__(self):
return self
async def __aexit__(self, *_args):
return False
async def post(self, *_args, **kwargs):
captured["json"] = kwargs.get("json")
return httpx.Response(
200,
json = {
"id": "chatcmpl-test",
"object": "chat.completion",
"created": 123,
"model": "gguf",
"choices": [
{
"index": 0,
"message": {"role": "assistant", "content": "OK"},
"finish_reason": "stop",
}
],
"usage": {
"prompt_tokens": 1,
"completion_tokens": 1,
"total_tokens": 2,
},
},
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod,
"nonstreaming_client",
lambda: FakeNonStreamingClient(),
)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
stream = True,
stream_options = {"include_usage": True},
tools = [
{
"type": "function",
"function": {
"name": "lookup",
"parameters": {"type": "object", "properties": {}},
},
}
],
)
await _openai_passthrough_non_streaming(
SimpleNamespace(
base_url = "http://llama.test",
context_length = 4096,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"gguf",
monitor_id = monitor_id,
)
assert captured["json"]["stream"] is False
assert "stream_options" not in captured["json"]
[entry] = monitor.snapshot()
assert entry["status"] == "completed"
asyncio.run(_run())
def test_passthrough_clean_eof_finalizes_monitor(self, monkeypatch):
async def _run():
result = await self._run_passthrough_stream(
monkeypatch,
['data: {"choices":[{"delta":{"content":"hello"}}]}'],
)
chunks = result.chunks
assert chunks[0] == 'data: {"choices":[{"delta":{"content":"hello"}}]}\n\n'
compact = "".join(chunks).replace(" ", "")
assert '"finish_reason":"stop"' in compact
assert chunks[-1] == "data: [DONE]\n\n"
[entry] = result.monitor.snapshot()
assert entry["status"] == "completed"
assert entry["reply"] == "hello"
assert result.monitor.active_count() == 0
asyncio.run(_run())
def test_passthrough_finish_without_done_closes_stream_early(self, monkeypatch):
# Some llama-server builds emit the finish chunk and then hold the HTTP
# stream open without sending [DONE]; the terminal classifier must end
# the client stream promptly instead of hanging on the open socket.
async def _run():
import routes.inference as inf_mod
class Request:
async def is_disconnected(self):
return False
async def fake_send(*_args, **_kwargs):
return httpx.Response(200, content = b"")
async def fake_items(*_args, **_kwargs):
yield 'data: {"choices":[{"index":0,"delta":{"content":"hi"},"finish_reason":null}]}'
yield 'data: {"choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}'
await asyncio.Event().wait() # upstream never closes
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "_send_stream_with_preheader_cancel", fake_send)
monkeypatch.setattr(inf_mod, "_aiter_llama_stream_items", fake_items)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
stream = True,
tools = [
{
"type": "function",
"function": {
"name": "lookup",
"parameters": {"type": "object", "properties": {}},
},
}
],
)
response = await _openai_passthrough_stream(
Request(),
threading.Event(),
SimpleNamespace(
base_url = "http://llama.test",
context_length = 4096,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"gguf",
"chatcmpl-test",
monitor_id = monitor_id,
)
async def _consume():
return [chunk async for chunk in response.body_iterator]
chunks = await asyncio.wait_for(_consume(), timeout = 2)
body = "".join(chunks)
assert '"finish_reason":"stop"' in body.replace(" ", "")
assert body.endswith("data: [DONE]\n\n")
[entry] = monitor.snapshot()
assert entry["status"] == "completed"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_passthrough_stall_after_finish_closes_cleanly(self, monkeypatch):
# include_usage keeps the stream open past the finish chunk waiting for
# the usage chunk; if that never arrives, the post-terminal grace path
# must close with a clean [DONE], not an in-band error.
async def _run():
import routes.inference as inf_mod
class Request:
async def is_disconnected(self):
return False
async def fake_send(*_args, **_kwargs):
return httpx.Response(200, content = b"")
async def fake_items(*_args, **_kwargs):
yield 'data: {"choices":[{"index":0,"delta":{"content":"hi"},"finish_reason":null}]}'
yield 'data: {"choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}'
raise httpx.ReadTimeout("usage chunk never arrived")
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "_send_stream_with_preheader_cancel", fake_send)
monkeypatch.setattr(inf_mod, "_aiter_llama_stream_items", fake_items)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
stream = True,
stream_options = {"include_usage": True},
tools = [
{
"type": "function",
"function": {
"name": "lookup",
"parameters": {"type": "object", "properties": {}},
},
}
],
)
response = await _openai_passthrough_stream(
Request(),
threading.Event(),
SimpleNamespace(
base_url = "http://llama.test",
context_length = 4096,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"gguf",
"chatcmpl-test",
monitor_id = monitor_id,
)
chunks = [chunk async for chunk in response.body_iterator]
body = "".join(chunks)
assert '"type":"api_error"' not in body.replace(" ", "")
assert body.endswith("data: [DONE]\n\n")
[entry] = monitor.snapshot()
assert entry["status"] == "completed"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_passthrough_stream_stall_after_data_emits_error(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class Request:
async def is_disconnected(self):
return False
async def fake_send(*_args, **_kwargs):
return httpx.Response(200, content = b"")
async def fake_items(*_args, **_kwargs):
yield 'data: {"choices":[{"delta":{"content":"hello"}}]}'
raise httpx.ReadTimeout("upstream went silent")
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "_send_stream_with_preheader_cancel", fake_send)
monkeypatch.setattr(inf_mod, "_aiter_llama_stream_items", fake_items)
monitor_id = monitor.start(
endpoint = "/v1/chat/completions",
method = "POST",
model = "gguf",
prompt = "hi",
)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
stream = True,
tools = [
{
"type": "function",
"function": {
"name": "lookup",
"parameters": {"type": "object", "properties": {}},
},
}
],
)
response = await _openai_passthrough_stream(
Request(),
threading.Event(),
SimpleNamespace(
base_url = "http://llama.test",
context_length = 4096,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
),
payload,
"gguf",
"chatcmpl-test",
monitor_id = monitor_id,
)
chunks = [chunk async for chunk in response.body_iterator]
body = "".join(chunks)
assert 'data: {"choices":[{"delta":{"content":"hello"}}]}\n\n' in body
assert '"finish_reason"' not in body.replace(" ", "")
assert '"type":"api_error"' in body.replace(" ", "")
assert "still processing the prompt" in body
assert body.endswith("data: [DONE]\n\n")
[entry] = monitor.snapshot()
assert entry["status"] == "error"
assert "still processing the prompt" in entry["error"]
assert entry["reply"] == "hello"
assert monitor.active_count() == 0
asyncio.run(_run())
class TestApiMonitorSafetensorsUsage:
class _Request:
state = SimpleNamespace()
url = SimpleNamespace(path = "/v1/chat/completions")
method = "POST"
def test_non_streaming_safetensors_records_usage(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class DummyBackend:
active_model_name = "safe-model"
models = {"safe-model": {"context_length": 2048}}
def generate_chat_response(self, *, stats_holder, **_kwargs):
stats_holder["stats"] = {
"usage": {
"prompt_tokens": 8,
"completion_tokens": 5,
"total_tokens": 13,
}
}
yield "safe reply"
def reset_generation_state(self, caller_cancel_event = None):
pass
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod,
"get_llama_cpp_backend",
lambda: SimpleNamespace(
is_loaded = False,
supports_tools = False,
is_vision = False,
context_length = None,
),
)
monkeypatch.setattr(inf_mod, "get_inference_backend", lambda: DummyBackend())
monkeypatch.setattr(
inf_mod,
"_detect_safetensors_features",
lambda *_args, **_kwargs: {"supports_tools": False},
)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
)
response = await openai_chat_completions(
payload,
request = self._Request(),
current_subject = "test",
)
body = json.loads(response.body)
assert body["choices"][0]["message"]["content"] == "safe reply"
[entry] = monitor.snapshot()
assert entry["status"] == "completed"
assert entry["reply"] == "safe reply"
assert entry["prompt_tokens"] == 8
assert entry["completion_tokens"] == 5
assert entry["total_tokens"] == 13
assert entry["context_length"] == 2048
asyncio.run(_run())
def test_non_streaming_safetensors_tool_cancel_records_cancelled(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
reset_tool_policy()
class DummyBackend:
active_model_name = "safe-model"
models = {"safe-model": {"context_length": 2048}}
def generate_chat_response(self, **_kwargs):
raise AssertionError("plain safetensors path should not be used")
def generate_chat_completion_with_tools(
self, *, cancel_event, stats_holder, **_kwargs
):
stats_holder["stats"] = {
"usage": {
"prompt_tokens": 8,
"completion_tokens": 5,
"total_tokens": 13,
}
}
yield {"type": "content", "text": "partial"}
cancel_event.set()
yield {"type": "content", "text": "ignored"}
def reset_generation_state(self, caller_cancel_event = None):
pass
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod,
"get_llama_cpp_backend",
lambda: SimpleNamespace(
is_loaded = False,
supports_tools = False,
is_vision = False,
context_length = None,
),
)
monkeypatch.setattr(inf_mod, "get_inference_backend", lambda: DummyBackend())
monkeypatch.setattr(
inf_mod,
"_detect_safetensors_features",
lambda *_args, **_kwargs: {"supports_tools": True},
)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
enable_tools = True,
enabled_tools = ["web_search"],
cancel_id = "safe-cancel",
)
response = await openai_chat_completions(
payload,
request = self._Request(),
current_subject = "test",
)
body = json.loads(response.body)
assert body["choices"][0]["message"]["content"] == "partial"
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert entry["reply"] == "partial"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_non_streaming_safetensors_tool_task_cancel_finalizes_monitor(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
reset_tool_policy()
reset_called = False
class DummyBackend:
active_model_name = "safe-model"
models = {"safe-model": {"context_length": 2048}}
def generate_chat_response(self, **_kwargs):
raise AssertionError("plain safetensors path should not be used")
def generate_chat_completion_with_tools(self, **_kwargs):
yield {"type": "content", "text": "unused"}
def reset_generation_state(self, caller_cancel_event = None):
nonlocal reset_called
reset_called = True
async def fake_to_thread(
func = None,
*_args,
**_kwargs,
):
# Only the generation hop should cancel; resolution runs before the row opens.
if getattr(func, "__name__", "") == "resolve_local_gguf":
return None
raise asyncio.CancelledError()
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod.asyncio, "to_thread", fake_to_thread)
monkeypatch.setattr(
inf_mod,
"get_llama_cpp_backend",
lambda: SimpleNamespace(
is_loaded = False,
supports_tools = False,
is_vision = False,
context_length = None,
),
)
monkeypatch.setattr(inf_mod, "get_inference_backend", lambda: DummyBackend())
monkeypatch.setattr(
inf_mod,
"_detect_safetensors_features",
lambda *_args, **_kwargs: {"supports_tools": True},
)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "hi")],
enable_tools = True,
enabled_tools = ["web_search"],
cancel_id = "safe-cancel",
)
with pytest.raises(asyncio.CancelledError):
await openai_chat_completions(
payload,
request = self._Request(),
current_subject = "test",
)
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert monitor.active_count() == 0
assert reset_called is True
asyncio.run(_run())
class TestApiMonitorAudioInput:
def _patch_audio_backend(self, monkeypatch, chunks):
import routes.inference as inf_mod
class DummyAudioBackend:
active_model_name = "audio-model"
models = {
"audio-model": {
"has_audio_input": True,
"audio_type": "audio-input",
}
}
def generate_audio_input_response(self, **_kwargs):
yield from chunks
monkeypatch.setattr(
inf_mod,
"get_llama_cpp_backend",
lambda: SimpleNamespace(is_loaded = False),
)
monkeypatch.setattr(
inf_mod,
"get_inference_backend",
lambda: DummyAudioBackend(),
)
monkeypatch.setattr(
inf_mod,
"_decode_audio_base64",
lambda _payload: object(),
)
return inf_mod
def test_audio_input_non_streaming_records_active_monitor(self, monkeypatch):
async def _run():
inf_mod = self._patch_audio_backend(monkeypatch, ["hello", " world"])
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "describe this audio")],
audio_base64 = "ZmFrZQ==",
)
request = SimpleNamespace(
state = SimpleNamespace(),
url = SimpleNamespace(path = "/v1/chat/completions"),
method = "POST",
)
response = await openai_chat_completions(
payload,
request = request,
current_subject = "test",
)
body = json.loads(response.body)
assert body["choices"][0]["message"]["content"] == "hello world"
[entry] = monitor.snapshot()
assert entry["status"] == "completed"
assert entry["reply"] == "hello world"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_audio_input_streaming_records_monitor_reply(self, monkeypatch):
async def _run():
inf_mod = self._patch_audio_backend(monkeypatch, ["hello", " world"])
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
async def is_disconnected():
return False
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "describe this audio")],
audio_base64 = "ZmFrZQ==",
stream = True,
)
request = SimpleNamespace(
state = SimpleNamespace(),
url = SimpleNamespace(path = "/v1/chat/completions"),
method = "POST",
is_disconnected = is_disconnected,
)
response = await openai_chat_completions(
payload,
request = request,
current_subject = "test",
)
chunks = []
async for chunk in response.body_iterator:
chunks.append(chunk.decode() if isinstance(chunk, bytes) else chunk)
assert chunks[-1] == "data: [DONE]\n\n"
[entry] = monitor.snapshot()
assert entry["status"] == "completed"
assert entry["reply"] == "hello world"
assert monitor.active_count() == 0
def failing_chunks():
yield "partial"
raise RuntimeError("generation failed")
self._patch_audio_backend(monkeypatch, failing_chunks())
error_monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", error_monitor)
error_response = await openai_chat_completions(
payload,
request = request,
current_subject = "test",
)
error_chunks = [
chunk.decode() if isinstance(chunk, bytes) else chunk
async for chunk in error_response.body_iterator
]
assert '"type": "server_error"' in error_chunks[-1]
assert error_chunks[-1].endswith("data: [DONE]\n\n")
[error_entry] = error_monitor.snapshot()
assert error_entry["status"] == "error"
assert error_monitor.active_count() == 0
asyncio.run(_run())
def test_non_gguf_tts_auto_route_records_monitor(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class DummyTtsBackend:
active_model_name = "tts-model"
models = {
"tts-model": {
"is_audio": True,
"audio_type": "snac",
}
}
async def fake_generate_audio(
_payload,
_request,
current_subject = None,
):
return inf_mod.JSONResponse(
content = {
"choices": [
{
"message": {
"content": "[Generated audio]",
}
}
]
}
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod,
"get_llama_cpp_backend",
lambda: SimpleNamespace(is_loaded = False),
)
monkeypatch.setattr(inf_mod, "get_inference_backend", lambda: DummyTtsBackend())
monkeypatch.setattr(inf_mod, "generate_audio", fake_generate_audio)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "say hello")],
)
request = SimpleNamespace(
state = SimpleNamespace(),
url = SimpleNamespace(path = "/v1/chat/completions"),
method = "POST",
)
response = await inf_mod.openai_chat_completions(
payload,
request = request,
current_subject = "test",
)
assert json.loads(response.body)["choices"][0]["message"]["content"] == (
"[Generated audio]"
)
[entry] = monitor.snapshot()
assert entry["status"] == "completed"
assert entry["model"] == "tts-model"
assert entry["reply"] == "[Generated audio]"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_non_gguf_tts_cancel_finalizes_monitor(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
class DummyTtsBackend:
active_model_name = "tts-model"
models = {
"tts-model": {
"is_audio": True,
"audio_type": "snac",
}
}
async def fake_generate_audio(
_payload,
_request,
current_subject = None,
):
raise asyncio.CancelledError()
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod,
"get_llama_cpp_backend",
lambda: SimpleNamespace(is_loaded = False),
)
monkeypatch.setattr(inf_mod, "get_inference_backend", lambda: DummyTtsBackend())
monkeypatch.setattr(inf_mod, "generate_audio", fake_generate_audio)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "say hello")],
)
request = SimpleNamespace(
state = SimpleNamespace(),
url = SimpleNamespace(path = "/v1/chat/completions"),
method = "POST",
)
with pytest.raises(asyncio.CancelledError):
await inf_mod.openai_chat_completions(
payload,
request = request,
current_subject = "test",
)
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert entry["model"] == "tts-model"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_gguf_tts_auto_route_records_monitor(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
async def fake_generate_audio(
_payload,
_request,
current_subject = None,
):
return inf_mod.JSONResponse(
content = {
"choices": [
{
"message": {
"content": "[Generated audio]",
}
}
]
}
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(
inf_mod,
"get_llama_cpp_backend",
lambda: SimpleNamespace(
is_loaded = True,
_is_audio = True,
model_identifier = "gguf-tts",
context_length = 2048,
),
)
monkeypatch.setattr(inf_mod, "generate_audio", fake_generate_audio)
payload = ChatCompletionRequest(
model = "default",
messages = [ChatMessage(role = "user", content = "say hello")],
)
request = SimpleNamespace(
state = SimpleNamespace(),
url = SimpleNamespace(path = "/v1/chat/completions"),
method = "POST",
)
await inf_mod.openai_chat_completions(
payload,
request = request,
current_subject = "test",
)
[entry] = monitor.snapshot()
assert entry["status"] == "completed"
assert entry["model"] == "gguf-tts"
assert entry["context_length"] == 2048
assert entry["reply"] == "[Generated audio]"
assert monitor.active_count() == 0
asyncio.run(_run())
# =====================================================================
# Responses API -> Chat Completions translation: chat_template_kwargs
# (e.g. {"enable_thinking": true}) sent via the Responses extra-body must
# reach the built ChatCompletionRequest's typed ``enable_thinking`` field,
# otherwise /v1/responses silently ignores reasoning control (issue #6198).
# =====================================================================
class TestResponsesChatTemplateKwargs:
_messages = [ChatMessage(role = "user", content = "What is 100 - 67?")]
class _Request:
app = SimpleNamespace(state = SimpleNamespace(llama_parallel_slots = 1))
state = SimpleNamespace()
url = SimpleNamespace(path = "/v1/responses")
method = "POST"
async def is_disconnected(self):
return False
def test_enable_thinking_lifted_from_extra_body(self):
payload = ResponsesRequest(
model = "qwen-local",
input = "What is 100 - 67?",
chat_template_kwargs = {"enable_thinking": True},
)
chat_req = _build_chat_request(payload, self._messages, stream = False)
assert chat_req.enable_thinking is True
def test_enable_thinking_false_lifted_from_extra_body(self):
payload = ResponsesRequest(
model = "qwen-local",
input = "hi",
chat_template_kwargs = {"enable_thinking": False},
)
chat_req = _build_chat_request(payload, self._messages, stream = True)
assert chat_req.enable_thinking is False
def test_no_chat_template_kwargs_leaves_enable_thinking_unset(self):
payload = ResponsesRequest(model = "qwen-local", input = "hi")
chat_req = _build_chat_request(payload, self._messages, stream = False)
assert chat_req.enable_thinking is None
def test_chat_template_kwargs_without_enable_thinking_is_ignored(self):
payload = ResponsesRequest(
model = "qwen-local",
input = "hi",
chat_template_kwargs = {"some_other_flag": True},
)
chat_req = _build_chat_request(payload, self._messages, stream = False)
assert chat_req.enable_thinking is None
def test_responses_stream_queued_request_sends_keepalive_before_upstream(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
async def fail_send(*_args, **_kwargs):
raise AssertionError("responses upstream must not start while queued")
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
base_url = "http://llama.responses.test",
context_length = 4096,
effective_parallel_slots = 1,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setenv(ADMISSION_KEEPALIVE_INTERVAL_ENV, "0.01")
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
monkeypatch.setattr(inf_mod, "_send_stream_with_preheader_cancel", fail_send)
queue = get_llama_admission_queue("http://llama.responses.test")
blocker = queue.reserve(capacity = 1, config = LlamaAdmissionConfig()).lease_nowait()
assert blocker is not None
monitor_id = monitor.start(
endpoint = "/v1/responses",
method = "POST",
model = "qwen-local",
prompt = "hi",
)
payload = ResponsesRequest(model = "qwen-local", input = "hi", stream = True)
response = await _responses_stream(
payload,
[ChatMessage(role = "user", content = "hi")],
self._Request(),
monitor_id,
)
iterator = response.body_iterator
try:
chunk = await asyncio.wait_for(iterator.__anext__(), timeout = 0.2)
assert chunk == ": keep-alive\n\n"
snapshot = queue.snapshot()
assert snapshot.active == 1
assert snapshot.queued == 1
finally:
aclose = getattr(iterator, "aclose", None)
if aclose is not None:
await aclose()
blocker.release()
snapshot = queue.snapshot()
assert snapshot.active == 0
assert snapshot.queued == 0
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert monitor.active_count() == 0
asyncio.run(_run())
def test_responses_stream_cancel_after_created_finalizes_monitor_and_slot(self, monkeypatch):
async def _run():
import routes.inference as inf_mod
async def fail_send(*_args, **_kwargs):
raise AssertionError("responses upstream must not start after created cancel")
backend = SimpleNamespace(
is_loaded = True,
is_vision = False,
base_url = "http://llama.responses.test",
context_length = 4096,
effective_parallel_slots = 1,
_request_reasoning_kwargs = lambda *_args, **_kwargs: None,
)
monitor = ApiMonitor(max_entries = 3)
monkeypatch.setattr(inf_mod, "api_monitor", monitor)
monkeypatch.setattr(inf_mod, "get_llama_cpp_backend", lambda: backend)
monkeypatch.setattr(inf_mod, "_send_stream_with_preheader_cancel", fail_send)
monitor_id = monitor.start(
endpoint = "/v1/responses",
method = "POST",
model = "qwen-local",
prompt = "hi",
)
payload = ResponsesRequest(model = "qwen-local", input = "hi", stream = True)
response = await _responses_stream(
payload,
[ChatMessage(role = "user", content = "hi")],
self._Request(),
monitor_id,
)
iterator = response.body_iterator
first = await asyncio.wait_for(iterator.__anext__(), timeout = 0.2)
assert "event: response.created" in first
with pytest.raises(asyncio.CancelledError):
await iterator.athrow(asyncio.CancelledError())
assert get_llama_admission_queue("http://llama.responses.test").snapshot().active == 0
[entry] = monitor.snapshot()
assert entry["status"] == "cancelled"
assert monitor.active_count() == 0
asyncio.run(_run())
# =====================================================================
# GGUF chat-template role alternation: coalesce orphaned user turns left
# behind when an empty assistant turn is dropped, so strict templates
# (Gemma 3, ...) do not 400 on a role-parity break.
# =====================================================================
class TestMergeUserContent:
def test_strings_join_with_blank_line(self):
assert _merge_user_content("hi", "again") == "hi\n\nagain"
def test_empty_sides_passthrough(self):
assert _merge_user_content("", "again") == "again"
assert _merge_user_content("hi", "") == "hi"
def test_multimodal_parts_concatenate(self):
img = {"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}}
out = _merge_user_content([{"type": "text", "text": "look"}, img], "and this?")
assert out == [
{"type": "text", "text": "look"},
img,
{"type": "text", "text": "and this?"},
]
class TestCoalesceConsecutiveUserTurns:
def test_merges_two_string_user_turns(self):
msgs = [
{"role": "user", "content": "hi"},
{"role": "user", "content": "again"},
]
assert _coalesce_consecutive_user_turns(msgs) == [
{"role": "user", "content": "hi\n\nagain"},
]
def test_merges_three_consecutive_user_turns(self):
msgs = [
{"role": "user", "content": "a"},
{"role": "user", "content": "b"},
{"role": "user", "content": "c"},
]
assert _coalesce_consecutive_user_turns(msgs) == [
{"role": "user", "content": "a\n\nb\n\nc"},
]
def test_alternating_history_is_unchanged(self):
msgs = [
{"role": "system", "content": "sys"},
{"role": "user", "content": "hi"},
{"role": "assistant", "content": "hello"},
{"role": "user", "content": "bye"},
]
assert _coalesce_consecutive_user_turns(msgs) == msgs
def test_assistant_and_tool_turns_untouched(self):
msgs = [
{"role": "user", "content": "weather?"},
{
"role": "assistant",
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {"name": "get_weather", "arguments": "{}"},
}
],
},
{"role": "tool", "tool_call_id": "call_1", "content": "{}"},
]
assert _coalesce_consecutive_user_turns(msgs) == msgs
def test_multimodal_parts_survive_merge(self):
img = {"type": "image_url", "image_url": {"url": "data:image/png;base64,AAAA"}}
msgs = [
{"role": "user", "content": [{"type": "text", "text": "look"}, img]},
{"role": "user", "content": "and this?"},
]
out = _coalesce_consecutive_user_turns(msgs)
assert len(out) == 1
assert out[0]["content"] == [
{"type": "text", "text": "look"},
img,
{"type": "text", "text": "and this?"},
]
def test_does_not_mutate_input(self):
msgs = [
{"role": "user", "content": "hi"},
{"role": "user", "content": "again"},
]
_coalesce_consecutive_user_turns(msgs)
assert msgs[0]["content"] == "hi"
class TestGgufChatHistoryAlternation:
def test_empty_assistant_turn_dropped_then_users_coalesced(self):
req = ChatCompletionRequest(
model = "default",
messages = [
ChatMessage(role = "user", content = "hi"),
ChatMessage(role = "assistant", content = ""),
ChatMessage(role = "user", content = "again"),
],
)
out, _ = _openai_messages_for_gguf_chat(req, is_vision = False)
roles = [m["role"] for m in out]
assert roles == ["user"]
assert out[0]["content"] == "hi\n\nagain"
def test_bare_stop_sentinel_also_coalesced(self):
req = ChatCompletionRequest(
model = "default",
messages = [
ChatMessage(role = "user", content = "hi"),
ChatMessage(role = "assistant"),
ChatMessage(role = "user", content = "again"),
],
)
out, _ = _openai_messages_for_gguf_chat(req, is_vision = False)
roles = [m["role"] for m in out]
assert all(roles[i] != roles[i + 1] for i in range(len(roles) - 1)), roles
assert roles == ["user"]
def test_system_prompt_preserved(self):
req = ChatCompletionRequest(
model = "default",
messages = [
ChatMessage(role = "system", content = "be brief"),
ChatMessage(role = "user", content = "hi"),
ChatMessage(role = "assistant", content = ""),
ChatMessage(role = "user", content = "again"),
],
)
out, _ = _openai_messages_for_gguf_chat(req, is_vision = False)
assert [m["role"] for m in out] == ["system", "user"]
assert out[1]["content"] == "hi\n\nagain"
def test_normal_history_unchanged(self):
req = ChatCompletionRequest(
model = "default",
messages = [
ChatMessage(role = "user", content = "hi"),
ChatMessage(role = "assistant", content = "hello"),
ChatMessage(role = "user", content = "again"),
],
)
out, _ = _openai_messages_for_gguf_chat(req, is_vision = False)
assert [m["role"] for m in out] == ["user", "assistant", "user"]
def test_tool_path_rebuild_stays_alternating(self):
# Tool path rebuilds via _set_or_prepend_system_message over the coalesced
# history, so it stays alternating too.
req = ChatCompletionRequest(
model = "default",
messages = [
ChatMessage(role = "user", content = "hi"),
ChatMessage(role = "assistant", content = ""),
ChatMessage(role = "user", content = "again"),
],
)
normalized, _ = _openai_messages_for_gguf_chat(req, is_vision = False)
rebuilt = _set_or_prepend_system_message(normalized, "You have access to tools.")
roles = [m["role"] for m in rebuilt]
assert roles == ["system", "user"]
assert all(roles[i] != roles[i + 1] for i in range(len(roles) - 1)), roles