fix(qwen3): preserve explicit stop-token causes

pegainfer-frontend/src/engine/driver.rs

@@ -108,6 +108,7 @@ mod tests {
     use super::super::step::Request;
     use super::super::step::RequestId;
     use super::super::step::Terminal;
+    use super::super::stop::StopPolicy;
     use super::*;
     use crate::engine::FinishReason;
 
@@ -163,6 +164,7 @@ mod tests {
         Request {
             prompt_tokens: vec![1, 2],
             params: crate::sampler::SamplingParams::default(),
+            stop_policy: StopPolicy::default(),
             max_tokens,
             lora_adapter: None,
             kv_transfer_params: None,
@@ -197,6 +199,7 @@ mod tests {
             terminal,
             Some(Terminal::Finished {
                 reason: FinishReason::Length,
+                stop_cause: None,
                 prompt_tokens: 2,
                 completion_tokens: 3,
             })

pegainfer-frontend/src/engine/ledger.rs

@@ -38,6 +38,7 @@ use super::step::RequestUpdate;
 use super::step::ScheduledInfo;
 use super::step::StepOutputs;
 use super::step::Terminal;
+use super::stop::StopCause;
 
 /// One open account: the request's admission facts and running tally. The
 /// payload is not here — it went to the scheduler at `submit`; the account is
@@ -239,12 +240,23 @@ impl RequestLedger {
 
     /// Finish the request. Token counts come from the ledger's tally.
     pub fn finish(&mut self, id: RequestId, reason: FinishReason) {
+        self.finish_with_cause(id, reason, None);
+    }
+
+    /// Finish a request while preserving a typed token-level stop cause.
+    pub fn finish_with_cause(
+        &mut self,
+        id: RequestId,
+        reason: FinishReason,
+        stop_cause: Option<StopCause>,
+    ) {
         let account = self.close(id);
         let AccountState::Active { completion_tokens } = account.state else {
             panic!("finish on {id} before admission");
         };
         self.statement.entry(id).terminal = Some(Terminal::Finished {
             reason,
+            stop_cause,
             prompt_tokens: account.prompt_len,
             completion_tokens,
         });
@@ -279,6 +291,16 @@ impl RequestLedger {
     /// this step — tokens included — folds into the returned message, so late
     /// delivery cannot reorder against the step stream.
     pub fn defer_finish(&mut self, id: RequestId, reason: FinishReason) -> DeferredFinish {
+        self.defer_finish_with_cause(id, reason, None)
+    }
+
+    /// Defer a finish while preserving a typed token-level stop cause.
+    pub fn defer_finish_with_cause(
+        &mut self,
+        id: RequestId,
+        reason: FinishReason,
+        stop_cause: Option<StopCause>,
+    ) -> DeferredFinish {
         let account = self.close(id);
         let AccountState::Active { completion_tokens } = account.state else {
             panic!("defer_finish on {id} before admission");
@@ -289,6 +311,7 @@ impl RequestLedger {
             .unwrap_or_else(|| RequestUpdate::empty(id));
         update.terminal = Some(Terminal::Finished {
             reason,
+            stop_cause,
             prompt_tokens: account.prompt_len,
             completion_tokens,
         });
@@ -374,6 +397,7 @@ mod tests {
     use super::super::request_lifecycle::StepReceiver;
     use super::super::step::Request;
     use super::super::step::Terminal;
+    use super::super::stop::StopPolicy;
     use super::super::wiring::SchedulerHandle;
     use super::super::wiring::scheduler_pair;
     use super::*;
@@ -382,6 +406,7 @@ mod tests {
         Request {
             prompt_tokens: prompt,
             params: crate::sampler::SamplingParams::default(),
+            stop_policy: StopPolicy::default(),
             max_tokens: 8,
             lora_adapter: None,
             kv_transfer_params: None,
@@ -402,7 +427,9 @@ mod tests {
         backend.ledger.admit(id);
         backend.ledger.push_tokens(id, &[10, 11], &[]);
         backend.ledger.set_cached_tokens(id, 2);
-        backend.ledger.finish(id, FinishReason::Stop);
+        backend
+            .ledger
+            .finish_with_cause(id, FinishReason::Stop, Some(StopCause::Token(11)));
         backend.ledger.commit_step();
 
         let mut steps = handle_steps(handle);
@@ -419,6 +446,7 @@ mod tests {
             update.terminal,
             Some(Terminal::Finished {
                 reason: FinishReason::Stop,
+                stop_cause: Some(StopCause::Token(11)),
                 prompt_tokens: 3,
                 completion_tokens: 2,
             })
@@ -479,7 +507,9 @@ mod tests {
         let id = backend.ledger.register(envelope).id;
         backend.ledger.admit(id);
         backend.ledger.push_tokens(id, &[7], &[]);
-        let deferred = backend.ledger.defer_finish(id, FinishReason::Length);
+        let deferred = backend
+            .ledger
+            .defer_finish_with_cause(id, FinishReason::Length, None);
         backend.ledger.commit_step();
 
         let mut steps = handle_steps(handle);
@@ -498,6 +528,7 @@ mod tests {
             update.terminal,
             Some(Terminal::Finished {
                 reason: FinishReason::Length,
+                stop_cause: None,
                 prompt_tokens: 2,
                 completion_tokens: 1,
             })

pegainfer-frontend/src/engine/mod.rs

@@ -39,6 +39,7 @@ mod request;
 mod request_lifecycle;
 mod sink;
 mod step;
+mod stop;
 mod wiring;
 
 pub use control::*;
@@ -52,4 +53,5 @@ pub use request::*;
 pub use request_lifecycle::*;
 pub use sink::*;
 pub use step::*;
+pub use stop::*;
 pub use wiring::*;

pegainfer-frontend/src/engine/step.rs

@@ -14,6 +14,8 @@ use std::time::Instant;
 
 use super::event::FinishReason;
 use super::event::TokenLogprob;
+use super::stop::StopCause;
+use super::stop::StopPolicy;
 
 /// In-process routing id for one generate request, minted by
 /// [`super::SchedulerHandle::submit`] from a per-scheduler counter. `Copy` and
@@ -49,6 +51,7 @@ impl std::fmt::Display for RequestId {
 pub struct Request {
     pub prompt_tokens: Vec<u32>,
     pub params: crate::sampler::SamplingParams,
+    pub stop_policy: StopPolicy,
     pub max_tokens: usize,
     pub lora_adapter: Option<String>,
     /// Opaque router/P-D metadata from the request's
@@ -228,6 +231,10 @@ impl fmt::Display for RejectReason {
 pub enum Terminal {
     Finished {
         reason: FinishReason,
+        /// Present for token-driven stop finishes. The triggering token remains
+        /// in `RequestUpdate.tokens`, with its real logprob in the matching
+        /// `RequestUpdate.logprobs` entry.
+        stop_cause: Option<StopCause>,
         prompt_tokens: usize,
         completion_tokens: usize,
     },

pegainfer-frontend/src/engine/stop.rs

@@ -0,0 +1,104 @@
+/// How a request treats end-of-sequence tokens.
+#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
+pub enum EosPolicy {
+    /// Do not stop on model EOS tokens.
+    Ignore,
+    /// Use the model executor's configured EOS set.
+    #[default]
+    ModelDefault,
+    /// Stop only on this protocol-provided primary EOS token.
+    Token(u32),
+}
+
+/// Request-scoped token stopping policy.
+///
+/// EOS is kept separate from caller stop tokens because the vLLM protocol
+/// reports them differently: EOS has no 'stop_reason', while a request stop
+/// reports the actual matching token ID.
+#[derive(Clone, Debug, Default, Eq, PartialEq)]
+pub struct StopPolicy {
+    pub eos: EosPolicy,
+    pub token_ids: Vec<u32>,
+}
+
+impl StopPolicy {
+    /// Classify a token using vLLM's priority: EOS first, then the request's
+    /// explicit stop-token set.
+    #[must_use]
+    pub fn classify(
+        &self,
+        token_id: u32,
+        is_model_eos: impl FnOnce(u32) -> bool,
+    ) -> Option<StopCause> {
+        let is_eos = match self.eos {
+            EosPolicy::Ignore => false,
+            EosPolicy::ModelDefault => is_model_eos(token_id),
+            EosPolicy::Token(eos_token_id) => token_id == eos_token_id,
+        };
+
+        if is_eos {
+            Some(StopCause::Eos(token_id))
+        } else if self.token_ids.contains(&token_id) {
+            Some(StopCause::Token(token_id))
+        } else {
+            None
+        }
+    }
+}
+
+/// The token-level cause of a normal stop finish.
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum StopCause {
+    /// A primary or model-default EOS token.
+    Eos(u32),
+    /// A token from the request's explicit stop-token set.
+    Token(u32),
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn model_default_classifies_model_eos() {
+        let policy = StopPolicy::default();
+
+        assert_eq!(
+            policy.classify(99, |token_id| token_id == 99),
+            Some(StopCause::Eos(99))
+        );
+    }
+
+    #[test]
+    fn ignored_eos_does_not_disable_an_explicit_stop() {
+        let policy = StopPolicy {
+            eos: EosPolicy::Ignore,
+            token_ids: vec![99],
+        };
+
+        assert_eq!(
+            policy.classify(99, |token_id| token_id == 99),
+            Some(StopCause::Token(99))
+        );
+    }
+
+    #[test]
+    fn eos_wins_when_the_same_id_is_also_an_explicit_stop() {
+        let policy = StopPolicy {
+            eos: EosPolicy::Token(99),
+            token_ids: vec![99],
+        };
+
+        assert_eq!(policy.classify(99, |_| false), Some(StopCause::Eos(99)));
+    }
+
+    #[test]
+    fn unmatched_token_does_not_stop() {
+        let policy = StopPolicy {
+            eos: EosPolicy::Token(99),
+            token_ids: vec![42],
+        };
+
+        assert_eq!(policy.classify(7, |_| false), None);
+    }
+}