feat(v1/sched): update_from_output + check_stop (schedule->execute->update loop)

Port scheduler.py::update_from_output + sched/utils.py::check_stop (T0 subset)
from vllm @ e24d1b24, closing the schedule->execute->update loop:

- sched/utils.{h,cpp}: check_stop(request, max_model_len). Precedence matched to
  the pin: min_tokens gate FIRST, then eos, stop_token_ids (sets stop_reason),
  then length cap (max_model_len / max_tokens). Repetition-detection deferred.
- Scheduler::update_from_output: append sampled token(s) per scheduled request,
  run check_stop after each (trim on stop), free KV + record finished_req_ids +
  erase from running/waiting/requests on finish, build EngineCoreOutputs.
  Deferred: logprobs, spec-decode, structured output, pooling, encoder,
  KV-connector, routed-experts, stats, resumable/streaming, client fan-out.
- request.h: un-defer stop_reason (optional<int>) for the stop-token path.
- Tests ported from test_scheduler.py: append, eos stop, max_tokens length cap,
  min_tokens gate, stop_token_ids/stop_reason, partial-prefill no-token,
  batched EngineCoreOutputs shape.

ctest 40/40 green; warnings-as-errors clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: mudler

CMakeLists.txt

@@ -70,6 +70,7 @@ add_library(vllm STATIC
   src/vllm/v1/core/sched/request_queue.cpp
   src/vllm/v1/core/sched/output.cpp
   src/vllm/v1/core/sched/scheduler.cpp
+  src/vllm/v1/core/sched/utils.cpp
   src/vllm/v1/engine/types.cpp
   src/vllm/v1/engine/detokenizer.cpp
   src/vt/dtype.cpp

include/vllm/v1/core/sched/scheduler.h

@@ -67,6 +67,7 @@
 #include "vllm/v1/core/kv_cache_manager.h"
 #include "vllm/v1/core/sched/output.h"
 #include "vllm/v1/core/sched/request_queue.h"
+#include "vllm/v1/engine/types.h"  // ModelRunnerOutput, EngineCoreOutputs
 #include "vllm/v1/kv_cache_interface.h"
 #include "vllm/v1/request.h"
 
@@ -104,6 +105,31 @@ class Scheduler {
   // schedule(): the core token-budget algorithm. See the file header.
   SchedulerOutput schedule();
 
+  // update_from_output: the final leg of the schedule -> execute -> update loop.
+  // For each scheduled request (iterating scheduler_output.num_scheduled_tokens,
+  // matching upstream's `for req_id ... in num_scheduled_tokens.items()`), it
+  // reads that request's sampled token(s) from model_runner_output via
+  // req_id_to_index, appends them one at a time running check_stop after each
+  // (trimming any tokens past a stop), and on a stop sets the finished status,
+  // frees the request's KV blocks, records its id in finished_req_ids, removes it
+  // from running_/waiting_ and erases it from the requests map. It builds one
+  // EngineCoreOutput per request that produced tokens or finished, and returns
+  // the batched EngineCoreOutputs for the step. A request still in chunked
+  // prefill receives an empty token list from the runner and stays running with
+  // no output. (Upstream scheduler.py::update_from_output, T0 subset.)
+  //
+  // DEFERRED (matches upstream structure): logprobs / prompt_logprobs,
+  // speculative-decode acceptance & num_computed rollback, structured-output
+  // grammar advance, pooling outputs, encoder-input free, KV-connector finish /
+  // invalid-block reload, routed-experts, num_nans_in_logits, spec/perf/spec-
+  // decoding stats, resumable / streaming sessions (so _handle_stopped_request
+  // is always "finished" here), and the per-client_index output fan-out (T0 has
+  // a single client, so a flat EngineCoreOutputs is returned rather than
+  // dict[client_index, EngineCoreOutputs]).
+  EngineCoreOutputs update_from_output(
+      const SchedulerOutput& scheduler_output,
+      const ModelRunnerOutput& model_runner_output);
+
   // get_num_unfinished_requests: len(waiting) + len(running) (T0 subset).
   int get_num_unfinished_requests() const;
   // get_request_counts: (num_running, num_waiting).

include/vllm/v1/core/sched/utils.h

@@ -0,0 +1,42 @@
+// Ported from: vllm/v1/core/sched/utils.py @ e24d1b24
+//
+// Scope (M1.4 Task 4): the token-level stop check the scheduler runs after each
+// sampled token. check_stop is the only piece of sched/utils.py the V1 engine's
+// update_from_output path needs at T0.
+//
+// DEFERRED (marked; the upstream file's other helpers are not needed at T0):
+//   - _has_repeating_pattern / check_sequence_repetition (the
+//     repetition-detection path — sampling_params.repetition_detection is a
+//     deferred SamplingParams field, so check_stop's repetition branch is
+//     omitted here).
+//   - remove_all (the list-removal helper) — update_from_output removes stopped
+//     requests from the running vector / waiting queue directly.
+//
+// IMPORTANT (stop precedence, verified against the pin — NOT the summary in the
+// plan): the pinned check_stop order is
+//   1. min_tokens gate: num_output_tokens < min_tokens -> return false FIRST
+//      (this gates *everything*, including the length cap, at this pin);
+//   2. eos: last token == eos_token_id -> FINISHED_STOPPED;
+//   3. stop_token_ids: last token in stop_token_ids -> FINISHED_STOPPED
+//      (stop_reason = the matched token id);
+//   4. length cap: num_tokens >= max_model_len OR num_output_tokens >= max_tokens
+//      -> FINISHED_LENGTH_CAPPED;
+//   5. repetition (DEFERRED).
+#ifndef VLLM_V1_CORE_SCHED_UTILS_H_
+#define VLLM_V1_CORE_SCHED_UTILS_H_
+
+#include "vllm/v1/request.h"
+
+namespace vllm::v1 {
+
+// check_stop(request, max_model_len): returns true and sets request.status
+// (and, for a stop_token_ids match, request.stop_reason) when the request has
+// hit a stop condition after its latest sampled token. Mirrors
+// vllm/v1/core/sched/utils.py::check_stop for the T0 subset (pooling and
+// repetition-detection branches deferred). Mutates the request exactly as
+// upstream does.
+bool check_stop(Request& request, int max_model_len);
+
+}  // namespace vllm::v1
+
+#endif  // VLLM_V1_CORE_SCHED_UTILS_H_

include/vllm/v1/request.h

@@ -15,7 +15,7 @@
 //   - prompt_embeds / prompt_is_token_ids / _prompt_embeds_per_block_hashes,
 //     mm_features (multimodal), pooling_params, structured_output_request,
 //     lora_request, cache_salt (prefix caching salt), events /
-//     stop_reason / kv_transfer_params, spec_token_ids, priority /
+//     kv_transfer_params, spec_token_ids, priority /
 //     client_index / __lt__ (priority scheduling), streaming / resumable
 //     state, prefill_stats, async-scheduling counters
 //     (num_output_placeholders, async_tokens_to_discard,
@@ -134,6 +134,15 @@ struct Request {
   std::vector<int32_t> output_token_ids;
   int num_computed_tokens = 0;
   RequestStatus status = RequestStatus::kWaiting;
+  // stop_reason (upstream Request.stop_reason: int | str | None = None). Set by
+  // check_stop (sched/utils) when a stop_token_ids match ends the request — it
+  // carries the matched token id — and read by update_from_output when it builds
+  // the request's EngineCoreOutput. Un-deferred from the Request deferred list
+  // because M1.4 Task 4 needs it. Repetition detection's string reason
+  // ("repetition_detected") is deferred with sampling_params.repetition_detection,
+  // so at T0 only the int (stop-token) form ever occurs — hence std::optional<int>
+  // rather than a variant<int, string>.
+  std::optional<int> stop_reason;
   double arrival_time = 0.0;
   // Set at construction from prompt_token_ids.size() (upstream:
   // length_from_prompt_token_ids_or_embeds).

src/vllm/v1/core/sched/scheduler.cpp

@@ -4,6 +4,7 @@
 
 #include <algorithm>
 #include <cassert>
+#include <cstdint>
 #include <map>
 #include <memory>
 #include <optional>
@@ -13,6 +14,8 @@
 #include <utility>
 #include <vector>
 
+#include "vllm/v1/core/sched/utils.h"  // check_stop
+
 namespace vllm::v1 {
 
 namespace {
@@ -316,6 +319,133 @@ SchedulerOutput Scheduler::schedule() {
   return scheduler_output;
 }
 
+EngineCoreOutputs Scheduler::update_from_output(
+    const SchedulerOutput& scheduler_output,
+    const ModelRunnerOutput& model_runner_output) {
+  const std::vector<std::vector<int32_t>>& sampled_token_ids =
+      model_runner_output.sampled_token_ids;
+  const std::map<std::string, int>& num_scheduled_tokens =
+      scheduler_output.num_scheduled_tokens;
+
+  std::vector<EngineCoreOutput> outputs;
+  // Requests that stopped this step, split by the queue they must be removed
+  // from (upstream stopped_running_reqs / stopped_preempted_reqs). The KV blocks
+  // are freed and finished_req_ids updated inside the loop, but the owning
+  // requests-map erase is deferred until after these pointers are used to filter
+  // running/waiting (so the Request* stays valid — upstream relies on Python GC).
+  std::set<Request*> stopped_running_reqs;
+  std::set<Request*> stopped_preempted_reqs;
+  std::vector<std::string> finished_ids_to_erase;
+
+  // NOTE(woosuk): upstream iterates num_scheduled_tokens.items() (dict/schedule
+  // order); std::map iterates in sorted key order. The set of outputs is the
+  // same — only their order in the returned vector differs, which is benign
+  // (each EngineCoreOutput is keyed by request_id).
+  for (const auto& [req_id, num_tokens_scheduled] : num_scheduled_tokens) {
+    assert(num_tokens_scheduled > 0);
+    (void)num_tokens_scheduled;
+
+    auto it = requests.find(req_id);
+    if (it == requests.end() || it->second->IsFinished()) {
+      // Already finished — e.g. aborted while the model was executing it.
+      continue;
+    }
+    Request* request = it->second.get();
+
+    const int req_index = model_runner_output.req_id_to_index.at(req_id);
+    // sampled_token_ids[req_index] if sampled_token_ids else []. A request still
+    // being prefilled gets an empty list from the runner.
+    std::vector<int32_t> new_token_ids =
+        sampled_token_ids.empty()
+            ? std::vector<int32_t>{}
+            : sampled_token_ids[static_cast<std::size_t>(req_index)];
+
+    // DEFERRED: speculative-decode acceptance / num_computed rollback; encoder-
+    // input free.
+
+    bool stopped = false;
+    const RequestStatus status_before_stop = request->status;
+
+    // _update_request_with_output: append each generated token, run check_stop
+    // after each, and trim any tokens generated past the stop.
+    if (!new_token_ids.empty()) {
+      for (std::size_t num_new = 1; num_new <= new_token_ids.size(); ++num_new) {
+        request->AppendOutputToken(new_token_ids[num_new - 1]);
+        stopped = check_stop(*request, max_model_len);
+        if (stopped) {
+          new_token_ids.resize(num_new);  // del new_token_ids[num_new:]
+          break;
+        }
+      }
+    }
+    // DEFERRED: pooling stop, structured-output grammar accept.
+
+    std::optional<FinishReason> finish_reason;
+    if (stopped) {
+      // Capture the finish reason before freeing (upstream captures it before
+      // _handle_stopped_request, which may reset the status for resumable reqs —
+      // resumable/streaming is deferred, so _handle_stopped_request is always
+      // "finished" at T0).
+      finish_reason = request->GetFinishedReason();
+      // _free_request + _free_blocks (T0 subset): free the KV blocks and record
+      // the finished id now; defer the requests-map erase (see above).
+      kv_cache_manager->free(*request);
+      finished_req_ids.insert(request->request_id);
+      finished_ids_to_erase.push_back(request->request_id);
+      if (status_before_stop == RequestStatus::kRunning) {
+        stopped_running_reqs.insert(request);
+      } else {
+        stopped_preempted_reqs.insert(request);
+      }
+    }
+
+    // DEFERRED: sample logprobs / prompt logprobs / num_nans_in_logits.
+
+    // Emit an EngineCoreOutput only when the request produced tokens or finished
+    // (upstream's `if new_token_ids or ... or stopped`). A partial-prefill
+    // request that produced neither is skipped: "EngineCore returns no partial
+    // prefill outputs".
+    if (!new_token_ids.empty() || stopped) {
+      EngineCoreOutput out;
+      out.request_id = req_id;
+      out.new_token_ids = new_token_ids;
+      out.finish_reason = finish_reason;
+      // stop_reason is int|str|None upstream; our EngineCoreOutput carries an
+      // optional<string> (see engine/types.h). Only a stop_token_ids match sets
+      // request.stop_reason at T0 — stringify that token id; otherwise nullopt.
+      if (request->stop_reason.has_value()) {
+        out.stop_reason = std::to_string(*request->stop_reason);
+      }
+      outputs.push_back(std::move(out));
+    }
+  }
+
+  // Remove the stopped requests from the running list and the waiting queue.
+  if (!stopped_running_reqs.empty()) {
+    running.erase(
+        std::remove_if(running.begin(), running.end(),
+                       [&](Request* r) {
+                         return stopped_running_reqs.count(r) > 0;
+                       }),
+        running.end());
+  }
+  if (!stopped_preempted_reqs.empty()) {
+    // Rare (a stopped-while-preempted request); remove each from waiting.
+    std::vector<Request*> to_remove(stopped_preempted_reqs.begin(),
+                                    stopped_preempted_reqs.end());
+    waiting->remove_requests(to_remove);
+  }
+  // Now that no queue references them, drop the owning entries (destroys the
+  // finished Request objects — upstream _free_blocks' `del self.requests[...]`).
+  for (const std::string& id : finished_ids_to_erase) {
+    requests.erase(id);
+  }
+
+  EngineCoreOutputs engine_core_outputs;
+  engine_core_outputs.outputs = std::move(outputs);
+  return engine_core_outputs;
+}
+
 CachedRequestData Scheduler::make_cached_request_data(
     const std::vector<Request*>& running_reqs,
     const std::vector<Request*>& resumed_reqs,

src/vllm/v1/core/sched/utils.cpp

@@ -0,0 +1,63 @@
+// Ported from: vllm/v1/core/sched/utils.py @ e24d1b24
+// See include/vllm/v1/core/sched/utils.h for the T0 scope + the (verified) stop
+// precedence.
+#include "vllm/v1/core/sched/utils.h"
+
+#include <cassert>
+
+#include "vllm/sampling_params.h"
+
+namespace vllm::v1 {
+
+bool check_stop(Request& request, int max_model_len) {
+  // assert not request.pooling_params — pooling is deferred at T0, so there is
+  // nothing to assert against here.
+  const SamplingParams& sampling_params = request.sampling_params;
+
+  // (1) min_tokens gate — FIRST at this pin, so it gates the length cap too:
+  // do not stop before at least min_tokens have been generated.
+  if (request.NumOutputTokens() < sampling_params.min_tokens) {
+    return false;
+  }
+
+  const int last_token_id = request.output_token_ids.back();
+
+  // (2) EOS. Upstream compares against sampling_params.eos_token_id, which is a
+  // *property* returning None when ignore_eos is set (sampling_params.py only
+  // assigns _eos_token_id `if not self.ignore_eos`). Our SamplingParams stores
+  // the raw eos id plus a separate ignore_eos flag, so we replicate the
+  // property's ignore_eos gate here — behaviorally identical to the pin.
+  if (!sampling_params.ignore_eos && sampling_params.eos_token_id.has_value() &&
+      last_token_id == *sampling_params.eos_token_id) {
+    request.status = RequestStatus::kFinishedStopped;
+    return true;
+  }
+
+  // (3) stop_token_ids. On a match, carry the matched token id as stop_reason
+  // (upstream `request.stop_reason = last_token_id`).
+  for (const int32_t stop_token_id : sampling_params.stop_token_ids) {
+    if (last_token_id == stop_token_id) {
+      request.status = RequestStatus::kFinishedStopped;
+      request.stop_reason = last_token_id;
+      return true;
+    }
+  }
+
+  // (4) length cap. request.max_tokens == sampling_params.max_tokens upstream
+  // (Request.__init__ asserts it is not None for sampling requests), so assert
+  // it here too rather than inventing a fallback.
+  assert(sampling_params.max_tokens.has_value() &&
+         "sampling request must have max_tokens set (Request.max_tokens)");
+  if (request.NumTokens() >= max_model_len ||
+      request.NumOutputTokens() >= *sampling_params.max_tokens) {
+    request.status = RequestStatus::kFinishedLengthCapped;
+    return true;
+  }
+
+  // (5) repetition detection — DEFERRED (sampling_params.repetition_detection is
+  // a deferred field). Upstream would check check_sequence_repetition here.
+
+  return false;
+}
+
+}  // namespace vllm::v1