From 77212d931e345befd15bed9c40aa2df7d791f224 Mon Sep 17 00:00:00 2001
From: Yi <hyi@amd.com>
Date: Mon, 2 Nov 2026 23:05:00 +0800
Subject: [PATCH 01/11] perf(api-server): stop keeping finished sequences,
 render chat templates off the event loop

Three costs in the OpenAI server that grow with the length of agentic
conversations, found on DSV4.1-Flash DEP4 c512 (MI455X) where throughput
fell from ~12k to ~5k tok/s/GPU over a 1 h run:

- CoreManager's output thread put every finished Sequence list on
  outputs_queue, which only offline LLMEngine.step()/generate() read. Online,
  each finished request (token arrays, block tables) stayed reachable for the
  life of the process: API RSS 3 -> 26 GB in 1 h, gen-2 GC pauses
  0.13 -> 0.92 s. The server now clears CoreManager.consumes_outputs_queue.
- apply_chat_template ran on the event loop for /v1/chat/completions,
  /v1/messages and /v1/responses. It now runs on the default executor
  (context variables carried), so the loop keeps reading requests.
- DeepSeek's encoding.py calls find_last_user_index once per rendered
  message, each a backward scan: O(n^2) per render. The loader memoizes it
  per (messages list, length), per thread.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 atom/entrypoints/openai/api_server.py        | 52 +++++++++++++-------
 atom/entrypoints/openai/chat_encoders.py     | 31 ++++++++++++
 atom/model_engine/engine_core_mgr.py         |  8 ++-
 tests/entrypoints/test_api_server_helpers.py | 42 ++++++++++++++++
 tests/entrypoints/test_chat_encoders.py      | 43 ++++++++++++++++
 5 files changed, 157 insertions(+), 19 deletions(-)

diff --git a/atom/entrypoints/openai/api_server.py b/atom/entrypoints/openai/api_server.py
index 5c539f07c2..550eb20949 100644
--- a/atom/entrypoints/openai/api_server.py
+++ b/atom/entrypoints/openai/api_server.py
@@ -19,6 +19,8 @@
 
 import asyncio
 import contextlib
+import contextvars
+import functools
 import json
 import logging
 import os
@@ -1741,6 +1743,29 @@ def _effective_thinking_enabled(kwargs):
     return effective.get(name, on_value) != off_value
 
 
+async def _render_chat_template(messages, *, tools, **kwargs):
+    """``apply_chat_template`` on a worker thread.
+
+    Rendering grows with the conversation (an agentic turn resends the whole
+    history), and on the event loop it holds up reading every other request:
+    under long agentic load the loop fell behind and requests waited unread in
+    the socket while the engine ran short of work. Context variables (request
+    id, tracing) follow the call.
+    """
+    return await asyncio.get_running_loop().run_in_executor(
+        None,
+        contextvars.copy_context().run,
+        functools.partial(
+            apply_chat_template,
+            tokenizer,
+            custom_message_encoder,
+            messages,
+            tools=tools,
+            **kwargs,
+        ),
+    )
+
+
 @app.post("/v1/chat/completions")
 async def chat_completions(request: ChatCompletionRequest, raw_request: Request):
     """Handle chat completion requests (OpenAI-compatible)."""
@@ -1825,12 +1850,8 @@ async def chat_completions(request: ChatCompletionRequest, raw_request: Request)
             prompt_or_tokens = pretokenized
             kv_transfer_params = _engine_kv_transfer_params(kv_transfer_params)
         else:
-            prompt_or_tokens = apply_chat_template(
-                tokenizer,
-                custom_message_encoder,
-                messages,
-                tools=request.tools,
-                **merged_kwargs,
+            prompt_or_tokens = await _render_chat_template(
+                messages, tools=request.tools, **merged_kwargs
             )
 
         # The K3 template may open the reasoning channel in text or token IDs.
@@ -2222,12 +2243,8 @@ async def anthropic_messages(request: AnthropicMessagesRequest, raw_request: Req
         # no such spelling, and are not answered here or on the chat path.
         if forbids_tool_calls(request.tool_choice):
             merged_kwargs["tool_choice"] = "none"
-        prompt = apply_chat_template(
-            tokenizer,
-            custom_message_encoder,
-            messages,
-            tools=anthropic_to_openai_tools(request.tools),
-            **merged_kwargs,
+        prompt = await _render_chat_template(
+            messages, tools=anthropic_to_openai_tools(request.tools), **merged_kwargs
         )
 
         generation_config = engine.config.generation_config
@@ -2649,12 +2666,8 @@ async def responses_create(raw_request: Request):
             rules=rules,
         )
 
-        prompt = apply_chat_template(
-            tokenizer,
-            custom_message_encoder,
-            messages,
-            tools=openai_tools,
-            **merged_kwargs,
+        prompt = await _render_chat_template(
+            messages, tools=openai_tools, **merged_kwargs
         )
 
         generation_config = engine.config.generation_config
@@ -3204,6 +3217,9 @@ def main():
     required_tool_channel = required_call_channel(tool_call_parser_cls, tokenizer)
 
     engine = engine_args.create_engine(tokenizer=tokenizer)
+    # Every output reaches its request through the per-sequence callbacks;
+    # nothing here reads the offline outputs queue.
+    engine.core_mgr.consumes_outputs_queue = False
     # Forced acceptance emits draft tokens that nothing verified, and once the
     # context is long enough those degenerate into the dialect's own channel
     # framing and nothing else -- read as structure, correctly, that leaves a
diff --git a/atom/entrypoints/openai/chat_encoders.py b/atom/entrypoints/openai/chat_encoders.py
index a4dbb38bb9..eb2536750a 100644
--- a/atom/entrypoints/openai/chat_encoders.py
+++ b/atom/entrypoints/openai/chat_encoders.py
@@ -10,12 +10,14 @@
 when one was found, or to ``tokenizer.apply_chat_template`` otherwise.
 """
 
+import functools
 import glob
 import importlib.util
 import json
 import logging
 import os
 import pathlib
+import threading
 from typing import Any
 
 from jinja2 import TemplateError
@@ -31,6 +33,34 @@
 logger = logging.getLogger("atom")
 
 
+def _memoize_last_user_index(mod) -> None:
+    """Cache ``mod.find_last_user_index`` per messages list.
+
+    DeepSeek's ``encoding.py`` calls it from ``render_message`` once for every
+    message, and each call scans the list backwards past the assistant and tool
+    turns, so one render costs O(n^2) in the number of messages (~135 in a long
+    agentic conversation). The answer only changes with the list, so it is
+    computed once per (list, length) and reused, per thread since renders run
+    on a thread pool.
+    """
+    orig = getattr(mod, "find_last_user_index", None)
+    if not callable(orig) or getattr(orig, "_atom_memoized", False):
+        return
+    local = threading.local()
+
+    @functools.wraps(orig)
+    def find_last_user_index(messages):
+        cached = getattr(local, "cached", None)
+        if cached is not None and cached[0] is messages and cached[1] == len(messages):
+            return cached[2]
+        result = orig(messages)
+        local.cached = (messages, len(messages), result)
+        return result
+
+    find_last_user_index._atom_memoized = True
+    mod.find_last_user_index = find_last_user_index
+
+
 def _load_encoder_from_dir(model_path: str) -> MessageEncoderAdapter | None:
     """Look for ``<model>/encoding/encoding_*.py`` and load ``encode_messages``.
 
@@ -63,6 +93,7 @@ def _load_encoder_from_dir(model_path: str) -> MessageEncoderAdapter | None:
         mod = importlib.util.module_from_spec(spec)
         spec.loader.exec_module(mod)
         raw = mod.encode_messages
+        _memoize_last_user_index(mod)
     except Exception:
         # Broad on purpose: this executes a file the operator supplied, so the
         # failure can be anything that module raises, and a bad encoder must
diff --git a/atom/model_engine/engine_core_mgr.py b/atom/model_engine/engine_core_mgr.py
index c1340b47d7..1f461057e9 100644
--- a/atom/model_engine/engine_core_mgr.py
+++ b/atom/model_engine/engine_core_mgr.py
@@ -191,6 +191,11 @@ class CoreManager:
     # to 0 (backpressure disabled).
     _bp_depth = 0
     _bp_offline_batch = False
+    # Whether something drains ``outputs_queue`` (offline ``LLMEngine.step`` /
+    # ``generate`` and the asyncio path do). An online front end that delivers
+    # every output through the per-sequence callbacks clears it, or each
+    # finished Sequence stays on the queue for the life of the process.
+    consumes_outputs_queue = True
     _bp_dispatch_paused = False
     _input_send_lock = Lock()
 
@@ -833,7 +838,8 @@ def process_outputs_socket():
                                     f"{self.label}: flush_stream_batch failed: {e}",
                                     exc_info=True,
                                 )
-                        self.outputs_queue.put_nowait(seqs)
+                        if self.consumes_outputs_queue:
+                            self.outputs_queue.put_nowait(seqs)
                         # These completions freed slots on their ranks — re-feed
                         # from the pending pool so a fast rank never drains to
                         # empty while peers still have work. No-op when
diff --git a/tests/entrypoints/test_api_server_helpers.py b/tests/entrypoints/test_api_server_helpers.py
index a0d26e0b34..9589672bef 100644
--- a/tests/entrypoints/test_api_server_helpers.py
+++ b/tests/entrypoints/test_api_server_helpers.py
@@ -874,3 +874,45 @@ async def drive():
 
     asyncio.run(drive())
     assert events == ["initialize", "serving", "shutdown"]
+
+
+def test_chat_template_renders_off_the_event_loop(monkeypatch):
+    # Rendering grows with the conversation; on the event loop it holds up
+    # reading every other request. It runs on a worker thread and keeps the
+    # caller's context variables.
+    import contextvars
+
+    marker = contextvars.ContextVar("marker", default=None)
+    seen = {}
+
+    def fake_apply(tokenizer, encoder, messages, *, tools=None, **kwargs):
+        seen["thread"] = threading.get_ident()
+        seen["marker"] = marker.get()
+        seen["kwargs"] = kwargs
+        return "prompt"
+
+    monkeypatch.setattr(api_server, "apply_chat_template", fake_apply)
+
+    async def go():
+        marker.set("req-1")
+        loop_thread = threading.get_ident()
+        out = await api_server._render_chat_template(
+            [{"role": "user", "content": "hi"}], tools=None, thinking=True
+        )
+        return out, loop_thread
+
+    out, loop_thread = asyncio.run(go())
+    assert out == "prompt"
+    assert seen["thread"] != loop_thread
+    assert seen["marker"] == "req-1"
+    assert seen["kwargs"] == {"thinking": True}
+
+
+def test_online_server_does_not_keep_finished_sequences():
+    # Only offline callers read CoreManager.outputs_queue; the server clears
+    # the flag so finished Sequences are not kept for the process lifetime.
+    from atom.model_engine.engine_core_mgr import CoreManager
+
+    assert CoreManager.consumes_outputs_queue is True
+    src = inspect.getsource(api_server.main)
+    assert "consumes_outputs_queue = False" in src
diff --git a/tests/entrypoints/test_chat_encoders.py b/tests/entrypoints/test_chat_encoders.py
index 324dfade46..e038ea30d2 100644
--- a/tests/entrypoints/test_chat_encoders.py
+++ b/tests/entrypoints/test_chat_encoders.py
@@ -389,3 +389,46 @@ def test_the_probe_returns_none_rather_than_raising(self):
     )
     def test_neither_startup_probe_raises(self, probe):
         assert probe(self.NoTemplate()) is None
+
+
+_COUNTING_ENCODER = """
+CALLS = [0]
+
+
+def find_last_user_index(messages):
+    CALLS[0] += 1
+    for i in range(len(messages) - 1, -1, -1):
+        if messages[i]["role"] == "user":
+            return i
+    return -1
+
+
+def encode_messages(messages, **kwargs):
+    idx = [find_last_user_index(messages) for _ in messages]
+    return f"{idx[0]}x{len(idx)}|calls={CALLS[0]}"
+"""
+
+
+def test_loader_memoizes_find_last_user_index(tmp_path):
+    # DeepSeek's encoder asks for the last user index once per rendered
+    # message; the loader caches it per messages list so a render is O(n).
+    encoding_dir = tmp_path / "encoding"
+    encoding_dir.mkdir()
+    (encoding_dir / "encoding_dsv4.py").write_text(_COUNTING_ENCODER, encoding="utf-8")
+
+    adapter = _load_encoder_from_dir(str(tmp_path))
+    assert adapter is not None
+    messages = [{"role": "user", "content": "a"}] + [
+        {"role": "assistant", "content": "b"}
+    ] * 9
+
+    first = apply_chat_template(
+        tokenizer=None, custom_encoder=adapter, messages=messages
+    )
+    assert first == "0x10|calls=1"
+    # The next turn appends to the history: a different answer, computed again.
+    messages.append({"role": "user", "content": "c"})
+    second = apply_chat_template(
+        tokenizer=None, custom_encoder=adapter, messages=messages
+    )
+    assert second == "10x11|calls=2"

From bd367f7b3b0da80e894108d40cdfe6655b7830d3 Mon Sep 17 00:00:00 2001
From: Yi <hyi@amd.com>
Date: Tue, 3 Nov 2026 00:39:24 +0800
Subject: [PATCH 02/11] review: clear the outputs flag in atomesh too, test the
 queue behaviour, drop the encoder memo

- atomesh's standalone front end also delivers through callbacks only, so it
  clears consumes_outputs_queue like the OpenAI server.
- The output-thread test drives a real ADD frame and checks the queue with the
  flag on and off, instead of grepping the source.
- Measured on the checkpoints' own encoding.py (V4-Flash, V4.1-Flash): the
  find_last_user_index memo is 0.97-1.03x up to 302 messages, because
  merge_tool_messages folds tool results into user turns and the backward scan
  stays short. Dropped.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 atom/entrypoints/atomesh/server.py           |  3 ++
 atom/entrypoints/openai/chat_encoders.py     | 31 --------------
 tests/entrypoints/test_api_server_helpers.py | 12 +++---
 tests/entrypoints/test_chat_encoders.py      | 43 --------------------
 tests/test_engine_core_mgr_shared_state.py   | 28 +++++++++++++
 5 files changed, 38 insertions(+), 79 deletions(-)

diff --git a/atom/entrypoints/atomesh/server.py b/atom/entrypoints/atomesh/server.py
index 8e5731851f..cf5e98f586 100644
--- a/atom/entrypoints/atomesh/server.py
+++ b/atom/entrypoints/atomesh/server.py
@@ -102,6 +102,9 @@ def initialize_engine(args: argparse.Namespace) -> tuple[Any, Any]:
     logger.info("Initializing engine with model %s...", args.model)
     engine_args = EngineArgs.from_cli_args(args)
     engine = engine_args.create_engine(tokenizer=tokenizer)
+    # Outputs reach requests through the per-sequence callbacks; nothing here
+    # reads the offline outputs queue.
+    engine.core_mgr.consumes_outputs_queue = False
     return engine, tokenizer
 
 
diff --git a/atom/entrypoints/openai/chat_encoders.py b/atom/entrypoints/openai/chat_encoders.py
index eb2536750a..a4dbb38bb9 100644
--- a/atom/entrypoints/openai/chat_encoders.py
+++ b/atom/entrypoints/openai/chat_encoders.py
@@ -10,14 +10,12 @@
 when one was found, or to ``tokenizer.apply_chat_template`` otherwise.
 """
 
-import functools
 import glob
 import importlib.util
 import json
 import logging
 import os
 import pathlib
-import threading
 from typing import Any
 
 from jinja2 import TemplateError
@@ -33,34 +31,6 @@
 logger = logging.getLogger("atom")
 
 
-def _memoize_last_user_index(mod) -> None:
-    """Cache ``mod.find_last_user_index`` per messages list.
-
-    DeepSeek's ``encoding.py`` calls it from ``render_message`` once for every
-    message, and each call scans the list backwards past the assistant and tool
-    turns, so one render costs O(n^2) in the number of messages (~135 in a long
-    agentic conversation). The answer only changes with the list, so it is
-    computed once per (list, length) and reused, per thread since renders run
-    on a thread pool.
-    """
-    orig = getattr(mod, "find_last_user_index", None)
-    if not callable(orig) or getattr(orig, "_atom_memoized", False):
-        return
-    local = threading.local()
-
-    @functools.wraps(orig)
-    def find_last_user_index(messages):
-        cached = getattr(local, "cached", None)
-        if cached is not None and cached[0] is messages and cached[1] == len(messages):
-            return cached[2]
-        result = orig(messages)
-        local.cached = (messages, len(messages), result)
-        return result
-
-    find_last_user_index._atom_memoized = True
-    mod.find_last_user_index = find_last_user_index
-
-
 def _load_encoder_from_dir(model_path: str) -> MessageEncoderAdapter | None:
     """Look for ``<model>/encoding/encoding_*.py`` and load ``encode_messages``.
 
@@ -93,7 +63,6 @@ def _load_encoder_from_dir(model_path: str) -> MessageEncoderAdapter | None:
         mod = importlib.util.module_from_spec(spec)
         spec.loader.exec_module(mod)
         raw = mod.encode_messages
-        _memoize_last_user_index(mod)
     except Exception:
         # Broad on purpose: this executes a file the operator supplied, so the
         # failure can be anything that module raises, and a bad encoder must
diff --git a/tests/entrypoints/test_api_server_helpers.py b/tests/entrypoints/test_api_server_helpers.py
index 9589672bef..9e4ad978b5 100644
--- a/tests/entrypoints/test_api_server_helpers.py
+++ b/tests/entrypoints/test_api_server_helpers.py
@@ -908,11 +908,13 @@ async def go():
     assert seen["kwargs"] == {"thinking": True}
 
 
-def test_online_server_does_not_keep_finished_sequences():
-    # Only offline callers read CoreManager.outputs_queue; the server clears
-    # the flag so finished Sequences are not kept for the process lifetime.
+def test_online_servers_do_not_keep_finished_sequences():
+    # Only offline callers read CoreManager.outputs_queue; both online front
+    # ends clear the flag (the queue behaviour itself is tested in
+    # test_engine_core_mgr_shared_state.py).
+    from atom.entrypoints.atomesh import server as atomesh_server
     from atom.model_engine.engine_core_mgr import CoreManager
 
     assert CoreManager.consumes_outputs_queue is True
-    src = inspect.getsource(api_server.main)
-    assert "consumes_outputs_queue = False" in src
+    for fn in (api_server.main, atomesh_server.initialize_engine):
+        assert "consumes_outputs_queue = False" in inspect.getsource(fn)
diff --git a/tests/entrypoints/test_chat_encoders.py b/tests/entrypoints/test_chat_encoders.py
index e038ea30d2..324dfade46 100644
--- a/tests/entrypoints/test_chat_encoders.py
+++ b/tests/entrypoints/test_chat_encoders.py
@@ -389,46 +389,3 @@ def test_the_probe_returns_none_rather_than_raising(self):
     )
     def test_neither_startup_probe_raises(self, probe):
         assert probe(self.NoTemplate()) is None
-
-
-_COUNTING_ENCODER = """
-CALLS = [0]
-
-
-def find_last_user_index(messages):
-    CALLS[0] += 1
-    for i in range(len(messages) - 1, -1, -1):
-        if messages[i]["role"] == "user":
-            return i
-    return -1
-
-
-def encode_messages(messages, **kwargs):
-    idx = [find_last_user_index(messages) for _ in messages]
-    return f"{idx[0]}x{len(idx)}|calls={CALLS[0]}"
-"""
-
-
-def test_loader_memoizes_find_last_user_index(tmp_path):
-    # DeepSeek's encoder asks for the last user index once per rendered
-    # message; the loader caches it per messages list so a render is O(n).
-    encoding_dir = tmp_path / "encoding"
-    encoding_dir.mkdir()
-    (encoding_dir / "encoding_dsv4.py").write_text(_COUNTING_ENCODER, encoding="utf-8")
-
-    adapter = _load_encoder_from_dir(str(tmp_path))
-    assert adapter is not None
-    messages = [{"role": "user", "content": "a"}] + [
-        {"role": "assistant", "content": "b"}
-    ] * 9
-
-    first = apply_chat_template(
-        tokenizer=None, custom_encoder=adapter, messages=messages
-    )
-    assert first == "0x10|calls=1"
-    # The next turn appends to the history: a different answer, computed again.
-    messages.append({"role": "user", "content": "c"})
-    second = apply_chat_template(
-        tokenizer=None, custom_encoder=adapter, messages=messages
-    )
-    assert second == "10x11|calls=2"
diff --git a/tests/test_engine_core_mgr_shared_state.py b/tests/test_engine_core_mgr_shared_state.py
index 1ad39a9d21..f7a6260c0d 100644
--- a/tests/test_engine_core_mgr_shared_state.py
+++ b/tests/test_engine_core_mgr_shared_state.py
@@ -130,6 +130,34 @@ def test_shutdown_frame_marks_unhealthy_before_cleanup(self):
         finally:
             sender.close(linger=0)
 
+    def test_finished_sequences_reach_the_outputs_queue_only_if_consumed(self):
+        # Online front ends deliver through the callbacks and clear the flag;
+        # otherwise every finished Sequence would stay queued for the life of
+        # the process.
+        for consumes in (True, False):
+            mgr = _bare_manager(CoreManager)
+            self.addCleanup(mgr.ctx.term)
+            mgr.consumes_outputs_queue = consumes
+            mgr._deliver_terminal_output = lambda seq: False
+            mgr._release_seq_load = lambda seq_id: None
+            mgr._shutdown_engine_core_rank = lambda rank: None
+            url = f"inproc://outputs-{uuid.uuid4().hex}"
+            receiver = mgr.ctx.socket(zmq.PULL)
+            receiver.bind(url)
+            sender = mgr.ctx.socket(zmq.PUSH)
+            sender.connect(url)
+            thread = mgr._create_output_thread(0, receiver, url + "-shutdown")
+            thread.start()
+            try:
+                seqs = [SimpleNamespace(id=7)]
+                sender.send(pickle.dumps((EngineCoreRequestType.ADD, seqs)))
+                sender.send(pickle.dumps((EngineCoreRequestType.SHUTDOWN, None)))
+                thread.join(timeout=3)
+                self.assertFalse(thread.is_alive())
+                self.assertEqual(mgr.outputs_queue.qsize(), 1 if consumes else 0)
+            finally:
+                sender.close(linger=0)
+
     def _terminate(self):
         for mgr in self.managers:
             mgr.ctx.term()

From 30368986bc2b4f86285647417021bcb6cf5aeb12 Mon Sep 17 00:00:00 2001
From: Yi <hyi@amd.com>
Date: Tue, 3 Nov 2026 12:29:35 +0800
Subject: [PATCH 03/11] review: no encode on the loop for messages/responses,
 keep_finished_outputs at construction, asyncio.to_thread

- /v1/messages and /v1/responses render with asyncio.to_thread and pass
  max_prompt_tokens down; the length check and truncation run in the
  preprocess call on the executor (_preprocess_prompt), from the engine's
  own num_prompt_tokens. Input/output token counts come from the engine, so
  neither endpoint encodes on the event loop any more.
- Config.keep_finished_outputs (next to asyncio_mode), set at construction
  through EngineArgs.create_engine(keep_finished_outputs=False) by both online
  front ends; CoreManager reads it in _init_shared_state. The class-level flag
  and the source-text test are gone.
- _render_chat_template is gone: asyncio.to_thread at the call sites.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 atom/config.py                                |   6 +
 atom/entrypoints/atomesh/server.py            |   6 +-
 atom/entrypoints/openai/api_server.py         | 227 +++++++++---------
 atom/model_engine/arg_utils.py                |  17 +-
 atom/model_engine/engine_core_mgr.py          |  11 +-
 tests/entrypoints/test_api_compatibility.py   |   2 +
 tests/entrypoints/test_api_server_helpers.py  |  78 +++---
 .../entrypoints/test_pd_messages_responses.py |  29 ++-
 tests/test_engine_core_mgr_shared_state.py    |  41 +++-
 9 files changed, 230 insertions(+), 187 deletions(-)

diff --git a/atom/config.py b/atom/config.py
index 638bed2304..2c6634a3c8 100644
--- a/atom/config.py
+++ b/atom/config.py
@@ -2103,6 +2103,12 @@ class Config:
     compilation_config: CompilationConfig = field(default_factory=CompilationConfig)
     quant_config: QuantizationConfig = field(init=False)
     asyncio_mode: bool = False
+    # Whether CoreManager also queues finished sequences on ``outputs_queue``
+    # for ``LLMEngine.step`` / ``generate`` and the asyncio path to read. Online
+    # front ends deliver every output through per-sequence callbacks and build
+    # the engine with this off; left on there, every finished Sequence stays
+    # on a queue nothing reads, for the life of the process.
+    keep_finished_outputs: bool = True
     mark_trace: bool = False
     load_dummy: str | None = None
     enable_expert_parallel: bool = False
diff --git a/atom/entrypoints/atomesh/server.py b/atom/entrypoints/atomesh/server.py
index cf5e98f586..535dfe0a1a 100644
--- a/atom/entrypoints/atomesh/server.py
+++ b/atom/entrypoints/atomesh/server.py
@@ -101,10 +101,8 @@ def initialize_engine(args: argparse.Namespace) -> tuple[Any, Any]:
 
     logger.info("Initializing engine with model %s...", args.model)
     engine_args = EngineArgs.from_cli_args(args)
-    engine = engine_args.create_engine(tokenizer=tokenizer)
-    # Outputs reach requests through the per-sequence callbacks; nothing here
-    # reads the offline outputs queue.
-    engine.core_mgr.consumes_outputs_queue = False
+    # Outputs reach requests through the per-sequence callbacks.
+    engine = engine_args.create_engine(tokenizer=tokenizer, keep_finished_outputs=False)
     return engine, tokenizer
 
 
diff --git a/atom/entrypoints/openai/api_server.py b/atom/entrypoints/openai/api_server.py
index 550eb20949..d2b6d63dba 100644
--- a/atom/entrypoints/openai/api_server.py
+++ b/atom/entrypoints/openai/api_server.py
@@ -19,8 +19,6 @@
 
 import asyncio
 import contextlib
-import contextvars
-import functools
 import json
 import logging
 import os
@@ -801,8 +799,12 @@ async def generate_async(
     dp_session_id: str | None = None,
     dp_parent_session_id: str | None = None,
     return_token_ids: bool = False,
+    max_prompt_tokens: int | None = None,
 ) -> AsyncGenerator[dict[str, Any], None]:
-    """Generate non-streaming output, optionally echoing local prompt token IDs."""
+    """Generate non-streaming output, optionally echoing local prompt token IDs.
+
+    ``max_prompt_tokens`` truncates, in the preprocess call
+    (``_preprocess_prompt``)."""
     token_queue: asyncio.Queue = asyncio.Queue()
     loop = asyncio.get_running_loop()
 
@@ -841,14 +843,18 @@ def completion_callback(request_output: RequestOutput):
         )
 
     def do_preprocess():
-        return engine.io_processor.preprocess(
+        return _preprocess_prompt(
+            lambda prompt: engine.io_processor.preprocess(
+                prompt,
+                sampling_params,
+                stream_callback=completion_callback,
+                kv_transfer_params=kv_transfer_params,
+                data_parallel_rank=data_parallel_rank,
+                dp_session_id=dp_session_id,
+                dp_parent_session_id=dp_parent_session_id,
+            ),
             prompt_or_tokens,
-            sampling_params,
-            stream_callback=completion_callback,
-            kv_transfer_params=kv_transfer_params,
-            data_parallel_rank=data_parallel_rank,
-            dp_session_id=dp_session_id,
-            dp_parent_session_id=dp_parent_session_id,
+            max_prompt_tokens,
         )
 
     seq = await loop.run_in_executor(None, do_preprocess)
@@ -1245,6 +1251,34 @@ def _prompt_headroom(
     return headroom
 
 
+def _max_prompt_tokens(
+    max_tokens: int, kv_transfer_params: dict[str, Any] | None
+) -> int:
+    """Longest prompt ``/v1/messages`` and ``/v1/responses`` keep; a longer one
+    is cut to this many tokens rather than refused."""
+    max_ctx = int(_get_engine_max_model_len() or 30720)
+    return max_ctx - _prompt_headroom(max_tokens, max_ctx, kv_transfer_params)
+
+
+def _preprocess_prompt(preprocess, prompt_or_tokens, max_prompt_tokens):
+    """``preprocess(prompt)`` on the executor, cutting a prompt over
+    ``max_prompt_tokens`` to that many tokens.
+
+    The prompt is tokenized once, by ``preprocess``, and checked by its
+    ``num_prompt_tokens``; only an over-long one is built a second time, from
+    its own first ``max_prompt_tokens`` ids."""
+    seq = preprocess(prompt_or_tokens)
+    if max_prompt_tokens is None or seq.num_prompt_tokens <= max_prompt_tokens:
+        return seq
+    logger.warning(
+        "Prompt too long (%d > %d), truncating",
+        seq.num_prompt_tokens,
+        max_prompt_tokens,
+    )
+    engine.io_processor.requests.pop(seq.id, None)
+    return preprocess(list(seq.prompt_token_ids[:max_prompt_tokens]))
+
+
 def _with_kv_transfer_params(response: dict, final_output: dict) -> dict:
     """Return the prefill's KV location to the PD router, as chat does."""
     if "kv_transfer_output_meta_info" in final_output:
@@ -1261,9 +1295,13 @@ async def setup_streaming_request(
     data_parallel_rank: int | None = None,
     dp_session_id: str | None = None,
     dp_parent_session_id: str | None = None,
+    max_prompt_tokens: int | None = None,
 ) -> tuple[int, StreamOutputCollector, int]:
     """Set up a streaming request with the engine.
 
+    ``max_prompt_tokens`` truncates, in the preprocess call
+    (``_preprocess_prompt``).
+
     Returns ``(seq_id, stream_collector, num_prompt_tokens)``.
     ``num_prompt_tokens`` is the engine-computed prompt length so the stream
     response generator does not have to re-tokenize the prompt on the event
@@ -1287,15 +1325,19 @@ def stream_callback(request_output: RequestOutput) -> None:
     executor_loop = asyncio.get_event_loop()
 
     def do_preprocess():
-        seq = engine.io_processor.preprocess(
+        seq = _preprocess_prompt(
+            lambda prompt: engine.io_processor.preprocess(
+                prompt,
+                sampling_params,
+                stream_callback=stream_callback,
+                kv_transfer_params=kv_transfer_params,
+                multimodal_data=multimodal_data,
+                data_parallel_rank=data_parallel_rank,
+                dp_session_id=dp_session_id,
+                dp_parent_session_id=dp_parent_session_id,
+            ),
             prompt_or_tokens,
-            sampling_params,
-            stream_callback=stream_callback,
-            kv_transfer_params=kv_transfer_params,
-            multimodal_data=multimodal_data,
-            data_parallel_rank=data_parallel_rank,
-            dp_session_id=dp_session_id,
-            dp_parent_session_id=dp_parent_session_id,
+            max_prompt_tokens,
         )
         _seq_id_to_request_id[seq.id] = request_id
         return seq
@@ -1743,29 +1785,6 @@ def _effective_thinking_enabled(kwargs):
     return effective.get(name, on_value) != off_value
 
 
-async def _render_chat_template(messages, *, tools, **kwargs):
-    """``apply_chat_template`` on a worker thread.
-
-    Rendering grows with the conversation (an agentic turn resends the whole
-    history), and on the event loop it holds up reading every other request:
-    under long agentic load the loop fell behind and requests waited unread in
-    the socket while the engine ran short of work. Context variables (request
-    id, tracing) follow the call.
-    """
-    return await asyncio.get_running_loop().run_in_executor(
-        None,
-        contextvars.copy_context().run,
-        functools.partial(
-            apply_chat_template,
-            tokenizer,
-            custom_message_encoder,
-            messages,
-            tools=tools,
-            **kwargs,
-        ),
-    )
-
-
 @app.post("/v1/chat/completions")
 async def chat_completions(request: ChatCompletionRequest, raw_request: Request):
     """Handle chat completion requests (OpenAI-compatible)."""
@@ -1850,8 +1869,15 @@ async def chat_completions(request: ChatCompletionRequest, raw_request: Request)
             prompt_or_tokens = pretokenized
             kv_transfer_params = _engine_kv_transfer_params(kv_transfer_params)
         else:
-            prompt_or_tokens = await _render_chat_template(
-                messages, tools=request.tools, **merged_kwargs
+            # Off the loop: rendering grows with the conversation, and on the
+            # loop it holds up reading every other request.
+            prompt_or_tokens = await asyncio.to_thread(
+                apply_chat_template,
+                tokenizer,
+                custom_message_encoder,
+                messages,
+                tools=request.tools,
+                **merged_kwargs,
             )
 
         # The K3 template may open the reasoning channel in text or token IDs.
@@ -2243,8 +2269,15 @@ async def anthropic_messages(request: AnthropicMessagesRequest, raw_request: Req
         # no such spelling, and are not answered here or on the chat path.
         if forbids_tool_calls(request.tool_choice):
             merged_kwargs["tool_choice"] = "none"
-        prompt = await _render_chat_template(
-            messages, tools=anthropic_to_openai_tools(request.tools), **merged_kwargs
+        # Off the loop, like the length check and truncation below, which
+        # happen in the preprocess call (`max_prompt_tokens`).
+        prompt = await asyncio.to_thread(
+            apply_chat_template,
+            tokenizer,
+            custom_message_encoder,
+            messages,
+            tools=anthropic_to_openai_tools(request.tools),
+            **merged_kwargs,
         )
 
         generation_config = engine.config.generation_config
@@ -2274,45 +2307,16 @@ async def anthropic_messages(request: AnthropicMessagesRequest, raw_request: Req
         request_id = uuid.uuid4().hex[:24]
         set_request_id(request_id)
         kv_transfer_params = _engine_kv_transfer_params(request.kv_transfer_params)
-        input_tokens = len(tokenizer.encode(prompt))
-
-        max_ctx = None
-        for _path in (
-            lambda: engine.config.max_model_len,
-            lambda: engine.model_config.max_model_len,
-            lambda: engine.scheduler.max_model_len,
-            lambda: engine.max_model_len,
-        ):
-            try:
-                _v = _path()
-                if _v:
-                    max_ctx = int(_v)
-                    break
-            except Exception:
-                logger.debug("max_ctx candidate rejected", exc_info=True)
-                continue
-        if not max_ctx:
-            max_ctx = 30720
-        logger.warning(f"[anthropic] resolved max_ctx={max_ctx}")
-        headroom = _prompt_headroom(request.max_tokens, max_ctx, kv_transfer_params)
-        max_input = max_ctx - headroom
-        if input_tokens > max_input:
-            logger.warning(
-                f"Prompt too long ({input_tokens} > {max_input}), truncating"
-            )
-            token_ids = tokenizer.encode(prompt)[:max_input]
-            prompt = tokenizer.decode(token_ids, skip_special_tokens=False)
-            input_tokens = max_input
+        max_input = _max_prompt_tokens(request.max_tokens, kv_transfer_params)
 
         if request.stream:
             # Streaming response
-            seq_id, stream_collector, _num_prompt_tokens = (
-                await setup_streaming_request(
-                    prompt,
-                    sampling_params,
-                    request_id,
-                    kv_transfer_params=kv_transfer_params,
-                )
+            seq_id, stream_collector, input_tokens = await setup_streaming_request(
+                prompt,
+                sampling_params,
+                request_id,
+                kv_transfer_params=kv_transfer_params,
+                max_prompt_tokens=max_input,
             )
 
             async def generate_anthropic_stream():
@@ -2488,12 +2492,14 @@ async def generate_anthropic_stream():
                 sampling_params,
                 request_id,
                 kv_transfer_params=kv_transfer_params,
+                max_prompt_tokens=max_input,
             ),
             raw_request,
             request_id,
         )
         if final_output is None:
             raise RuntimeError("No output generated")
+        input_tokens = final_output["num_tokens_input"]
 
         raw_text = final_output["text"]
         # Separating is unconditional -- the same call the chat path makes,
@@ -2517,7 +2523,7 @@ async def generate_anthropic_stream():
         if drop_reasoning:
             events = [e for e in events if e[0] != "reasoning"]
         _, tool_calls = flatten_tool_events(events)
-        output_tokens = len(tokenizer.encode(raw_text))
+        output_tokens = final_output["num_tokens_output"]
         cache_read_input_tokens = final_output.get("num_cached_tokens", 0)
         response = build_anthropic_response(
             request_id=request_id,
@@ -2666,8 +2672,15 @@ async def responses_create(raw_request: Request):
             rules=rules,
         )
 
-        prompt = await _render_chat_template(
-            messages, tools=openai_tools, **merged_kwargs
+        # Off the loop, like the length check and truncation below, which
+        # happen in the preprocess call (`max_prompt_tokens`).
+        prompt = await asyncio.to_thread(
+            apply_chat_template,
+            tokenizer,
+            custom_message_encoder,
+            messages,
+            tools=openai_tools,
+            **merged_kwargs,
         )
 
         generation_config = engine.config.generation_config
@@ -2701,38 +2714,16 @@ async def responses_create(raw_request: Request):
 
         request_id = f"resp_{uuid.uuid4().hex[:24]}"
         set_request_id(request_id)
-        input_tokens = len(tokenizer.encode(prompt))
-
-        max_ctx = 30720
-        for obj in (
-            getattr(engine, "config", None),
-            getattr(engine, "model_config", None),
-            getattr(engine, "scheduler", None),
-            engine,
-        ):
-            value = getattr(obj, "max_model_len", None) if obj is not None else None
-            if value:
-                max_ctx = int(value)
-                break
-        headroom = _prompt_headroom(max_tokens, max_ctx, kv_transfer_params)
-        max_input = max_ctx - headroom
-        if input_tokens > max_input:
-            logger.warning(
-                f"Prompt too long ({input_tokens} > {max_input}), truncating"
-            )
-            token_ids = tokenizer.encode(prompt)[:max_input]
-            prompt = tokenizer.decode(token_ids, skip_special_tokens=False)
-            input_tokens = max_input
+        max_input = _max_prompt_tokens(max_tokens, kv_transfer_params)
 
         created_at = int(time.time())
         if request.stream:
-            seq_id, stream_collector, _num_prompt_tokens = (
-                await setup_streaming_request(
-                    prompt,
-                    sampling_params,
-                    request_id,
-                    kv_transfer_params=kv_transfer_params,
-                )
+            seq_id, stream_collector, input_tokens = await setup_streaming_request(
+                prompt,
+                sampling_params,
+                request_id,
+                kv_transfer_params=kv_transfer_params,
+                max_prompt_tokens=max_input,
             )
 
             async def generate_responses_stream():
@@ -2831,12 +2822,14 @@ async def generate_responses_stream():
                 sampling_params,
                 request_id,
                 kv_transfer_params=kv_transfer_params,
+                max_prompt_tokens=max_input,
             ),
             raw_request,
             request_id,
         )
         if final_output is None:
             raise RuntimeError("No output generated")
+        input_tokens = final_output["num_tokens_input"]
 
         raw_text = final_output["text"]
         events = read_whole_blocks(
@@ -2849,7 +2842,7 @@ async def generate_responses_stream():
             openai_tools,
             suppress_calls=forbids_tool_calls(request.tool_choice),
         )
-        output_tokens = len(tokenizer.encode(raw_text))
+        output_tokens = final_output["num_tokens_output"]
         cache_read_input_tokens = final_output.get("num_cached_tokens", 0)
         response = build_responses_response(
             request_id=request_id,
@@ -3216,10 +3209,8 @@ def main():
     )
     required_tool_channel = required_call_channel(tool_call_parser_cls, tokenizer)
 
-    engine = engine_args.create_engine(tokenizer=tokenizer)
-    # Every output reaches its request through the per-sequence callbacks;
-    # nothing here reads the offline outputs queue.
-    engine.core_mgr.consumes_outputs_queue = False
+    # Every output reaches its request through the per-sequence callbacks.
+    engine = engine_args.create_engine(tokenizer=tokenizer, keep_finished_outputs=False)
     # Forced acceptance emits draft tokens that nothing verified, and once the
     # context is long enough those degenerate into the dialect's own channel
     # framing and nothing else -- read as structure, correctly, that leaves a
diff --git a/atom/model_engine/arg_utils.py b/atom/model_engine/arg_utils.py
index 092be91e19..5e942a3819 100644
--- a/atom/model_engine/arg_utils.py
+++ b/atom/model_engine/arg_utils.py
@@ -945,6 +945,17 @@ def _get_engine_kwargs(self) -> dict:
 
         return kwargs
 
-    def create_engine(self, tokenizer=None) -> LLMEngine:
-        """Create and return an LLMEngine instance with the configured parameters."""
-        return LLMEngine(self.model, tokenizer=tokenizer, **self._get_engine_kwargs())
+    def create_engine(
+        self, tokenizer=None, *, keep_finished_outputs: bool = True
+    ) -> LLMEngine:
+        """Create and return an LLMEngine instance with the configured parameters.
+
+        ``keep_finished_outputs=False`` is for online front ends, which take
+        every output through per-sequence callbacks (``Config.keep_finished_outputs``).
+        """
+        return LLMEngine(
+            self.model,
+            tokenizer=tokenizer,
+            keep_finished_outputs=keep_finished_outputs,
+            **self._get_engine_kwargs(),
+        )
diff --git a/atom/model_engine/engine_core_mgr.py b/atom/model_engine/engine_core_mgr.py
index 1f461057e9..47fe2c17be 100644
--- a/atom/model_engine/engine_core_mgr.py
+++ b/atom/model_engine/engine_core_mgr.py
@@ -191,11 +191,6 @@ class CoreManager:
     # to 0 (backpressure disabled).
     _bp_depth = 0
     _bp_offline_batch = False
-    # Whether something drains ``outputs_queue`` (offline ``LLMEngine.step`` /
-    # ``generate`` and the asyncio path do). An online front end that delivers
-    # every output through the per-sequence callbacks clears it, or each
-    # finished Sequence stays on the queue for the life of the process.
-    consumes_outputs_queue = True
     _bp_dispatch_paused = False
     _input_send_lock = Lock()
 
@@ -236,6 +231,10 @@ def _init_shared_state(
         )
         self.ctx = zmq.Context(io_threads=2)
         self.outputs_queue = queue.Queue[list[Sequence]]()
+        # Off for online front ends, which read every output through the
+        # per-sequence callbacks: nothing would drain the queue
+        # (`Config.keep_finished_outputs`).
+        self.keep_finished_outputs = getattr(config, "keep_finished_outputs", True)
         # Utility replies are matched to their caller by request id, not by
         # position on a shared queue, where a late or unasked-for reply became
         # the next caller's answer.
@@ -838,7 +837,7 @@ def process_outputs_socket():
                                     f"{self.label}: flush_stream_batch failed: {e}",
                                     exc_info=True,
                                 )
-                        if self.consumes_outputs_queue:
+                        if self.keep_finished_outputs:
                             self.outputs_queue.put_nowait(seqs)
                         # These completions freed slots on their ranks — re-feed
                         # from the pending pool so a fast rank never drains to
diff --git a/tests/entrypoints/test_api_compatibility.py b/tests/entrypoints/test_api_compatibility.py
index 6bf3595d11..eba8323960 100644
--- a/tests/entrypoints/test_api_compatibility.py
+++ b/tests/entrypoints/test_api_compatibility.py
@@ -1014,6 +1014,8 @@ async def generate(*args, **kwargs):
             "text": "partial" if partial else "",
             "finish_reason": reason,
             "num_cached_tokens": 0,
+            "num_tokens_input": 2,
+            "num_tokens_output": 1 if partial else 0,
         }
 
     monkeypatch.setattr(api_server, "generate_async", generate)
diff --git a/tests/entrypoints/test_api_server_helpers.py b/tests/entrypoints/test_api_server_helpers.py
index 9e4ad978b5..e87908693d 100644
--- a/tests/entrypoints/test_api_server_helpers.py
+++ b/tests/entrypoints/test_api_server_helpers.py
@@ -75,7 +75,7 @@ def add_cli_args(cls, parser):
         def from_cli_args(cls, args):
             return cls()
 
-        def create_engine(self, tokenizer=None):
+        def create_engine(self, tokenizer=None, **kwargs):
             return None
 
     _try_import_else_stub(
@@ -876,45 +876,41 @@ async def drive():
     assert events == ["initialize", "serving", "shutdown"]
 
 
-def test_chat_template_renders_off_the_event_loop(monkeypatch):
-    # Rendering grows with the conversation; on the event loop it holds up
-    # reading every other request. It runs on a worker thread and keeps the
-    # caller's context variables.
-    import contextvars
+class _FakeSeq:
+    def __init__(self, seq_id, token_ids):
+        self.id = seq_id
+        self.prompt_token_ids = list(token_ids)
+        self.num_prompt_tokens = len(token_ids)
 
-    marker = contextvars.ContextVar("marker", default=None)
-    seen = {}
 
-    def fake_apply(tokenizer, encoder, messages, *, tools=None, **kwargs):
-        seen["thread"] = threading.get_ident()
-        seen["marker"] = marker.get()
-        seen["kwargs"] = kwargs
-        return "prompt"
-
-    monkeypatch.setattr(api_server, "apply_chat_template", fake_apply)
-
-    async def go():
-        marker.set("req-1")
-        loop_thread = threading.get_ident()
-        out = await api_server._render_chat_template(
-            [{"role": "user", "content": "hi"}], tools=None, thinking=True
-        )
-        return out, loop_thread
-
-    out, loop_thread = asyncio.run(go())
-    assert out == "prompt"
-    assert seen["thread"] != loop_thread
-    assert seen["marker"] == "req-1"
-    assert seen["kwargs"] == {"thinking": True}
-
-
-def test_online_servers_do_not_keep_finished_sequences():
-    # Only offline callers read CoreManager.outputs_queue; both online front
-    # ends clear the flag (the queue behaviour itself is tested in
-    # test_engine_core_mgr_shared_state.py).
-    from atom.entrypoints.atomesh import server as atomesh_server
-    from atom.model_engine.engine_core_mgr import CoreManager
-
-    assert CoreManager.consumes_outputs_queue is True
-    for fn in (api_server.main, atomesh_server.initialize_engine):
-        assert "consumes_outputs_queue = False" in inspect.getsource(fn)
+def test_prompt_is_checked_and_truncated_in_the_preprocess_call(monkeypatch):
+    # /v1/messages and /v1/responses: no encode on the event loop. The length
+    # check reads the engine's own token count from the one preprocess call,
+    # and only an over-long prompt is built again, from its first ids.
+    requests = {}
+    monkeypatch.setattr(
+        api_server,
+        "engine",
+        SimpleNamespace(io_processor=SimpleNamespace(requests=requests)),
+    )
+    built = []
+
+    def preprocess(prompt):
+        tokens = list(range(10)) if isinstance(prompt, str) else prompt
+        seq = _FakeSeq(len(built), tokens)
+        built.append(prompt)
+        requests[seq.id] = seq
+        return seq
+
+    seq = api_server._preprocess_prompt(preprocess, "short", 16)
+    assert built == ["short"] and seq.num_prompt_tokens == 10
+    seq = api_server._preprocess_prompt(preprocess, "short", None)
+    assert seq.num_prompt_tokens == 10
+
+    built.clear()
+    requests.clear()
+    seq = api_server._preprocess_prompt(preprocess, "long prompt", 4)
+    assert built == ["long prompt", [0, 1, 2, 3]]
+    assert seq.prompt_token_ids == [0, 1, 2, 3]
+    # The discarded first build is not left registered.
+    assert list(requests) == [seq.id]
diff --git a/tests/entrypoints/test_pd_messages_responses.py b/tests/entrypoints/test_pd_messages_responses.py
index b77c8ac683..0f2afc4e36 100644
--- a/tests/entrypoints/test_pd_messages_responses.py
+++ b/tests/entrypoints/test_pd_messages_responses.py
@@ -49,19 +49,34 @@ def server(monkeypatch):
     """Stub the engine and record what each endpoint hands to it."""
     calls = SimpleNamespace(generate=[], setup=[], template="<rendered prompt>")
 
+    def encode(text, **_kw):
+        return [0] * (LONG_PROMPT_TOKENS if text == "<long prompt>" else 2)
+
+    def engine_prompt(prompt, kwargs):
+        # What preprocess builds: the prompt, cut to `max_prompt_tokens`
+        # (shown as `<n>` for an n-token cut) when it is longer.
+        limit = kwargs.get("max_prompt_tokens")
+        n = len(encode(prompt))
+        return (f"<{limit}>", limit) if limit is not None and n > limit else (prompt, n)
+
     async def fake_generate_async(prompt, _sampling, _request_id, **kwargs):
-        calls.generate.append((prompt, kwargs))
-        output = {"text": "hello", "finish_reason": "eos", "num_cached_tokens": 0}
+        sent, n = engine_prompt(prompt, kwargs)
+        calls.generate.append((sent, kwargs))
+        output = {
+            "text": "hello",
+            "finish_reason": "eos",
+            "num_cached_tokens": 0,
+            "num_tokens_input": n,
+            "num_tokens_output": 1,
+        }
         if kwargs.get("kv_transfer_params") is not None:
             output["kv_transfer_output_meta_info"] = PREFILL_OUTPUT
         yield output
 
     async def fake_setup_streaming_request(prompt, _sampling, _request_id, **kwargs):
-        calls.setup.append((prompt, kwargs))
-        return 1, _FinishedCollector(), 2
-
-    def encode(text, **_kw):
-        return [0] * (LONG_PROMPT_TOKENS if text == "<long prompt>" else 2)
+        sent, n = engine_prompt(prompt, kwargs)
+        calls.setup.append((sent, kwargs))
+        return 1, _FinishedCollector(), n
 
     monkeypatch.setattr(api_server, "model_name", "m")
     monkeypatch.setattr(api_server, "_request_logger", None)
diff --git a/tests/test_engine_core_mgr_shared_state.py b/tests/test_engine_core_mgr_shared_state.py
index f7a6260c0d..363114ab4f 100644
--- a/tests/test_engine_core_mgr_shared_state.py
+++ b/tests/test_engine_core_mgr_shared_state.py
@@ -76,6 +76,7 @@ def _bare_manager(cls, **kwargs):
             dp_load_balance="round_robin",
             tp_world_size=1,
             prefill_context_parallel_size=1,
+            keep_finished_outputs=kwargs.get("keep_finished_outputs", True),
         ),
         label=cls.__name__,
         local_engine_count=kwargs.get("local_engine_count", 2),
@@ -130,14 +131,13 @@ def test_shutdown_frame_marks_unhealthy_before_cleanup(self):
         finally:
             sender.close(linger=0)
 
-    def test_finished_sequences_reach_the_outputs_queue_only_if_consumed(self):
-        # Online front ends deliver through the callbacks and clear the flag;
-        # otherwise every finished Sequence would stay queued for the life of
-        # the process.
-        for consumes in (True, False):
-            mgr = _bare_manager(CoreManager)
+    def test_finished_sequences_are_queued_only_if_the_config_keeps_them(self):
+        # Online front ends build the engine with keep_finished_outputs=False
+        # and take outputs through the callbacks; otherwise every finished
+        # Sequence would stay queued for the life of the process.
+        for keep in (True, False):
+            mgr = _bare_manager(CoreManager, keep_finished_outputs=keep)
             self.addCleanup(mgr.ctx.term)
-            mgr.consumes_outputs_queue = consumes
             mgr._deliver_terminal_output = lambda seq: False
             mgr._release_seq_load = lambda seq_id: None
             mgr._shutdown_engine_core_rank = lambda rank: None
@@ -154,7 +154,7 @@ def test_finished_sequences_reach_the_outputs_queue_only_if_consumed(self):
                 sender.send(pickle.dumps((EngineCoreRequestType.SHUTDOWN, None)))
                 thread.join(timeout=3)
                 self.assertFalse(thread.is_alive())
-                self.assertEqual(mgr.outputs_queue.qsize(), 1 if consumes else 0)
+                self.assertEqual(mgr.outputs_queue.qsize(), 1 if keep else 0)
             finally:
                 sender.close(linger=0)
 
@@ -217,3 +217,28 @@ def test_subclass_does_not_override_the_shared_initialiser(self):
 
 if __name__ == "__main__":
     unittest.main()
+
+
+def test_online_front_ends_build_the_engine_without_kept_outputs(monkeypatch):
+    # Decided at construction: create_engine hands the choice to Config, and
+    # the default keeps the offline / asyncio behaviour.
+    import dataclasses
+
+    from atom.config import Config
+    from atom.model_engine import arg_utils
+
+    default = {f.name: f.default for f in dataclasses.fields(Config)}
+    assert default["keep_finished_outputs"] is True
+    seen = {}
+
+    def fake_engine(model, tokenizer=None, **kwargs):
+        seen.update(kwargs)
+
+    monkeypatch.setattr(arg_utils, "LLMEngine", fake_engine)
+    monkeypatch.setattr(arg_utils.EngineArgs, "_get_engine_kwargs", lambda self: {})
+    args = arg_utils.EngineArgs.__new__(arg_utils.EngineArgs)
+    args.model = "m"
+    args.create_engine(keep_finished_outputs=False)
+    assert seen == {"keep_finished_outputs": False}
+    args.create_engine()
+    assert seen == {"keep_finished_outputs": True}

From a7b1b7c1b1aa8b978a23af832a29a6c7b6e04cc1 Mon Sep 17 00:00:00 2001
From: Yi <hyi@amd.com>
Date: Tue, 3 Nov 2026 12:30:17 +0800
Subject: [PATCH 04/11] docs(core-mgr): why the stalled-run error goes to
 outputs_queue

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 atom/model_engine/engine_core_mgr.py | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/atom/model_engine/engine_core_mgr.py b/atom/model_engine/engine_core_mgr.py
index 47fe2c17be..3c473fe608 100644
--- a/atom/model_engine/engine_core_mgr.py
+++ b/atom/model_engine/engine_core_mgr.py
@@ -1278,6 +1278,11 @@ def _bp_fail_stalled_run(self) -> None:
         ``LLMEngine.generate`` waits while the engine is alive and any request
         is unfinished. A dropped seq never finishes, so that wait never ends.
         The seqs are still in ``_bp_pending``; this only surfaces the error.
+
+        Only offline seqs are ever in the pool (``add_request`` keeps online
+        requests out of it), and offline ``generate`` reads ``outputs_queue``.
+        If online seqs could wait there, this error would have to reach them
+        through their callbacks instead.
         """
         pending = len(getattr(self, "_bp_pending", ()))
         logger.error(

From b7387b91e8bf1db836d5e3a6b43a1a3771f7ff03 Mon Sep 17 00:00:00 2001
From: Yi <hyi@amd.com>
Date: Tue, 3 Nov 2026 13:06:10 +0800
Subject: [PATCH 05/11] review: seed the reasoning channel from the text the
 engine runs; pin the front end

- _PromptCap carries the cap into the preprocess call and, when the prompt is
  cut, the decoded text of what the engine runs (decoded on the executor);
  /v1/messages and /v1/responses seed the reasoning channel from it.
- Tests: create_engine's test renamed to what it checks; a new test drives
  atomesh's initialize_engine and asserts keep_finished_outputs=False.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 atom/entrypoints/openai/api_server.py         | 75 ++++++++++++-------
 tests/entrypoints/test_api_server_helpers.py  | 14 +++-
 .../entrypoints/test_pd_messages_responses.py |  5 +-
 tests/test_engine_core_mgr_shared_state.py    | 28 ++++++-
 4 files changed, 89 insertions(+), 33 deletions(-)

diff --git a/atom/entrypoints/openai/api_server.py b/atom/entrypoints/openai/api_server.py
index d2b6d63dba..171b0081b7 100644
--- a/atom/entrypoints/openai/api_server.py
+++ b/atom/entrypoints/openai/api_server.py
@@ -799,12 +799,11 @@ async def generate_async(
     dp_session_id: str | None = None,
     dp_parent_session_id: str | None = None,
     return_token_ids: bool = False,
-    max_prompt_tokens: int | None = None,
+    prompt_cap: "_PromptCap | None" = None,
 ) -> AsyncGenerator[dict[str, Any], None]:
     """Generate non-streaming output, optionally echoing local prompt token IDs.
 
-    ``max_prompt_tokens`` truncates, in the preprocess call
-    (``_preprocess_prompt``)."""
+    ``prompt_cap`` truncates, in the preprocess call (``_preprocess_prompt``)."""
     token_queue: asyncio.Queue = asyncio.Queue()
     loop = asyncio.get_running_loop()
 
@@ -854,7 +853,7 @@ def do_preprocess():
                 dp_parent_session_id=dp_parent_session_id,
             ),
             prompt_or_tokens,
-            max_prompt_tokens,
+            prompt_cap,
         )
 
     seq = await loop.run_in_executor(None, do_preprocess)
@@ -1251,32 +1250,47 @@ def _prompt_headroom(
     return headroom
 
 
-def _max_prompt_tokens(
+class _PromptCap:
+    """The longest prompt ``/v1/messages`` and ``/v1/responses`` keep: a longer
+    one is cut to ``tokens`` rather than refused. After the preprocess call,
+    ``cut_text`` is the text the engine runs if the prompt was cut, for the
+    readers that seed from the prompt's end (the reasoning channel)."""
+
+    __slots__ = ("tokens", "cut_text")
+
+    def __init__(self, tokens: int):
+        self.tokens = tokens
+        self.cut_text: str | None = None
+
+
+def _prompt_cap(
     max_tokens: int, kv_transfer_params: dict[str, Any] | None
-) -> int:
-    """Longest prompt ``/v1/messages`` and ``/v1/responses`` keep; a longer one
-    is cut to this many tokens rather than refused."""
+) -> _PromptCap:
     max_ctx = int(_get_engine_max_model_len() or 30720)
-    return max_ctx - _prompt_headroom(max_tokens, max_ctx, kv_transfer_params)
+    return _PromptCap(
+        max_ctx - _prompt_headroom(max_tokens, max_ctx, kv_transfer_params)
+    )
 
 
-def _preprocess_prompt(preprocess, prompt_or_tokens, max_prompt_tokens):
-    """``preprocess(prompt)`` on the executor, cutting a prompt over
-    ``max_prompt_tokens`` to that many tokens.
+def _preprocess_prompt(preprocess, prompt_or_tokens, prompt_cap):
+    """``preprocess(prompt)`` on the executor, cutting a prompt longer than
+    ``prompt_cap.tokens``.
 
     The prompt is tokenized once, by ``preprocess``, and checked by its
     ``num_prompt_tokens``; only an over-long one is built a second time, from
-    its own first ``max_prompt_tokens`` ids."""
+    its own first ids."""
     seq = preprocess(prompt_or_tokens)
-    if max_prompt_tokens is None or seq.num_prompt_tokens <= max_prompt_tokens:
+    if prompt_cap is None or seq.num_prompt_tokens <= prompt_cap.tokens:
         return seq
     logger.warning(
         "Prompt too long (%d > %d), truncating",
         seq.num_prompt_tokens,
-        max_prompt_tokens,
+        prompt_cap.tokens,
     )
     engine.io_processor.requests.pop(seq.id, None)
-    return preprocess(list(seq.prompt_token_ids[:max_prompt_tokens]))
+    token_ids = list(seq.prompt_token_ids[: prompt_cap.tokens])
+    prompt_cap.cut_text = tokenizer.decode(token_ids, skip_special_tokens=False)
+    return preprocess(token_ids)
 
 
 def _with_kv_transfer_params(response: dict, final_output: dict) -> dict:
@@ -1295,12 +1309,11 @@ async def setup_streaming_request(
     data_parallel_rank: int | None = None,
     dp_session_id: str | None = None,
     dp_parent_session_id: str | None = None,
-    max_prompt_tokens: int | None = None,
+    prompt_cap: "_PromptCap | None" = None,
 ) -> tuple[int, StreamOutputCollector, int]:
     """Set up a streaming request with the engine.
 
-    ``max_prompt_tokens`` truncates, in the preprocess call
-    (``_preprocess_prompt``).
+    ``prompt_cap`` truncates, in the preprocess call (``_preprocess_prompt``).
 
     Returns ``(seq_id, stream_collector, num_prompt_tokens)``.
     ``num_prompt_tokens`` is the engine-computed prompt length so the stream
@@ -1337,7 +1350,7 @@ def do_preprocess():
                 dp_parent_session_id=dp_parent_session_id,
             ),
             prompt_or_tokens,
-            max_prompt_tokens,
+            prompt_cap,
         )
         _seq_id_to_request_id[seq.id] = request_id
         return seq
@@ -2270,7 +2283,7 @@ async def anthropic_messages(request: AnthropicMessagesRequest, raw_request: Req
         if forbids_tool_calls(request.tool_choice):
             merged_kwargs["tool_choice"] = "none"
         # Off the loop, like the length check and truncation below, which
-        # happen in the preprocess call (`max_prompt_tokens`).
+        # happen in the preprocess call (`prompt_cap`).
         prompt = await asyncio.to_thread(
             apply_chat_template,
             tokenizer,
@@ -2307,7 +2320,7 @@ async def anthropic_messages(request: AnthropicMessagesRequest, raw_request: Req
         request_id = uuid.uuid4().hex[:24]
         set_request_id(request_id)
         kv_transfer_params = _engine_kv_transfer_params(request.kv_transfer_params)
-        max_input = _max_prompt_tokens(request.max_tokens, kv_transfer_params)
+        prompt_cap = _prompt_cap(request.max_tokens, kv_transfer_params)
 
         if request.stream:
             # Streaming response
@@ -2316,8 +2329,10 @@ async def anthropic_messages(request: AnthropicMessagesRequest, raw_request: Req
                 sampling_params,
                 request_id,
                 kv_transfer_params=kv_transfer_params,
-                max_prompt_tokens=max_input,
+                prompt_cap=prompt_cap,
             )
+            # What the engine runs, if the prompt was cut.
+            prompt = prompt_cap.cut_text or prompt
 
             async def generate_anthropic_stream():
                 # Unconditional, like the chat path and like both upstreams:
@@ -2492,7 +2507,7 @@ async def generate_anthropic_stream():
                 sampling_params,
                 request_id,
                 kv_transfer_params=kv_transfer_params,
-                max_prompt_tokens=max_input,
+                prompt_cap=prompt_cap,
             ),
             raw_request,
             request_id,
@@ -2500,6 +2515,7 @@ async def generate_anthropic_stream():
         if final_output is None:
             raise RuntimeError("No output generated")
         input_tokens = final_output["num_tokens_input"]
+        prompt = prompt_cap.cut_text or prompt
 
         raw_text = final_output["text"]
         # Separating is unconditional -- the same call the chat path makes,
@@ -2673,7 +2689,7 @@ async def responses_create(raw_request: Request):
         )
 
         # Off the loop, like the length check and truncation below, which
-        # happen in the preprocess call (`max_prompt_tokens`).
+        # happen in the preprocess call (`prompt_cap`).
         prompt = await asyncio.to_thread(
             apply_chat_template,
             tokenizer,
@@ -2714,7 +2730,7 @@ async def responses_create(raw_request: Request):
 
         request_id = f"resp_{uuid.uuid4().hex[:24]}"
         set_request_id(request_id)
-        max_input = _max_prompt_tokens(max_tokens, kv_transfer_params)
+        prompt_cap = _prompt_cap(max_tokens, kv_transfer_params)
 
         created_at = int(time.time())
         if request.stream:
@@ -2723,8 +2739,10 @@ async def responses_create(raw_request: Request):
                 sampling_params,
                 request_id,
                 kv_transfer_params=kv_transfer_params,
-                max_prompt_tokens=max_input,
+                prompt_cap=prompt_cap,
             )
+            # What the engine runs, if the prompt was cut.
+            prompt = prompt_cap.cut_text or prompt
 
             async def generate_responses_stream():
                 reasoning_filter = reasoning_channel(
@@ -2822,7 +2840,7 @@ async def generate_responses_stream():
                 sampling_params,
                 request_id,
                 kv_transfer_params=kv_transfer_params,
-                max_prompt_tokens=max_input,
+                prompt_cap=prompt_cap,
             ),
             raw_request,
             request_id,
@@ -2830,6 +2848,7 @@ async def generate_responses_stream():
         if final_output is None:
             raise RuntimeError("No output generated")
         input_tokens = final_output["num_tokens_input"]
+        prompt = prompt_cap.cut_text or prompt
 
         raw_text = final_output["text"]
         events = read_whole_blocks(
diff --git a/tests/entrypoints/test_api_server_helpers.py b/tests/entrypoints/test_api_server_helpers.py
index e87908693d..932ac56daa 100644
--- a/tests/entrypoints/test_api_server_helpers.py
+++ b/tests/entrypoints/test_api_server_helpers.py
@@ -893,6 +893,11 @@ def test_prompt_is_checked_and_truncated_in_the_preprocess_call(monkeypatch):
         "engine",
         SimpleNamespace(io_processor=SimpleNamespace(requests=requests)),
     )
+    monkeypatch.setattr(
+        api_server,
+        "tokenizer",
+        SimpleNamespace(decode=lambda ids, **_kw: f"<{len(ids)} ids>"),
+    )
     built = []
 
     def preprocess(prompt):
@@ -902,15 +907,20 @@ def preprocess(prompt):
         requests[seq.id] = seq
         return seq
 
-    seq = api_server._preprocess_prompt(preprocess, "short", 16)
+    cap = api_server._PromptCap(16)
+    seq = api_server._preprocess_prompt(preprocess, "short", cap)
     assert built == ["short"] and seq.num_prompt_tokens == 10
+    assert cap.cut_text is None
     seq = api_server._preprocess_prompt(preprocess, "short", None)
     assert seq.num_prompt_tokens == 10
 
     built.clear()
     requests.clear()
-    seq = api_server._preprocess_prompt(preprocess, "long prompt", 4)
+    cap = api_server._PromptCap(4)
+    seq = api_server._preprocess_prompt(preprocess, "long prompt", cap)
     assert built == ["long prompt", [0, 1, 2, 3]]
     assert seq.prompt_token_ids == [0, 1, 2, 3]
+    # The text the engine runs, for the reasoning channel's seed.
+    assert cap.cut_text == "<4 ids>"
     # The discarded first build is not left registered.
     assert list(requests) == [seq.id]
diff --git a/tests/entrypoints/test_pd_messages_responses.py b/tests/entrypoints/test_pd_messages_responses.py
index 0f2afc4e36..b5b112e2cb 100644
--- a/tests/entrypoints/test_pd_messages_responses.py
+++ b/tests/entrypoints/test_pd_messages_responses.py
@@ -53,9 +53,10 @@ def encode(text, **_kw):
         return [0] * (LONG_PROMPT_TOKENS if text == "<long prompt>" else 2)
 
     def engine_prompt(prompt, kwargs):
-        # What preprocess builds: the prompt, cut to `max_prompt_tokens`
+        # What preprocess builds: the prompt, cut to `prompt_cap.tokens`
         # (shown as `<n>` for an n-token cut) when it is longer.
-        limit = kwargs.get("max_prompt_tokens")
+        cap = kwargs.get("prompt_cap")
+        limit = None if cap is None else cap.tokens
         n = len(encode(prompt))
         return (f"<{limit}>", limit) if limit is not None and n > limit else (prompt, n)
 
diff --git a/tests/test_engine_core_mgr_shared_state.py b/tests/test_engine_core_mgr_shared_state.py
index 363114ab4f..143f7978d3 100644
--- a/tests/test_engine_core_mgr_shared_state.py
+++ b/tests/test_engine_core_mgr_shared_state.py
@@ -219,7 +219,7 @@ def test_subclass_does_not_override_the_shared_initialiser(self):
     unittest.main()
 
 
-def test_online_front_ends_build_the_engine_without_kept_outputs(monkeypatch):
+def test_create_engine_hands_keep_finished_outputs_to_the_config(monkeypatch):
     # Decided at construction: create_engine hands the choice to Config, and
     # the default keeps the offline / asyncio behaviour.
     import dataclasses
@@ -242,3 +242,29 @@ def fake_engine(model, tokenizer=None, **kwargs):
     assert seen == {"keep_finished_outputs": False}
     args.create_engine()
     assert seen == {"keep_finished_outputs": True}
+
+
+def test_atomesh_front_end_builds_the_engine_without_kept_outputs(monkeypatch):
+    # An online front end: every output reaches its request through the
+    # callbacks, so it must not ask the engine to keep finished sequences.
+    from atom.entrypoints.atomesh import server as atomesh_server
+    from atom.model_engine import arg_utils, llm_engine
+
+    seen = {}
+
+    class FakeArgs:
+        @classmethod
+        def from_cli_args(cls, args):
+            return cls()
+
+        def create_engine(self, **kwargs):
+            seen.update(kwargs)
+            return "engine"
+
+    monkeypatch.setattr(arg_utils, "EngineArgs", FakeArgs)
+    monkeypatch.setattr(llm_engine, "_load_tokenizer", lambda *a: "tok")
+    engine, _ = atomesh_server.initialize_engine(
+        SimpleNamespace(model="m", trust_remote_code=False)
+    )
+    assert engine == "engine"
+    assert seen == {"tokenizer": "tok", "keep_finished_outputs": False}

From 6ca4ec8601a67e8b25dd96c2de3c1cbadb98fef5 Mon Sep 17 00:00:00 2001
From: Yi <hyi@amd.com>
Date: Tue, 3 Nov 2026 13:46:38 +0800
Subject: [PATCH 06/11] style: sort _PromptCap.__slots__ (ruff RUF023)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 atom/entrypoints/openai/api_server.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/atom/entrypoints/openai/api_server.py b/atom/entrypoints/openai/api_server.py
index 171b0081b7..3f57c5af42 100644
--- a/atom/entrypoints/openai/api_server.py
+++ b/atom/entrypoints/openai/api_server.py
@@ -1256,7 +1256,7 @@ class _PromptCap:
     ``cut_text`` is the text the engine runs if the prompt was cut, for the
     readers that seed from the prompt's end (the reasoning channel)."""
 
-    __slots__ = ("tokens", "cut_text")
+    __slots__ = ("cut_text", "tokens")
 
     def __init__(self, tokens: int):
         self.tokens = tokens

From 991517cf502aa18c227acaa3c6890d4a6da2cde6 Mon Sep 17 00:00:00 2001
From: Yi <hyi@amd.com>
Date: Tue, 3 Nov 2026 20:58:26 +0800
Subject: [PATCH 07/11] review: /v1/responses seeds its reasoning channel from
 the text the engine runs

- With a schema, the channel goes into the sampling params before the engine
  call, so the prompt is cut first (_cut_prompt, off the loop) and the engine
  gets the kept ids.
- Without one, the preprocess call cuts it and the channel is re-seeded from
  cut_text afterwards (_reseed_after_cut), as /v1/messages does.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 atom/entrypoints/openai/api_server.py        | 41 +++++++++++++++++---
 tests/entrypoints/test_api_server_helpers.py | 29 ++++++++++++++
 2 files changed, 65 insertions(+), 5 deletions(-)

diff --git a/atom/entrypoints/openai/api_server.py b/atom/entrypoints/openai/api_server.py
index 1cc4cf7b4d..2a2c6fc959 100644
--- a/atom/entrypoints/openai/api_server.py
+++ b/atom/entrypoints/openai/api_server.py
@@ -1323,6 +1323,28 @@ def _prompt_cap(
     )
 
 
+def _cut_prompt(prompt: str, prompt_cap: _PromptCap) -> tuple[str | list[int], str]:
+    """For a caller that needs the cut text before the engine call: returns
+    what to send (the prompt, or the kept ids when it is over the cap) and the
+    text the engine will run. Tokenizes, so run it off the event loop."""
+    token_ids = tokenizer.encode(prompt)
+    if len(token_ids) <= prompt_cap.tokens:
+        return prompt, prompt
+    token_ids = token_ids[: prompt_cap.tokens]
+    return token_ids, tokenizer.decode(token_ids, skip_special_tokens=False)
+
+
+def _reseed_after_cut(channel, prompt_cap: _PromptCap, template_kwargs):
+    """The reasoning channel for the text the engine ran, when the preprocess
+    call cut the prompt (only the unstructured path gets there)."""
+    if prompt_cap.cut_text is None:
+        return channel
+    return reasoning_channel(
+        prompt_starts_in_reasoning(prompt_cap.cut_text),
+        template_kwargs=template_kwargs,
+    )
+
+
 def _preprocess_prompt(preprocess, prompt_or_tokens, prompt_cap):
     """``preprocess(prompt)`` on the executor, cutting a prompt longer than
     ``prompt_cap.tokens``.
@@ -2854,9 +2876,16 @@ async def responses_create(raw_request: Request):
         prompt_cap = _prompt_cap(sampling_params.max_tokens, kv_transfer_params)
 
         created_at = int(time.time())
-        # Decided before the engine call, since structured output configures
-        # the sampling params from it; so for a prompt over the cap it reads
-        # the template's ending, not the cut text the engine runs.
+        engine_prompt = prompt
+        if schema is not None:
+            # Structured output builds the channel into the sampling params
+            # before the engine call, so the cut has to come first here:
+            # tokenized off the loop, and the engine gets the kept ids.
+            engine_prompt, prompt = await asyncio.to_thread(
+                _cut_prompt, prompt, prompt_cap
+            )
+        # Without a schema the cut happens in the preprocess call, and the
+        # channel is re-seeded from `cut_text` after it (below).
         _reasoning = reasoning_channel(
             prompt_starts_in_reasoning(prompt), template_kwargs=merged_kwargs
         )
@@ -2873,12 +2902,13 @@ async def responses_create(raw_request: Request):
 
         if request.stream:
             seq_id, stream_collector, input_tokens = await setup_streaming_request(
-                prompt,
+                engine_prompt,
                 sampling_params,
                 request_id,
                 kv_transfer_params=kv_transfer_params,
                 prompt_cap=prompt_cap,
             )
+            _reasoning = _reseed_after_cut(_reasoning, prompt_cap, merged_kwargs)
 
             async def generate_responses_stream():
                 reasoning_filter = _reasoning.stream()
@@ -2986,7 +3016,7 @@ async def generate_responses_stream():
 
         final_output = await _run_nonstream_with_disconnect(
             generate_async(
-                prompt,
+                engine_prompt,
                 sampling_params,
                 request_id,
                 kv_transfer_params=kv_transfer_params,
@@ -2998,6 +3028,7 @@ async def generate_responses_stream():
         if final_output is None:
             raise RuntimeError("No output generated")
         input_tokens = final_output["num_tokens_input"]
+        _reasoning = _reseed_after_cut(_reasoning, prompt_cap, merged_kwargs)
 
         raw_text = final_output["text"]
         events = read_whole_blocks(
diff --git a/tests/entrypoints/test_api_server_helpers.py b/tests/entrypoints/test_api_server_helpers.py
index 5f664333c7..7502596fb3 100644
--- a/tests/entrypoints/test_api_server_helpers.py
+++ b/tests/entrypoints/test_api_server_helpers.py
@@ -926,6 +926,35 @@ def preprocess(prompt):
     assert list(requests) == [seq.id]
 
 
+def test_responses_channel_follows_the_text_the_engine_runs(monkeypatch):
+    # Structured: cut before the engine call (off the loop), so the channel
+    # the sampling params are built from reads the kept text. Unstructured:
+    # re-seeded from `cut_text` after the preprocess call cut it.
+    monkeypatch.setattr(
+        api_server,
+        "tokenizer",
+        SimpleNamespace(
+            encode=lambda text, **_kw: list(range(len(text))),
+            decode=lambda ids, **_kw: "x" * len(ids),
+        ),
+    )
+    cap = api_server._PromptCap(4)
+    assert api_server._cut_prompt("abc", cap) == ("abc", "abc")
+    assert api_server._cut_prompt("abcdefgh", cap) == ([0, 1, 2, 3], "xxxx")
+
+    seeds = []
+    monkeypatch.setattr(
+        api_server, "prompt_starts_in_reasoning", lambda text: seeds.append(text)
+    )
+    monkeypatch.setattr(
+        api_server, "reasoning_channel", lambda start, **_kw: ("reseeded", start)
+    )
+    assert api_server._reseed_after_cut("orig", cap, {}) == "orig"
+    cap.cut_text = "kept"
+    assert api_server._reseed_after_cut("orig", cap, {})[0] == "reseeded"
+    assert seeds == ["kept"]
+
+
 class TestDeploymentDefaultOutputCap:
     @pytest.mark.parametrize("explicit", [False, True])
     def test_small_deployment_distinguishes_default_from_explicit(

From bf3f38ad53dcf22920a6779e5ee94ebc5e07a23d Mon Sep 17 00:00:00 2001
From: Yi <hyi@amd.com>
Date: Tue, 3 Nov 2026 21:11:12 +0800
Subject: [PATCH 08/11] review: warn when a structured prompt is cut; drive the
 structured over-cap case

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 atom/entrypoints/openai/api_server.py         |  3 ++
 .../entrypoints/test_pd_messages_responses.py | 38 +++++++++++++++++++
 2 files changed, 41 insertions(+)

diff --git a/atom/entrypoints/openai/api_server.py b/atom/entrypoints/openai/api_server.py
index 2a2c6fc959..1aff1a95b5 100644
--- a/atom/entrypoints/openai/api_server.py
+++ b/atom/entrypoints/openai/api_server.py
@@ -1330,6 +1330,9 @@ def _cut_prompt(prompt: str, prompt_cap: _PromptCap) -> tuple[str | list[int], s
     token_ids = tokenizer.encode(prompt)
     if len(token_ids) <= prompt_cap.tokens:
         return prompt, prompt
+    logger.warning(
+        "Prompt too long (%d > %d), truncating", len(token_ids), prompt_cap.tokens
+    )
     token_ids = token_ids[: prompt_cap.tokens]
     return token_ids, tokenizer.decode(token_ids, skip_special_tokens=False)
 
diff --git a/tests/entrypoints/test_pd_messages_responses.py b/tests/entrypoints/test_pd_messages_responses.py
index b5b112e2cb..32abf505cf 100644
--- a/tests/entrypoints/test_pd_messages_responses.py
+++ b/tests/entrypoints/test_pd_messages_responses.py
@@ -224,3 +224,41 @@ def test_without_handoff_truncation_still_follows_max_tokens(
             self._sent_prompt(server, endpoint, 4096, None)
             == f"<{MAX_CTX - MAX_CTX // 8}>"
         )
+
+
+class TestStructuredResponsesCutFirst:
+    """A structured /v1/responses request builds its reasoning channel into the
+    sampling params before the engine call, so an over-long prompt is cut
+    first and the channel is seeded from the kept text."""
+
+    def test_over_the_cap_the_engine_gets_the_kept_ids(self, server, monkeypatch):
+        server.template = "<long prompt>"
+        configured = []
+        seeds = []
+
+        async def configure(_sampling, schema, channel, **_kw):
+            configured.append(schema)
+
+        monkeypatch.setattr(api_server, "_configure_structured_output", configure)
+        monkeypatch.setattr(
+            api_server,
+            "prompt_starts_in_reasoning",
+            lambda text: seeds.append(text) or False,
+        )
+        monkeypatch.setattr(api_server, "validate_output", lambda *a, **kw: None)
+        _responses(
+            {
+                "model": "m",
+                "input": "hi",
+                "max_output_tokens": 4096,
+                "text": {"format": {"type": "json_object"}},
+            }
+        )
+
+        sent, kwargs = server.generate[0]
+        kept = kwargs["prompt_cap"].tokens
+        assert configured == ['{"type": "object"}']
+        # The engine is handed the kept ids, already under the cap.
+        assert sent == [0] * kept
+        # The channel was seeded from the cut text, before the engine call.
+        assert seeds[0] == f"<{kept}>"

From 25cb4331ed9237b75a722ea4c53a7659ed46d5a0 Mon Sep 17 00:00:00 2001
From: Yi <hyi@amd.com>
Date: Tue, 3 Nov 2026 21:24:57 +0800
Subject: [PATCH 09/11] test: patch the validator responses actually calls and
 assert 200

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 tests/entrypoints/test_pd_messages_responses.py | 8 ++++++--
 1 file changed, 6 insertions(+), 2 deletions(-)

diff --git a/tests/entrypoints/test_pd_messages_responses.py b/tests/entrypoints/test_pd_messages_responses.py
index 32abf505cf..12b1de9bdc 100644
--- a/tests/entrypoints/test_pd_messages_responses.py
+++ b/tests/entrypoints/test_pd_messages_responses.py
@@ -245,8 +245,11 @@ async def configure(_sampling, schema, channel, **_kw):
             "prompt_starts_in_reasoning",
             lambda text: seeds.append(text) or False,
         )
-        monkeypatch.setattr(api_server, "validate_output", lambda *a, **kw: None)
-        _responses(
+        # The fake engine answers "hello", which is not JSON.
+        monkeypatch.setattr(
+            api_server, "validate_responses_output", lambda *a, **kw: None
+        )
+        response = _responses(
             {
                 "model": "m",
                 "input": "hi",
@@ -255,6 +258,7 @@ async def configure(_sampling, schema, channel, **_kw):
             }
         )
 
+        assert response.status_code == 200, response.text
         sent, kwargs = server.generate[0]
         kept = kwargs["prompt_cap"].tokens
         assert configured == ['{"type": "object"}']

From 97861062c2e9f630e3aec50a5e19789019146d7c Mon Sep 17 00:00:00 2001
From: Yi <hyi@amd.com>
Date: Tue, 3 Nov 2026 23:26:24 +0800
Subject: [PATCH 10/11] review: one prompt-preparation step;
 keep_finished_outputs off by default

- /v1/messages and /v1/responses: _prepare_prompt renders, tokenizes through
  the engine's own path (InputOutputProcessor.encode_prompt: segment cache
  and block-hash chain) and cuts to the context, once, on the executor,
  before the structured/reasoning channels are built. The engine gets the
  encoding and does not tokenize again. A context with no room for a prompt,
  or an empty cut, is a 400. _PromptCap, _cut_prompt, _reseed_after_cut, the
  preprocess rebuild and the prompt_cap parameters are gone.
- Config.keep_finished_outputs defaults to False; its readers turn it on
  (LLMEngine.add_request, asyncio_mode). Online front ends pass nothing;
  the create_engine / atomesh plumbing and their tests are gone. Read
  directly from the config.
- _bp_fail_stalled_run ends pending seqs that wait on a callback through it,
  as abort does, instead of a docstring promise.
- Tests: the fakes only count what the engine is handed; the prepare step is
  tested directly; trimmed duplicate comments.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 atom/config.py                                |   9 +-
 atom/entrypoints/atomesh/server.py            |   3 +-
 atom/entrypoints/openai/api_server.py         | 229 +++++++-----------
 atom/model_engine/arg_utils.py                |  17 +-
 atom/model_engine/engine_core_mgr.py          |  26 +-
 atom/model_engine/llm_engine.py               |  54 +++--
 tests/entrypoints/test_api_compatibility.py   |   3 +-
 tests/entrypoints/test_api_server_helpers.py  | 151 ++++++------
 .../test_kimi_k3_request_compat.py            |   3 +-
 .../entrypoints/test_pd_messages_responses.py |  52 ++--
 tests/test_distributed_store.py               |   6 +-
 tests/test_dp_backpressure.py                 |  27 ++-
 tests/test_engine_core_mgr_shared_state.py    |  70 +-----
 tests/test_llm_engine_add_request_fanout.py   |  16 ++
 14 files changed, 304 insertions(+), 362 deletions(-)

diff --git a/atom/config.py b/atom/config.py
index 6dfe206739..dd3add9415 100644
--- a/atom/config.py
+++ b/atom/config.py
@@ -2108,12 +2108,9 @@ class Config:
     compilation_config: CompilationConfig = field(default_factory=CompilationConfig)
     quant_config: QuantizationConfig = field(init=False)
     asyncio_mode: bool = False
-    # Whether CoreManager also queues finished sequences on ``outputs_queue``
-    # for ``LLMEngine.step`` / ``generate`` and the asyncio path to read. Online
-    # front ends deliver every output through per-sequence callbacks and build
-    # the engine with this off; left on there, every finished Sequence stays
-    # on a queue nothing reads, for the life of the process.
-    keep_finished_outputs: bool = True
+    # Queue finished sequences on outputs_queue; only its readers turn it on
+    # (LLMEngine.add_request for step / generate, and asyncio_mode).
+    keep_finished_outputs: bool = False
     mark_trace: bool = False
     load_dummy: str | None = None
     enable_expert_parallel: bool = False
diff --git a/atom/entrypoints/atomesh/server.py b/atom/entrypoints/atomesh/server.py
index 85cf343b2f..b7c722d03f 100644
--- a/atom/entrypoints/atomesh/server.py
+++ b/atom/entrypoints/atomesh/server.py
@@ -101,8 +101,7 @@ def initialize_engine(args: argparse.Namespace) -> tuple[Any, Any]:
 
     logger.info("Initializing engine with model %s...", args.model)
     engine_args = EngineArgs.from_cli_args(args)
-    # Outputs reach requests through the per-sequence callbacks.
-    engine = engine_args.create_engine(tokenizer=tokenizer, keep_finished_outputs=False)
+    engine = engine_args.create_engine(tokenizer=tokenizer)
     return engine, tokenizer
 
 
diff --git a/atom/entrypoints/openai/api_server.py b/atom/entrypoints/openai/api_server.py
index a0b3877945..e3854179ec 100644
--- a/atom/entrypoints/openai/api_server.py
+++ b/atom/entrypoints/openai/api_server.py
@@ -51,6 +51,7 @@
 from atom.model_engine.arg_utils import EngineArgs
 from atom.model_engine.llm_engine import load_tokenizer
 from atom.model_engine.request import RequestOutput
+from atom.model_engine.segment_tokenizer import PrefixEncoding
 from atom.model_engine.sequence import new_token_ids
 from atom.multimodal.processing import prepare_multimodal_inputs
 from atom.multimodal.registry import get_native_multimodal_processor
@@ -861,11 +862,8 @@ async def generate_async(
     dp_session_id: str | None = None,
     dp_parent_session_id: str | None = None,
     return_token_ids: bool = False,
-    prompt_cap: "_PromptCap | None" = None,
 ) -> AsyncGenerator[dict[str, Any], None]:
-    """Generate non-streaming output, optionally echoing local prompt token IDs.
-
-    ``prompt_cap`` truncates, in the preprocess call (``_preprocess_prompt``)."""
+    """Generate non-streaming output, optionally echoing local prompt token IDs."""
     token_queue: asyncio.Queue = asyncio.Queue()
     loop = asyncio.get_running_loop()
 
@@ -904,18 +902,14 @@ def completion_callback(request_output: RequestOutput):
         )
 
     def do_preprocess():
-        return _preprocess_prompt(
-            lambda prompt: engine.io_processor.preprocess(
-                prompt,
-                sampling_params,
-                stream_callback=completion_callback,
-                kv_transfer_params=kv_transfer_params,
-                data_parallel_rank=data_parallel_rank,
-                dp_session_id=dp_session_id,
-                dp_parent_session_id=dp_parent_session_id,
-            ),
+        return engine.io_processor.preprocess(
             prompt_or_tokens,
-            prompt_cap,
+            sampling_params,
+            stream_callback=completion_callback,
+            kv_transfer_params=kv_transfer_params,
+            data_parallel_rank=data_parallel_rank,
+            dp_session_id=dp_session_id,
+            dp_parent_session_id=dp_parent_session_id,
         )
 
     seq = await loop.run_in_executor(None, do_preprocess)
@@ -1313,72 +1307,48 @@ def _prompt_headroom(
     return headroom
 
 
-class _PromptCap:
-    """The longest prompt ``/v1/messages`` and ``/v1/responses`` keep: a longer
-    one is cut to ``tokens`` rather than refused. After the preprocess call,
-    ``cut_text`` is the text the engine runs if the prompt was cut, for the
-    readers that seed from the prompt's end (the reasoning channel)."""
-
-    __slots__ = ("cut_text", "tokens")
-
-    def __init__(self, tokens: int):
-        self.tokens = tokens
-        self.cut_text: str | None = None
-
-
-def _prompt_cap(
-    max_tokens: int, kv_transfer_params: dict[str, Any] | None
-) -> _PromptCap:
-    max_ctx = int(_get_engine_max_model_len() or 30720)
-    return _PromptCap(
-        max_ctx - _prompt_headroom(max_tokens, max_ctx, kv_transfer_params)
-    )
-
-
-def _cut_prompt(prompt: str, prompt_cap: _PromptCap) -> tuple[str | list[int], str]:
-    """For a caller that needs the cut text before the engine call: returns
-    what to send (the prompt, or the kept ids when it is over the cap) and the
-    text the engine will run. Tokenizes, so run it off the event loop."""
-    token_ids = tokenizer.encode(prompt)
-    if len(token_ids) <= prompt_cap.tokens:
-        return prompt, prompt
-    logger.warning(
-        "Prompt too long (%d > %d), truncating", len(token_ids), prompt_cap.tokens
-    )
-    token_ids = token_ids[: prompt_cap.tokens]
-    return token_ids, tokenizer.decode(token_ids, skip_special_tokens=False)
-
-
-def _reseed_after_cut(channel, prompt_cap: _PromptCap, template_kwargs):
-    """The reasoning channel for the text the engine ran, when the preprocess
-    call cut the prompt (only the unstructured path gets there)."""
-    if prompt_cap.cut_text is None:
-        return channel
-    return reasoning_channel(
-        prompt_starts_in_reasoning(prompt_cap.cut_text),
-        template_kwargs=template_kwargs,
-    )
-
-
-def _preprocess_prompt(preprocess, prompt_or_tokens, prompt_cap):
-    """``preprocess(prompt)`` on the executor, cutting a prompt longer than
-    ``prompt_cap.tokens``.
-
-    The prompt is tokenized once, by ``preprocess``, and checked by its
-    ``num_prompt_tokens``; only an over-long one is built a second time, from
-    its own first ids."""
-    seq = preprocess(prompt_or_tokens)
-    if prompt_cap is None or seq.num_prompt_tokens <= prompt_cap.tokens:
-        return seq
-    logger.warning(
-        "Prompt too long (%d > %d), truncating",
-        seq.num_prompt_tokens,
-        prompt_cap.tokens,
+def _prepare_prompt(
+    messages,
+    template_kwargs: dict[str, Any],
+    *,
+    tools,
+    preserve_tool_call_arguments: bool,
+    max_tokens: int,
+    kv_transfer_params: dict[str, Any] | None,
+):
+    """Render, tokenize the engine's way and cut to the context, in one
+    executor step for ``/v1/messages`` and ``/v1/responses``.
+
+    Returns ``(engine_input, text)``: what preprocess takes in place of the
+    prompt (no second tokenize), and the text the engine runs, which seeds the
+    reasoning channel. A prompt longer than the context minus the output
+    headroom is cut from the end; a context with no room for any prompt is
+    refused (ValueError, a 400)."""
+    prompt = apply_chat_template(
+        tokenizer,
+        custom_message_encoder,
+        messages,
+        tools=tools,
+        preserve_tool_call_arguments=preserve_tool_call_arguments,
+        **template_kwargs,
     )
-    engine.io_processor.requests.pop(seq.id, None)
-    token_ids = list(seq.prompt_token_ids[: prompt_cap.tokens])
-    prompt_cap.cut_text = tokenizer.decode(token_ids, skip_special_tokens=False)
-    return preprocess(token_ids)
+    encoded = engine.io_processor.encode_prompt(prompt)
+    max_ctx = int(_get_engine_max_model_len() or 30720)
+    cap = max_ctx - _prompt_headroom(max_tokens, max_ctx, kv_transfer_params)
+    if cap < 1:
+        raise ValueError(
+            f"max_tokens={max_tokens} leaves no room for a prompt in this "
+            f"model's {max_ctx}-token context"
+        )
+    token_ids = encoded.token_ids if isinstance(encoded, PrefixEncoding) else encoded
+    if len(token_ids) <= cap:
+        return encoded, prompt
+    logger.warning("Prompt too long (%d > %d), truncating", len(token_ids), cap)
+    token_ids = list(token_ids[:cap])
+    text = tokenizer.decode(token_ids, skip_special_tokens=False)
+    if not text:
+        raise ValueError("The prompt is empty after truncation to the context")
+    return token_ids, text
 
 
 def _with_kv_transfer_params(response: dict, final_output: dict) -> dict:
@@ -1397,12 +1367,9 @@ async def setup_streaming_request(
     data_parallel_rank: int | None = None,
     dp_session_id: str | None = None,
     dp_parent_session_id: str | None = None,
-    prompt_cap: "_PromptCap | None" = None,
 ) -> tuple[int, StreamOutputCollector, int]:
     """Set up a streaming request with the engine.
 
-    ``prompt_cap`` truncates, in the preprocess call (``_preprocess_prompt``).
-
     Returns ``(seq_id, stream_collector, num_prompt_tokens)``.
     ``num_prompt_tokens`` is the engine-computed prompt length so the stream
     response generator does not have to re-tokenize the prompt on the event
@@ -1426,19 +1393,15 @@ def stream_callback(request_output: RequestOutput) -> None:
     executor_loop = asyncio.get_event_loop()
 
     def do_preprocess():
-        seq = _preprocess_prompt(
-            lambda prompt: engine.io_processor.preprocess(
-                prompt,
-                sampling_params,
-                stream_callback=stream_callback,
-                kv_transfer_params=kv_transfer_params,
-                multimodal_data=multimodal_data,
-                data_parallel_rank=data_parallel_rank,
-                dp_session_id=dp_session_id,
-                dp_parent_session_id=dp_parent_session_id,
-            ),
+        seq = engine.io_processor.preprocess(
             prompt_or_tokens,
-            prompt_cap,
+            sampling_params,
+            stream_callback=stream_callback,
+            kv_transfer_params=kv_transfer_params,
+            multimodal_data=multimodal_data,
+            data_parallel_rank=data_parallel_rank,
+            dp_session_id=dp_session_id,
+            dp_parent_session_id=dp_parent_session_id,
         )
         _seq_id_to_request_id[seq.id] = request_id
         return seq
@@ -2006,8 +1969,6 @@ async def chat_completions(request: ChatCompletionRequest, raw_request: Request)
             prompt_or_tokens = pretokenized
             kv_transfer_params = _engine_kv_transfer_params(kv_transfer_params)
         else:
-            # Off the loop: rendering grows with the conversation, and on the
-            # loop it holds up reading every other request.
             prompt_or_tokens = await asyncio.to_thread(
                 apply_chat_template,
                 tokenizer,
@@ -2436,18 +2397,6 @@ async def anthropic_messages(request: AnthropicMessagesRequest, raw_request: Req
         # no such spelling, and are not answered here or on the chat path.
         if forbids_tool_calls(request.tool_choice):
             merged_kwargs["tool_choice"] = "none"
-        # Off the loop, like the length check and truncation below, which
-        # happen in the preprocess call (`prompt_cap`).
-        prompt = await asyncio.to_thread(
-            apply_chat_template,
-            tokenizer,
-            custom_message_encoder,
-            messages,
-            tools=anthropic_to_openai_tools(request.tools),
-            preserve_tool_call_arguments=preserve_tool_call_arguments,
-            **merged_kwargs,
-        )
-
         generation_config = engine.config.generation_config
         model_temperature = getattr(generation_config, "temperature", None)
         model_top_p = getattr(generation_config, "top_p", None)
@@ -2476,19 +2425,24 @@ async def anthropic_messages(request: AnthropicMessagesRequest, raw_request: Req
         request_id = uuid.uuid4().hex[:24]
         set_request_id(request_id)
         kv_transfer_params = _engine_kv_transfer_params(request.kv_transfer_params)
-        prompt_cap = _prompt_cap(sampling_params.max_tokens, kv_transfer_params)
+        engine_input, prompt = await asyncio.to_thread(
+            _prepare_prompt,
+            messages,
+            merged_kwargs,
+            tools=anthropic_to_openai_tools(request.tools),
+            preserve_tool_call_arguments=preserve_tool_call_arguments,
+            max_tokens=sampling_params.max_tokens,
+            kv_transfer_params=kv_transfer_params,
+        )
 
         if request.stream:
             # Streaming response
             seq_id, stream_collector, input_tokens = await setup_streaming_request(
-                prompt,
+                engine_input,
                 sampling_params,
                 request_id,
                 kv_transfer_params=kv_transfer_params,
-                prompt_cap=prompt_cap,
             )
-            # What the engine runs, if the prompt was cut.
-            prompt = prompt_cap.cut_text or prompt
 
             async def generate_anthropic_stream():
                 # Unconditional, like the chat path and like both upstreams:
@@ -2659,11 +2613,10 @@ async def generate_anthropic_stream():
         # Non-streaming response
         final_output = await _run_nonstream_with_disconnect(
             generate_async(
-                prompt,
+                engine_input,
                 sampling_params,
                 request_id,
                 kv_transfer_params=kv_transfer_params,
-                prompt_cap=prompt_cap,
             ),
             raw_request,
             request_id,
@@ -2671,7 +2624,6 @@ async def generate_anthropic_stream():
         if final_output is None:
             raise RuntimeError("No output generated")
         input_tokens = final_output["num_tokens_input"]
-        prompt = prompt_cap.cut_text or prompt
 
         raw_text = final_output["text"]
         # Separating is unconditional -- the same call the chat path makes,
@@ -2845,18 +2797,6 @@ async def responses_create(raw_request: Request):
             rules=rules,
         )
 
-        # Off the loop, like the length check and truncation below, which
-        # happen in the preprocess call (`prompt_cap`).
-        prompt = await asyncio.to_thread(
-            apply_chat_template,
-            tokenizer,
-            custom_message_encoder,
-            messages,
-            tools=openai_tools,
-            preserve_tool_call_arguments=preserve_tool_call_arguments,
-            **merged_kwargs,
-        )
-
         generation_config = engine.config.generation_config
         model_temperature = getattr(generation_config, "temperature", None)
         model_top_p = getattr(generation_config, "top_p", None)
@@ -2893,19 +2833,17 @@ async def responses_create(raw_request: Request):
         schema = schema_from_format((text_config or {}).get("format"), responses=True)
         request_id = f"resp_{uuid.uuid4().hex[:24]}"
         set_request_id(request_id)
-        prompt_cap = _prompt_cap(sampling_params.max_tokens, kv_transfer_params)
+        engine_input, prompt = await asyncio.to_thread(
+            _prepare_prompt,
+            messages,
+            merged_kwargs,
+            tools=openai_tools,
+            preserve_tool_call_arguments=preserve_tool_call_arguments,
+            max_tokens=sampling_params.max_tokens,
+            kv_transfer_params=kv_transfer_params,
+        )
 
         created_at = int(time.time())
-        engine_prompt = prompt
-        if schema is not None:
-            # Structured output builds the channel into the sampling params
-            # before the engine call, so the cut has to come first here:
-            # tokenized off the loop, and the engine gets the kept ids.
-            engine_prompt, prompt = await asyncio.to_thread(
-                _cut_prompt, prompt, prompt_cap
-            )
-        # Without a schema the cut happens in the preprocess call, and the
-        # channel is re-seeded from `cut_text` after it (below).
         _reasoning = reasoning_channel(
             prompt_starts_in_reasoning(prompt), template_kwargs=merged_kwargs
         )
@@ -2922,13 +2860,11 @@ async def responses_create(raw_request: Request):
 
         if request.stream:
             seq_id, stream_collector, input_tokens = await setup_streaming_request(
-                engine_prompt,
+                engine_input,
                 sampling_params,
                 request_id,
                 kv_transfer_params=kv_transfer_params,
-                prompt_cap=prompt_cap,
             )
-            _reasoning = _reseed_after_cut(_reasoning, prompt_cap, merged_kwargs)
 
             async def generate_responses_stream():
                 reasoning_filter = _reasoning.stream()
@@ -3036,11 +2972,10 @@ async def generate_responses_stream():
 
         final_output = await _run_nonstream_with_disconnect(
             generate_async(
-                engine_prompt,
+                engine_input,
                 sampling_params,
                 request_id,
                 kv_transfer_params=kv_transfer_params,
-                prompt_cap=prompt_cap,
             ),
             raw_request,
             request_id,
@@ -3048,7 +2983,6 @@ async def generate_responses_stream():
         if final_output is None:
             raise RuntimeError("No output generated")
         input_tokens = final_output["num_tokens_input"]
-        _reasoning = _reseed_after_cut(_reasoning, prompt_cap, merged_kwargs)
 
         raw_text = final_output["text"]
         events = read_whole_blocks(
@@ -3443,8 +3377,7 @@ def main():
     preserve_tool_call_arguments = template_keeps_tool_call_arguments(tools_probe)
     required_tool_channel = required_call_channel(tool_call_parser_cls, tokenizer)
 
-    # Every output reaches its request through the per-sequence callbacks.
-    engine = engine_args.create_engine(tokenizer=tokenizer, keep_finished_outputs=False)
+    engine = engine_args.create_engine(tokenizer=tokenizer)
     # Forced acceptance emits draft tokens that nothing verified, and once the
     # context is long enough those degenerate into the dialect's own channel
     # framing and nothing else -- read as structure, correctly, that leaves a
diff --git a/atom/model_engine/arg_utils.py b/atom/model_engine/arg_utils.py
index dbb903e887..fba9769adf 100644
--- a/atom/model_engine/arg_utils.py
+++ b/atom/model_engine/arg_utils.py
@@ -958,17 +958,6 @@ def _get_engine_kwargs(self) -> dict:
 
         return kwargs
 
-    def create_engine(
-        self, tokenizer=None, *, keep_finished_outputs: bool = True
-    ) -> LLMEngine:
-        """Create and return an LLMEngine instance with the configured parameters.
-
-        ``keep_finished_outputs=False`` is for online front ends, which take
-        every output through per-sequence callbacks (``Config.keep_finished_outputs``).
-        """
-        return LLMEngine(
-            self.model,
-            tokenizer=tokenizer,
-            keep_finished_outputs=keep_finished_outputs,
-            **self._get_engine_kwargs(),
-        )
+    def create_engine(self, tokenizer=None) -> LLMEngine:
+        """Create and return an LLMEngine instance with the configured parameters."""
+        return LLMEngine(self.model, tokenizer=tokenizer, **self._get_engine_kwargs())
diff --git a/atom/model_engine/engine_core_mgr.py b/atom/model_engine/engine_core_mgr.py
index 0ca98b853c..44329db0ed 100644
--- a/atom/model_engine/engine_core_mgr.py
+++ b/atom/model_engine/engine_core_mgr.py
@@ -231,10 +231,7 @@ def _init_shared_state(
         )
         self.ctx = zmq.Context(io_threads=2)
         self.outputs_queue = queue.Queue[list[Sequence]]()
-        # Off for online front ends, which read every output through the
-        # per-sequence callbacks: nothing would drain the queue
-        # (`Config.keep_finished_outputs`).
-        self.keep_finished_outputs = getattr(config, "keep_finished_outputs", True)
+        self.keep_finished_outputs = config.keep_finished_outputs or config.asyncio_mode
         # Utility replies are matched to their caller by request id, not by
         # position on a shared queue, where a late or unasked-for reply became
         # the next caller's answer.
@@ -1311,12 +1308,23 @@ def _bp_fail_stalled_run(self) -> None:
         ``LLMEngine.generate`` waits while the engine is alive and any request
         is unfinished. A dropped seq never finishes, so that wait never ends.
         The seqs are still in ``_bp_pending``; this only surfaces the error.
-
-        Only offline seqs are ever in the pool (``add_request`` keeps online
-        requests out of it), and offline ``generate`` reads ``outputs_queue``.
-        If online seqs could wait there, this error would have to reach them
-        through their callbacks instead.
+        A pending seq with a stream callback is not reading that queue, so it
+        is taken out and ended through its callback, as abort does.
         """
+        with self._lb_lock:
+            waiting = [s for s in self._bp_pending if s.id in self._seq_id_to_callback]
+            for seq in waiting:
+                self._bp_pending.remove(seq)
+        delivered = False
+        for seq in waiting:
+            if not getattr(seq, "leave_reason", None):
+                seq.leave_reason = "aborted"
+            delivered |= self._deliver_terminal_output(seq)
+        if delivered and self._flush_stream_batch_fn is not None:
+            try:
+                self._flush_stream_batch_fn()
+            except Exception as e:  # noqa: BLE001 - the error below must still go out
+                logger.warning("%s: flush_stream_batch failed: %s", self.label, e)
         pending = len(getattr(self, "_bp_pending", ()))
         logger.error(
             "%s: backpressure send failed with nothing in flight; "
diff --git a/atom/model_engine/llm_engine.py b/atom/model_engine/llm_engine.py
index d3d2426714..50df10485f 100644
--- a/atom/model_engine/llm_engine.py
+++ b/atom/model_engine/llm_engine.py
@@ -14,7 +14,7 @@
 from atom.config import Config
 from atom.model_engine.engine_core_mgr import CoreManager, DisaggCoreManager
 from atom.model_engine.prefix_hash import hash_block_size_for, ship_prompt_hash_chain
-from atom.model_engine.segment_tokenizer import SegmentTokenizer
+from atom.model_engine.segment_tokenizer import PrefixEncoding, SegmentTokenizer
 from atom.model_engine.sequence import Sequence
 from atom.multimodal.registry import get_mrope_input_positions
 from atom.sampling_params import RequiredChannel, SamplingParams
@@ -330,6 +330,7 @@ def add_request(
                 parent_request_id=request_id,
             )
             reqs.extend(fanout)
+        self.core_mgr.keep_finished_outputs = True
         self.core_mgr.add_request(reqs)
 
     def step(self) -> list[Sequence]:
@@ -813,7 +814,7 @@ def _per_req_cache_model_types() -> frozenset[str]:
 
     def preprocess(
         self,
-        prompt_or_tokens: str | list[int],
+        prompt_or_tokens: str | list[int] | PrefixEncoding,
         sampling_params: SamplingParams,
         stream_callback=None,
         kv_transfer_params=None,
@@ -874,9 +875,29 @@ def _engine_channel(self, channel: RequiredChannel) -> RequiredChannel:
             )
         return self._required_channel
 
+    def encode_prompt(
+        self, prompt: str, *, multimodal: bool = False
+    ) -> PrefixEncoding | list[int]:
+        """Tokenize ``prompt`` the way ``preprocess`` does (segment cache and,
+        when enabled, the prompt's block-hash chain). ``preprocess`` takes the
+        result in place of the text without tokenizing again."""
+        segment_tokenizer = getattr(self, "segment_tokenizer", None)
+        if segment_tokenizer is None:
+            return self.tokenizer.encode(prompt)
+        if (
+            getattr(self, "_reuse_block_hashes", False)
+            and self.prefix_hash_block_size
+            and envs.ATOM_FRONTEND_PREFIX_HASH
+            and not multimodal
+        ):
+            return segment_tokenizer.encode_with_prefix(
+                prompt, self.prefix_hash_block_size
+            )
+        return segment_tokenizer.encode(prompt)
+
     def preprocess_fanout(
         self,
-        prompt_or_tokens: str | list[int],
+        prompt_or_tokens: str | list[int] | PrefixEncoding,
         sampling_params: SamplingParams,
         stream_callback=None,
         stream_callbacks: list | None = None,
@@ -918,25 +939,16 @@ def preprocess_fanout(
             )
         n = max(1, int(getattr(sampling_params, "n", 1)))
 
-        prefix_encoding = None
-        segment_tokenizer = getattr(self, "segment_tokenizer", None)
-        if not isinstance(prompt_or_tokens, str):
-            tokens = prompt_or_tokens
-        elif segment_tokenizer is not None:
-            if (
-                getattr(self, "_reuse_block_hashes", False)
-                and self.prefix_hash_block_size
-                and envs.ATOM_FRONTEND_PREFIX_HASH
-                and multimodal_data is None
-            ):
-                prefix_encoding = segment_tokenizer.encode_with_prefix(
-                    prompt_or_tokens, self.prefix_hash_block_size
-                )
-                tokens = prefix_encoding.token_ids
-            else:
-                tokens = segment_tokenizer.encode(prompt_or_tokens)
+        if isinstance(prompt_or_tokens, str):
+            prompt_or_tokens = self.encode_prompt(
+                prompt_or_tokens, multimodal=multimodal_data is not None
+            )
+        if isinstance(prompt_or_tokens, PrefixEncoding):
+            prefix_encoding = prompt_or_tokens
+            tokens = prefix_encoding.token_ids
         else:
-            tokens = self.tokenizer.encode(prompt_or_tokens)
+            prefix_encoding = None
+            tokens = prompt_or_tokens
         mrope_positions = None
         mrope_position_delta = 0
         if multimodal_data is not None:
diff --git a/tests/entrypoints/test_api_compatibility.py b/tests/entrypoints/test_api_compatibility.py
index 7305181419..6233ef01c8 100644
--- a/tests/entrypoints/test_api_compatibility.py
+++ b/tests/entrypoints/test_api_compatibility.py
@@ -72,7 +72,8 @@ def encode(messages, **kwargs):
         "engine": SimpleNamespace(
             config=SimpleNamespace(
                 generation_config=SimpleNamespace(), max_model_len=8192
-            )
+            ),
+            io_processor=SimpleNamespace(encode_prompt=lambda text: [1, 2]),
         ),
     }.items():
         monkeypatch.setattr(api_server, name, value)
diff --git a/tests/entrypoints/test_api_server_helpers.py b/tests/entrypoints/test_api_server_helpers.py
index 7502596fb3..b6b8a73a87 100644
--- a/tests/entrypoints/test_api_server_helpers.py
+++ b/tests/entrypoints/test_api_server_helpers.py
@@ -876,83 +876,82 @@ async def drive():
     assert events == ["initialize", "serving", "shutdown"]
 
 
-class _FakeSeq:
-    def __init__(self, seq_id, token_ids):
-        self.id = seq_id
-        self.prompt_token_ids = list(token_ids)
-        self.num_prompt_tokens = len(token_ids)
-
-
-def test_prompt_is_checked_and_truncated_in_the_preprocess_call(monkeypatch):
-    # /v1/messages and /v1/responses: no encode on the event loop. The length
-    # check reads the engine's own token count from the one preprocess call,
-    # and only an over-long prompt is built again, from its first ids.
-    requests = {}
-    monkeypatch.setattr(
-        api_server,
-        "engine",
-        SimpleNamespace(io_processor=SimpleNamespace(requests=requests)),
-    )
-    monkeypatch.setattr(
-        api_server,
-        "tokenizer",
-        SimpleNamespace(decode=lambda ids, **_kw: f"<{len(ids)} ids>"),
-    )
-    built = []
-
-    def preprocess(prompt):
-        tokens = list(range(10)) if isinstance(prompt, str) else prompt
-        seq = _FakeSeq(len(built), tokens)
-        built.append(prompt)
-        requests[seq.id] = seq
-        return seq
-
-    cap = api_server._PromptCap(16)
-    seq = api_server._preprocess_prompt(preprocess, "short", cap)
-    assert built == ["short"] and seq.num_prompt_tokens == 10
-    assert cap.cut_text is None
-    seq = api_server._preprocess_prompt(preprocess, "short", None)
-    assert seq.num_prompt_tokens == 10
-
-    built.clear()
-    requests.clear()
-    cap = api_server._PromptCap(4)
-    seq = api_server._preprocess_prompt(preprocess, "long prompt", cap)
-    assert built == ["long prompt", [0, 1, 2, 3]]
-    assert seq.prompt_token_ids == [0, 1, 2, 3]
-    # The text the engine runs, for the reasoning channel's seed.
-    assert cap.cut_text == "<4 ids>"
-    # The discarded first build is not left registered.
-    assert list(requests) == [seq.id]
-
-
-def test_responses_channel_follows_the_text_the_engine_runs(monkeypatch):
-    # Structured: cut before the engine call (off the loop), so the channel
-    # the sampling params are built from reads the kept text. Unstructured:
-    # re-seeded from `cut_text` after the preprocess call cut it.
-    monkeypatch.setattr(
-        api_server,
-        "tokenizer",
-        SimpleNamespace(
-            encode=lambda text, **_kw: list(range(len(text))),
-            decode=lambda ids, **_kw: "x" * len(ids),
-        ),
-    )
-    cap = api_server._PromptCap(4)
-    assert api_server._cut_prompt("abc", cap) == ("abc", "abc")
-    assert api_server._cut_prompt("abcdefgh", cap) == ([0, 1, 2, 3], "xxxx")
+class TestPreparePrompt:
+    """`_prepare_prompt`: render, tokenize the engine's way, cut to the
+    context, on the executor, for /v1/messages and /v1/responses."""
 
-    seeds = []
-    monkeypatch.setattr(
-        api_server, "prompt_starts_in_reasoning", lambda text: seeds.append(text)
-    )
-    monkeypatch.setattr(
-        api_server, "reasoning_channel", lambda start, **_kw: ("reseeded", start)
-    )
-    assert api_server._reseed_after_cut("orig", cap, {}) == "orig"
-    cap.cut_text = "kept"
-    assert api_server._reseed_after_cut("orig", cap, {})[0] == "reseeded"
-    assert seeds == ["kept"]
+    @pytest.fixture
+    def prepare(self, monkeypatch):
+        from atom.model_engine.segment_tokenizer import PrefixEncoding
+
+        encoded = {}
+
+        def encode_prompt(text):
+            # One id per character; with a block-hash chain when asked.
+            ids = list(range(len(text)))
+            return PrefixEncoding(ids, "chain") if encoded.get("chain") else ids
+
+        monkeypatch.setattr(
+            api_server,
+            "engine",
+            SimpleNamespace(
+                config=SimpleNamespace(max_model_len=100),
+                io_processor=SimpleNamespace(encode_prompt=encode_prompt),
+            ),
+        )
+        monkeypatch.setattr(
+            api_server,
+            "tokenizer",
+            SimpleNamespace(decode=lambda ids, **_kw: "k" * len(ids)),
+        )
+        monkeypatch.setattr(api_server, "custom_message_encoder", None)
+        rendered = {}
+        monkeypatch.setattr(
+            api_server,
+            "apply_chat_template",
+            lambda _tok, _enc, messages, **kwargs: rendered.setdefault(
+                "text", messages
+            ),
+        )
+
+        def run(prompt, max_tokens, chain=False, kv=None):
+            encoded["chain"] = chain
+            rendered.clear()
+            return api_server._prepare_prompt(
+                prompt,
+                {},
+                tools=None,
+                preserve_tool_call_arguments=False,
+                max_tokens=max_tokens,
+                kv_transfer_params=kv,
+            )
+
+        return run
+
+    def test_a_prompt_that_fits_goes_to_the_engine_as_encoded(self, prepare):
+        engine_input, text = prepare("x" * 50, max_tokens=10, chain=True)
+        # The engine's own encoding, chain included, so preprocess reuses it.
+        assert engine_input.token_ids == list(range(50))
+        assert engine_input.block_hashes == "chain"
+        assert text == "x" * 50
+
+    def test_an_over_long_prompt_is_cut_to_the_context(self, prepare):
+        # 100-token context, 10 reserved for output: 90 kept.
+        engine_input, text = prepare("x" * 150, max_tokens=10, chain=True)
+        assert engine_input == list(range(90))
+        assert text == "k" * 90
+
+    @pytest.mark.parametrize("max_tokens", [100, 1024])
+    def test_no_room_for_a_prompt_is_refused(self, prepare, max_tokens):
+        with pytest.raises(ValueError, match="no room for a prompt"):
+            prepare("x" * 5, max_tokens=max_tokens)
+
+    def test_an_empty_cut_is_refused(self, prepare, monkeypatch):
+        monkeypatch.setattr(
+            api_server, "tokenizer", SimpleNamespace(decode=lambda ids, **_kw: "")
+        )
+        with pytest.raises(ValueError, match="empty after truncation"):
+            prepare("x" * 150, max_tokens=10)
 
 
 class TestDeploymentDefaultOutputCap:
diff --git a/tests/entrypoints/test_kimi_k3_request_compat.py b/tests/entrypoints/test_kimi_k3_request_compat.py
index e7030eaad8..4de97961d7 100644
--- a/tests/entrypoints/test_kimi_k3_request_compat.py
+++ b/tests/entrypoints/test_kimi_k3_request_compat.py
@@ -407,7 +407,8 @@ async def fake_generate_async(_prompt, sampling, _request_id, **_kwargs):
         SimpleNamespace(
             config=SimpleNamespace(
                 generation_config=SimpleNamespace(), max_model_len=8192
-            )
+            ),
+            io_processor=SimpleNamespace(encode_prompt=lambda _text: [1, 2]),
         ),
     )
     monkeypatch.setattr(api_server, "generate_async", fake_generate_async)
diff --git a/tests/entrypoints/test_pd_messages_responses.py b/tests/entrypoints/test_pd_messages_responses.py
index 12b1de9bdc..7fd8ddc04e 100644
--- a/tests/entrypoints/test_pd_messages_responses.py
+++ b/tests/entrypoints/test_pd_messages_responses.py
@@ -52,22 +52,15 @@ def server(monkeypatch):
     def encode(text, **_kw):
         return [0] * (LONG_PROMPT_TOKENS if text == "<long prompt>" else 2)
 
-    def engine_prompt(prompt, kwargs):
-        # What preprocess builds: the prompt, cut to `prompt_cap.tokens`
-        # (shown as `<n>` for an n-token cut) when it is longer.
-        cap = kwargs.get("prompt_cap")
-        limit = None if cap is None else cap.tokens
-        n = len(encode(prompt))
-        return (f"<{limit}>", limit) if limit is not None and n > limit else (prompt, n)
-
+    # Only what the engine does: count what it is handed. Any cut is the
+    # production code's (`_prepare_prompt`).
     async def fake_generate_async(prompt, _sampling, _request_id, **kwargs):
-        sent, n = engine_prompt(prompt, kwargs)
-        calls.generate.append((sent, kwargs))
+        calls.generate.append((prompt, kwargs))
         output = {
             "text": "hello",
             "finish_reason": "eos",
             "num_cached_tokens": 0,
-            "num_tokens_input": n,
+            "num_tokens_input": len(prompt),
             "num_tokens_output": 1,
         }
         if kwargs.get("kv_transfer_params") is not None:
@@ -75,9 +68,8 @@ async def fake_generate_async(prompt, _sampling, _request_id, **kwargs):
         yield output
 
     async def fake_setup_streaming_request(prompt, _sampling, _request_id, **kwargs):
-        sent, n = engine_prompt(prompt, kwargs)
-        calls.setup.append((sent, kwargs))
-        return 1, _FinishedCollector(), n
+        calls.setup.append((prompt, kwargs))
+        return 1, _FinishedCollector(), len(prompt)
 
     monkeypatch.setattr(api_server, "model_name", "m")
     monkeypatch.setattr(api_server, "_request_logger", None)
@@ -99,7 +91,8 @@ async def fake_setup_streaming_request(prompt, _sampling, _request_id, **kwargs)
         SimpleNamespace(
             config=SimpleNamespace(
                 generation_config=SimpleNamespace(), max_model_len=MAX_CTX
-            )
+            ),
+            io_processor=SimpleNamespace(encode_prompt=encode),
         ),
     )
     monkeypatch.setattr(api_server, "generate_async", fake_generate_async)
@@ -211,7 +204,7 @@ def test_with_handoff_the_prompt_does_not_depend_on_max_tokens(
         prefill = self._sent_prompt(server, endpoint, 1, HANDOFF)
         decode = self._sent_prompt(server, endpoint, 4096, HANDOFF)
 
-        assert prefill == decode == f"<{MAX_CTX - MAX_CTX // 8}>"
+        assert len(prefill) == len(decode) == MAX_CTX - MAX_CTX // 8
 
     @pytest.mark.parametrize("endpoint", ["messages", "responses"])
     def test_without_handoff_truncation_still_follows_max_tokens(
@@ -219,17 +212,28 @@ def test_without_handoff_truncation_still_follows_max_tokens(
     ):
         server.template = "<long prompt>"
 
-        assert self._sent_prompt(server, endpoint, 1, None) == "<long prompt>"
+        # max_tokens=1 leaves room for the whole prompt; 4096 does not.
+        assert len(self._sent_prompt(server, endpoint, 1, None)) == LONG_PROMPT_TOKENS
         assert (
-            self._sent_prompt(server, endpoint, 4096, None)
-            == f"<{MAX_CTX - MAX_CTX // 8}>"
+            len(self._sent_prompt(server, endpoint, 4096, None))
+            == MAX_CTX - MAX_CTX // 8
         )
 
+    @pytest.mark.parametrize("endpoint", ["messages", "responses"])
+    def test_input_tokens_are_the_engine_count_of_what_it_ran(self, server, endpoint):
+        server.template = "<long prompt>"
+        if endpoint == "messages":
+            usage = _messages(max_tokens=4096)["usage"]
+        else:
+            body = {"model": "m", "input": "hi", "max_output_tokens": 4096}
+            usage = _responses(body).json()["usage"]
+        assert usage["input_tokens"] == MAX_CTX - MAX_CTX // 8
+
 
 class TestStructuredResponsesCutFirst:
     """A structured /v1/responses request builds its reasoning channel into the
-    sampling params before the engine call, so an over-long prompt is cut
-    first and the channel is seeded from the kept text."""
+    sampling params before the engine call; both are seeded from the text the
+    engine runs."""
 
     def test_over_the_cap_the_engine_gets_the_kept_ids(self, server, monkeypatch):
         server.template = "<long prompt>"
@@ -259,10 +263,10 @@ async def configure(_sampling, schema, channel, **_kw):
         )
 
         assert response.status_code == 200, response.text
-        sent, kwargs = server.generate[0]
-        kept = kwargs["prompt_cap"].tokens
+        sent, _kwargs = server.generate[0]
+        kept = MAX_CTX - MAX_CTX // 8
         assert configured == ['{"type": "object"}']
-        # The engine is handed the kept ids, already under the cap.
+        # The engine is handed the kept ids, and does not tokenize again.
         assert sent == [0] * kept
         # The channel was seeded from the cut text, before the engine call.
         assert seeds[0] == f"<{kept}>"
diff --git a/tests/test_distributed_store.py b/tests/test_distributed_store.py
index 6910fd527e..a82b7c3365 100644
--- a/tests/test_distributed_store.py
+++ b/tests/test_distributed_store.py
@@ -31,7 +31,11 @@ def manager(monkeypatch):
         monkeypatch.delenv(name, raising=False)
     manager = CoreManager.__new__(CoreManager)
     manager._init_shared_state(
-        SimpleNamespace(dp_load_balance="round_robin"),
+        SimpleNamespace(
+            dp_load_balance="round_robin",
+            keep_finished_outputs=False,
+            asyncio_mode=False,
+        ),
         label="test",
         local_engine_count=0,
     )
diff --git a/tests/test_dp_backpressure.py b/tests/test_dp_backpressure.py
index 8abca2f278..86f25da871 100644
--- a/tests/test_dp_backpressure.py
+++ b/tests/test_dp_backpressure.py
@@ -121,7 +121,12 @@ def finish_one_on_rank(rank):
 def _init_with_strategy(strategy):
     mgr = CoreManager.__new__(CoreManager)
     mgr._init_shared_state(
-        SimpleNamespace(dp_load_balance=strategy, max_num_seqs=1536),
+        SimpleNamespace(
+            dp_load_balance=strategy,
+            max_num_seqs=1536,
+            keep_finished_outputs=False,
+            asyncio_mode=False,
+        ),
         label="test",
         local_engine_count=2,
     )
@@ -309,3 +314,23 @@ def release_loop():
     assert pending == []
     assert sorted(sent) == sorted(seq.id for seq in seqs)
     assert len(sent) == len(set(sent))
+
+
+def test_a_stalled_run_ends_pending_seqs_that_wait_on_a_callback():
+    # The stalled-run error goes to outputs_queue, which a callback-driven
+    # client never reads: such a pending seq is ended through its callback.
+    mgr = make_mgr(n_ranks=2, depth=1)
+    mgr.outputs_queue = queue.Queue()
+    delivered = {}
+    seqs = [Seq() for _ in range(4)]
+    seqs[-1].stream_callback = lambda out: delivered.update(
+        {out.request_id: out.finish_reason}
+    )
+    mgr.add_request(seqs)
+    mgr.fail_ranks.update({0, 1})
+    for sid in list(mgr._seq_load):
+        mgr._release_seq_load(sid)
+    mgr._bp_dispatch_or_refill(raise_on_error=False)
+    assert delivered == {seqs[-1].id: "aborted"}
+    assert all(seq.id != seqs[-1].id for seq in mgr._bp_pending)
+    assert isinstance(mgr.outputs_queue.get_nowait(), RuntimeError)
diff --git a/tests/test_engine_core_mgr_shared_state.py b/tests/test_engine_core_mgr_shared_state.py
index 5854981b26..2cf00bfe1f 100644
--- a/tests/test_engine_core_mgr_shared_state.py
+++ b/tests/test_engine_core_mgr_shared_state.py
@@ -76,7 +76,8 @@ def _bare_manager(cls, **kwargs):
             dp_load_balance="round_robin",
             tp_world_size=1,
             prefill_context_parallel_size=1,
-            keep_finished_outputs=kwargs.get("keep_finished_outputs", True),
+            keep_finished_outputs=kwargs.get("keep_finished_outputs", False),
+            asyncio_mode=kwargs.get("asyncio_mode", False),
         ),
         label=cls.__name__,
         local_engine_count=kwargs.get("local_engine_count", 2),
@@ -132,11 +133,13 @@ def test_shutdown_frame_marks_unhealthy_before_cleanup(self):
             sender.close(linger=0)
 
     def test_finished_sequences_are_queued_only_if_the_config_keeps_them(self):
-        # Online front ends build the engine with keep_finished_outputs=False
-        # and take outputs through the callbacks; otherwise every finished
-        # Sequence would stay queued for the life of the process.
-        for keep in (True, False):
-            mgr = _bare_manager(CoreManager, keep_finished_outputs=keep)
+        # Off by default: online front ends take every output through the
+        # callbacks, and a queued Sequence would stay for the process's life.
+        # asyncio_mode reads the queue, so it keeps them too.
+        for keep, asyncio_mode in ((True, False), (False, True), (False, False)):
+            mgr = _bare_manager(
+                CoreManager, keep_finished_outputs=keep, asyncio_mode=asyncio_mode
+            )
             self.addCleanup(mgr.ctx.term)
             mgr._deliver_terminal_output = lambda seq: False
             mgr._release_seq_load = lambda seq_id: None
@@ -154,7 +157,9 @@ def test_finished_sequences_are_queued_only_if_the_config_keeps_them(self):
                 sender.send(pickle.dumps((EngineCoreRequestType.SHUTDOWN, None)))
                 thread.join(timeout=3)
                 self.assertFalse(thread.is_alive())
-                self.assertEqual(mgr.outputs_queue.qsize(), 1 if keep else 0)
+                self.assertEqual(
+                    mgr.outputs_queue.qsize(), 1 if keep or asyncio_mode else 0
+                )
             finally:
                 sender.close(linger=0)
 
@@ -217,54 +222,3 @@ def test_subclass_does_not_override_the_shared_initialiser(self):
 
 if __name__ == "__main__":
     unittest.main()
-
-
-def test_create_engine_hands_keep_finished_outputs_to_the_config(monkeypatch):
-    # Decided at construction: create_engine hands the choice to Config, and
-    # the default keeps the offline / asyncio behaviour.
-    import dataclasses
-
-    from atom.config import Config
-    from atom.model_engine import arg_utils
-
-    default = {f.name: f.default for f in dataclasses.fields(Config)}
-    assert default["keep_finished_outputs"] is True
-    seen = {}
-
-    def fake_engine(model, tokenizer=None, **kwargs):
-        seen.update(kwargs)
-
-    monkeypatch.setattr(arg_utils, "LLMEngine", fake_engine)
-    monkeypatch.setattr(arg_utils.EngineArgs, "_get_engine_kwargs", lambda self: {})
-    args = arg_utils.EngineArgs.__new__(arg_utils.EngineArgs)
-    args.model = "m"
-    args.create_engine(keep_finished_outputs=False)
-    assert seen == {"keep_finished_outputs": False}
-    args.create_engine()
-    assert seen == {"keep_finished_outputs": True}
-
-
-def test_atomesh_front_end_builds_the_engine_without_kept_outputs(monkeypatch):
-    # An online front end: every output reaches its request through the
-    # callbacks, so it must not ask the engine to keep finished sequences.
-    from atom.entrypoints.atomesh import server as atomesh_server
-    from atom.model_engine import arg_utils, llm_engine
-
-    seen = {}
-
-    class FakeArgs:
-        @classmethod
-        def from_cli_args(cls, args):
-            return cls()
-
-        def create_engine(self, **kwargs):
-            seen.update(kwargs)
-            return "engine"
-
-    monkeypatch.setattr(arg_utils, "EngineArgs", FakeArgs)
-    monkeypatch.setattr(llm_engine, "load_tokenizer", lambda *a: "tok")
-    engine, _ = atomesh_server.initialize_engine(
-        SimpleNamespace(model="m", trust_remote_code=False)
-    )
-    assert engine == "engine"
-    assert seen == {"tokenizer": "tok", "keep_finished_outputs": False}
diff --git a/tests/test_llm_engine_add_request_fanout.py b/tests/test_llm_engine_add_request_fanout.py
index cbbb58f66d..8de7db4a7d 100644
--- a/tests/test_llm_engine_add_request_fanout.py
+++ b/tests/test_llm_engine_add_request_fanout.py
@@ -167,3 +167,19 @@ def test_preprocess_still_refuses_n_greater_than_one():
         io.preprocess("hello", SamplingParams(n=2))
 
     assert io.preprocess("hello", SamplingParams(n=1)).choice_index == 0
+
+
+def test_offline_add_turns_on_the_finished_outputs_queue_before_submitting():
+    # `step` / `generate` read outputs_queue, so the offline entry point turns
+    # it on (Config.keep_finished_outputs defaults to off for online servers)
+    # before any of its sequences can finish.
+    engine, _io, _submitted = _engine()
+    seen = []
+    engine.core_mgr.keep_finished_outputs = False
+    engine.core_mgr.add_request = lambda seqs: seen.append(
+        engine.core_mgr.keep_finished_outputs
+    )
+
+    LLMEngine.add_request(engine, ["hello"], SamplingParams())
+
+    assert seen == [True]

From 5cfc14eeb532c0071cc7603beede6beab2f6a048 Mon Sep 17 00:00:00 2001
From: Yi <hyi@amd.com>
Date: Tue, 3 Nov 2026 23:40:03 +0800
Subject: [PATCH 11/11] review: widen prompt annotations to PrefixEncoding

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 atom/entrypoints/openai/api_server.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/atom/entrypoints/openai/api_server.py b/atom/entrypoints/openai/api_server.py
index e3854179ec..d988ba65a1 100644
--- a/atom/entrypoints/openai/api_server.py
+++ b/atom/entrypoints/openai/api_server.py
@@ -854,7 +854,7 @@ async def _submit_engine_requests(seqs):
 
 
 async def generate_async(
-    prompt_or_tokens: str | list[int],
+    prompt_or_tokens: str | list[int] | PrefixEncoding,
     sampling_params: SamplingParams,
     request_id: str,
     kv_transfer_params: dict[str, Any] | None = None,
@@ -1315,7 +1315,7 @@ def _prepare_prompt(
     preserve_tool_call_arguments: bool,
     max_tokens: int,
     kv_transfer_params: dict[str, Any] | None,
-):
+) -> tuple[PrefixEncoding | list[int], str]:
     """Render, tokenize the engine's way and cut to the context, in one
     executor step for ``/v1/messages`` and ``/v1/responses``.
 
@@ -1359,7 +1359,7 @@ def _with_kv_transfer_params(response: dict, final_output: dict) -> dict:
 
 
 async def setup_streaming_request(
-    prompt_or_tokens: str | list[int],
+    prompt_or_tokens: str | list[int] | PrefixEncoding,
     sampling_params: SamplingParams,
     request_id: str,
     kv_transfer_params: dict[str, Any] | None = None,