Preserve speculative finish correctness across reasoning and stop strings

ReasonerGrammarObject wraps an inner grammar, but disaggregated prebuilt replay checks the wrapper current_token to avoid accepting an already accepted token twice. Track the token on the wrapper itself so reasoning grammar follows the same contract as XGrammar.

Speculative decode can accept a stop string and EOS in one step. Check stop strings before token-based EOS after sanitizing invalid token ids, and set finished_len at the matched stop position so trailing accepted tokens do not leak.

Constraint: Current branch predates upstream helper methods for locating string stop positions, so the stop-string fix is manually ported instead of cherry-picked.

Rejected: Direct cherry-pick of bbc853df46 | current schedule_batch.py lacks the upstream helper context.

Confidence: high

Scope-risk: moderate

Directive: Keep vocab-boundary sanitization before string decoding; do not move token-based EOS ahead of stop-string checks without a same-step speculative regression test.

Tested: RED/GREEN remote pytest in cjy-glm5-new for constrained current_token and stop-string speculative tests

Tested: git diff --check; py_compile for reasoner_grammar_backend.py and schedule_batch.py

Not-tested: Full scheduler/disaggregation integration suite

test/registered/unit/constrained/test_reasoner_grammar_backend.py

@@ -0,0 +1,69 @@
+import unittest
+
+from sglang.srt.constrained.reasoner_grammar_backend import ReasonerGrammarObject
+
+
+class _InnerGrammar:
+    def __init__(self):
+        self.accepted_tokens = []
+
+    def accept_token(self, token: int):
+        self.accepted_tokens.append(token)
+
+    def is_terminated(self):
+        return False
+
+    def rollback(self, k: int):
+        self.accepted_tokens = self.accepted_tokens[:-k]
+
+    def allocate_vocab_mask(self, vocab_size: int, batch_size: int, device):
+        return None
+
+    def fill_vocab_mask(self, vocab_mask, idx: int):
+        return None
+
+    def move_vocab_mask(self, vocab_mask, device):
+        return vocab_mask
+
+    @property
+    def apply_vocab_mask(self):
+        return None
+
+    def copy(self):
+        ret = _InnerGrammar()
+        ret.accepted_tokens = list(self.accepted_tokens)
+        return ret
+
+    @property
+    def finished(self):
+        return False
+
+
+class TestReasonerGrammarCurrentToken(unittest.TestCase):
+    def test_wrapper_tracks_current_token_for_disagg_prebuilt_dedup(self):
+        inner = _InnerGrammar()
+        grammar = ReasonerGrammarObject(inner, think_end_id=7)
+        grammar.maybe_init_reasoning(True)
+
+        grammar.accept_token(10)  # still in thinking; inner grammar is untouched
+        self.assertEqual(grammar.current_token, 10)
+        self.assertEqual(inner.accepted_tokens, [])
+
+        grammar.accept_token(7)  # exits thinking
+        self.assertEqual(grammar.current_token, 7)
+        self.assertEqual(inner.accepted_tokens, [])
+
+        grammar.accept_token(58)  # generation token accepted by inner grammar
+        self.assertEqual(grammar.current_token, 58)
+        self.assertEqual(inner.accepted_tokens, [58])
+
+        # Mirrors disaggregation/decode_schedule_batch_mixin.py:process_prebuilt.
+        # Once current_token is tracked on the wrapper, a re-prebuilt request
+        # skips re-accepting the already accepted token.
+        if grammar.current_token is None:
+            grammar.accept_token(58)
+        self.assertEqual(inner.accepted_tokens, [58])
+
+
+if __name__ == "__main__":
+    unittest.main()

test/registered/unit/managers/test_stop_str_speculative.py

@@ -0,0 +1,61 @@
+import unittest
+from array import array
+
+from sglang.srt.managers.schedule_batch import Req
+from sglang.srt.sampling.sampling_params import SamplingParams
+
+
+STOP_ID = 1
+EOS_ID = 2
+ID_TO_TEXT = {
+    STOP_ID: "STOP",
+    EOS_ID: "",
+    **{i: chr(ord("a") + i % 26) for i in range(10, 40)},
+}
+
+
+class _FakeTokenizer:
+    eos_token_id = EOS_ID
+    additional_stop_token_ids = None
+
+    def decode(self, ids):
+        return "".join(ID_TO_TEXT[int(i)] for i in ids)
+
+
+class _NormalizeTokenizer:
+    def encode(self, text, add_special_tokens=False):
+        return list(range(len(text)))
+
+
+def _make_req(output_ids, *, stop=None, eos_token_ids=frozenset()):
+    sampling_params = SamplingParams(max_new_tokens=1000, stop=stop)
+    sampling_params.normalize(tokenizer=_NormalizeTokenizer())
+    req = Req(
+        rid="r",
+        origin_input_text="",
+        origin_input_ids=array("q", [0]),
+        sampling_params=sampling_params,
+        eos_token_ids=set(eos_token_ids),
+        vocab_size=10_000,
+    )
+    req.tokenizer = _FakeTokenizer()
+    req.output_ids = array("q", output_ids)
+    return req
+
+
+class TestStopStrSpeculative(unittest.TestCase):
+    def test_stop_str_wins_over_eos_in_same_spec_step(self):
+        # A speculative step may accept both the stop string and EOS. The stop
+        # string must finish first so finished_len trims at STOP instead of EOS.
+        req = _make_req([10, 11, STOP_ID, EOS_ID], stop=["STOP"], eos_token_ids={EOS_ID})
+
+        req.check_finished(new_accepted_len=4)
+
+        self.assertTrue(req.finished())
+        self.assertEqual(req.finished_reason.matched, "STOP")
+        self.assertEqual(req.finished_len, 3)
+        self.assertEqual(list(req.output_ids_through_stop), [10, 11, STOP_ID])
+
+
+if __name__ == "__main__":
+    unittest.main()