diff --git a/python/freetoken/core.py b/python/freetoken/core.py
index a7811ed712..bbba592378 100644
--- a/python/freetoken/core.py
+++ b/python/freetoken/core.py
@@ -54,6 +54,9 @@ class SamplingParams:
# reports the chosen token's raw (pre-temperature) logprob and top-k alternatives.
logprobs: bool = False
top_logprobs: int = 0
+ # Set by the tokenizer worker when the rendered prompt leaves a block open, so
+ # the detokenizer counts an unclosed generation as reasoning tokens.
+ starts_in_reasoning: bool = False
@property
def is_greedy(self) -> bool:
diff --git a/python/freetoken/engine/sample.py b/python/freetoken/engine/sample.py
index 9d12b5f571..62bdd29cc8 100644
--- a/python/freetoken/engine/sample.py
+++ b/python/freetoken/engine/sample.py
@@ -57,7 +57,10 @@ class BatchSamplingArgs:
greedy_mask: torch.Tensor | None = None
penalties: list[RowPenalty] = field(default_factory=list)
plan: LogitsPlan | None = None
- logprob_rows: torch.Tensor | None = None
+ # Rows that asked for logprobs, as an int64 device index built on the host (no
+ # nonzero sync). None with logprob_all_rows set when every row asks: no gather then.
+ logprob_row_idx: torch.Tensor | None = None
+ logprob_all_rows: bool = False
max_top_logprobs: int = 0
@@ -244,9 +247,12 @@ def prepare(self, batch: Batch) -> BatchSamplingArgs:
)
plan = self._plan(batch, params)
want_logprobs = [p.logprobs for p in params]
- logprob_rows = (
- make_device_tensor(want_logprobs, torch.bool, self.device)
- if any(want_logprobs)
+ logprob_all_rows = bool(want_logprobs) and all(want_logprobs)
+ logprob_row_idx = (
+ make_device_tensor(
+ [i for i, want in enumerate(want_logprobs) if want], torch.int64, self.device
+ )
+ if any(want_logprobs) and not logprob_all_rows
else None
)
max_top_logprobs = min(
@@ -257,7 +263,8 @@ def prepare(self, batch: Batch) -> BatchSamplingArgs:
temperatures=None,
penalties=penalties,
plan=plan,
- logprob_rows=logprob_rows,
+ logprob_row_idx=logprob_row_idx,
+ logprob_all_rows=logprob_all_rows,
max_top_logprobs=max_top_logprobs,
)
@@ -289,7 +296,8 @@ def prepare(self, batch: Batch) -> BatchSamplingArgs:
greedy_mask=greedy_mask,
penalties=penalties,
plan=plan,
- logprob_rows=logprob_rows,
+ logprob_row_idx=logprob_row_idx,
+ logprob_all_rows=logprob_all_rows,
max_top_logprobs=max_top_logprobs,
)
@@ -324,46 +332,50 @@ def compute_logprobs(
sampled_tokens: torch.Tensor,
args: BatchSamplingArgs,
) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor] | None:
- if args.logprob_rows is None:
- return None
-
- requested_rows = torch.nonzero(args.logprob_rows, as_tuple=False).flatten()
- if requested_rows.numel() == 0:
+ rows = args.logprob_row_idx
+ if not args.logprob_all_rows and (rows is None or rows.numel() == 0):
return None
- request_logits = logits.index_select(0, requested_rows).float()
+ request_logits = (logits if rows is None else logits.index_select(0, rows)).float()
# Reported values are raw model logprobs (pre-temperature log_softmax over logits).
request_logprobs = torch.log_softmax(request_logits, dim=-1)
- request_tokens = sampled_tokens.to(dtype=torch.long, device=logits.device).index_select(
- 0, requested_rows
- )
- request_row_idx = torch.arange(requested_rows.numel(), device=logits.device)
+ request_tokens = sampled_tokens.to(dtype=torch.long, device=logits.device)
+ if rows is not None:
+ request_tokens = request_tokens.index_select(0, rows)
+ request_row_idx = torch.arange(request_logprobs.shape[0], device=logits.device)
request_chosen_logprobs = request_logprobs[request_row_idx, request_tokens]
- chosen_logprobs = torch.full(
- (logits.shape[0],), float("nan"), dtype=torch.float32, device=logits.device
- )
- chosen_logprobs.index_copy_(0, requested_rows, request_chosen_logprobs)
+ if rows is None:
+ chosen_logprobs = request_chosen_logprobs
+ else:
+ chosen_logprobs = torch.full(
+ (logits.shape[0],), float("nan"), dtype=torch.float32, device=logits.device
+ )
+ chosen_logprobs.index_copy_(0, rows, request_chosen_logprobs)
if args.max_top_logprobs > 0:
request_top_logprobs, request_top_ids = torch.topk(
request_logprobs, k=args.max_top_logprobs, dim=-1
)
- top_ids = torch.full(
- (logits.shape[0], args.max_top_logprobs),
- -1,
- dtype=torch.int32,
- device=logits.device,
- )
- top_logprobs = torch.full(
- (logits.shape[0], args.max_top_logprobs),
- float("-inf"),
- dtype=torch.float32,
- device=logits.device,
- )
- top_ids[requested_rows] = request_top_ids.to(torch.int32)
- top_logprobs[requested_rows] = request_top_logprobs
+ if rows is None:
+ top_ids = request_top_ids.to(torch.int32)
+ top_logprobs = request_top_logprobs
+ else:
+ top_ids = torch.full(
+ (logits.shape[0], args.max_top_logprobs),
+ -1,
+ dtype=torch.int32,
+ device=logits.device,
+ )
+ top_logprobs = torch.full(
+ (logits.shape[0], args.max_top_logprobs),
+ float("-inf"),
+ dtype=torch.float32,
+ device=logits.device,
+ )
+ top_ids[rows] = request_top_ids.to(torch.int32)
+ top_logprobs[rows] = request_top_logprobs
else:
top_ids = torch.empty((logits.shape[0], 0), dtype=torch.int32, device=logits.device)
top_logprobs = torch.empty((logits.shape[0], 0), dtype=torch.float32, device=logits.device)
diff --git a/python/freetoken/message/frontend.py b/python/freetoken/message/frontend.py
index d70453e910..35620c7e8e 100644
--- a/python/freetoken/message/frontend.py
+++ b/python/freetoken/message/frontend.py
@@ -62,9 +62,13 @@ class UserReply(BaseFrontendMsg):
# On the finished reply: tokens generated up to and including the reasoning end tag
# (0 when the model emitted none) -> usage.completion_tokens_details.reasoning_tokens.
reasoning_tokens: int = 0
- # Neutral sampled-token logprobs entry for this token (see
- # tokenizer.detokenize.build_logprobs_entry); None when the request did not ask.
- logprobs: dict | None = None
+ # Neutral sampled-token logprobs entries released with this reply (see
+ # tokenizer.detokenize.build_logprobs_entry), normally this token's own; None when
+ # there are none. Each entry's "text_offset" is where its token starts in the generated
+ # text. With stop strings an entry waits for its token's text, and entries
+ # of tokens a matched stop string trimmed arrive marked "stop_trimmed"; the token it
+ # ends inside carries "stop_visible", the length of its text that was released.
+ logprobs: list[dict] | None = None
@dataclass
diff --git a/python/freetoken/message/tokenizer.py b/python/freetoken/message/tokenizer.py
index 1b3b49eeeb..b5c24a6e10 100644
--- a/python/freetoken/message/tokenizer.py
+++ b/python/freetoken/message/tokenizer.py
@@ -40,6 +40,8 @@ class DetokenizeMsg(BaseTokenizerMsg):
keep_stop_str: bool = False
# Decode this request with skip_special_tokens (SamplingParams.skip_special_tokens).
skip_special_tokens: bool = False
+ # The prompt left a reasoning block open (SamplingParams.starts_in_reasoning).
+ starts_in_reasoning: bool = False
# Sampled-token logprobs (None unless the request asked): the chosen token's raw
# logprob and the top alternatives, already cut to this request's top_logprobs.
chosen_logprob: float | None = None
diff --git a/python/freetoken/scheduler/scheduler.py b/python/freetoken/scheduler/scheduler.py
index cb6085d344..358ce31355 100644
--- a/python/freetoken/scheduler/scheduler.py
+++ b/python/freetoken/scheduler/scheduler.py
@@ -414,6 +414,7 @@ def _process_last_data(self, last_data: ForwardData | None) -> None:
stop_strs=req.sampling_params.stop_strs or None,
keep_stop_str=req.sampling_params.include_stop_str_in_output,
skip_special_tokens=req.sampling_params.skip_special_tokens,
+ starts_in_reasoning=req.sampling_params.starts_in_reasoning,
prefill_ms=(
0.0 if started_at is None
else (time.monotonic() - started_at) * 1000.0
diff --git a/python/freetoken/server/api_models.py b/python/freetoken/server/api_models.py
index 21b1d1f860..0da3d09e77 100644
--- a/python/freetoken/server/api_models.py
+++ b/python/freetoken/server/api_models.py
@@ -77,9 +77,9 @@ class ChatCompletionRequest(BaseModel):
presence_penalty: float = Field(default=0.0, ge=-2.0, le=2.0, allow_inf_nan=False)
frequency_penalty: float = Field(default=0.0, ge=-2.0, le=2.0, allow_inf_nan=False)
# Sampled-token logprobs (OpenAI chat semantics): top_logprobs (0..20) requires
- # logprobs=true; entries cover generated tokens only (no prompt logprobs).
- # Fail-closed (400) when a reasoning parser or tool parsing is active: entries
- # must align 1:1 with visible content tokens (see _chat_logprobs_conflict).
+ # logprobs=true; entries cover message-content tokens only (no prompt, reasoning or
+ # tool-call tokens). 400 for parser families whose hidden tokens the segmenter
+ # cannot tell apart from content (see _chat_logprobs_conflict).
logprobs: bool = False
top_logprobs: int | None = None
chat_template_kwargs: dict[str, Any] = Field(default_factory=dict)
diff --git a/python/freetoken/server/function_call_parser.py b/python/freetoken/server/function_call_parser.py
index 88cc65b367..e45acea77b 100644
--- a/python/freetoken/server/function_call_parser.py
+++ b/python/freetoken/server/function_call_parser.py
@@ -261,6 +261,11 @@ class BaseFormatDetector(ABC):
# bot_token is a parse trigger, not a uniqueness claim.
toolcall_opener: str | None = None
+ # True when detect_and_parse replays the streaming state machine: its normal_text
+ # is then already every piece of text the stream emits (before, between and after
+ # calls), so parse_non_stream must return it as is.
+ one_shot_replays_stream = False
+
def __init__(self):
# Streaming state management
# Buffer for accumulating incomplete patterns that arrive across multiple streaming chunks
@@ -778,6 +783,12 @@ def parse_streaming_increment(self, new_text: str, tools: List[Tool]) -> Streami
+# A bare invoke opener's name: bounded so a literal "]{1,128})>\s*")
+_PS_BARE_NAME_PARTIAL_RE = re.compile(r"[^\s<>]{0,128}")
+
+
class InvokeParamStreamMixin:
"""Value-level streaming for invoke/parameter block formats (qwen3_coder,
minimax — the same shape vLLM's ParserEngine covers): text outside blocks
@@ -793,6 +804,13 @@ class InvokeParamStreamMixin:
_ps_trim_single True: at most ONE leading/trailing trim char (qwen3_coder)
_ps_missing_type schema type assumed for undeclared params ("string" streams,
"loose" buffers and loose-parses — per-family legacy typing)
+ _ps_trim_after_block True: whitespace after a closed block is held, then
+ dropped if another block or the end follows, kept if
+ text does -- the same for any chunking (qwen3_coder)
+ _ps_bare_invoke True: an invoke without the wrapper block is a call when it
+ names a declared tool and its body starts (qwen3_coder)
+ _ps_literal_open True: a block opener whose next non-whitespace text cannot
+ start an invoke or the block close is text (qwen3_coder)
"""
_ps_outer_open: str = ""
@@ -800,9 +818,15 @@ class InvokeParamStreamMixin:
_ps_trim: str = "\n"
_ps_trim_single: bool = False
_ps_missing_type: str = "string"
+ _ps_trim_after_block: bool = False
+ _ps_bare_invoke: bool = False
+ _ps_literal_open: bool = False
def _ps_reset(self) -> None:
self._ps_mode = "idle"
+ self._ps_after_block = False
+ self._ps_bare = False
+ self._ps_after_bare = False
self._ps_key = ""
self._ps_lead = "{"
self._ps_emitted_any = False
@@ -829,6 +853,58 @@ def _ps_trim_trailing(self, text: str) -> str:
return text[:-1] if text[-1:] and text[-1:] in self._ps_trim else text
return text.rstrip(self._ps_trim)
+ def _ps_bare_status(self, text: str) -> str:
+ """Classify the bare invoke opener ``text`` starts with: "call", "partial"
+ (may still become one) or "literal". A declared name is required -- not
+ the unknown-tool forwarding a wrapped block gets -- and the body must
+ start, so the format quoted in prose or code stays text."""
+ rest = text[len(self._ps_invoke_open_prefix):]
+ m = _PS_BARE_NAME_RE.match(rest)
+ if m is None:
+ return "partial" if _PS_BARE_NAME_PARTIAL_RE.fullmatch(rest) else "literal"
+ if m.group(1) not in self._tool_indices:
+ return "literal"
+ body = rest[m.end():]
+ for tok in (self._ps_param_open_prefix, self._ps_invoke_close):
+ if body.startswith(tok):
+ return "call"
+ if tok.startswith(body):
+ return "partial"
+ return "literal"
+
+ def _ps_find_bare(self, buf: str, limit: int) -> Tuple[int, str]:
+ """First bare invoke opener before ``limit`` that is not literal."""
+ i = buf.find(self._ps_invoke_open_prefix)
+ while i != -1 and i < limit:
+ status = self._ps_bare_status(buf[i:])
+ if status != "literal":
+ return i, status
+ i = buf.find(self._ps_invoke_open_prefix, i + 1)
+ return -1, ""
+
+ def _ps_opens_block(self, text: str) -> str:
+ """Whether ``text`` starts a block: "yes", "no" or "maybe" (too short to tell)."""
+ if text.startswith(self._ps_outer_open):
+ if not self._ps_literal_open:
+ return "yes"
+ body = text[len(self._ps_outer_open):].lstrip()
+ # A repeated opener is a stutter: the block ignores it, as before.
+ for tok in (self._ps_invoke_open_prefix, self._ps_outer_close, self._ps_outer_open):
+ if body.startswith(tok):
+ return "yes"
+ if tok.startswith(body):
+ return "maybe"
+ return "no"
+ if self._ps_outer_open.startswith(text):
+ return "maybe"
+ if self._ps_bare_invoke:
+ prefix = self._ps_invoke_open_prefix
+ if text.startswith(prefix):
+ return {"call": "yes", "partial": "maybe"}.get(self._ps_bare_status(text), "no")
+ if prefix.startswith(text):
+ return "maybe"
+ return "no"
+
def parse_streaming_increment(self, new_text: str, tools: List[Tool]) -> StreamingParseResult:
self._buffer += new_text
if not hasattr(self, "_tool_indices"):
@@ -866,9 +942,42 @@ def _update_prev() -> None:
if mode == "idle":
if calls:
break # text after a call defers to the next step (wire order)
+ if self._ps_after_block:
+ rest = buf.lstrip()
+ if self._ps_after_bare:
+ if rest.startswith(self._ps_outer_close):
+ # The model dropped the opener but kept the closer.
+ self._ps_after_bare = False
+ self._buffer = rest[len(self._ps_outer_close):]
+ continue
+ if self._ps_outer_close.startswith(rest):
+ break
+ self._ps_after_bare = False
+ opens = self._ps_opens_block(rest)
+ if opens == "maybe":
+ break # a separator or a text gap: undecided until more arrives
+ self._ps_after_block = False
+ if opens == "yes":
+ buf = self._buffer = rest
pos = buf.find(self._ps_outer_open)
+ if self._ps_bare_invoke:
+ bare, status = self._ps_find_bare(buf, len(buf) if pos == -1 else pos)
+ if bare > 0:
+ normal_parts.append(buf[:bare])
+ self._buffer = buf[bare:]
+ continue
+ if bare == 0:
+ if status == "partial":
+ break
+ self._ps_bare = True
+ self._ps_mode = "block" # parses the invoke at the buffer head
+ continue
if pos == -1:
hold = self._ends_with_partial_token(buf, self._ps_outer_open)
+ if self._ps_bare_invoke:
+ hold = max(
+ hold, self._ends_with_partial_token(buf, self._ps_invoke_open_prefix)
+ )
release = buf[: len(buf) - hold] if hold else buf
if release:
normal_parts.append(release)
@@ -878,7 +987,13 @@ def _update_prev() -> None:
normal_parts.append(buf[:pos])
self._buffer = buf[pos:]
continue
+ opens = self._ps_opens_block(buf)
+ if opens == "maybe":
+ break
self._buffer = buf[len(self._ps_outer_open):]
+ if opens == "no":
+ normal_parts.append(self._ps_outer_open) # the tag named in prose
+ continue
self._ps_mode = "block"
continue
@@ -888,6 +1003,7 @@ def _update_prev() -> None:
if close != -1 and (inv == -1 or close < inv):
self._buffer = buf[close + len(self._ps_outer_close):]
self._ps_mode = "idle"
+ self._ps_after_block = self._ps_trim_after_block
continue
if inv != -1:
m = self._ps_invoke_open_re.search(buf, inv)
@@ -939,7 +1055,14 @@ def _update_prev() -> None:
while len(self.streamed_args_for_tool) <= self.current_tool_id:
self.streamed_args_for_tool.append("")
self._buffer = buf[e + len(self._ps_invoke_close):]
- self._ps_mode = "block"
+ if self._ps_bare:
+ # A bare invoke is its own block: back to text after it.
+ self._ps_bare = False
+ self._ps_mode = "idle"
+ self._ps_after_block = self._ps_trim_after_block
+ self._ps_after_bare = True
+ else:
+ self._ps_mode = "block"
continue
if p != -1:
m = self._ps_param_open_re.search(buf, p)
@@ -1023,6 +1146,8 @@ def _update_prev() -> None:
def finish_streaming(self) -> str:
residual, self._buffer = self._buffer, ""
mode = getattr(self, "_ps_mode", "idle")
+ if getattr(self, "_ps_after_block", False) and not residual.strip():
+ residual = ""
self._ps_reset()
if mode != "idle" or (self._ps_outer_open and self._ps_outer_open in residual):
return ""
@@ -2016,9 +2141,13 @@ def finish_streaming(self) -> str:
class Qwen3CoderDetector(InvokeParamStreamMixin, BaseFormatDetector):
toolcall_opener = ""
+ one_shot_replays_stream = True
_ps_trim = "\n"
_ps_trim_single = True
_ps_missing_type = "string"
+ _ps_trim_after_block = True
+ _ps_bare_invoke = True
+ _ps_literal_open = True
"""
Detector for Qwen3-Coder XML-style function call format.
@@ -2052,8 +2181,6 @@ def __init__(self):
self.tool_call_separator = "\n"
# Regex patterns
- self.tool_call_block_regex = re.compile(r"(.*?)", re.DOTALL)
- self.function_regex = re.compile(r"||(?=)|$)", re.DOTALL
)
@@ -2063,7 +2190,8 @@ def __init__(self):
self._ps_outer_open = ""
self._ps_outer_close = ""
self._ps_invoke_open_prefix = "]*)>")
+ # No "<" in a name: an opener cut off before its ">" must not run into .
+ self._ps_invoke_open_re = re.compile(r"]*)>")
self._ps_invoke_close = ""
self._ps_param_open_prefix = "]*)>")
@@ -2074,44 +2202,6 @@ def has_tool_call(self, text: str) -> bool:
return " Optional[ToolCallItem]:
- """Parse a single ... block into a ToolCallItem."""
- try:
- end_index = function_str.index(">")
- except ValueError:
- return None
-
- func_name = function_str[:end_index].strip()
- tool_indices = self._get_tool_indices(tools)
- if func_name not in tool_indices and not _should_forward_unknown_tool(func_name):
- logger.warning(f"Model attempted to call undefined function: {func_name}")
- return None
-
- parameters_text = function_str[end_index + 1 :]
- param_config = self._get_param_config(func_name, tools)
- param_dict = {}
-
- for match in self.parameter_regex.findall(parameters_text):
- try:
- idx = match.index(">")
- except ValueError:
- continue
- param_name = match[:idx].strip()
- param_value = match[idx + 1 :]
- # Strip leading/trailing newlines from value
- if param_value.startswith("\n"):
- param_value = param_value[1:]
- if param_value.endswith("\n"):
- param_value = param_value[:-1]
-
- param_dict[param_name] = self._convert_param_value(param_value, param_name, param_config, func_name)
-
- return ToolCallItem(
- tool_index=tool_indices.get(func_name, 0),
- name=func_name,
- parameters=json.dumps(param_dict, ensure_ascii=False),
- )
-
def _build_partial_arguments_json(self, func_name: str, partial_body: str, tools: List[Tool]) -> Optional[str]:
"""Build the current argument JSON from a partial XML tool-call body."""
param_matches = self.parameter_regex.findall(partial_body)
@@ -2155,28 +2245,38 @@ def _build_partial_arguments_json(self, func_name: str, partial_body: str, tools
return json.dumps(param_dict, ensure_ascii=False)
def detect_and_parse(self, text: str, tools: List[Tool]) -> StreamingParseResult:
- idx = text.find(self.bot_token)
- normal_text = text[:idx].strip() if idx != -1 else text
-
- if " StreamingParseResult:
if not self.tools:
return StreamingParseResult(normal_text=full_text, calls=[])
parsed_result = self.detector.detect_and_parse(full_text, self.tools)
+ if self.detector.one_shot_replays_stream:
+ return parsed_result
tool_call_list = parsed_result.calls
if tool_call_list:
# Keep text after the LAST tool block: the streaming path emits it, so
diff --git a/python/freetoken/server/generation.py b/python/freetoken/server/generation.py
index 00d94abebf..e2d4d343a4 100644
--- a/python/freetoken/server/generation.py
+++ b/python/freetoken/server/generation.py
@@ -37,8 +37,16 @@
_TemplateError = ()
from .function_call_parser import FunctionCallParser, TOOLS_TAG_LIST, ToolCallItem
+from .logprobs import (
+ LogprobMarkers,
+ LogprobSegmenter,
+ resolve_logprob_markers,
+ untrimmed,
+)
from .reasoning_parser import (
DSV4_SPECIAL_TOKENS,
+ THINK_END_TOKEN,
+ THINK_START_TOKEN,
ReasoningParser,
build_reasoning_parser,
strip_special_tokens,
@@ -386,6 +394,9 @@ class GenDone:
cached_tokens: int = 0
reasoning_tokens: int = 0
timings: GenTimings = field(default_factory=GenTimings)
+ # Content logprob entries no content delta carried (text held to the end, or
+ # dropped as a trailing separator); None when there are none.
+ logprobs: list[dict] | None = None
GenEvent = ReasoningDelta | ContentDelta | ToolCallStart | ToolCallArgsDelta | ToolCallsDelta | GenDone
@@ -403,8 +414,8 @@ class GenResult:
cached_tokens: int = 0
reasoning_tokens: int = 0
timings: GenTimings = field(default_factory=GenTimings)
- # Neutral sampled-token logprob entries, one per sampled token (empty when the
- # request did not ask).
+ # Neutral logprob entries of the message-content tokens (empty when the request did
+ # not ask).
logprobs: list[dict] = field(default_factory=list)
@@ -740,8 +751,13 @@ async def prerender_error(spec: GenSpec, state: Any) -> GenerationError | None:
def _make_reasoning_parser(spec: GenSpec, state: Any) -> ReasoningParser | None:
"""Build a reasoning parser for this generation, or None if the server has no
- reasoning parser configured. ``force_reasoning`` matches the encode-side
- thinking mode so chat-mode content is never mislabeled as reasoning."""
+ reasoning parser configured."""
+ return build_reasoning_parser(state.config, _force_reasoning(spec, state))
+
+
+def _force_reasoning(spec: GenSpec, state: Any) -> bool:
+ """Whether the generation starts inside reasoning: matches the encode-side thinking
+ mode so chat-mode content is never mislabeled as reasoning."""
parser_name = getattr(state.config, "reasoning_parser", None)
if parser_name == "qwen3":
# The qwen3 chat template opens an implicit (thinking on) unless
@@ -776,7 +792,7 @@ def _make_reasoning_parser(spec: GenSpec, state: Any) -> ReasoningParser | None:
force_reasoning = (
resolve_thinking_mode(spec.chat_template_kwargs, spec.template_tools) == "thinking"
)
- return build_reasoning_parser(state.config, force_reasoning)
+ return force_reasoning
def _split_reasoning(text: str, spec: GenSpec, state: Any) -> tuple[str, str]:
@@ -817,11 +833,75 @@ def _parse_tool_response(
return None
parser = _make_tool_parser(spec, state)
result = parser.parse_non_stream(text)
- if not result.calls:
+ # A detector that replays its stream can drop markup without producing a call
+ # (an empty block): the one-shot content must drop it too.
+ if not result.calls and result.normal_text == text:
return None
return result.normal_text, result.calls
+_UNRESOLVED = object()
+
+
+async def chat_logprob_markers(state: Any) -> LogprobMarkers | None:
+ """The token ids chat logprobs are segmented on, resolved once per server from the
+ frontend tokenizer; None when this server has none."""
+ cached = getattr(state, "_logprob_markers", _UNRESOLVED)
+ if cached is not _UNRESOLVED:
+ return cached
+ markers = None
+ build = getattr(state, "frontend_tokenizer", None)
+ if build is not None:
+ try:
+ manager = await asyncio.to_thread(build)
+ hidden = _QWEN_SEMANTIC_HIDDEN_TOKENS if _uses_qwen_semantic_protocol(state) else ()
+ markers = await asyncio.to_thread(
+ resolve_logprob_markers,
+ manager.tokenizer, THINK_START_TOKEN, THINK_END_TOKEN, hidden,
+ )
+ except Exception: # noqa: BLE001 -- server fault: retried on the next request
+ return None
+ state._logprob_markers = markers
+ return markers
+
+
+def chat_logprobs_supported(state: Any, parse_tools: bool, markers: LogprobMarkers | None) -> bool:
+ """Whether chat logprobs can be confined to message content for this server's parsers:
+ the qwen3 reasoning split needs single-id think markers, and tool markup needs a
+ detector whose one-shot parse replays its stream (the segmenter shadows that stream)."""
+ config = state.config
+ reasoning = getattr(config, "reasoning_parser", None)
+ if reasoning not in (None, "qwen3"):
+ return False
+ if reasoning and (markers is None or markers.think_start is None or markers.think_end is None):
+ return False
+ if parse_tools:
+ name = getattr(config, "tool_call_parser", None)
+ detector = FunctionCallParser.ToolCallParserEnum.get(name)
+ if detector is None or not (detector.one_shot_replays_stream and detector.supports_streaming):
+ return False
+ return True
+
+
+def _wants_logprobs(spec: GenSpec) -> bool:
+ return bool(getattr(spec.sampling_params, "logprobs", False))
+
+
+async def _make_logprob_segmenter(spec: GenSpec, state: Any) -> LogprobSegmenter | None:
+ """The segmenter for a request that asked for logprobs, or None when its entries
+ cannot be confined to message content (they are then all dropped)."""
+ markers = await chat_logprob_markers(state)
+ if not chat_logprobs_supported(state, spec.parse_tools, markers):
+ return None
+ reasoning = getattr(state.config, "reasoning_parser", None) is not None
+ return LogprobSegmenter(
+ markers or LogprobMarkers(None, None, frozenset()),
+ reasoning=reasoning,
+ force_reasoning=reasoning and _force_reasoning(spec, state),
+ tool_parser=_make_tool_parser(spec, state) if spec.parse_tools else None,
+ )
+
+
def _valid_json(text: str) -> bool:
if not text:
return False
@@ -1065,6 +1145,10 @@ def _content_delta(text: str) -> ContentDelta:
pending_logprobs = []
return ContentDelta(text, logprobs=logprobs)
+ # Content entries ride the next content delta; reasoning and tool-call entries are
+ # never collected, so a hidden token's entry cannot surface on a visible chunk.
+ segmenter = await _make_logprob_segmenter(spec, state) if _wants_logprobs(spec) else None
+
tool_parser: FunctionCallParser | None = None
if parse_tools:
try:
@@ -1074,14 +1158,9 @@ def _content_delta(text: str) -> ContentDelta:
if candidate is not None and candidate.supports_streaming():
tool_parser = candidate
frag_stable = tool_parser.args_fragments_prefix_stable() if tool_parser else True
-
- # Logprob entries are collected only on the passthrough path: with a reasoning
- # parser or tool parsing active, text can be hidden, held back, or reclassified,
- # so an entry could come to describe a token that never reaches visible content.
- # The API layer fail-closes those request combinations; this guard is the
- # structural half of the contract — a hidden-token entry must never ride a
- # later visible delta, whatever the caller.
- collect_logprobs = reasoning_parser is None and tool_parser is None and not parse_tools
+ # A detector that replays its stream for the one-shot parse already decides which
+ # whitespace after a call is a separator; dropping more here would split the modes.
+ detector_owns_ws = tool_parser is not None and tool_parser.detector.one_shot_replays_stream
# Streaming tool-call assembly: detectors emit fragments (name first, then
# argument diffs); they accumulate here and the call is emitted complete when
@@ -1105,7 +1184,7 @@ def _semantic_content_events(items: list[Any]) -> list[GenEvent]:
else:
out.append(item)
if isinstance(item, ToolCallsDelta):
- suppress_ws = True
+ suppress_ws = not detector_owns_ws
return out
def _filter_tool_text(piece: str) -> list[GenEvent]:
@@ -1183,8 +1262,8 @@ def _route_tool_text(piece: str) -> list[GenEvent]:
completion_tokens += ack.completion_tokens_delta
cached_tokens += ack.cached_tokens
reasoning_tokens = max(reasoning_tokens, getattr(ack, "reasoning_tokens", 0))
- if collect_logprobs and getattr(ack, "logprobs", None) is not None:
- pending_logprobs.append(ack.logprobs)
+ if segmenter is not None:
+ pending_logprobs.extend(segmenter.feed(untrimmed(getattr(ack, "logprobs", None))))
content_delta = ack.incremental_output
if reasoning_parser is not None and content_delta:
reasoning_delta, content_delta = reasoning_parser.parse_stream_chunk(content_delta)
@@ -1207,6 +1286,9 @@ def _route_tool_text(piece: str) -> list[GenEvent]:
engine_matched_stop = getattr(ack, "matched_stop", None)
break
+ if segmenter is not None:
+ pending_logprobs.extend(segmenter.finish())
+
# Drain residue held in the reasoning parser (a deferred tool block, or a
# trailing partial token) so it is not silently dropped.
if reasoning_parser is not None:
@@ -1274,9 +1356,10 @@ def _route_tool_text(piece: str) -> list[GenEvent]:
normal_text = _filter_content(normal_text)
if normal_text:
yield _content_delta(normal_text)
- yield ToolCallsDelta(tool_calls)
- if finish_reason != "length":
- finish_reason = "tool_calls"
+ if tool_calls:
+ yield ToolCallsDelta(tool_calls)
+ if finish_reason != "length":
+ finish_reason = "tool_calls"
elif parse_tools and pending:
stripped_pending = _filter_content(pending)
if stripped_pending:
@@ -1293,12 +1376,12 @@ def _route_tool_text(piece: str) -> list[GenEvent]:
if trailing_content:
yield _content_delta(trailing_content)
- # Entries without a content delta are intentionally dropped in streaming mode.
yield GenDone(
finish_reason, prompt_tokens, completion_tokens,
matched_stop=engine_matched_stop, cached_tokens=cached_tokens,
reasoning_tokens=reasoning_tokens,
timings=timer.finish(),
+ logprobs=pending_logprobs or None,
)
@@ -1307,6 +1390,7 @@ async def _generate_full_impl(uid: int, spec: GenSpec, state: Any, timer: _Timer
tool calls, strip special tokens. The adapters format the GenResult into their wire."""
full_content = ""
logprob_entries: list[dict] = []
+ wants_logprobs = _wants_logprobs(spec)
prompt_tokens = 0
completion_tokens = 0
cached_tokens = 0
@@ -1322,13 +1406,18 @@ async def _generate_full_impl(uid: int, spec: GenSpec, state: Any, timer: _Timer
cached_tokens += ack.cached_tokens
reasoning_tokens = max(reasoning_tokens, getattr(ack, "reasoning_tokens", 0))
full_content += ack.incremental_output
- if getattr(ack, "logprobs", None) is not None:
- logprob_entries.append(ack.logprobs)
+ if wants_logprobs:
+ logprob_entries.extend(untrimmed(getattr(ack, "logprobs", None)))
if ack.finished:
engine_finish_reason = getattr(ack, "finish_reason", None)
engine_matched_stop = getattr(ack, "matched_stop", None)
break
+ if wants_logprobs:
+ segmenter = await _make_logprob_segmenter(spec, state)
+ logprob_entries = (
+ segmenter.feed(logprob_entries) + segmenter.finish() if segmenter is not None else []
+ )
reasoning_text, content_text = _split_reasoning(full_content, spec, state)
# Engine reason ("stop"/"length"); a tool call overrides it, but a truncation (length) wins.
finish_reason = engine_finish_reason or "stop"
@@ -1336,7 +1425,7 @@ async def _generate_full_impl(uid: int, spec: GenSpec, state: Any, timer: _Timer
parsed = _parse_tool_response(content_text, spec, state)
if parsed is not None:
content_text, tool_calls = parsed
- if finish_reason != "length":
+ if tool_calls and finish_reason != "length":
finish_reason = "tool_calls"
specials = _leaked_special_tokens(state)
diff --git a/python/freetoken/server/logprobs.py b/python/freetoken/server/logprobs.py
index 8618569544..c3d22141a7 100644
--- a/python/freetoken/server/logprobs.py
+++ b/python/freetoken/server/logprobs.py
@@ -2,6 +2,8 @@
from __future__ import annotations
+from collections import deque
+from dataclasses import dataclass
from typing import Any
@@ -21,6 +23,11 @@ def completion_logprobs_error(req: Any) -> str | None:
return None
+def untrimmed(entries: list[dict] | None) -> list[dict]:
+ """A reply's entries minus those of tokens a matched stop string cut from the text."""
+ return [entry for entry in entries or () if not entry.get("stop_trimmed")]
+
+
def chat_content_entry(entry: dict) -> dict:
return {
"token": entry["token"],
@@ -37,24 +44,210 @@ def chat_content_entry(entry: dict) -> dict:
}
-def completions_logprobs(entries: list[dict], start_offset: int = 0) -> dict:
+def completions_logprobs(entries: list[dict], text_len: int | None = None) -> dict:
+ """Each entry's detokenizer-stamped ``text_offset`` is where its token starts in the
+ generated text (the pieces of a split multi-byte character share its start). Given the
+ finished text's ``text_len``, offsets are clipped to it, since a matched stop string
+ trims tokens off the end."""
tokens: list[str] = []
token_logprobs: list[float] = []
top_logprobs: list[dict[str, float]] = []
- text_offset: list[int] = []
- offset = start_offset
for entry in entries:
- token = entry["token"]
- tokens.append(token)
+ tokens.append(entry["token"])
token_logprobs.append(entry["logprob"])
- top_logprobs.append({candidate["token"]: candidate["logprob"] for candidate in entry["top"]})
- text_offset.append(offset)
- offset += len(token)
+ top: dict[str, float] = {}
+ for candidate in entry["top"]:
+ # The legacy shape keys alternatives by text, and partial-UTF-8 candidates
+ # all decode to U+FFFD: keep the most likely one per key.
+ key, logprob = candidate["token"], candidate["logprob"]
+ if key not in top or logprob > top[key]:
+ top[key] = logprob
+ top_logprobs.append(top)
return {
"tokens": tokens,
"token_logprobs": token_logprobs,
"top_logprobs": top_logprobs,
- "text_offset": text_offset,
+ "text_offset": [
+ entry["text_offset"] if text_len is None else min(entry["text_offset"], text_len)
+ for entry in entries
+ ],
}
+
+
+def chat_logprobs_object(entries: list[dict]) -> dict:
+ return {"content": [chat_content_entry(entry) for entry in entries], "refusal": None}
+
+
+@dataclass(frozen=True)
+class LogprobMarkers:
+ """Token ids the chat segmenter keys on, resolved once per server. A think marker is
+ None unless the tokenizer encodes it as exactly one id that decodes back to it."""
+
+ think_start: int | None
+ think_end: int | None
+ # Special and protocol-hidden tokens: never message content.
+ hidden: frozenset[int]
+
+
+def resolve_logprob_markers(
+ tokenizer: Any, think_start: str, think_end: str, hidden_tokens: tuple[str, ...]
+) -> LogprobMarkers:
+ def single_id(marker: str) -> int | None:
+ ids = tokenizer.encode(marker, add_special_tokens=False)
+ if len(ids) == 1 and tokenizer.decode(ids) == marker:
+ return int(ids[0])
+ return None
+
+ hidden = {int(tid) for tid in getattr(tokenizer, "all_special_ids", None) or ()}
+ hidden.update(tid for tid in map(single_id, hidden_tokens) if tid is not None)
+ return LogprobMarkers(single_id(think_start), single_id(think_end), frozenset(hidden))
+
+
+def _visible_text(entry: dict) -> str:
+ """The part of an entry's token that reached the output text."""
+ return entry["token"][: entry.get("stop_visible")]
+
+
+class LogprobSegmenter:
+ """Picks a chat generation's message-content logprob entries out of the sampled ones.
+
+ A pure function of the entry sequence (token ids and single-token strings), so the
+ streaming and non-streaming paths select the same entries for any ack chunking.
+ The reasoning split is keyed on the think marker ids and mirrors the qwen3 parser:
+ a leading ```` (after whitespace) opens reasoning once, the first ````
+ ends it for good, and content after it is whitespace-trimmed at both ends. Tool
+ markup is decided by a shadow tool-call parser fed one token string at a time: its
+ held buffer is always a suffix of what it was fed, so every resolved character is
+ either emitted as text or consumed as markup, and a token is content when any of
+ its characters is emitted.
+ """
+
+ def __init__(
+ self,
+ markers: LogprobMarkers,
+ *,
+ reasoning: bool,
+ force_reasoning: bool,
+ tool_parser: Any | None = None,
+ ) -> None:
+ self._markers = markers
+ self._state = "head" if reasoning else "content"
+ self._force = force_reasoning
+ self._trim = False # content follows a real : trim it at both ends
+ self._content_started = False
+ self._held: list[dict] = [] # whitespace-only tokens awaiting the next text
+ self._tools = tool_parser
+ self._queue: deque[list] = deque() # [entry, start, end, is_content] being resolved
+ self._fed = 0
+ self._out: list[dict] = []
+
+ def feed(self, entries: list[dict]) -> list[dict]:
+ """Content entries that became final, in generation order."""
+ for entry in entries:
+ # Hidden ids skip the parsers whole: the final EOS adds no text to the output.
+ if entry["token_id"] not in self._markers.hidden:
+ self._split_reasoning(entry)
+ out, self._out = self._out, []
+ return out
+
+ def finish(self) -> list[dict]:
+ held, self._held = self._held, []
+ if self._state == "head" and not self._force:
+ for entry in held:
+ self._content(entry)
+ if self._tools is not None:
+ det = self._tools.detector
+ base = self._fed - len(det._buffer)
+ region = det._buffer
+ if self._tools.recover_truncated_call():
+ self._resolve(base, region, "")
+ base, region = self._fed, ""
+ self._resolve(base, region, self._tools.finish_stream())
+ out, self._out = self._out, []
+ return out
+
+ def _split_reasoning(self, entry: dict) -> None:
+ # A token a stop string cut inside is only its visible text, never a marker.
+ tid = -1 if "stop_visible" in entry else entry["token_id"]
+ text = _visible_text(entry)
+ blank = not text.strip()
+ if self._state == "head":
+ if tid == self._markers.think_start:
+ self._held = []
+ self._state = "reasoning"
+ elif blank:
+ self._held.append(entry)
+ elif self._force:
+ self._held = []
+ self._state = "reasoning"
+ self._split_reasoning(entry)
+ else:
+ self._state = "content"
+ held, self._held = self._held, []
+ for e in held + [entry]:
+ self._content(e)
+ elif self._state == "reasoning":
+ if tid == self._markers.think_end:
+ self._state = "content"
+ self._trim = True
+ elif not self._trim:
+ self._content(entry)
+ elif blank:
+ if self._content_started:
+ self._held.append(entry)
+ elif not self._content_started:
+ # The reasoning parser strips this leading whitespace before the tool
+ # parser sees it: " <" opening a bare call is markup, not a space.
+ self._content_started = True
+ self._content(entry, text.lstrip())
+ else:
+ held, self._held = self._held, []
+ for e in held + [entry]:
+ self._content(e)
+
+ def _content(self, entry: dict, text: str | None = None) -> None:
+ if self._tools is None:
+ self._out.append(entry)
+ return
+ text = _visible_text(entry) if text is None else text
+ held = self._tools.detector._buffer
+ base = self._fed - len(held)
+ self._queue.append([entry, self._fed, self._fed + len(text), False])
+ self._fed += len(text)
+ normal = "".join(
+ payload for kind, payload in self._tools.parse_stream_events(text) if kind == "text"
+ )
+ resolved = self._fed - len(self._tools.detector._buffer) - base
+ region = (held + text)[:resolved] if normal else ""
+ self._resolve(base, region, normal, resolved)
+
+ def _resolve(self, base: int, region: str, normal: str, length: int | None = None) -> None:
+ """``region`` (``length`` chars from offset ``base``) left the parser's buffer,
+ emitting ``normal``: mark the tokens it emitted from, then release resolved ones.
+ ``region`` itself is only needed when something was emitted."""
+ if normal:
+ if region.startswith(normal):
+ emitted = [(0, len(normal))]
+ elif region.endswith(normal):
+ # Text after markup in one token, as when a bare call's ">" merges with
+ # the next character (">n"): a leftmost match could land inside the markup.
+ emitted = [(len(region) - len(normal), len(region))]
+ else: # not seen in practice: fall back to the leftmost alignment
+ emitted, i = [], 0
+ for ch in normal:
+ j = region.find(ch, i)
+ if j == -1:
+ break
+ emitted.append((j, j + 1))
+ i = j + 1
+ for item in self._queue:
+ start, end = item[1] - base, item[2] - base
+ if any(lo < end and start < hi for lo, hi in emitted):
+ item[3] = True
+ resolved = base + (len(region) if length is None else length)
+ while self._queue and self._queue[0][2] <= resolved:
+ entry, _, _, is_content = self._queue.popleft()
+ if is_content:
+ self._out.append(entry)
diff --git a/python/freetoken/server/openai_api.py b/python/freetoken/server/openai_api.py
index 0fc8113bc8..4b43074b6e 100644
--- a/python/freetoken/server/openai_api.py
+++ b/python/freetoken/server/openai_api.py
@@ -27,10 +27,11 @@
from .function_call_parser import ToolCallItem
from .request_logger import log_request
from .logprobs import (
- chat_content_entry,
chat_logprobs_error,
+ chat_logprobs_object,
completion_logprobs_error,
completions_logprobs,
+ untrimmed,
)
from .generation import (
DEFAULT_MAX_OUTPUT_TOKENS,
@@ -44,6 +45,8 @@
ToolCallsDelta,
ToolCallStart,
build_metrics,
+ chat_logprob_markers,
+ chat_logprobs_supported,
generate_events,
generate_full,
metrics_enabled,
@@ -261,27 +264,25 @@ def _client_disconnected_response() -> JSONResponse:
)
-def _chat_logprobs_conflict(req: ChatCompletionRequest, state: Any) -> str | None:
- """Fail-close chat logprobs when a semantic parsing layer can hide or reclassify
- generated tokens. A reasoning parser routes tokens out of visible content (and its
- stream buffering re-chunks text), and tool parsing consumes tokens into tool_calls,
- so logprob entries could not be aligned 1:1 with visible content tokens — and hidden
- reasoning-token strings must never surface through logprobs entries. The raw
- /v1/completions path has no semantic layer and keeps full logprobs support."""
+async def _chat_logprobs_conflict(req: ChatCompletionRequest, state: Any) -> str | None:
+ """Refuse chat logprobs when the server's parsers can hide or reclassify generated
+ tokens in a way the logprob segmenter cannot follow: entries must cover message
+ content only, never reasoning or tool-call tokens. The raw /v1/completions path has
+ no semantic layer and keeps full logprobs support."""
if not req.logprobs:
return None
- if getattr(state.config, "reasoning_parser", None):
+ markers = await chat_logprob_markers(state)
+ if not chat_logprobs_supported(state, False, markers):
return (
- "logprobs on /v1/chat/completions are not supported when the server runs a "
- "reasoning parser: reasoning tokens are hidden from message content, so "
- "logprob entries cannot be aligned with it. Use /v1/completions for raw "
- "token logprobs."
+ "logprobs on /v1/chat/completions are not supported with the "
+ f"{state.config.reasoning_parser!r} reasoning parser: reasoning tokens cannot "
+ "be told apart from message content. Use /v1/completions for raw token logprobs."
)
- if _should_parse_tools(req):
+ if not chat_logprobs_supported(state, _should_parse_tools(req), markers):
return (
- "logprobs with tool parsing on /v1/chat/completions are not supported: "
- "tool-call tokens are consumed into tool_calls, so logprob entries cannot "
- "be aligned with message content. Send tool_choice='none' or use "
+ "logprobs with tool parsing on /v1/chat/completions are not supported for the "
+ f"{state.config.tool_call_parser!r} tool-call parser: tool-call tokens cannot be "
+ "told apart from message content. Send tool_choice='none' or use "
"/v1/completions for raw token logprobs."
)
return None
@@ -298,7 +299,7 @@ async def handle_chat_completion(
logprobs_error = chat_logprobs_error(req)
if logprobs_error is not None:
return create_error_response(logprobs_error, param="top_logprobs")
- logprobs_conflict = _chat_logprobs_conflict(req, state)
+ logprobs_conflict = await _chat_logprobs_conflict(req, state)
if logprobs_conflict is not None:
return create_error_response(logprobs_conflict, param="logprobs")
if _response_format_unsupported(req.response_format):
@@ -399,10 +400,13 @@ async def handle_chat_completion(
message["reasoning_content"] = result.reasoning
if result.tool_calls:
message["tool_calls"] = _tool_calls_to_openai(result.tool_calls)
- choice: dict[str, Any] = {"index": index, "message": message, "finish_reason": result.finish_reason}
- if req.logprobs:
- choice["logprobs"] = {"content": [chat_content_entry(e) for e in result.logprobs]}
- choices.append(choice)
+ choices.append({
+ "index": index,
+ "message": message,
+ # OpenAI always sends the key, null when logprobs were not requested
+ "logprobs": chat_logprobs_object(result.logprobs) if req.logprobs else None,
+ "finish_reason": result.finish_reason,
+ })
first = results[0]
response: dict[str, Any] = {
@@ -487,7 +491,7 @@ async def stream_chat_completion_chunks(
elif isinstance(ev, ContentDelta):
choice: dict[str, Any] = {"delta": {"content": ev.text}, "index": index, "finish_reason": None}
if ev.logprobs:
- choice["logprobs"] = {"content": [chat_content_entry(e) for e in ev.logprobs]}
+ choice["logprobs"] = chat_logprobs_object(ev.logprobs)
yield _sse(_chat_chunk(req, uid, [choice]))
elif isinstance(ev, ToolCallStart):
open_tool = {
@@ -568,7 +572,10 @@ async def stream_chat_completion_chunks(
completion_tokens = ev.completion_tokens
cached_tokens = ev.cached_tokens
reasoning_tokens = ev.reasoning_tokens
- yield _sse(_chat_chunk(req, uid, [{"delta": {}, "index": index, "finish_reason": ev.finish_reason}]))
+ done: dict[str, Any] = {"delta": {}, "index": index, "finish_reason": ev.finish_reason}
+ if ev.logprobs:
+ done["logprobs"] = chat_logprobs_object(ev.logprobs)
+ yield _sse(_chat_chunk(req, uid, [done]))
timings = ev.timings
usage = _usage(
@@ -666,8 +673,8 @@ async def _drain_one(uid: int, index: int) -> dict[str, Any] | JSONResponse:
completion_tokens += ack.completion_tokens_delta
cached_tokens += ack.cached_tokens
text += ack.incremental_output
- if req.logprobs is not None and ack.logprobs is not None:
- entries.append(ack.logprobs)
+ if req.logprobs is not None:
+ entries.extend(untrimmed(ack.logprobs))
if ack.finished:
finish_reason = getattr(ack, "finish_reason", None) or "stop"
break
@@ -675,7 +682,9 @@ async def _drain_one(uid: int, index: int) -> dict[str, Any] | JSONResponse:
"index": index,
"text": echo_texts[index // req.n] + text,
"finish_reason": finish_reason,
- "logprobs": completions_logprobs(entries) if req.logprobs is not None else None,
+ "logprobs": (
+ completions_logprobs(entries, len(text)) if req.logprobs is not None else None
+ ),
"_usage": (prompt_tokens, completion_tokens, cached_tokens),
}
@@ -727,7 +736,10 @@ async def stream_completion_chunks(
echo_text: str = "",
) -> AsyncIterator[bytes]:
prompt_tokens = 0
- text_offset = 0
+ sent_len = 0
+ # Logprob entries of acks with no visible text yet (a held partial UTF-8 character, an
+ # EOS), carried to the next chunk that has text; a reply can bring several at once.
+ pending: list[dict] = []
completion_tokens = 0
cached_tokens = 0
finish_reason = "stop"
@@ -743,7 +755,10 @@ async def stream_completion_chunks(
prompt_tokens += ack.prompt_tokens_delta
completion_tokens += ack.completion_tokens_delta
cached_tokens += ack.cached_tokens
+ if req.logprobs is not None:
+ pending.extend(untrimmed(ack.logprobs))
if ack.incremental_output:
+ sent_len += len(ack.incremental_output)
yield _sse(
_completion_chunk(
req, uid,
@@ -751,20 +766,35 @@ async def stream_completion_chunks(
"text": ack.incremental_output,
"index": index,
"finish_reason": None,
+ # Mid-stream, a stamp past sent_len is held-back text still to come.
"logprobs": (
- completions_logprobs([ack.logprobs], text_offset)
- if req.logprobs is not None and ack.logprobs is not None
+ completions_logprobs(pending, sent_len if ack.finished else None)
+ if pending
else None
),
}],
)
)
- text_offset += len(ack.incremental_output)
+ pending = []
if ack.finished:
finish_reason = getattr(ack, "finish_reason", None) or "stop"
break
- yield _sse(_completion_chunk(req, uid, [{"text": "", "index": index, "finish_reason": finish_reason, "logprobs": None}]))
+ # Entries whose text never showed (EOS, a stop string, a trailing partial character)
+ # ride the finish chunk, so the joined stream matches the non-stream list.
+ yield _sse(
+ _completion_chunk(
+ req, uid,
+ [{
+ "text": "",
+ "index": index,
+ "finish_reason": finish_reason,
+ "logprobs": (
+ completions_logprobs(pending, sent_len) if pending else None
+ ),
+ }],
+ )
+ )
usage = _usage(prompt_tokens, completion_tokens, _reported_cached(state, cached_tokens))
if not terminal:
yield _StreamUsage(usage)
diff --git a/python/freetoken/server/reasoning_parser.py b/python/freetoken/server/reasoning_parser.py
index 6788675ca9..c571b97a79 100644
--- a/python/freetoken/server/reasoning_parser.py
+++ b/python/freetoken/server/reasoning_parser.py
@@ -393,7 +393,7 @@ def _scan(self, text: str, *, hold_partial: bool) -> tuple[str, str]:
class ThinkReasoningParser(BaseReasoningParser):
"""Generic ````/```` reasoning parser for models that wrap their
- chain-of-thought in think tags (Qwen3/3.5, GLM-4.x, MiniMax-M2). No DSML tool
+ chain-of-thought in think tags (GLM-4.x, MiniMax-M2). No DSML tool
marker — reasoning ends at ````.
Assumes the model closes ```` before any tool call. These families do
@@ -412,6 +412,113 @@ def __init__(self, force_reasoning: bool = False, stream_reasoning: bool = True)
)
+class Qwen3ReasoningParser(BaseReasoningParser):
+ """Qwen3 ```` parser with one positional rule for both paths.
+
+ Reasoning is open from the start when the template pre-opened it
+ (``force_reasoning``) or when the generation itself opens with ````
+ (stripped once). It ends at the FIRST ```` and is never re-entered:
+ every later ````/```` -- quoted in prose, code or tool-call
+ arguments -- is literal content. Reasoning is whitespace-trimmed at both
+ ends, and so is content that follows a real ````; streaming holds
+ whitespace-only tails so its deltas join to exactly the one-shot result.
+ The one-shot parse replays the streaming machine, so the two cannot drift.
+ """
+
+ def __init__(self, force_reasoning: bool = False, stream_reasoning: bool = True) -> None:
+ super().__init__(
+ think_start_token=THINK_START_TOKEN,
+ think_end_token=THINK_END_TOKEN,
+ force_reasoning=force_reasoning,
+ stream_reasoning=stream_reasoning,
+ )
+ # The generation start, held while it can still grow into ws* + "".
+ self._head: Optional[str] = ""
+ self._trim_content = False
+ self._reasoning_started = False
+ self._content_started = False
+ self._unstreamed_reasoning = ""
+
+ def detect_and_parse(self, text: str) -> ReasoningParseResult:
+ clone = type(self)(
+ force_reasoning=self.force_reasoning, stream_reasoning=self.stream_reasoning
+ )
+ first = clone.parse_streaming_increment(text)
+ rest = clone.flush()
+ return ReasoningParseResult(
+ reasoning_text=first.reasoning_text + rest.reasoning_text,
+ normal_text=first.normal_text + rest.normal_text,
+ )
+
+ def parse_streaming_increment(self, new_text: str) -> ReasoningParseResult:
+ if self._head is not None:
+ self._head += new_text
+ head = self._head.lstrip()
+ if head.startswith(self.think_start_token):
+ self._head = None
+ self._in_reasoning = True
+ return self._consume(head[len(self.think_start_token):], final=False)
+ if not head or self.think_start_token.startswith(head):
+ return ReasoningParseResult()
+ new_text, self._head = self._head, None
+ return self._consume(new_text, final=False)
+
+ def flush(self) -> ReasoningParseResult:
+ text, self._head = self._head or "", None
+ return self._consume(text, final=True)
+
+ def _reasoning_piece(self, piece: str, closing: bool) -> str:
+ if not self._reasoning_started:
+ piece = piece.lstrip()
+ self._reasoning_started = bool(piece)
+ if self.stream_reasoning:
+ return piece
+ self._unstreamed_reasoning += piece
+ if not closing:
+ return ""
+ piece, self._unstreamed_reasoning = self._unstreamed_reasoning, ""
+ return piece
+
+ def _consume(self, text: str, final: bool) -> ReasoningParseResult:
+ buf = self._buffer + text
+ self._buffer = ""
+ reasoning = ""
+ if self._in_reasoning:
+ end = buf.find(self.think_end_token)
+ if end == -1:
+ if final:
+ return ReasoningParseResult(
+ reasoning_text=self._reasoning_piece(buf.rstrip(), closing=True)
+ )
+ safe, _ = self._split_trailing_partial(buf)
+ body = safe.rstrip()
+ self._buffer = buf[len(body):]
+ return ReasoningParseResult(
+ reasoning_text=self._reasoning_piece(body, closing=False)
+ )
+ reasoning = self._reasoning_piece(buf[:end].rstrip(), closing=True)
+ self._in_reasoning = False
+ self._trim_content = True
+ buf = buf[end + len(self.think_end_token):]
+ if not self._trim_content:
+ return ReasoningParseResult(reasoning_text=reasoning, normal_text=buf)
+ if not self._content_started:
+ buf = buf.lstrip()
+ self._content_started = bool(buf)
+ body = buf.rstrip()
+ if not final:
+ self._buffer = buf[len(body):]
+ return ReasoningParseResult(reasoning_text=reasoning, normal_text=body)
+
+ def _split_trailing_partial(self, text: str) -> tuple[str, str]:
+ # Only is tracked: a later is literal, never a marker.
+ token = self.think_end_token
+ for k in range(min(len(token) - 1, len(text)), 0, -1):
+ if text.endswith(token[:k]):
+ return text[:-k], text[-k:]
+ return text, ""
+
+
class MiniMaxM3ReasoningParser(BaseReasoningParser):
"""Reasoning parser for MiniMax-M3's ``...`` protocol.
@@ -876,7 +983,7 @@ class ReasoningParser:
ReasoningParserEnum: Dict[str, Type[BaseReasoningParser]] = {
"deepseekv32": DeepSeekV32ReasoningParser,
"gpt_oss": GptOssHarmonyReasoningParser,
- "qwen3": ThinkReasoningParser,
+ "qwen3": Qwen3ReasoningParser,
"glm": ThinkReasoningParser,
"minimax": ThinkReasoningParser,
"minimax_m3": MiniMaxM3ReasoningParser,
diff --git a/python/freetoken/server/responses_api.py b/python/freetoken/server/responses_api.py
index a84c68cb6d..4240080510 100644
--- a/python/freetoken/server/responses_api.py
+++ b/python/freetoken/server/responses_api.py
@@ -467,6 +467,7 @@ def build_responses_response(
usage=_usage(
result.prompt_tokens, result.completion_tokens,
result.cached_tokens if cache_report else 0,
+ reasoning_tokens=result.reasoning_tokens,
),
incomplete_reason="max_output_tokens" if truncated else None,
metrics=build_metrics(
@@ -504,7 +505,9 @@ def _response_obj(
return response
-def _usage(prompt_tokens: int, completion_tokens: int, cached_tokens: int = 0) -> ResponseUsage:
+def _usage(
+ prompt_tokens: int, completion_tokens: int, cached_tokens: int = 0, reasoning_tokens: int = 0
+) -> ResponseUsage:
# input_tokens stays inclusive of the cached prefix (OpenAI semantics);
# cached_tokens is nonzero only under --enable-cache-report.
return ResponseUsage(
@@ -512,7 +515,7 @@ def _usage(prompt_tokens: int, completion_tokens: int, cached_tokens: int = 0) -
output_tokens=completion_tokens,
total_tokens=prompt_tokens + completion_tokens,
input_tokens_details=InputTokensDetails(cached_tokens=cached_tokens, cache_write_tokens=0),
- output_tokens_details=OutputTokensDetails(reasoning_tokens=0),
+ output_tokens_details=OutputTokensDetails(reasoning_tokens=reasoning_tokens),
)
@@ -550,6 +553,7 @@ def snapshot(status, output, usage=None, incomplete_reason=None, request_metrics
usage_pt = 0
usage_ct = 0
usage_cached = 0
+ usage_reasoning = 0
def close_current() -> list[str]:
nonlocal current, output_index
@@ -735,6 +739,7 @@ def args_delta_frame(fragment: str) -> str:
finish_reason = ev.finish_reason
usage_pt, usage_ct = ev.prompt_tokens, ev.completion_tokens
usage_cached = ev.cached_tokens if cache_report else 0
+ usage_reasoning = ev.reasoning_tokens
request_metrics = build_metrics(
prompt_tokens=ev.prompt_tokens,
completion_tokens=ev.completion_tokens,
@@ -748,7 +753,7 @@ def args_delta_frame(fragment: str) -> str:
type="response.incomplete", sequence_number=seq.next(),
response=snapshot(
"incomplete", output_items,
- usage=_usage(usage_pt, usage_ct, usage_cached),
+ usage=_usage(usage_pt, usage_ct, usage_cached, usage_reasoning),
incomplete_reason="max_output_tokens",
request_metrics=request_metrics,
),
@@ -758,7 +763,7 @@ def args_delta_frame(fragment: str) -> str:
type="response.completed", sequence_number=seq.next(),
response=snapshot(
"completed", output_items,
- usage=_usage(usage_pt, usage_ct, usage_cached),
+ usage=_usage(usage_pt, usage_ct, usage_cached, usage_reasoning),
request_metrics=request_metrics,
),
))
@@ -772,7 +777,7 @@ def args_delta_frame(fragment: str) -> str:
type="response.failed", sequence_number=seq.next(),
response=_response_obj(
response_id, created, req.model, output_items, status="failed",
- usage=_usage(usage_pt, usage_ct, usage_cached),
+ usage=_usage(usage_pt, usage_ct, usage_cached, usage_reasoning),
# codex reads this code to tell a blown context window from a generic failure.
# model_construct because ResponseError.code is a closed Literal in the SDK.
error=ResponseError.model_construct(
@@ -787,7 +792,7 @@ def args_delta_frame(fragment: str) -> str:
type="response.failed", sequence_number=seq.next(),
response=_response_obj(
response_id, created, req.model, output_items, status="failed",
- usage=_usage(usage_pt, usage_ct, usage_cached),
+ usage=_usage(usage_pt, usage_ct, usage_cached, usage_reasoning),
error=ResponseError(code="server_error", message=str(exc)),
),
))
diff --git a/python/freetoken/tokenizer/detokenize.py b/python/freetoken/tokenizer/detokenize.py
index 2eb3029df0..44c3149948 100644
--- a/python/freetoken/tokenizer/detokenize.py
+++ b/python/freetoken/tokenizer/detokenize.py
@@ -1,5 +1,7 @@
-from dataclasses import dataclass
-from typing import Dict, FrozenSet, List
+import json
+import weakref
+from dataclasses import dataclass, field
+from typing import Any, Dict, FrozenSet, List
from freetoken.message import DetokenizeMsg
from transformers import PreTrainedTokenizerBase
@@ -73,6 +75,45 @@ class DecodeStatus:
sent_offset: int # length of sent out string
skip_special: bool = False # decode this request with skip_special_tokens
think_end_at: int | None = None # generated tokens up to and incl. the reasoning end tag
+ in_reasoning: bool = False # the prompt or the output opened a reasoning block
+ generated: int = 0 # sampled tokens, EOS included
+ text_end: int = 0 # decoded text so far, before holdback: where the next token starts
+ # (text start, entry) of logprob entries whose token text is still held for a stop string
+ held_logprobs: List[tuple] = field(default_factory=list)
+
+ def reasoning_tokens(self) -> int:
+ if self.think_end_at is not None:
+ return self.think_end_at
+ # stopped inside the block (max_tokens, or EOS before the end tag): all of it
+ return self.generated if self.in_reasoning else 0
+
+ def release_logprobs(
+ self, entry: dict | None, token_start: int, emit_end: int, text_len: int, flush: bool
+ ) -> List[dict]:
+ """Hold this token's logprob entry and release those whose text is now sent."""
+ if entry is not None:
+ self.held_logprobs.append((token_start, {**entry, "text_offset": token_start}))
+ out: List[dict] = []
+ # A held token's text ends where the next one's starts.
+ starts = [start for start, _ in self.held_logprobs]
+ ends = starts[1:] + [text_len] if starts else []
+ if flush:
+ trimmed = emit_end < text_len
+ for (start, held), end in zip(self.held_logprobs, ends, strict=True):
+ if trimmed and start >= emit_end:
+ held = {**held, "stop_trimmed": True}
+ elif trimmed and emit_end < end:
+ held = {**held, "stop_visible": emit_end - start}
+ out.append(held)
+ self.held_logprobs = []
+ elif self.held_logprobs:
+ # Wait for all of a token's text, so a stop match inside it can still mark it.
+ n = 0
+ while n < len(ends) and starts[n] < emit_end and ends[n] <= emit_end:
+ n += 1
+ out = [held for _, held in self.held_logprobs[:n]]
+ self.held_logprobs = self.held_logprobs[n:]
+ return out
def update(self, read_str: str, surr_str: str) -> str:
new_text = read_str[len(surr_str) :]
@@ -99,6 +140,7 @@ def __init__(
tokenizer: PreTrainedTokenizerBase,
eos_token_ids: FrozenSet[int] | None = None,
think_end_id: int | None = None,
+ think_start_id: int | None = None,
) -> None:
# uid -> DecodeStatus
self.decode_map: Dict[int, DecodeStatus] = {}
@@ -109,6 +151,7 @@ def __init__(
# The reasoning end tag's id (e.g. ````): tokens generated up to and
# including it are reported as usage.completion_tokens_details.reasoning_tokens.
self.think_end_id = think_end_id if think_end_id is None else int(think_end_id)
+ self.think_start_id = think_start_id if think_start_id is None else int(think_start_id)
def discard(self, uid: int) -> None:
"""Drop a uid's decode state without a finished DetokenizeMsg. An aborted or errored
@@ -131,13 +174,42 @@ def _batch_decode(self, ids: List[List[int]], skip: List[bool]) -> List[str]:
out[i] = t
return [t if t is not None else "" for t in out]
- def detokenize_with_meta(self, msgs: List[DetokenizeMsg]) -> tuple[List[str], List[int]]:
- """Incremental text per message plus, for a finished message, its reasoning token
- count (0 otherwise or when the model emitted no reasoning end tag)."""
+ def detokenize_with_meta(
+ self, msgs: List[DetokenizeMsg]
+ ) -> tuple[List[str], List[int], List[int]]:
+ """Incremental text per message; for a finished message, its reasoning token count
+ (0 otherwise or when the generation never was in a reasoning block); and where each
+ message's token starts in the generated text. That offset ignores stop-string
+ holdback, which delays when text is sent but not where it lands."""
+ texts, counts, _, offsets = self._detokenize(msgs, [None] * len(msgs))
+ return texts, counts, offsets
+
+ def detokenize_with_logprobs(
+ self, msgs: List[DetokenizeMsg], entries: List[dict | None]
+ ) -> tuple[List[str], List[int], List[List[dict]]]:
+ """``detokenize_with_meta`` plus the logprob entries each reply carries, each stamped
+ with its token's ``text_offset``. With stop strings, an entry waits for the reply that
+ releases all its token's text, so it never precedes text that may still be trimmed;
+ entries of tokens a matched stop string trimmed come on the finished reply marked
+ ``stop_trimmed``, and that of a token it cut inside with ``stop_visible``, the length
+ of the token's text left visible."""
+ texts, counts, released, _ = self._detokenize(msgs, entries)
+ return texts, counts, released
+
+ def _detokenize(
+ self, msgs: List[DetokenizeMsg], entries: List[dict | None]
+ ) -> tuple[List[str], List[int], List[List[dict]], List[int]]:
# Each message must advance its request's decode state before the next one.
if len({msg.uid for msg in msgs}) != len(msgs):
- parts = [self.detokenize_with_meta([msg]) for msg in msgs]
- return [p[0][0] for p in parts], [p[1][0] for p in parts]
+ parts = [
+ self._detokenize([msg], [entry]) for msg, entry in zip(msgs, entries, strict=True)
+ ]
+ return (
+ [p[0][0] for p in parts],
+ [p[1][0] for p in parts],
+ [p[2][0] for p in parts],
+ [p[3][0] for p in parts],
+ )
read_ids: List[List[int]] = []
surr_ids: List[List[int]] = []
skip: List[bool] = []
@@ -150,8 +222,10 @@ def detokenize_with_meta(self, msgs: List[DetokenizeMsg]) -> tuple[List[str], Li
surr_offset=0,
sent_offset=0,
skip_special=bool(getattr(msg, "skip_special_tokens", False)),
+ in_reasoning=bool(getattr(msg, "starts_in_reasoning", False)),
)
s = self.decode_map[msg.uid]
+ s.generated += 1
if not (
msg.finished and not msg.matched_stop and msg.next_token in self.eos_token_ids
):
@@ -162,6 +236,8 @@ def detokenize_with_meta(self, msgs: List[DetokenizeMsg]) -> tuple[List[str], Li
and msg.next_token == self.think_end_id
):
s.think_end_at = len(s.decoded_ids)
+ if s.think_end_at is None and msg.next_token == self.think_start_id:
+ s.in_reasoning = True
read_ids.append(s.decoded_ids[s.surr_offset :])
surr_ids.append(s.decoded_ids[s.surr_offset : s.read_offset])
skip.append(s.skip_special)
@@ -171,9 +247,16 @@ def detokenize_with_meta(self, msgs: List[DetokenizeMsg]) -> tuple[List[str], Li
incremental_strs: List[str] = []
reasoning_tokens: List[int] = []
- for msg, read_str, surr_str in zip(msgs, read_texts, surr_texts, strict=True):
+ released: List[List[dict]] = []
+ text_offsets: List[int] = []
+ for msg, entry, read_str, surr_str in zip(
+ msgs, entries, read_texts, surr_texts, strict=True
+ ):
s = self.decode_map[msg.uid]
+ token_start = s.text_end
+ text_offsets.append(token_start)
output_str = s.update(read_str, surr_str)
+ s.text_end = max(s.text_end, len(output_str))
prev_sent = s.sent_offset
if msg.finished:
@@ -197,11 +280,83 @@ def detokenize_with_meta(self, msgs: List[DetokenizeMsg]) -> tuple[List[str], Li
incremental_output = output_str[prev_sent:emit_end] if emit_end > prev_sent else ""
s.sent_offset = max(prev_sent, emit_end)
incremental_strs.append(incremental_output)
- reasoning_tokens.append((s.think_end_at or 0) if msg.finished else 0)
+ reasoning_tokens.append(s.reasoning_tokens() if msg.finished else 0)
+ if entry is not None or s.held_logprobs:
+ flush = msg.finished or not msg.stop_strs
+ out = s.release_logprobs(entry, token_start, emit_end, len(output_str), flush)
+ else:
+ out = [] # no logprobs asked for: skip the hold bookkeeping
+ released.append(out)
if msg.finished:
del self.decode_map[msg.uid]
- return incremental_strs, reasoning_tokens
+ return incremental_strs, reasoning_tokens, released, text_offsets
+
+
+def _gpt2_byte_decoder() -> dict[str, int]:
+ """Inverse of GPT-2's ``bytes_to_unicode``: the printable stand-in char -> its byte."""
+ bs = [*range(ord("!"), ord("~") + 1), *range(ord("¡"), ord("¬") + 1), *range(ord("®"), ord("ÿ") + 1)]
+ cs = bs[:]
+ n = 0
+ for b in range(256):
+ if b not in bs:
+ bs.append(b)
+ cs.append(256 + n)
+ n += 1
+ return {chr(c): b for b, c in zip(bs, cs, strict=True)}
+
+
+class _TokenBytes:
+ """A token id's own bytes. A single-id decode cannot give them for a piece of a
+ multi-byte character (it decodes to U+FFFD), so read the vocab piece instead."""
+
+ def __init__(self, tokenizer: Any) -> None:
+ decoder: dict = {}
+ backend = getattr(tokenizer, "backend_tokenizer", None)
+ try:
+ decoder = json.loads(backend.decoder.__getstate__())
+ except Exception: # noqa: BLE001 -- no fast backend: fall back to the decoded text
+ pass
+ kinds = {d.get("type") for d in decoder.get("decoders", [decoder])}
+ self.byte_level = _gpt2_byte_decoder() if "ByteLevel" in kinds else None
+ self.byte_fallback = "ByteFallback" in kinds
+ added = getattr(tokenizer, "added_tokens_decoder", None) or {}
+ self.added = {int(tid): str(getattr(tok, "content", tok)) for tid, tok in added.items()}
+ self.convert = getattr(tokenizer, "convert_ids_to_tokens", None)
+
+ def __call__(self, tid: int, text: str) -> bytes:
+ if tid in self.added:
+ # the GPT-2 table cannot invert an added token's raw (e.g. full-width) chars
+ return self.added[tid].encode("utf-8")
+ try:
+ piece = self.convert(tid) if self.convert is not None else None
+ except Exception: # noqa: BLE001 -- out-of-vocab id
+ piece = None
+ if isinstance(piece, str):
+ if self.byte_level is not None:
+ try:
+ return bytes(self.byte_level[c] for c in piece)
+ except KeyError:
+ pass
+ elif self.byte_fallback and len(piece) == 6 and piece.startswith("<0x") and piece.endswith(">"):
+ try:
+ return bytes([int(piece[3:5], 16)])
+ except ValueError:
+ pass
+ return text.encode("utf-8")
+
+
+_TOKEN_BYTES: "weakref.WeakKeyDictionary[Any, _TokenBytes]" = weakref.WeakKeyDictionary()
+
+
+def _token_bytes(tokenizer: Any) -> _TokenBytes:
+ try:
+ cached = _TOKEN_BYTES.get(tokenizer)
+ if cached is None:
+ cached = _TOKEN_BYTES[tokenizer] = _TokenBytes(tokenizer)
+ except TypeError: # not weak-referenceable or unhashable
+ return _TokenBytes(tokenizer)
+ return cached
def build_logprobs_entry(
@@ -213,21 +368,27 @@ def build_logprobs_entry(
) -> dict:
"""Neutral sampled-token logprobs entry (UserReply.logprobs). Values are the raw
pre-temperature logprobs the sampler computed; token text comes from a single-id
- decode, with UTF-8 bytes alongside so clients can reassemble partial-UTF-8 pieces."""
+ decode, and ``bytes`` are the token's own bytes, so clients can reassemble a
+ character split across tokens. Ids past the real vocab (the padded tail of the
+ logits) report an empty token."""
+ token_bytes = _token_bytes(tokenizer)
def _fields(tid: int) -> tuple[str, list[int]]:
- text = tokenizer.decode([tid])
- return text, list(text.encode("utf-8"))
+ try:
+ text = tokenizer.decode([tid])
+ except Exception: # noqa: BLE001 -- a padded id some tokenizers refuse to decode
+ text = ""
+ return text, list(token_bytes(tid, text))
- token, token_bytes = _fields(token_id)
+ token, data = _fields(token_id)
top = []
for tid, logprob in zip(top_ids or [], top_logprobs or []):
- text, data = _fields(int(tid))
- top.append({"token_id": int(tid), "token": text, "bytes": data, "logprob": float(logprob)})
+ text, cand = _fields(int(tid))
+ top.append({"token_id": int(tid), "token": text, "bytes": cand, "logprob": float(logprob)})
return {
"token_id": token_id,
"token": token,
- "bytes": token_bytes,
+ "bytes": data,
"logprob": float(chosen_logprob),
"top": top,
}
diff --git a/python/freetoken/tokenizer/server.py b/python/freetoken/tokenizer/server.py
index 4c650042b4..43c5515761 100644
--- a/python/freetoken/tokenizer/server.py
+++ b/python/freetoken/tokenizer/server.py
@@ -84,7 +84,11 @@ def _send_generation_replies(
def _reasoning_end_id(tokenizer) -> int | None:
"""The id of the reasoning end tag when the vocabulary has one (```` for the
Qwen / DeepSeek / GLM families), else None. Only used to count reasoning tokens."""
- for tag in ("", "<|end_of_thinking|>"):
+ return _vocab_tag_id(tokenizer, ("", "<|end_of_thinking|>"))
+
+
+def _vocab_tag_id(tokenizer, tags: tuple[str, ...]) -> int | None:
+ for tag in tags:
try:
tid = tokenizer.convert_tokens_to_ids(tag)
except Exception: # noqa: BLE001 -- a tokenizer without the tag answers unk or raises
@@ -94,10 +98,20 @@ def _reasoning_end_id(tokenizer) -> int | None:
return None
+def _prompt_opens_reasoning(input_ids: torch.Tensor, start_id: int, end_id: int) -> bool:
+ """The prompt's last reasoning tag is an opening one: the template pre-opened the
+ block (qwen3 with thinking on), so generation starts inside it. A raw prompt that just
+ mentions a literal trips this too; harmless, since only the chat and responses
+ endpoints, whose prompts are always templated, report reasoning_tokens."""
+ tags = torch.nonzero((input_ids == start_id) | (input_ids == end_id)).flatten()
+ return tags.numel() > 0 and int(input_ids[tags[-1]]) == start_id
+
+
def _tokenize_requests(
tokenize_manager: Any,
messages: List[TokenizeMsg],
logger: Any,
+ reasoning_tags: tuple[int, int] | None = None,
) -> tuple[List[UserMsg], List[UserReply]]:
"""Tokenize independently, returning backend work plus terminal frontend errors.
@@ -132,6 +146,10 @@ def _tokenize_requests(
)
)
continue
+ if reasoning_tags is not None:
+ user_msg.sampling_params.starts_in_reasoning = _prompt_opens_reasoning(
+ user_msg.input_ids, *reasoning_tags
+ )
backend.append(user_msg)
return backend, errors
@@ -163,10 +181,18 @@ def tokenize_worker(
from .tokenize import TokenizeManager
tokenize_manager = TokenizeManager(tokenizer, get_mm_processor(tokenizer_path, mm))
+ think_start_id = _vocab_tag_id(tokenizer, ("",))
+ think_end_id = _reasoning_end_id(tokenizer)
+ reasoning_tags = (
+ (think_start_id, think_end_id)
+ if think_start_id is not None and think_end_id is not None
+ else None
+ )
detokenize_manager = DetokenizeManager(
tokenizer,
load_eos_token_ids(tokenizer_path, tokenizer),
- think_end_id=_reasoning_end_id(tokenizer),
+ think_end_id=think_end_id,
+ think_start_id=think_start_id,
)
if ack_queue is not None:
@@ -224,7 +250,21 @@ def tokenize_worker(
)
sampled_replies: List[UserReply] = []
if len(detokenize_msg) > 0:
- replies, reasoning_counts = detokenize_manager.detokenize_with_meta(detokenize_msg)
+ entries = [
+ build_logprobs_entry(
+ detokenize_manager.tokenizer,
+ msg.next_token,
+ msg.chosen_logprob,
+ msg.top_ids,
+ msg.top_logprobs,
+ )
+ if msg.chosen_logprob is not None
+ else None
+ for msg in detokenize_msg
+ ]
+ replies, reasoning_counts, released = detokenize_manager.detokenize_with_logprobs(
+ detokenize_msg, entries
+ )
sampled_replies = [
UserReply(
uid=msg.uid,
@@ -242,22 +282,10 @@ def tokenize_worker(
swa_total_tokens=msg.swa_total_tokens,
gpu_mem_bytes=msg.gpu_mem_bytes,
prefill_ms=msg.prefill_ms,
- # Stop-string trimming can hide final visible text; keep one logprob
- # entry per sampled token.
- logprobs=(
- build_logprobs_entry(
- detokenize_manager.tokenizer,
- msg.next_token,
- msg.chosen_logprob,
- msg.top_ids,
- msg.top_logprobs,
- )
- if msg.chosen_logprob is not None
- else None
- ),
+ logprobs=reply_logprobs or None,
)
- for msg, reply, reasoning_tokens in zip(
- detokenize_msg, replies, reasoning_counts, strict=True
+ for msg, reply, reasoning_tokens, reply_logprobs in zip(
+ detokenize_msg, replies, reasoning_counts, released, strict=True
)
]
@@ -279,7 +307,9 @@ def tokenize_worker(
# Tokenize per-message so a single un-renderable request (e.g. a chat template
# that rejects the message layout) becomes a terminal error reply for THAT uid
# instead of an uncaught exception that kills the worker and bricks the server.
- backend, errors = _tokenize_requests(tokenize_manager, tokenize_msg, logger)
+ backend, errors = _tokenize_requests(
+ tokenize_manager, tokenize_msg, logger, reasoning_tags
+ )
if errors:
send_frontend.put(
errors[0] if len(errors) == 1 else BatchFrontendMsg(data=errors)
diff --git a/tests/engine/test_sample_logprobs.py b/tests/engine/test_sample_logprobs.py
index 88fffcc2ed..02dff731ef 100644
--- a/tests/engine/test_sample_logprobs.py
+++ b/tests/engine/test_sample_logprobs.py
@@ -20,7 +20,7 @@ def test_compute_logprobs_matches_raw_log_softmax_and_sorted_top() -> None:
sampled_tokens = torch.tensor([2, 3, 0], dtype=torch.long)
args = BatchSamplingArgs(
temperatures=torch.tensor([1.0, 1.0, 1.0], dtype=torch.float32),
- logprob_rows=torch.tensor([True, False, True], dtype=torch.bool),
+ logprob_row_idx=torch.tensor([0, 2]),
max_top_logprobs=3,
)
@@ -58,16 +58,15 @@ def test_compute_logprobs_ignores_temperatures() -> None:
dtype=torch.float32,
)
sampled_tokens = torch.tensor([1, 2], dtype=torch.long)
- logprob_rows = torch.tensor([True, True], dtype=torch.bool)
cold = BatchSamplingArgs(
temperatures=torch.full((2,), 0.3, dtype=torch.float32),
- logprob_rows=logprob_rows,
+ logprob_all_rows=True,
max_top_logprobs=2,
)
hot = BatchSamplingArgs(
temperatures=torch.full((2,), 2.5, dtype=torch.float32),
- logprob_rows=logprob_rows,
+ logprob_all_rows=True,
max_top_logprobs=2,
)
@@ -89,7 +88,7 @@ def test_compute_logprobs_returns_none_without_requested_rows() -> None:
sampled_tokens = torch.tensor([0, 1], dtype=torch.long)
args = BatchSamplingArgs(
temperatures=torch.ones(2),
- logprob_rows=torch.zeros(2, dtype=torch.bool),
+ logprob_row_idx=torch.empty(0, dtype=torch.int64),
max_top_logprobs=2,
)
@@ -109,7 +108,7 @@ def test_compute_logprobs_handles_zero_top_logprobs() -> None:
sampled_tokens = torch.tensor([3, 0, 1], dtype=torch.long)
args = BatchSamplingArgs(
temperatures=torch.full((3,), 1.0),
- logprob_rows=torch.tensor([True, False, True], dtype=torch.bool),
+ logprob_row_idx=torch.tensor([0, 2]),
max_top_logprobs=0,
)
@@ -166,3 +165,57 @@ def test_compute_logprobs_with_prepare_keeps_max_top_logprobs_clamped(monkeypatc
assert top_logprobs.shape == (2, sampler.vocab_size)
assert torch.equal(top_ids[1], torch.full((sampler.vocab_size,), -1, dtype=torch.int32))
assert torch.isneginf(top_logprobs[1]).all()
+
+
+def _reference_logprobs(logits, sampled_tokens, want, k):
+ """The pre-index implementation (bool mask + torch.nonzero), kept as the oracle."""
+ mask = torch.tensor(want, dtype=torch.bool)
+ rows = torch.nonzero(mask, as_tuple=False).flatten()
+ if rows.numel() == 0:
+ return None
+ logprobs = torch.log_softmax(logits.index_select(0, rows).float(), dim=-1)
+ chosen = torch.full((logits.shape[0],), float("nan"))
+ chosen.index_copy_(
+ 0, rows, logprobs[torch.arange(rows.numel()), sampled_tokens.long().index_select(0, rows)]
+ )
+ top_ids = torch.full((logits.shape[0], k), -1, dtype=torch.int32)
+ top_logprobs = torch.full((logits.shape[0], k), float("-inf"))
+ if k > 0:
+ values, ids = torch.topk(logprobs, k=k, dim=-1)
+ top_ids[rows] = ids.to(torch.int32)
+ top_logprobs[rows] = values
+ return chosen, top_ids, top_logprobs
+
+
+def test_compute_logprobs_builds_row_index_on_host(monkeypatch) -> None:
+ # A nonzero over a CUDA mask is a host-device sync on every logprobs step; the row
+ # index must come from prepare's host list instead, with values unchanged.
+ import freetoken.engine.sample as sample_mod
+
+ monkeypatch.setattr(
+ sample_mod, "make_device_tensor",
+ lambda data, dtype, device: torch.tensor(data, dtype=dtype),
+ )
+ sampler = Sampler(torch.device("cpu"), vocab_size=6)
+ logits = torch.randn(4, 6, generator=torch.Generator().manual_seed(0))
+ sampled_tokens = torch.tensor([5, 0, 3, 1], dtype=torch.int32)
+
+ for want in ([True, False, True, False], [True] * 4, [False] * 4):
+ batch = SimpleNamespace(
+ reqs=[
+ SimpleNamespace(sampling_params=SamplingParams(logprobs=w, top_logprobs=3 if w else 0))
+ for w in want
+ ]
+ )
+ args = sampler.prepare(batch)
+ expected = _reference_logprobs(logits, sampled_tokens, want, 3)
+ with monkeypatch.context() as m:
+ m.setattr(torch, "nonzero", lambda *a, **kw: (_ for _ in ()).throw(AssertionError("sync")))
+ result = sampler.compute_logprobs(logits, sampled_tokens, args)
+ if expected is None:
+ assert result is None
+ continue
+ assert result is not None
+ for got, ref in zip(result, expected, strict=True):
+ assert got.shape == ref.shape and got.dtype == ref.dtype
+ assert torch.equal(got.nan_to_num(-7.0), ref.nan_to_num(-7.0))
diff --git a/tests/scheduler/test_cost_accounting_core.py b/tests/scheduler/test_cost_accounting_core.py
index 512bfa4049..045b9924be 100644
--- a/tests/scheduler/test_cost_accounting_core.py
+++ b/tests/scheduler/test_cost_accounting_core.py
@@ -60,6 +60,28 @@ def tokenize(self, messages):
assert errors == [] # in particular, no early prompt_tokens_delta UserReply
+def test_tokenization_flags_a_prompt_that_leaves_reasoning_open():
+ # The detokenizer cannot see the prompt; the tokenizer worker marks a template that
+ # pre-opened (5) so an unclosed generation counts as reasoning (6 = ).
+ prompts = {
+ 1: [10, 5, 11], # thinking on: the generation prompt ends inside
+ 2: [10, 5, 6, 11], # thinking off: the template pre-closed the block
+ 3: [5, 7, 6, 10, 5, 12], # an earlier turn's block is closed, the last one open
+ 4: [10, 11], # no reasoning tags
+ }
+
+ class Tokenizer:
+ def tokenize(self, messages):
+ (msg,) = messages
+ ids = torch.tensor(prompts[msg.uid], dtype=torch.int32)
+ return [UserMsg(uid=msg.uid, input_ids=ids, sampling_params=msg.sampling_params)]
+
+ backend, _ = _tokenize_requests(
+ Tokenizer(), [_tokenize_msg(uid) for uid in prompts], _Logger(), (5, 6)
+ )
+ assert [m.sampling_params.starts_in_reasoning for m in backend] == [True, False, True, False]
+
+
def test_tokenization_failure_and_empty_prompt_are_terminal_without_usage():
class Tokenizer:
def tokenize(self, messages):
diff --git a/tests/server/test_function_call_parser.py b/tests/server/test_function_call_parser.py
index 8af5bc88b6..3ba0fb0e9e 100644
--- a/tests/server/test_function_call_parser.py
+++ b/tests/server/test_function_call_parser.py
@@ -365,6 +365,35 @@ def test_dsv41_dsml_tags_lead_with_a_space():
assert json.loads("".join(c.parameters for c in calls if c.name is None)) == {"pattern": "*.py", "path": "/src"}
+def _qwen3_coder_read(path):
+ return (
+ "\n\n\n"
+ f"{path}\n\n\n"
+ )
+
+
+def test_qwen3_coder_one_shot_keeps_text_between_calls():
+ text = "Hi" + _qwen3_coder_read("/a") + "\nThen more.\n" + _qwen3_coder_read("/b") + "\n"
+ result = FunctionCallParser(OPENCODE_TOOLS, tool_call_parser="qwen3_coder").parse_non_stream(text)
+ assert result.normal_text == "Hi\nThen more.\n"
+ assert [json.loads(c.parameters) for c in result.calls] == [{"filePath": "/a"}, {"filePath": "/b"}]
+
+ parser = FunctionCallParser(OPENCODE_TOOLS, tool_call_parser="qwen3_coder")
+ texts, calls = _feed(parser, text)
+ assert "".join(texts) + parser.finish_stream() == result.normal_text
+ assert [c.name for c in calls if c.name] == ["read", "read"]
+
+
+def test_qwen3_coder_unclosed_function_tag_is_not_a_call():
+ # The name stops at the next tag: never "x\n\n")])
+ assert calls == [] and parser.recover_truncated_call() == []
+
+
def test_streaming_support_flags():
# Every registered detector is incremental-safe (buffered fallback remains as
# the escape hatch for future formats, exercised via monkeypatch in
diff --git a/tests/server/test_logprobs_api.py b/tests/server/test_logprobs_api.py
index 8cf0387c60..f983e6d9b5 100644
--- a/tests/server/test_logprobs_api.py
+++ b/tests/server/test_logprobs_api.py
@@ -2,9 +2,14 @@
import asyncio
import json
+import os
+import random
from types import SimpleNamespace
+import pytest
+
from freetoken.message import TokenizeMsg, UserReply
+from freetoken.server.generation import generate_full
from freetoken.server.openai_api import (
ChatCompletionRequest,
CompletionRequest,
@@ -20,16 +25,79 @@ def run(coro):
return asyncio.run(coro)
+THINK, END_THINK, TOOL_CALL, END_TOOL_CALL = 248068, 248069, 248058, 248059
+IM_END, ENDOFTEXT = 248046, 248044
+_MARKERS = {
+ "": THINK,
+ "": END_THINK,
+ "": TOOL_CALL,
+ "": END_TOOL_CALL,
+ "<|im_end|>": IM_END,
+ "<|endoftext|>": ENDOFTEXT,
+}
+
+
+class QwenTokenizer:
+ """Qwen3.8's marker ids; any other text encodes one id per character."""
+
+ all_special_ids = [ENDOFTEXT, IM_END]
+
+ def __init__(self, single_id_markers: bool = True) -> None:
+ self.single = single_id_markers
+
+ def encode(self, text: str, add_special_tokens: bool = False) -> list[int]:
+ if self.single and text in _MARKERS:
+ return [_MARKERS[text]]
+ return [ord(ch) for ch in text]
+
+ def decode(self, ids: list[int]) -> str:
+ names = {v: k for k, v in _MARKERS.items()}
+ return "".join(names.get(i, chr(i) if i < 0x110000 else "") for i in ids)
+
+
+def token_replies(
+ tokens: list[str], *, finish_reason: str = "stop", texts: list[str] | None = None
+) -> list[UserReply]:
+ """One reply per sampled token, each carrying that token's entry (EOS appended for a
+ "stop" finish). ``texts`` overrides the text each reply releases, the way the
+ detokenizer holds back partial characters and stop prefixes."""
+ tokens = tokens + (["<|im_end|>"] if finish_reason == "stop" else [])
+ if texts is None:
+ texts = ["" if t in ("<|im_end|>", "<|endoftext|>") else t for t in tokens]
+ replies = [UserReply(uid=42, incremental_output="", finished=False, prompt_tokens_delta=3)]
+ for i, (token, text) in enumerate(zip(tokens, texts, strict=True)):
+ last = i == len(tokens) - 1
+ replies.append(
+ UserReply(
+ uid=42,
+ incremental_output=text,
+ finished=last,
+ finish_reason=finish_reason if last else None,
+ completion_tokens_delta=1,
+ logprobs=[entry(_MARKERS.get(token, 1000 + i), token, -0.01 * (i + 1))],
+ )
+ )
+ return replies
+
+
class FakeState:
- def __init__(self, replies: list[UserReply], reasoning_parser: str | None = None) -> None:
+ def __init__(
+ self,
+ replies: list[UserReply],
+ reasoning_parser: str | None = None,
+ tool_call_parser: str = "llama3",
+ tokenizer: QwenTokenizer | None = None,
+ ) -> None:
self.config = SimpleNamespace(
model_path="/models/unit-model",
served_model_name="unit-model",
- tool_call_parser="llama3",
+ tool_call_parser=tool_call_parser,
reasoning_parser=reasoning_parser,
)
self.replies = replies
self.sent: TokenizeMsg | None = None
+ if tokenizer is not None:
+ self.frontend_tokenizer = lambda: SimpleNamespace(tokenizer=tokenizer)
def new_user(self) -> int:
return 42
@@ -43,7 +111,7 @@ async def wait_for_ack(self, uid: int):
yield reply
-def reply(text: str, *, finished: bool = False, logprobs: dict | None = None) -> UserReply:
+def reply(text: str, *, finished: bool = False, logprobs: list[dict] | None = None) -> UserReply:
return UserReply(
uid=42,
incremental_output=text,
@@ -55,10 +123,11 @@ def reply(text: str, *, finished: bool = False, logprobs: dict | None = None) ->
)
-def entry(token_id: int, token: str, logprob: float) -> dict:
+def entry(token_id: int, token: str, logprob: float, *, text_offset: int = 0) -> dict:
return {
"token_id": token_id,
"token": token,
+ "text_offset": text_offset,
"bytes": list(token.encode("utf-8")),
"logprob": logprob,
"top": [
@@ -105,7 +174,7 @@ def test_chat_non_stream_logprobs() -> None:
handle_chat_completion(
chat_request(logprobs=True, top_logprobs=2),
None,
- FakeState([reply("Hello", logprobs=first), reply("!", finished=True, logprobs=second)]),
+ FakeState([reply("Hello", logprobs=[first]), reply("!", finished=True, logprobs=[second])]),
{},
)
)
@@ -136,18 +205,20 @@ def test_chat_non_stream_logprobs() -> None:
handle_chat_completion(
chat_request(),
None,
- FakeState([reply("Hello", finished=True, logprobs=first)]),
+ FakeState([reply("Hello", finished=True, logprobs=[first])]),
{},
)
)
- assert without_logprobs["choices"][0].get("logprobs") is None
+ # OpenAI sends the key with null rather than leaving it out
+ assert "logprobs" in without_logprobs["choices"][0]
+ assert without_logprobs["choices"][0]["logprobs"] is None
def test_chat_stream_logprobs_follow_content_deltas() -> None:
first = entry(1, "Hello", -0.1)
state = FakeState(
[
- reply("Hello", logprobs=first),
+ reply("Hello", logprobs=[first]),
reply(" world", finished=True),
]
)
@@ -168,14 +239,14 @@ def test_chat_stream_logprobs_follow_content_deltas() -> None:
def test_completion_logprobs_non_stream_and_stream() -> None:
first = entry(1, "Hi", -0.1)
- second = entry(2, "!", -0.2)
+ second = entry(2, "!", -0.2, text_offset=2)
req = CompletionRequest(model="client-model", prompt="hello", logprobs=2, max_tokens=8)
result = run(
handle_completion(
req,
None,
- FakeState([reply("Hi", logprobs=first), reply("!", finished=True, logprobs=second)]),
+ FakeState([reply("Hi", logprobs=[first]), reply("!", finished=True, logprobs=[second])]),
{},
)
)
@@ -201,7 +272,7 @@ def test_completion_logprobs_non_stream_and_stream() -> None:
max_tokens=8,
stream=True,
),
- FakeState([reply("Hi", finished=True, logprobs=first)]),
+ FakeState([reply("Hi", finished=True, logprobs=[first])]),
)
)
)
@@ -215,6 +286,255 @@ def test_completion_logprobs_non_stream_and_stream() -> None:
}
+def _crab_replies() -> list[UserReply]:
+ # 🦀 is three byte-level tokens: the first two acks carry no visible text yet.
+ pieces = [
+ reply("", logprobs=[entry(9008, "�", -0.1)]),
+ reply("", logprobs=[entry(99, "�", -0.2)]),
+ reply("🦀", logprobs=[entry(222, "�", -0.3)]),
+ reply(" x", logprobs=[entry(1, " x", -0.4, text_offset=1)]),
+ reply("", finished=True, logprobs=[entry(2, "<|im_end|>", -0.5, text_offset=3)]),
+ ]
+ return pieces
+
+
+def test_completion_text_offset_counts_visible_text() -> None:
+ req = CompletionRequest(model="client-model", prompt="hello", logprobs=1, max_tokens=8)
+ result = run(handle_completion(req, None, FakeState(_crab_replies()), {}))
+ logprobs = result["choices"][0]["logprobs"]
+ assert result["choices"][0]["text"] == "🦀 x"
+ # The split character's pieces all point at where it starts, not at
+ # len("�") steps that drift past the real text.
+ assert logprobs["text_offset"] == [0, 0, 0, 1, 3]
+
+
+def _offset_replies(acks: list[tuple[str, list[tuple[str, int]]]]) -> list[UserReply]:
+ """One reply per (text, [(token, text_offset), ...]) ack, entries in generation order."""
+ replies, n = [], 0
+ for i, (text, tokens) in enumerate(acks):
+ entries = []
+ for token, offset in tokens:
+ entries.append(entry(n, token, -0.1, text_offset=offset))
+ n += 1
+ replies.append(reply(text, finished=i == len(acks) - 1, logprobs=entries or None))
+ return replies
+
+
+# "Answer: 4\nQuestion 2" under stop=["\n\n"]: the "\n" ack sends nothing (it could start
+# the stop), and "Question" releases it along with its own text. "own" gives every ack
+# its own entry; "held" is what the detokenizer sends, holding the "\n" entry until the
+# reply that releases its text, which then carries two entries.
+_HOLDBACK_ACKS = {
+ "own": [("Answer", [("Answer", 0)]), (":", [(":", 6)]), (" ", [(" ", 7)]), ("4", [("4", 8)]),
+ ("", [("\n", 9)]), ("\nQuestion", [("Question", 10)]), (" ", [(" ", 18)]),
+ ("2", [("2", 19)])],
+ "held": [("Answer", [("Answer", 0)]), (":", [(":", 6)]), (" ", [(" ", 7)]), ("4", [("4", 8)]),
+ ("", []), ("\nQuestion", [("\n", 9), ("Question", 10)]), (" ", [(" ", 18)]),
+ ("2", [("2", 19)])],
+}
+
+
+def _joined_stream_logprobs(replies: list[UserReply], **kwargs) -> dict[str, list]:
+ req = CompletionRequest(model="client-model", prompt="hello", max_tokens=8, stream=True, **kwargs)
+ events = parse_sse(run(_collect(stream_completion_chunks(42, req, FakeState(replies)))))
+ joined: dict[str, list] = {"tokens": [], "token_logprobs": [], "top_logprobs": [], "text_offset": []}
+ for event in events:
+ if isinstance(event, dict) and event["choices"] and event["choices"][0]["logprobs"]:
+ for key in joined:
+ joined[key] += event["choices"][0]["logprobs"][key]
+ return joined
+
+
+@pytest.mark.parametrize("shape", ["own", "held"])
+def test_completion_text_offset_survives_stop_prefix_holdback(shape: str) -> None:
+ req = CompletionRequest(model="client-model", prompt="hello", logprobs=0, max_tokens=8, stop=["\n\n"])
+ replies = _offset_replies(_HOLDBACK_ACKS[shape])
+ choice = run(handle_completion(req, None, FakeState(replies), {}))["choices"][0]
+ logprobs = choice["logprobs"]
+ assert choice["text"] == "Answer: 4\nQuestion 2"
+ assert logprobs["tokens"] == ["Answer", ":", " ", "4", "\n", "Question", " ", "2"]
+ assert logprobs["text_offset"] == [0, 6, 7, 8, 9, 10, 18, 19]
+ for token, offset in zip(logprobs["tokens"], logprobs["text_offset"], strict=True):
+ assert choice["text"][offset:].startswith(token)
+ assert _joined_stream_logprobs(replies, logprobs=0, stop=["\n\n"]) == logprobs
+
+
+def test_completion_text_offset_clipped_to_stop_trimmed_text() -> None:
+ # "a\n\n" under stop=["\n\n"]: the second "\n" completes the stop and both are trimmed.
+ # Unmarked, their offsets are clipped so they cannot point past the returned text.
+ replies = [
+ reply("a", logprobs=[entry(1, "a", -0.1)]),
+ reply("", logprobs=[entry(2, "\n", -0.2, text_offset=1)]),
+ reply("", finished=True, logprobs=[entry(2, "\n", -0.3, text_offset=2)]),
+ ]
+ req = CompletionRequest(model="client-model", prompt="hello", logprobs=0, max_tokens=8, stop=["\n\n"])
+ choice = run(handle_completion(req, None, FakeState(replies), {}))["choices"][0]
+ assert choice["text"] == "a"
+ assert choice["logprobs"]["text_offset"] == [0, 1, 1]
+ assert _joined_stream_logprobs(replies, logprobs=0, stop=["\n\n"]) == choice["logprobs"]
+
+ # As the detokenizer sends them: held with their text, then marked and dropped.
+ trimmed = [
+ {**entry(2, "\n", -0.2, text_offset=1), "stop_trimmed": True},
+ {**entry(2, "\n", -0.3, text_offset=2), "stop_trimmed": True},
+ ]
+ marked = [replies[0], reply("", logprobs=None), reply("", finished=True, logprobs=trimmed)]
+ choice = run(handle_completion(req, None, FakeState(marked), {}))["choices"][0]
+ assert choice["logprobs"]["tokens"] == ["a"]
+ assert choice["logprobs"]["text_offset"] == [0]
+ assert _joined_stream_logprobs(marked, logprobs=0, stop=["\n\n"]) == choice["logprobs"]
+
+
+# stop=["aab"] over "aaax": two "a"s stay held while the first is sent. "own": the held
+# tokens' entries ride their own acks, so their offsets run past the text sent so far;
+# that is where they land, not an overrun. "held": the detokenizer's shape, each entry
+# waiting for its text, so the finished reply carries three.
+_AHEAD_ACKS = {
+ "own": [("", [("a", 0)]), ("", [("a", 1)]), ("a", [("a", 2)]), ("aax", [("x", 3)])],
+ "held": [("", []), ("", []), ("a", [("a", 0)]), ("aax", [("a", 1), ("a", 2), ("x", 3)])],
+}
+
+
+@pytest.mark.parametrize("shape", ["own", "held"])
+def test_completion_stream_text_offset_ahead_of_sent_text(shape: str) -> None:
+ req = CompletionRequest(model="client-model", prompt="hello", logprobs=0, max_tokens=8, stop=["aab"])
+ replies = _offset_replies(_AHEAD_ACKS[shape])
+ choice = run(handle_completion(req, None, FakeState(replies), {}))["choices"][0]
+ assert choice["text"] == "aaax"
+ assert choice["logprobs"]["tokens"] == ["a", "a", "a", "x"]
+ assert choice["logprobs"]["text_offset"] == [0, 1, 2, 3]
+ assert _joined_stream_logprobs(replies, logprobs=0, stop=["aab"]) == choice["logprobs"]
+
+
+def test_completion_text_offset_real_tokenizer_with_stops() -> None:
+ # The real Qwen3.8 tokenizer through the real detokenizer: every token's offset must
+ # point at its own text, stream and non-stream alike, with entries held for their text
+ # and those of stop-trimmed tokens dropped.
+ path = os.path.expanduser("~/models/Qwen3.8-Flash-Next-NVFP4-FTW-fp8")
+ if not os.path.exists(os.path.join(path, "tokenizer.json")):
+ pytest.skip("local Qwen3.8 tokenizer not present")
+ from transformers import AutoTokenizer
+
+ from freetoken.message import DetokenizeMsg
+ from freetoken.tokenizer.detokenize import DetokenizeManager, build_logprobs_entry
+
+ tokenizer = AutoTokenizer.from_pretrained(path)
+ cases = [
+ ("Answer: 4\nQuestion 2", "\n\n", None),
+ ("Q: hi\nA: Question time 🦀 ok", "Question:", None),
+ # the stop matches on the last token: the "\n\n" token it trimmed loses its entry
+ ("Answer: 4\n\n", "\n\n", "\n\n"),
+ # the stop cuts inside " hello": that entry stays, " world" (trimmed) does not
+ ("Say hello world", "lo w", "lo w"),
+ ]
+ held = False
+ for text, stop, matched in cases:
+ ids = tokenizer.encode(text, add_special_tokens=False)
+ manager = DetokenizeManager(tokenizer, frozenset({248046}))
+ replies = []
+ for i, tid in enumerate(ids):
+ last = i == len(ids) - 1
+ msg = DetokenizeMsg(
+ uid=42, next_token=tid, finished=last, stop_strs=[stop],
+ matched_stop=matched if last else None,
+ )
+ (out,), _, (released,) = manager.detokenize_with_logprobs(
+ [msg], [build_logprobs_entry(tokenizer, tid, -0.1, [], [])]
+ )
+ held |= len(released) > 1
+ replies.append(reply(out, finished=last, logprobs=released or None))
+ visible = text[: text.find(matched)] if matched else text
+ req = CompletionRequest(model="client-model", prompt="p", logprobs=0, max_tokens=64, stop=[stop])
+ choice = run(handle_completion(req, None, FakeState(replies), {}))["choices"][0]
+ assert choice["text"] == visible
+ tokens, offsets = choice["logprobs"]["tokens"], choice["logprobs"]["text_offset"]
+ assert len(tokens) == len(ids) - (1 if matched else 0)
+ for token, offset in zip(tokens, offsets, strict=True):
+ if "\ufffd" not in token:
+ # a token's text sits at its offset; only one the stop cut ends the text early
+ seg = visible[offset : offset + len(token)]
+ assert token.startswith(seg), (text, token, offset)
+ assert seg == token or offset + len(seg) == len(visible), (text, token, offset)
+ assert offsets == sorted(offsets)
+ assert _joined_stream_logprobs(replies, logprobs=0, stop=[stop]) == choice["logprobs"]
+ assert held, "no reply carried more than one entry"
+
+
+def test_completion_top_logprobs_key_collision_keeps_most_likely() -> None:
+ # Partial-UTF-8 candidates all decode to U+FFFD and collide on the dict key.
+ colliding = entry(9008, "�", -1.0)
+ colliding["top"] = [
+ {"token_id": 9008, "token": "�", "bytes": [240, 159], "logprob": -1.0},
+ {"token_id": 9009, "token": "�", "bytes": [240, 160], "logprob": -2.0},
+ {"token_id": 5, "token": "a", "bytes": [97], "logprob": -3.0},
+ ]
+ req = CompletionRequest(model="client-model", prompt="hello", logprobs=3, max_tokens=8)
+ result = run(handle_completion(req, None, FakeState([reply("x", finished=True, logprobs=[colliding])]), {}))
+ assert result["choices"][0]["logprobs"]["top_logprobs"] == [{"�": -1.0, "a": -3.0}]
+
+
+def test_completion_stream_logprobs_match_non_stream() -> None:
+ # Acks with no visible text must not lose their entries: they ride the next chunk
+ # with text (or the finish chunk), so the joined stream equals the non-stream list.
+ req = CompletionRequest(model="client-model", prompt="hello", logprobs=1, max_tokens=8)
+ expected = run(handle_completion(req, None, FakeState(_crab_replies()), {}))
+ stream_req = CompletionRequest(model="client-model", prompt="hello", logprobs=1, max_tokens=8, stream=True)
+ events = parse_sse(run(_collect(stream_completion_chunks(42, stream_req, FakeState(_crab_replies())))))
+ joined: dict[str, list] = {"tokens": [], "token_logprobs": [], "top_logprobs": [], "text_offset": []}
+ for event in events:
+ if isinstance(event, dict) and event["choices"] and event["choices"][0]["logprobs"]:
+ for key in joined:
+ joined[key] += event["choices"][0]["logprobs"][key]
+ assert joined == expected["choices"][0]["logprobs"]
+ assert len(joined["tokens"]) == 5
+
+
+def _stop_trimmed_replies() -> list[UserReply]:
+ # "Hi", then a stop string "##" sampled as "#" "#": the detokenizer withheld both
+ # entries with the text and releases them, marked, on the finished reply.
+ trimmed = [{**entry(3, "#", -0.3), "stop_trimmed": True}, {**entry(3, "#", -0.4), "stop_trimmed": True}]
+ return [
+ reply("Hi", logprobs=[entry(1, "Hi", -0.1)]),
+ reply("", logprobs=None),
+ reply("", finished=True, logprobs=trimmed),
+ ]
+
+
+def test_completion_logprobs_drop_stop_trimmed_entries() -> None:
+ req = CompletionRequest(model="client-model", prompt="hello", logprobs=1, max_tokens=8, stop=["##"])
+ result = run(handle_completion(req, None, FakeState(_stop_trimmed_replies()), {}))
+ assert result["choices"][0]["logprobs"]["tokens"] == ["Hi"]
+
+ stream_req = req.model_copy(update={"stream": True})
+ events = parse_sse(run(_collect(stream_completion_chunks(42, stream_req, FakeState(_stop_trimmed_replies())))))
+ tokens = [
+ token
+ for event in events
+ if isinstance(event, dict) and event["choices"] and event["choices"][0]["logprobs"]
+ for token in event["choices"][0]["logprobs"]["tokens"]
+ ]
+ assert tokens == ["Hi"]
+
+
+def test_chat_logprobs_drop_stop_trimmed_entries() -> None:
+ req = chat_request(logprobs=True, stop=["##"])
+ result = run(handle_chat_completion(req, None, FakeState(_stop_trimmed_replies()), {}))
+ assert [e["token"] for e in result["choices"][0]["logprobs"]["content"]] == ["Hi"]
+
+ stream_req = chat_request(logprobs=True, stop=["##"], stream=True)
+ spec = chat_request_to_genspec(stream_req, {})
+ events = parse_sse(
+ run(_collect(stream_chat_completion_chunks(42, stream_req, FakeState(_stop_trimmed_replies()), spec)))
+ )
+ tokens = [
+ e["token"]
+ for event in events
+ if isinstance(event, dict) and event["choices"] and event["choices"][0].get("logprobs")
+ for e in event["choices"][0]["logprobs"]["content"]
+ ]
+ assert tokens == ["Hi"]
+
+
def test_logprobs_validation_errors() -> None:
chat_top_out_of_range = run(
handle_chat_completion(chat_request(logprobs=True, top_logprobs=25), None, FakeState([]), {})
@@ -247,61 +567,292 @@ def test_logprobs_validation_errors() -> None:
assert completion_echo.status_code == 400
-def test_chat_logprobs_fail_closed_under_semantic_parsing() -> None:
- # A server-side reasoning parser hides tokens from message content, so
- # logprob entries cannot be aligned with it: reject up front, stream and
- # non-stream alike, before any engine work is submitted.
- with_parser = FakeState([], reasoning_parser="qwen3")
- for stream in (False, True):
- resp = run(
- handle_chat_completion(chat_request(logprobs=True, stream=stream), None, with_parser, {})
- )
- assert resp.status_code == 400
- assert json.loads(resp.body)["error"]["param"] == "logprobs"
-
- # Tool parsing consumes tokens into tool_calls -- same conflict.
- tool = {"type": "function", "function": {"name": "f", "parameters": {"type": "object"}}}
- tools_resp = run(
- handle_chat_completion(chat_request(logprobs=True, tools=[tool]), None, FakeState([]), {})
+TOOL = {
+ "type": "function",
+ "function": {
+ "name": "read",
+ "parameters": {"type": "object", "properties": {"path": {"type": "string"}}},
+ },
+}
+CALL = [
+ "", "\n", "<", "function", "=read", ">", "\n", "<", "parameter", "=path", ">",
+ "\n", "a", ".py", "\n", "", "\n", "", "\n",
+ "",
+]
+
+
+def qwen_state(replies: list[UserReply], **kwargs) -> FakeState:
+ """The production stack: qwen3 reasoning parser and qwen3_coder tool parser."""
+ return FakeState(
+ replies,
+ reasoning_parser="qwen3",
+ tool_call_parser="qwen3_coder",
+ tokenizer=kwargs.pop("tokenizer", QwenTokenizer()),
+ **kwargs,
)
- assert tools_resp.status_code == 400
- assert json.loads(tools_resp.body)["error"]["param"] == "logprobs"
- # tool_choice="none" disables parsing, so logprobs stay available.
+
+def _chat_both(replies: list[UserReply], **req_kwargs) -> tuple[dict, list[dict]]:
+ """The non-streaming response and the streamed chunk choices for the same replies."""
+ req = chat_request(logprobs=True, **req_kwargs)
+ result = run(handle_chat_completion(req, None, qwen_state(replies), {}))
+ req = chat_request(logprobs=True, stream=True, **req_kwargs)
+ spec = chat_request_to_genspec(req, {})
+ chunks = stream_chat_completion_chunks(42, req, qwen_state(replies), spec)
+ events = parse_sse(run(_collect(chunks)))
+ choices = [e["choices"][0] for e in events if isinstance(e, dict) and e.get("choices")]
+ return result, choices
+
+
+def _streamed_tokens(choices: list[dict]) -> list[str]:
+ return [e["token"] for c in choices if c.get("logprobs") for e in c["logprobs"]["content"]]
+
+
+def _content_tokens(result: dict) -> list[str]:
+ return [e["token"] for e in result["choices"][0]["logprobs"]["content"]]
+
+
+def test_chat_logprobs_exclude_reasoning() -> None:
+ replies = token_replies(["thought", " more", "", "\n\n", "Hello", " world"])
+ result, choices = _chat_both(replies)
+
+ assert result["choices"][0]["message"]["reasoning_content"] == "thought more"
+ assert result["choices"][0]["logprobs"] == {
+ "content": result["choices"][0]["logprobs"]["content"],
+ "refusal": None,
+ }
+ assert _content_tokens(result) == ["Hello", " world"]
+ assert _streamed_tokens(choices) == ["Hello", " world"]
+ for choice in choices:
+ if choice["delta"].get("reasoning_content"):
+ assert "logprobs" not in choice
+ if choice.get("logprobs"):
+ assert choice["logprobs"]["refusal"] is None
+
+
+def test_chat_logprobs_exclude_tool_call_markup() -> None:
+ replies = token_replies(["plan", "", "\n\n", "Sure", "."] + CALL, finish_reason="stop")
+ result, choices = _chat_both(replies, tools=[TOOL])
+
+ assert result["choices"][0]["message"]["tool_calls"][0]["function"]["name"] == "read"
+ assert _content_tokens(result) == ["Sure", "."]
+ assert _streamed_tokens(choices) == ["Sure", "."]
+ for choice in choices:
+ if choice["delta"].get("tool_calls"):
+ assert "logprobs" not in choice
+
+
+def test_chat_logprobs_bare_function_call_is_not_content() -> None:
+ # The qwen3_coder parser takes a declared without its
+ # wrapper as a call; its markup tokens are not content either.
+ call = CALL[2:-2]
+ replies = token_replies(["plan", "", "Doing", " it", ".\n"] + call + ["\n", "ok"])
+ result, choices = _chat_both(replies, tools=[TOOL])
+
+ assert result["choices"][0]["message"]["tool_calls"]
+ assert result["choices"][0]["message"]["content"] == "Doing it.\n\nok"
+ assert _content_tokens(result) == ["Doing", " it", ".\n", "\n", "ok"]
+ assert _streamed_tokens(choices) == _content_tokens(result)
+
+
+def test_chat_logprobs_text_merged_into_a_bare_call_closer_is_content() -> None:
+ # Qwen3.8 merges the closing ">" with the next character (">n"): the emitted "n"
+ # sits at the end of "n", not on the "n" inside "function".
+ call = CALL[2:-3] + [">n"]
+ replies = token_replies(["plan", ""] + call)
+ result, choices = _chat_both(replies, tools=[TOOL])
+
+ assert result["choices"][0]["message"]["tool_calls"]
+ assert result["choices"][0]["message"]["content"] == "n"
+ assert _content_tokens(result) == [">n"]
+ assert _streamed_tokens(choices) == _content_tokens(result)
+
+
+def test_chat_logprobs_leading_space_merged_into_a_bare_call_opener_is_not_content() -> None:
+ # The reasoning parser strips content's leading whitespace before the tool parser
+ # sees it, so the space of " <" is never message text.
+ call = [" <"] + CALL[3:-2]
+ replies = token_replies(["x", ""] + call)
+ result, choices = _chat_both(replies, tools=[TOOL])
+
+ assert result["choices"][0]["message"]["tool_calls"]
+ assert not result["choices"][0]["message"]["content"]
+ assert _content_tokens(result) == []
+ assert _streamed_tokens(choices) == []
+
+
+def _stop_cut_replies(tokens: list[str], visible: int) -> list[UserReply]:
+ """A stop string matched inside the last token: only its first ``visible`` chars
+ were released, and the detokenizer says so on its entry."""
+ texts = tokens[:-1] + [tokens[-1][:visible]]
+ replies = token_replies(tokens, finish_reason="length", texts=texts)
+ replies[-1].finish_reason = "stop"
+ replies[-1].logprobs = [{**replies[-1].logprobs[0], "stop_visible": visible}]
+ return replies
+
+
+def test_chat_logprobs_follow_the_visible_part_of_a_token_a_stop_string_cut() -> None:
+ # Only the trailing space of " !" survives the stop "!", and content trimming drops it.
+ result, choices = _chat_both(_stop_cut_replies(["x", "", "Hi", " !"], 1), stop=["!"])
+ assert result["choices"][0]["message"]["content"] == "Hi"
+ assert _content_tokens(result) == ["Hi"]
+ assert _streamed_tokens(choices) == ["Hi"]
+
+ # A stop "_c" leaves " as text, not a call opener.
+ replies = _stop_cut_replies(["x", "", "Use", " ", ""], 5)
+ result, choices = _chat_both(replies, stop=["_c"], tools=[TOOL])
+ assert result["choices"][0]["message"]["content"] == "Use "]
+ assert _streamed_tokens(choices) == _content_tokens(result)
+
+
+def test_chat_logprobs_literal_markers_in_content_stay_content() -> None:
+ # Only a marker that switches state is markup: a later or a
+ # named in prose is message text, and so is its entry.
+ replies = token_replies(
+ ["plan", "", "Use", " ", " and", " ", " tags", "."]
+ )
+ result, choices = _chat_both(replies, tools=[TOOL])
+
+ content = result["choices"][0]["message"]["content"]
+ assert content == "Use and tags."
+ assert "".join(_content_tokens(result)) == content
+ assert _streamed_tokens(choices) == _content_tokens(result)
+
+
+def test_chat_logprobs_ride_the_delta_that_releases_a_held_marker_prefix() -> None:
+ # The tool parser holds the "<" of " <" as a possible : the token's
+ # entry waits until all its text is out, then rides that delta, in order.
+ replies = token_replies(["", "a", " <", " b"])
+ _, choices = _chat_both(replies, tools=[TOOL])
+
+ content_chunks = [c for c in choices if c["delta"].get("content")]
+ assert [c["delta"]["content"] for c in content_chunks] == ["a", " ", "< b"]
+ assert [
+ [e["token"] for e in c["logprobs"]["content"]] if c.get("logprobs") else None
+ for c in content_chunks
+ ] == [["a"], None, [" <", " b"]]
+
+
+def test_chat_logprobs_trailing_whitespace_is_dropped() -> None:
+ # Content after is trimmed at both ends, and so are its whitespace entries;
+ # whitespace between words is kept.
+ replies = token_replies(["", "\n\n", "Hi", "\n", "\n", "there", "\n\n"])
+ result, choices = _chat_both(replies)
+
+ assert result["choices"][0]["message"]["content"] == "Hi\n\nthere"
+ assert _content_tokens(result) == ["Hi", "\n", "\n", "there"]
+ assert _streamed_tokens(choices) == _content_tokens(result)
+
+
+@pytest.mark.parametrize("seed", range(12))
+def test_chat_logprobs_stream_matches_non_stream_for_any_chunking(seed: int) -> None:
+ # The detokenizer can release a token's text on a later reply (a partial character,
+ # a stop prefix), leaving replies with no text at all: the streamed entries, joined
+ # across content chunks and the finish chunk, must still equal the one-shot list.
+ tokens = (
+ [" ", "", "plan", " it", "", "\n\n", "Hi", " ", "<", "b", ">", "\n\n"]
+ + CALL
+ + ["\n", "Then", " more", "\n"]
+ )
+ full = "".join(tokens)
+ rng = random.Random(seed)
+ cuts, end, prev = [], 0, 0
+ for token in tokens:
+ end += len(token)
+ prev = rng.randint(prev, end)
+ cuts.append(prev)
+ cuts[-1] = len(full)
+ texts = [full[a:b] for a, b in zip([0] + cuts[:-1], cuts, strict=True)] + [""]
+ replies = token_replies(tokens, texts=texts)
+
+ result, choices = _chat_both(replies, tools=[TOOL])
+ assert result["choices"][0]["message"]["content"] == "Hi \n\n\nThen more"
+ assert _content_tokens(result) == ["Hi", " ", "<", "b", ">", "\n\n", "\n", "Then", " more"]
+ assert _streamed_tokens(choices) == _content_tokens(result)
+
+
+def test_chat_logprobs_start_in_content_when_thinking_is_off() -> None:
+ replies = token_replies([" ", "Hi", " ", " there"])
+ result, choices = _chat_both(replies, chat_template_kwargs={"enable_thinking": False})
+
+ assert result["choices"][0]["message"]["content"] == " Hi there"
+ assert _content_tokens(result) == [" ", "Hi", " ", " there"]
+ assert _streamed_tokens(choices) == _content_tokens(result)
+
+
+def test_chat_logprobs_truncated_reasoning_has_no_content_entries() -> None:
+ replies = token_replies(["still", " thinking"], finish_reason="length")
+ result, choices = _chat_both(replies)
+
+ assert result["choices"][0]["finish_reason"] == "length"
+ assert result["choices"][0]["logprobs"] == {"content": [], "refusal": None}
+ assert _streamed_tokens(choices) == []
+
+
+def test_chat_logprobs_drop_the_eos_entry() -> None:
+ # The end-of-turn token is not message text; non-streaming used to list it.
+ replies = token_replies(["Hi", "!"])
+ state = FakeState(replies, tokenizer=QwenTokenizer())
+ result = run(handle_chat_completion(chat_request(logprobs=True), None, state, {}))
+ assert _content_tokens(result) == ["Hi", "!"]
+
+
+def test_chat_logprobs_unsupported_parser_families_still_400() -> None:
+ def status(state: FakeState, **kwargs) -> int | None:
+ resp = run(handle_chat_completion(chat_request(logprobs=True, **kwargs), None, state, {}))
+ if isinstance(resp, dict):
+ return None
+ assert json.loads(resp.body)["error"]["param"] == "logprobs"
+ return resp.status_code
+
+ # A reasoning parser whose split the segmenter does not mirror.
+ assert status(FakeState([], reasoning_parser="glm", tokenizer=QwenTokenizer())) == 400
+ # qwen3 whose think markers are not single tokens, or with no tokenizer to tell.
+ multi = QwenTokenizer(single_id_markers=False)
+ assert status(FakeState([], reasoning_parser="qwen3", tokenizer=multi)) == 400
+ assert status(FakeState([], reasoning_parser="qwen3")) == 400
+ for stream in (False, True):
+ assert status(FakeState([], reasoning_parser="qwen3"), stream=stream) == 400
+ # A tool-call parser the segmenter cannot shadow...
+ assert status(FakeState([], tokenizer=QwenTokenizer()), tools=[TOOL]) == 400
+ # ...unless tool parsing is off for the request.
ok = run(
handle_chat_completion(
- chat_request(logprobs=True, tools=[tool], tool_choice="none"),
+ chat_request(logprobs=True, tools=[TOOL], tool_choice="none"),
None,
- FakeState([reply("Hi", finished=True, logprobs=entry(1, "Hi", -0.1))]),
+ FakeState(token_replies(["Hi"]), tokenizer=QwenTokenizer()),
{},
)
)
- assert ok["choices"][0]["logprobs"]["content"][0]["token"] == "Hi"
+ assert _content_tokens(ok) == ["Hi"]
-def test_reasoning_logprob_is_not_carried_to_content_delta() -> None:
- # The generation-layer guard behind the API fail-close: even when a caller
- # bypasses request validation, a hidden reasoning token's entry is dropped,
- # never attached to a later visible delta (where its token string would leak).
+def test_chat_logprobs_are_dropped_when_the_segmenter_cannot_build() -> None:
+ # The generation-layer guard behind the API check: a caller that bypasses it still
+ # never gets a reasoning token's entry on a visible delta.
state = FakeState(
- [
- reply("thought", logprobs=entry(1, "thought", -0.1)),
- reply("answer", finished=True),
- ],
- reasoning_parser="qwen3",
+ token_replies(["thought", "", "answer"]), reasoning_parser="qwen3"
)
req = chat_request(stream=True, logprobs=True, top_logprobs=2)
spec = chat_request_to_genspec(req, {})
events = parse_sse(run(_collect(stream_chat_completion_chunks(42, req, state, spec))))
- content_choice = next(
- event["choices"][0]
- for event in events
- if isinstance(event, dict)
- and event["choices"]
- and event["choices"][0]["delta"].get("content") == "answer"
+ choices = [e["choices"][0] for e in events if isinstance(e, dict) and e.get("choices")]
+ assert any(c["delta"].get("content") == "answer" for c in choices)
+ assert _streamed_tokens(choices) == []
+
+
+def test_generate_full_ignores_entries_when_logprobs_were_not_requested() -> None:
+ # The protocol-neutral result carries entries only for a request that asked for them,
+ # so an adapter reading it never sees unsegmented (reasoning) entries.
+ state = FakeState(
+ token_replies(["thought", "", "answer"]), reasoning_parser="qwen3"
)
- assert "logprobs" not in content_choice
+ spec = chat_request_to_genspec(chat_request(), {})
+ result = run(generate_full(42, spec, state))
+ assert result.content == "answer"
+ assert result.logprobs == []
async def _collect(iterator):
diff --git a/tests/server/test_qwen_semantic_special_tokens.py b/tests/server/test_qwen_semantic_special_tokens.py
index 11e35ccb3e..da20fdde7a 100644
--- a/tests/server/test_qwen_semantic_special_tokens.py
+++ b/tests/server/test_qwen_semantic_special_tokens.py
@@ -118,8 +118,10 @@ def test_stream_and_nonstream_match_while_preserving_semantic_markers_and_litera
"<|box_start|>(1,2),(3,4)<|box_end|>"
"<|quad_start|>(1,2),(3,4),(5,6),(7,8)<|quad_end|>"
)
+ # The leaked marker sits after the think block: qwen3 opens on only as the
+ # first non-whitespace text, so "<|im_start|>" keeps "" as reasoning.
text = (
- "<|im_start|>plananswer"
+ "plan<|im_start|>answer"
"<|vision_start|><|image_pad|><|vision_end|>"
f"{grounding} literal \"<|im_start|>\" and `<|image_pad|>`<|im_end|>"
)
diff --git a/tests/server/test_reasoning_parser.py b/tests/server/test_reasoning_parser.py
index 05c207d0e1..2580af83b9 100644
--- a/tests/server/test_reasoning_parser.py
+++ b/tests/server/test_reasoning_parser.py
@@ -44,10 +44,25 @@ class TestDeepSeekV32ReasoningParser(_ReasoningContract):
class TestThinkReasoningParser(_ReasoningContract):
+ name = "glm"
+ opener = ""
+ closer = ""
+
+
+class TestQwen3ReasoningParser(_ReasoningContract):
name = "qwen3"
opener = ""
closer = ""
+ def test_markers_after_the_first_close_are_literal(self):
+ text = "plan\n\nSTART = \"\"\nEND = \"\"\n"
+ expected = ("plan", 'START = ""\nEND = ""')
+ parser = ReasoningParser(self.name, force_reasoning=True)
+ assert parser.parse_non_stream(text) == expected
+ parser = ReasoningParser(self.name, force_reasoning=True)
+ parts = [parser.parse_stream_chunk(char) for char in text] + [parser.flush()]
+ assert tuple("".join(pair[i] for pair in parts) for i in (0, 1)) == expected
+
class TestMiniMaxM3ReasoningParser(_ReasoningContract):
name = "minimax_m3"
diff --git a/tests/server/test_responses_api.py b/tests/server/test_responses_api.py
index 6e7721b891..95fbba3ef1 100644
--- a/tests/server/test_responses_api.py
+++ b/tests/server/test_responses_api.py
@@ -338,8 +338,9 @@ def test_stream_tool_call_events():
# Route smoke tests
# --------------------------------------------------------------------------- #
class FakeState:
- def __init__(self, outputs, finish_reason=None, cached_tokens=0, prefill_ms=0.0):
+ def __init__(self, outputs, finish_reason=None, cached_tokens=0, prefill_ms=0.0, reasoning_tokens=0):
self._outputs = outputs
+ self._reasoning_tokens = reasoning_tokens # stamped on the terminal ack
self._finish_reason = finish_reason # stamped on the terminal ack
self._cached_tokens = cached_tokens # stamped on the first ack (admission reply)
self._prefill_ms = prefill_ms # the scheduler stamps it on the first generated token
@@ -366,6 +367,7 @@ async def wait_for_ack(self, uid):
finish_reason=self._finish_reason if finished else None,
prompt_tokens_delta=pt, completion_tokens_delta=ct,
cached_tokens=self._cached_tokens if i == 0 else 0,
+ reasoning_tokens=self._reasoning_tokens if finished else 0,
prefill_ms=self._prefill_ms if i == 0 else 0.0)
async def stream_with_cancellation(self, gen, request, uid):
@@ -780,6 +782,27 @@ def test_stream_length_truncation_emits_incomplete_event():
assert final["incomplete_details"]["reason"] == "max_output_tokens"
+def test_route_reports_reasoning_tokens():
+ # Codex shows usage.output_tokens_details.reasoning_tokens; it was hardcoded to 0.
+ def usage(stream):
+ client = _client(FakeState([("Hello", True, 5, 9)], reasoning_tokens=7))
+ r = client.post("/v1/responses", json={"model": "gpt-x", "input": "hi", "stream": stream})
+ assert r.status_code == 200, r.text
+ if not stream:
+ return r.json()["usage"]
+ final = [
+ json.loads(line[len("data:"):])
+ for line in r.text.splitlines()
+ if line.startswith("data:") and "response.completed" in line
+ ]
+ return final[-1]["response"]["usage"]
+
+ for stream in (False, True):
+ u = usage(stream)
+ assert u["output_tokens"] == 9
+ assert u["output_tokens_details"]["reasoning_tokens"] == 7
+
+
def test_route_length_truncation_reports_incomplete():
client = _client(FakeState([("partial", True, 5, 8192)], finish_reason="length"))
r = client.post("/v1/responses", json={"model": "gpt-x", "input": "hi"})
diff --git a/tests/server/test_streaming_model_matrix.py b/tests/server/test_streaming_model_matrix.py
index eb6e3a2ed2..2226fe7fe5 100644
--- a/tests/server/test_streaming_model_matrix.py
+++ b/tests/server/test_streaming_model_matrix.py
@@ -708,3 +708,212 @@ def norm(s):
assert norm(_content(events)) == norm(full.content)
assert [c.name for c in _calls(events)] == [c.name for c in full.tool_calls] == ["read"]
+
+
+# --------------------------------------------------------------------------- #
+# Axis 3: production Qwen3.8 parsers (qwen3 + qwen3_coder), stream == one-shot
+# --------------------------------------------------------------------------- #
+QWEN_TOOLS = TOOLS + [
+ {
+ "type": "function",
+ "function": {
+ "name": "edit",
+ "parameters": {
+ "type": "object",
+ "properties": {"path": {"type": "string"}, "new": {"type": "string"}},
+ },
+ },
+ },
+]
+
+
+def _qwen_parity(pieces, *, thinking=True, finish_reason=None):
+ """Feed token texts one per ack to the streaming path and joined to the
+ non-streaming path; assert both expose the same content, reasoning and calls
+ (exactly, no strip) and return them."""
+ spec = GenSpec(
+ messages=[], sampling_params=None, parser_tools=QWEN_TOOLS, template_tools=QWEN_TOOLS,
+ chat_template_kwargs={} if thinking else {"enable_thinking": False},
+ )
+
+ def state(chunks):
+ return FakeState(chunks, tool_call_parser="qwen3_coder", reasoning_parser="qwen3",
+ finish_reason=finish_reason)
+
+ async def stream(chunks):
+ return [ev async for ev in generate_events(42, spec, state(chunks))]
+
+ def calls(items):
+ return [(c.name, json.loads(c.parameters)) for c in items]
+
+ full = asyncio.run(generate_full(42, spec, state(["".join(pieces)])))
+ one_shot = (full.content, full.reasoning, calls(full.tool_calls), full.finish_reason)
+ # One token per delta (what the detokenizer sends), then the worst case: one char.
+ for chunks in (list(pieces), list("".join(pieces))):
+ events = asyncio.run(stream(chunks))
+ streamed = (_content(events), _reasoning(events), calls(_calls(events)),
+ events[-1].finish_reason)
+ assert streamed == one_shot
+ return one_shot
+
+
+EDIT_WITH_THINK_LITERALS = [
+ "plan", "", "\n\n", "Edit:", "", "\n\n\n",
+ 'START = "', "", '"\nEND = "', "", '"\n', "\n\n",
+ "",
+]
+
+
+def test_qwen_think_literals_in_tool_arguments_stay_in_the_call():
+ # Editing a parser's own source: the markers are argument data, and reasoning
+ # ended at the first for good.
+ content, reasoning, calls, finish = _qwen_parity(EDIT_WITH_THINK_LITERALS)
+ assert reasoning == "plan"
+ assert content == "Edit:"
+ assert calls == [("edit", {"new": 'START = ""\nEND = ""'})]
+ assert finish == "tool_calls"
+
+
+@pytest.mark.parametrize("pieces, expected", [
+ (["r", "", "\n\n", "Use ", "", " tags", " and ", "", "."],
+ ("Use tags and .", "r")),
+ (["r", "", "\n\n", "if a ", "<", " b:", " pass"], ("if a < b: pass", "r")),
+ (["r", "\n", "", "\n\nAnswer", " here", "\n"], ("Answer here", "r")),
+ (["\n", "r", "\n\n", "s", "", "\n\n", "A", "\n\n", "B"], ("A\n\nB", "r\n\ns")),
+ (["", "\n", "r", "", "\n\n", "a"], ("a", "r")),
+], ids=["think-literals-in-content", "code-lt", "ws-glued", "ws-separate", "explicit-think"])
+def test_qwen_reasoning_split_parity(pieces, expected):
+ content, reasoning, calls, _ = _qwen_parity(pieces)
+ assert (content, reasoning) == expected
+ assert calls == []
+
+
+@pytest.mark.parametrize("pieces, expected", [
+ (["Use", " ", " here", " ", " too"], "Use here too"),
+ (["\n", "Plain", " answer", "\n"], "\nPlain answer\n"),
+], ids=["think-literals", "literal-ws"])
+def test_qwen_start_in_content_is_literal(pieces, expected):
+ # enable_thinking=false: the template closed the think block, so nothing the
+ # model writes (except an explicit leading ) reopens it.
+ content, reasoning, calls, _ = _qwen_parity(pieces, thinking=False)
+ assert (content, reasoning, calls) == (expected, "", [])
+
+
+def test_qwen_start_in_content_explicit_leading_think():
+ content, reasoning, _, _ = _qwen_parity(["", "r", "", "\n\n", "a"], thinking=False)
+ assert (content, reasoning) == ("a", "r")
+
+
+def test_qwen_truncated_in_reasoning():
+ content, reasoning, calls, finish = _qwen_parity(
+ ["thinking", " more", "\n", "", "\n\n", "", "\n"] + _qwen_call("/a"))
+ assert (content, calls) == ("", [("read", {"filePath": "/a"})])
+
+
+def test_qwen_tool_call_only_turn():
+ content, reasoning, calls, finish = _qwen_parity([
+ "r", "\n", "", "\n\n", "", "\n", "", "\n",
+ "", "\n/tmp/test_calc.py\n", "", "\n", "",
+ "\n", "",
+ ])
+ assert (content, reasoning, finish) == ("", "r", "tool_calls")
+ assert calls == [("read", READ_ARGS)]
+
+
+def _qwen_call(path):
+ return ["", "\n", "", "\n", "", f"\n{path}\n",
+ "", "\n", "", "\n", ""]
+
+
+@pytest.mark.parametrize("between, after, expected", [
+ (["\nThen more.\n"], [], "\nThen more.\n"),
+ (["\n", "Then", " more.", "\n"], ["\n"], "\nThen more.\n"),
+ (["\n\n"], [" Done", "."], " Done."),
+ (["\n", "\n"], [" ", "Done", "."], " Done."),
+ (["\n"], ["\n\n", "", "x"], "\n\n
x"),
+ (["\n"], [" ", "
", " generic"], " generic"),
+ (["\n"], ["\n", ""], "\n"),
+ (["\n"], ["\n", "", " tags"], "\n tags"),
+], ids=["glued", "separate", "ws-only-glued-tail", "ws-only-separate-tail", "ws-then-tag",
+ "space-then-tag", "stray-close", "literal-open"])
+def test_qwen_text_around_calls_parity(between, after, expected):
+ # Text before, between and after calls is content in both modes, verbatim.
+ # A whitespace-only gap between calls or after the last one is markup
+ # separation and is dropped, however it is chunked.
+ pieces = (["r", "", "\n\n", "Let me look.", "\n\n"] + _qwen_call("/a") + between
+ + _qwen_call("/b") + after)
+ content, reasoning, calls, finish = _qwen_parity(pieces)
+ assert content == "Let me look.\n\n" + expected
+ assert calls == [("read", {"filePath": "/a"}), ("read", {"filePath": "/b"})]
+ assert (reasoning, finish) == ("r", "tool_calls")
+
+
+def test_qwen_empty_tool_block_is_dropped_in_both_modes():
+ content, _, calls, finish = _qwen_parity(
+ ["r", "", "\n\n", "A", "", "\n", "", "\n", "B"])
+ assert (content, calls, finish) == ("A\nB", [], "stop")
+
+
+def test_qwen_truncated_inside_function_tag_is_not_recovered():
+ # Cut off by max_tokens inside "\n", ">x<", "div>", " tags", " and splits", " them."],
+ ["Wrap calls in ", "", "...", "", " tags, then answer."],
+ ["Calls:\n\n", "", "\n", '{"name": "read"}', "\n", ""],
+], ids=["opener", "pair", "not-an-invoke"])
+def test_qwen_tool_call_tag_in_prose_stays_content(pieces):
+ # followed by anything but an invoke or its close is the tag
+ # named in text, like a bare ", "\n\n"] + pieces)
+ assert (content, calls, finish) == ("".join(pieces), [], "stop")
+
+
+BARE_EDIT = ["<", "function", "=", "edit", ">\n", "\n", "a.py\n", "\n",
+ ""]
+
+
+def test_qwen_bare_function_is_a_call_in_both_modes():
+ # The model skipped : an agent still needs the call executed, and
+ # neither mode may leave the markup in content.
+ content, _, calls, finish = _qwen_parity(
+ ["r", "", "\n\n", "Doing it.\n"] + BARE_EDIT + ["\n", "Done", "."])
+ assert (content, finish) == ("Doing it.\n\nDone.", "tool_calls")
+ assert calls == [("edit", {"path": "a.py"})]
+
+
+def test_qwen_bare_function_after_a_wrapped_call():
+ content, _, calls, _ = _qwen_parity(
+ ["r", "", "\n\n"] + _qwen_call("/a") + ["\n"] + BARE_EDIT)
+ assert content == ""
+ assert calls == [("read", {"filePath": "/a"}), ("edit", {"path": "a.py"})]
+
+
+def test_qwen_bare_function_swallows_a_stray_close():
+ content, _, calls, _ = _qwen_parity(
+ ["r", "", "\n\n", "Doing it.\n"] + BARE_EDIT + ["\n", "", "\n", "Done"])
+ assert content == "Doing it.\n\nDone"
+ assert calls == [("edit", {"path": "a.py"})]
+
+
+@pytest.mark.parametrize("pieces", [
+ ["Write ", "<", "function", "=", "name", ">", " tags."],
+ ["<", "function", "=", "nope", ">\n", "\n", "1\n", "\n", ""],
+ ["x ", "<", "function", "=", "read", ">", " y"],
+ ['p = "', "<", "function", "=", '"', " + name"],
+ ["if a ", "<", "f", "(b):", " pass"],
+ ["tail ", "<", "function"],
+], ids=["undeclared-name", "unknown-tool", "no-body", "code-literal", "lt-call", "truncated-prefix"])
+def test_qwen_bare_function_lookalikes_stay_content(pieces):
+ content, _, calls, finish = _qwen_parity(["r", "", "\n\n"] + pieces)
+ assert (content, calls, finish) == ("".join(pieces), [], "stop")
diff --git a/tests/tokenizer/test_detokenize_extras.py b/tests/tokenizer/test_detokenize_extras.py
index 4717617b9a..01e02e63ee 100644
--- a/tests/tokenizer/test_detokenize_extras.py
+++ b/tests/tokenizer/test_detokenize_extras.py
@@ -6,11 +6,12 @@
from freetoken.message import DetokenizeMsg
from freetoken.tokenizer.detokenize import DetokenizeManager
-EOS, SPECIAL, THINK_END = 0, 99, 7
+EOS, SPECIAL, THINK_END, THINK_START = 0, 99, 7, 8
class _Tok:
- """Token ids decode to letters; 99 is a special marker, 7 the reasoning end tag."""
+ """Token ids decode to letters; 99 is a special marker, 7 / 8 the reasoning end /
+ start tags."""
eos_token_id = EOS
unk_token_id = -1
@@ -23,6 +24,8 @@ def decode(self, ids, skip_special_tokens=False):
out.append("")
elif t == THINK_END:
out.append("")
+ elif t == THINK_START:
+ out.append("")
else:
out.append(chr(ord("a") + t))
return "".join(out)
@@ -39,7 +42,7 @@ def _run(manager, uid, tokens, *, finished_kwargs=None, **msg_kwargs):
kwargs = dict(msg_kwargs)
if last and finished_kwargs:
kwargs.update(finished_kwargs)
- strs, counts = manager.detokenize_with_meta(
+ strs, counts, _ = manager.detokenize_with_meta(
[DetokenizeMsg(uid=uid, next_token=t, finished=last, **kwargs)]
)
text += strs[0]
@@ -81,6 +84,27 @@ def test_reasoning_tokens_count_up_to_the_end_tag():
assert none == 0
+def test_reasoning_tokens_when_generation_stops_inside_the_block():
+ # The prompt pre-opened (qwen3, thinking on) and the output never closed it:
+ # every completion token was reasoning, whether max_tokens or EOS ended it.
+ manager = DetokenizeManager(
+ _Tok(), frozenset({EOS}), think_end_id=THINK_END, think_start_id=THINK_START
+ )
+ _, by_length = _run(manager, 1, [1, 2, 3], starts_in_reasoning=True)
+ assert by_length == 3
+ _, by_eos = _run(manager, 2, [1, 2, EOS], starts_in_reasoning=True)
+ assert by_eos == 3
+ # the model opened the block itself
+ _, self_opened = _run(manager, 3, [THINK_START, 1, 2])
+ assert self_opened == 3
+ # thinking off: no block, no reasoning
+ _, off = _run(manager, 4, [1, 2, 3])
+ assert off == 0
+ # a closed block still counts up to the end tag
+ _, closed = _run(manager, 5, [1, THINK_END, 2, 3], starts_in_reasoning=True)
+ assert closed == 2
+
+
def test_a_batch_with_one_uid_twice_advances_its_state_in_order():
# each message must see the decode state the previous one left, and the reasoning
# counts must stay aligned with the texts
@@ -91,7 +115,156 @@ def test_a_batch_with_one_uid_twice_advances_its_state_in_order():
DetokenizeMsg(uid=2, next_token=3, finished=False),
DetokenizeMsg(uid=1, next_token=2, finished=True),
]
- texts, counts = manager.detokenize_with_meta(msgs)
+ texts, counts, offsets = manager.detokenize_with_meta(msgs)
assert texts == ["b", "", "d", "c"]
assert counts == [0, 0, 0, 2]
+ assert offsets == [0, 1, 0, 9]
assert manager.detokenize([DetokenizeMsg(uid=2, next_token=4, finished=True)]) == ["e"]
+
+
+def test_text_offset_is_where_the_token_starts_despite_holdback():
+ # "c" is held back as a possible start of the stop "cx", then sent with "d": its
+ # offset is still where it starts, not how much text had been sent at its ack.
+ manager = DetokenizeManager(_Tok(), frozenset({EOS}))
+ sent, offsets = [], []
+ for i, t in enumerate([1, 2, 3, EOS]):
+ strs, _, offs = manager.detokenize_with_meta(
+ [DetokenizeMsg(uid=1, next_token=t, finished=i == 3, stop_strs=["cx"])]
+ )
+ sent.append(strs[0])
+ offsets.append(offs[0])
+ assert sent == ["b", "", "cd", ""]
+ assert offsets == [0, 1, 2, 3]
+
+
+PARTIAL, COMPLETES = 50, 51
+
+
+class _ByteTok(_Tok):
+ """50 is the first half of a two-token character: alone it decodes to U+FFFD."""
+
+ def decode(self, ids, skip_special_tokens=False):
+ out = []
+ for i, t in enumerate(ids):
+ if t == PARTIAL:
+ out.append("" if i + 1 < len(ids) and ids[i + 1] == COMPLETES else "�")
+ elif t == COMPLETES:
+ out.append("é" if i and ids[i - 1] == PARTIAL else "�")
+ else:
+ out.append(super().decode([t], skip_special_tokens))
+ return "".join(out)
+
+
+def _run_logprobs(manager, tokens, *, finished_kwargs=None, **msg_kwargs):
+ """(text, entry token ids) per reply, each token's entry carrying its own id."""
+ replies = []
+ for i, t in enumerate(tokens):
+ last = i == len(tokens) - 1
+ kwargs = dict(msg_kwargs)
+ if last and finished_kwargs:
+ kwargs.update(finished_kwargs)
+ texts, _, released = manager.detokenize_with_logprobs(
+ [DetokenizeMsg(uid=1, next_token=t, finished=last, **kwargs)], [{"token_id": t}]
+ )
+ replies.append(
+ (texts[0], [(e["token_id"], bool(e.get("stop_trimmed"))) for e in released[0]])
+ )
+ return replies
+
+
+def test_logprob_entries_of_stop_trimmed_tokens_are_marked():
+ manager = DetokenizeManager(_Tok(), frozenset({EOS}))
+ assert _run_logprobs(
+ manager, [1, 2, 3], stop_strs=["cd"], finished_kwargs={"matched_stop": "cd"}
+ ) == [("b", [(1, False)]), ("", []), ("", [(2, True), (3, True)])]
+
+
+def test_logprob_entry_of_a_token_a_stop_string_cuts_inside_carries_its_visible_length():
+ # "bc" with stop "k>c": " token stays visible.
+ manager = DetokenizeManager(_Tok(), frozenset({EOS}))
+ final = None
+ for i, t in enumerate([1, THINK_END, 2]):
+ last = i == 2
+ _, _, released = manager.detokenize_with_logprobs(
+ [DetokenizeMsg(uid=1, next_token=t, finished=last, stop_strs=["k>c"],
+ matched_stop="k>c" if last else None)],
+ [{"token_id": t}],
+ )
+ final = released[0]
+ # each entry also carries where its token starts in the text ("b" is 1 char, "" 8)
+ assert final == [
+ {"token_id": THINK_END, "text_offset": 1, "stop_visible": 6},
+ {"token_id": 2, "text_offset": 9, "stop_trimmed": True},
+ ]
+
+
+def test_logprob_entry_waits_for_its_text_when_a_stop_prefix_is_released():
+ # "b" is held as a prefix of "bc"; the second "b" releases the first, but its own
+ # text is held, so its entry must not ride that reply -- the stop then trims it.
+ manager = DetokenizeManager(_Tok(), frozenset({EOS}))
+ assert _run_logprobs(
+ manager, [1, 1, 2], stop_strs=["bc"], finished_kwargs={"matched_stop": "bc"}
+ ) == [("", []), ("b", [(1, False)]), ("", [(1, True), (2, True)])]
+
+
+def test_logprob_entries_kept_with_the_stop_string_or_without_a_match():
+ manager = DetokenizeManager(_Tok(), frozenset({EOS}))
+ kept = _run_logprobs(
+ manager, [1, 2, 3], stop_strs=["cd"], keep_stop_str=True,
+ finished_kwargs={"matched_stop": "cd"},
+ )
+ assert kept[-1] == ("cd", [(2, False), (3, False)])
+ # finished by length while "c" was held as a stop prefix: nothing was trimmed
+ by_length = _run_logprobs(manager, [1, 2], stop_strs=["cd"])
+ assert by_length == [("b", [(1, False)]), ("c", [(2, False)])]
+ # no stop strings: every entry rides its own reply, as before
+ plain = _run_logprobs(manager, [1, 2, EOS], finished_kwargs={"finish_reason": "stop"})
+ assert plain == [("b", [(1, False)]), ("c", [(2, False)]), ("", [(EOS, False)])]
+
+
+def test_logprob_entry_of_a_partial_character_rides_the_reply_that_completes_it():
+ manager = DetokenizeManager(_ByteTok(), frozenset({EOS}))
+ assert _run_logprobs(manager, [1, PARTIAL, COMPLETES, 2], stop_strs=["zz"]) == [
+ ("b", [(1, False)]),
+ ("", []),
+ ("é", [(PARTIAL, False), (COMPLETES, False)]),
+ ("c", [(2, False)]),
+ ]
+
+
+def test_logprob_entries_released_together_keep_their_own_text_offsets():
+ # "c" is held as a prefix of "cx" and released with "d": that reply carries both
+ # entries, each stamped with where its own token starts, not one per-reply offset.
+ manager = DetokenizeManager(_Tok(), frozenset({EOS}))
+ replies = []
+ for i, t in enumerate([1, 2, 3, EOS]):
+ texts, _, released = manager.detokenize_with_logprobs(
+ [DetokenizeMsg(uid=1, next_token=t, finished=i == 3, stop_strs=["cx"])],
+ [{"token_id": t}],
+ )
+ replies.append((texts[0], [(e["token_id"], e["text_offset"]) for e in released[0]]))
+ assert replies == [("b", [(1, 0)]), ("", []), ("cd", [(2, 1), (3, 2)]), ("", [(EOS, 3)])]
+
+
+def test_held_logprob_entries_release_on_a_message_without_its_own_entry():
+ # "c" is held as a prefix of "cx"; the messages that release it carry no entry
+ # (and skip the hold bookkeeping when nothing is held), yet "c"'s entry still rides them.
+ def run(tokens, entries, last_finishes):
+ manager = DetokenizeManager(_Tok(), frozenset({EOS}))
+ replies = []
+ for i, (t, entry) in enumerate(zip(tokens, entries, strict=True)):
+ finished = last_finishes and i == len(tokens) - 1
+ texts, _, released = manager.detokenize_with_logprobs(
+ [DetokenizeMsg(uid=1, next_token=t, finished=finished, stop_strs=["cx"])], [entry]
+ )
+ replies.append((texts[0], [e["token_id"] for e in released[0]]))
+ assert manager.decode_map.get(1) is None or not manager.decode_map[1].held_logprobs
+ return replies
+
+ e = [{"token_id": t} for t in (1, 2)]
+ # released mid-stream by "d", whose message has no entry
+ assert run([1, 2, 3], [*e, None], False) == [("b", [1]), ("", []), ("cd", [2])]
+ # flushed by the finishing message, which has no entry
+ assert run([1, 2, EOS], [*e, None], True) == [("b", [1]), ("", []), ("c", [2])]
+ # a request without logprobs never carries any
+ assert run([1, 2, 3], [None] * 3, True) == [("b", []), ("", []), ("cd", [])]
diff --git a/tests/tokenizer/test_logprobs_entry.py b/tests/tokenizer/test_logprobs_entry.py
index 89cf35cb5e..978824a34d 100644
--- a/tests/tokenizer/test_logprobs_entry.py
+++ b/tests/tokenizer/test_logprobs_entry.py
@@ -1,3 +1,7 @@
+import os
+
+import pytest
+
from freetoken.tokenizer.detokenize import build_logprobs_entry
@@ -57,3 +61,53 @@ def test_build_logprobs_entry_empty_top() -> None:
)
assert entry["top"] == []
+
+
+def _byte_level_tokenizer():
+ """A tiny GPT-2-style ByteLevel BPE tokenizer built in memory (no download): the 256
+ byte stand-ins, one merge ("ðŁ", the first two bytes of most emoji) and one added
+ token whose content the byte table cannot invert."""
+ from tokenizers import AddedToken, Tokenizer, decoders, models, pre_tokenizers
+ from transformers import PreTrainedTokenizerFast
+
+ vocab = {c: i for i, c in enumerate(sorted(pre_tokenizers.ByteLevel.alphabet()))}
+ vocab["ðŁ"] = len(vocab)
+ tk = Tokenizer(models.BPE(vocab=vocab, merges=[("ð", "Ł")]))
+ tk.pre_tokenizer = pre_tokenizers.ByteLevel(add_prefix_space=False, use_regex=False)
+ tk.decoder = decoders.ByteLevel()
+ tk.add_tokens([AddedToken("<｜end｜>", special=False)])
+ return PreTrainedTokenizerFast(tokenizer_object=tk)
+
+
+def test_logprobs_bytes_rebuild_a_split_character() -> None:
+ # 🦀 is three byte-level tokens that each decode to U+FFFD; their bytes must still
+ # concatenate back to the character.
+ tokenizer = _byte_level_tokenizer()
+ ids = tokenizer.encode("🦀", add_special_tokens=False)
+ assert len(ids) == 3
+ entries = [build_logprobs_entry(tokenizer, i, -0.5, [i], [-0.5]) for i in ids]
+ assert all(e["token"] == "�" for e in entries)
+ assert bytes(b for e in entries for b in e["bytes"]) == "🦀".encode("utf-8")
+ assert [e["top"][0]["bytes"] for e in entries] == [e["bytes"] for e in entries]
+
+
+def test_logprobs_bytes_added_and_padded_ids() -> None:
+ tokenizer = _byte_level_tokenizer()
+ added = tokenizer.convert_tokens_to_ids("<｜end｜>")
+ padded = len(tokenizer) + 10 # the logits' padded tail: no vocab entry at all
+ entry = build_logprobs_entry(tokenizer, added, -0.1, [added, padded], [-0.1, -9.0])
+ assert entry["token"] == "<｜end｜>"
+ assert entry["bytes"] == list("<｜end｜>".encode("utf-8"))
+ assert entry["top"][1] == {"token_id": padded, "token": "", "bytes": [], "logprob": -9.0}
+
+
+def test_logprobs_bytes_real_qwen_tokenizer() -> None:
+ path = os.path.expanduser("~/models/Qwen3.8-Flash-Next-NVFP4-FTW-fp8")
+ if not os.path.exists(os.path.join(path, "tokenizer.json")):
+ pytest.skip("local Qwen3.8 tokenizer not present")
+ from transformers import AutoTokenizer
+
+ tokenizer = AutoTokenizer.from_pretrained(path)
+ entries = [build_logprobs_entry(tokenizer, i, -0.1, [248300], [-9.0]) for i in (9008, 99, 222)]
+ assert bytes(b for e in entries for b in e["bytes"]) == "🦀".encode("utf-8")
+ assert entries[0]["top"][0]["bytes"] == []