ceos/unsloth
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

studio: scope agentic-loop boundary marker so normal completions stream cleanly

Browse source 
Codex P2 on #5549 flagged that the cycle-5 cursor-reset (efa43c4) keyed
on every empty-status event, but generate_chat_completion_with_tools
emits empty status events in five places, only two of which are real
iteration boundaries:

  - Line 4497: emitted right before `continue` after a re-prompt /
    auto-continue. The next iteration starts a fresh assistant turn.
    BOUNDARY: reset cursor.
  - Line 4766: emitted right before `continue` after a tool call. The
    next iteration regenerates with tool results in history. BOUNDARY:
    reset cursor.
  - Line 4501: emitted at metadata-yield after normal streaming. Stream
    is about to end, no new iteration follows. NOT a boundary.
  - Line 4584: emitted in DRAINING-no-tool-call fallback path. Stream
    is about to end with buffered content_accum. NOT a boundary.
  - Line 4794: emitted at the final exit of the generator. NOT a
    boundary.

Treating all five as boundaries gave every Anthropic-streaming response
an extra content_block_stop + content_block_start pair around its final
text and around every tool call.

Fix by tagging the two real boundary sites with `"boundary": True` and
tightening both the Anthropic emitter (`anthropic_compat.py`) and the
OpenAI-compat tool path + Anthropic non-streaming path
(`routes/inference.py`) to reset the cumulative-text cursor only when
that flag is set. Plain empty-status events keep their existing badge-
clear semantics on the frontend (`tool_status` SSE with content "").

Add two regression tests in
`backend/tests/test_anthropic_messages.py::TestAnthropicStreamEmitter`:

  - test_boundary_flag_closes_block_and_resets_cursor: a boundary=True
    status closes the open text block and the next content delta
    streams from zero.
  - test_empty_status_without_boundary_does_not_close_block: a plain
    empty status leaves block_index unchanged and the next content
    delta is diffed against the previous text length.

107 tests pass across the four anthropic + trailing-plan test files.
This commit is contained in:
danielhanchen 2026-05-19 02:39:35 +00:00
commit 0cc69de210
4 changed files with 80 additions and 21 deletions
Download patch file Download diff file Expand all files Collapse all files

26
studio/backend/core/inference/anthropic_compat.py
View file

@ -253,17 +253,23 @@ class AnthropicStreamEmitter:
elif etype == "metadata":
self._usage = event.get("usage", {})
return []
elif etype == "status" and not event.get("text"):
# Auto-continue boundary marker emitted by
# generate_chat_completion_with_tools — the next "content"
# event resets to a fresh cumulative baseline, so close any
# open text block and clear the diff cursor. Without this
# the next continuation gets diffed against the previous
# turn's length (shorter continuations are dropped, longer
# ones lose their prefix).
elif etype == "status" and event.get("boundary"):
# Iteration-boundary marker emitted by
# generate_chat_completion_with_tools when a fresh model
# turn is about to begin (after an auto-continue re-prompt
# or after a tool result). The next "content" event resets
# to a fresh cumulative baseline, so we close any open text
# block and clear the diff cursor. Without this the next
# continuation gets diffed against the previous turn's
# length (shorter continuations are dropped, longer ones
# lose their prefix). Non-boundary empty status events
# (UI badge clears at normal completion, draining-no-tool
# fallbacks, final stream-end yields) do NOT reach this
# branch and so do NOT produce spurious extra
# content_block_start/stop pairs.
return self._handle_boundary()
# Other status events (tool progress text) have no Anthropic
# equivalent.
# Other status events (tool progress text, non-boundary badge
# clears) have no Anthropic equivalent.
return []
def _handle_boundary(self) -> list[str]:

11
studio/backend/core/inference/llama_cpp.py
View file

@ -4485,7 +4485,10 @@ class LlamaCppBackend:
_it_r = _iter_timings or {}
_accumulated_predicted_ms += _it_r.get("predicted_ms", 0)
_accumulated_predicted_n += _it_r.get("predicted_n", 0)
yield {"type": "status", "text": ""}
# boundary=True: the next agentic iteration
# starts a fresh assistant turn, so adapters
# must reset their cumulative cursor here.
yield {"type": "status", "text": "", "boundary": True}
continue
# Content was already streamed. Yield metadata.
@ -4753,8 +4756,10 @@ class LlamaCppBackend:
tool_msg["tool_call_id"] = tool_call_id
conversation.append(tool_msg)
# Clear tool status badge before next generation iteration
yield {"type": "status", "text": ""}
# Clear tool status badge before next generation iteration.
# boundary=True: the model is about to start a fresh turn
# so cumulative-text adapters must reset their cursor.
yield {"type": "status", "text": "", "boundary": True}
# Continue the loop to let model respond with context
continue

21
studio/backend/routes/inference.py
View file

@ -2459,11 +2459,15 @@ async def openai_chat_completions(
break
if event["type"] == "status":
# Empty status marks an iteration boundary
# in the GGUF tool loop (e.g. after a
# re-prompt). Reset the cumulative cursor
# so the next assistant turn streams cleanly.
if not event["text"]:
# boundary=True flags a true iteration
# boundary (auto-continue re-prompt or
# post-tool resume). Reset the cumulative
# cursor only then; non-boundary empty
# status events (UI badge clears at normal
# stream end) keep the existing cursor so
# we do not spuriously re-emit a duplicate
# prefix on the next "content" yield.
if event.get("boundary"):
prev_text = ""
# Emit tool status as a custom SSE event
# (including empty ones to clear UI badges)
@ -4370,12 +4374,13 @@ async def _anthropic_tool_non_streaming(run_gen, message_id, model_name):
)
elif etype == "tool_end":
prev_text = ""
elif etype == "status" and not event.get("text"):
# Auto-continue boundary marker: the next content event
elif etype == "status" and event.get("boundary"):
# Iteration-boundary marker: the next content event
# restarts the cumulative diff baseline, so reset prev_text
# the same way tool_end does. Without this a shorter
# continuation gets dropped entirely and a longer one
# loses its prefix.
# loses its prefix. Plain empty-status events (UI badge
# clears at normal stream end) do not match this branch.
prev_text = ""
elif etype == "metadata":
usage = event.get("usage", {})

43
studio/backend/tests/test_anthropic_messages.py
View file

@ -629,6 +629,49 @@ class TestAnthropicStreamEmitter:
parsed = json.loads(events[0].split("data: ")[1])
assert parsed["delta"]["text"] == "After tool"
def test_boundary_flag_closes_block_and_resets_cursor(self):
"""An iteration-boundary status (boundary=True) must close the
open text block, open a fresh one, and reset _prev_text so the
next content delta starts from zero."""
e = AnthropicStreamEmitter()
e.start("msg_1", "m")
e.feed({"type": "content", "text": "first turn"})
boundary = e.feed({"type": "status", "text": "", "boundary": True})
# Boundary must produce content_block_stop + content_block_start
# so the next text lives in a new block.
joined = "\n".join(boundary)
assert "content_block_stop" in joined
assert "content_block_start" in joined
# Next content delta must include the full "second turn", not a
# diff against the previous turn's length.
nxt = e.feed({"type": "content", "text": "second turn"})
parsed = json.loads(nxt[0].split("data: ")[1])
assert parsed["delta"]["text"] == "second turn"
def test_empty_status_without_boundary_does_not_close_block(self):
"""A non-boundary empty-status event (UI badge clear at normal
stream end, draining fallbacks, final status yields in
llama_cpp.py at lines 4501, 4584, 4794) must NOT close the
current text block or reset _prev_text - otherwise every normal
Anthropic response gets extra content_block_start/stop pairs
around its final text. Regression test for PR 5549 codex P2."""
e = AnthropicStreamEmitter()
e.start("msg_1", "m")
block_before = e.block_index
e.feed({"type": "content", "text": "hello "})
# Plain empty status (no boundary flag) -> no extra SSE events.
out = e.feed({"type": "status", "text": ""})
assert out == []
# block_index must not have advanced (no close+reopen happened).
assert e.block_index == block_before
# Next content delta is diffed against "hello ", so we only emit
# " world" (the new suffix).
nxt = e.feed({"type": "content", "text": "hello world"})
parsed = json.loads(nxt[0].split("data: ")[1])
assert parsed["delta"]["text"] == "world"
# =====================================================================
# Pass-through emitter tests (client-side tool execution path)