feat(observability): add exclusive frontend TTFT breakdown

components/src/dynamo/frontend/prepost.py

@@ -38,6 +38,7 @@
     should_attempt_structural_tag,
 )
 from .thinking import apply_default_thinking_mode_to_template_kwargs
+from .ttft import ttft_stage
 from .utils import legacy_guided_decoding
 
 get_model_structural_tag: Any
@@ -747,6 +748,7 @@ async def preprocess_chat_request(
     structural_tag_scope: str = "auto",
     structural_tag_schema: str = "auto",
     structural_tag_excludes_reasoning: bool = False,
+    context: Any | None = None,
 ) -> PreprocessResult:
     validated_request = _validate_chat_completion_request(request)
     assistant_guided_decoding = _build_assistant_guided_decoding(validated_request)
@@ -803,17 +805,18 @@ async def preprocess_chat_request(
     # Render first: whether generation starts inside a reasoning block depends
     # on the rendered prompt (e.g. a template that ends the generation prompt
     # with an open think tag), and the tool-call grammar must match it.
-    _, engine_prompt = await renderer.render_messages_async(messages, chat_params)
+    with ttft_stage(context, "template_tokenize"):
+        _, engine_prompt = await renderer.render_messages_async(messages, chat_params)
 
-    if "prompt_token_ids" in engine_prompt:
-        tokens = list(engine_prompt["prompt_token_ids"])
-    else:
-        async_tokenizer = _get_async_tokenizer(tokenizer)
-        encoded = await async_tokenizer(
-            engine_prompt["prompt"],
-            add_special_tokens=request_for_sampling.add_special_tokens,
-        )
-        tokens = list(encoded.input_ids)
+        if "prompt_token_ids" in engine_prompt:
+            tokens = list(engine_prompt["prompt_token_ids"])
+        else:
+            async_tokenizer = _get_async_tokenizer(tokenizer)
+            encoded = await async_tokenizer(
+                engine_prompt["prompt"],
+                add_special_tokens=request_for_sampling.add_special_tokens,
+            )
+            tokens = list(encoded.input_ids)
 
     tool_guided_decoding = build_tool_call_guided_decoding(
         request_for_sampling,

components/src/dynamo/frontend/sglang_prepost.py

@@ -39,6 +39,7 @@
     should_attempt_structural_tag,
 )
 from .thinking import apply_default_thinking_mode_to_template_kwargs
+from .ttft import ttft_stage
 from .utils import PreprocessError, legacy_guided_decoding, random_call_id
 
 logger = logging.getLogger(__name__)
@@ -509,6 +510,7 @@ def _render_deepseek_v4_prompt_token_ids(
     messages: list[dict[str, Any]],
     tokenizer,
     template_tools: list[dict[str, Any]] | None,
+    context: Any | None = None,
 ) -> list[int]:
     """Render DeepSeek-V4 prompt token ids via SGLang's ``encoding_dsv4``.
 
@@ -568,12 +570,13 @@ def _render_deepseek_v4_prompt_token_ids(
     if reasoning_effort not in ("max", "high", None):
         reasoning_effort = None
 
-    prompt = encode_messages(
-        encoding_messages,
-        thinking_mode=thinking_mode,
-        reasoning_effort=reasoning_effort,
-    )
-    return _normalize_prompt_token_ids(tokenizer.encode(prompt))
+    with ttft_stage(context, "template_tokenize"):
+        prompt = encode_messages(
+            encoding_messages,
+            thinking_mode=thinking_mode,
+            reasoning_effort=reasoning_effort,
+        )
+        return _normalize_prompt_token_ids(tokenizer.encode(prompt))
 
 
 @lru_cache(maxsize=64)
@@ -874,6 +877,7 @@ def preprocess_chat_request(
     structural_tag_mode: str = "off",
     structural_tag_scope: str = "auto",
     structural_tag_schema: str = "auto",
+    context: Any | None = None,
 ) -> SglangPreprocessResult:
     """Preprocess a chat request using SGLang tokenizer and parser APIs.
 
@@ -945,6 +949,7 @@ def preprocess_chat_request(
             messages=messages,
             tokenizer=tokenizer,
             template_tools=template_tools,
+            context=context,
         )
     else:
         # Build template kwargs -- single call for rendering + tokenization
@@ -968,14 +973,15 @@ def preprocess_chat_request(
 
         template_messages = _normalize_messages_for_template(messages, tokenizer)
 
-        try:
-            rendered = tokenizer.apply_chat_template(
-                template_messages, **template_kwargs
-            )
-        except (TemplateError, TypeError) as exc:
-            # Jinja filters such as tojson can raise TypeError for invalid inputs.
-            raise PreprocessError(str(exc)) from exc
-        prompt_token_ids = _normalize_prompt_token_ids(rendered)
+        with ttft_stage(context, "template_tokenize"):
+            try:
+                rendered = tokenizer.apply_chat_template(
+                    template_messages, **template_kwargs
+                )
+            except (TemplateError, TypeError) as exc:
+                # Jinja filters such as tojson can raise TypeError for invalid inputs.
+                raise PreprocessError(str(exc)) from exc
+            prompt_token_ids = _normalize_prompt_token_ids(rendered)
 
     response_format_guided_decoding = build_response_format_guided_decoding(request)
     tool_call_guided_decoding = build_tool_call_guided_decoding(

components/src/dynamo/frontend/sglang_processor.py

@@ -14,6 +14,7 @@
 from collections.abc import AsyncGenerator
 from concurrent.futures import ProcessPoolExecutor
 from concurrent.futures import wait as _futures_wait
+from contextlib import contextmanager
 from dataclasses import dataclass
 from typing import Any, cast
 
@@ -42,6 +43,7 @@
 )
 from .structural_tag_policy import runtime_structural_tag_options
 from .thinking import runtime_default_thinking_mode
+from .ttft import ttft_stage
 from .utils import (
     PreprocessError,
     as_error_envelope,
@@ -316,6 +318,24 @@ class SglangPreprocessWorkerResult:
     # a reasoning parser in that case so the pool path matches the inline
     # path byte-for-byte.
     effective_reasoning_parser_name: str | None = None
+    # Same-host monotonic timestamps cross the process boundary without
+    # serializing the native request context.
+    template_tokenize_intervals_ns: list[tuple[int, int]] | None = None
+
+
+class _WorkerPreprocessTiming:
+    def __init__(self) -> None:
+        self.intervals_ns: list[tuple[int, int]] = []
+
+    @contextmanager
+    def start_ttft_stage(self, stage: str):
+        if stage != "template_tokenize":
+            raise ValueError(f"Unsupported TTFT stage: {stage}")
+        start_ns = time.monotonic_ns()
+        try:
+            yield
+        finally:
+            self.intervals_ns.append((start_ns, time.monotonic_ns()))
 
 
 def _init_worker(
@@ -353,8 +373,10 @@ def _preprocess_worker(
     request: dict[str, Any],
     model_name: str,
     eos_token_ids: list[int] | None,
+    record_ttft_timing: bool = False,
 ) -> SglangPreprocessWorkerResult:
     """Preprocess a request in a worker process and return a picklable result."""
+    timing = _WorkerPreprocessTiming() if record_ttft_timing else None
     pre = preprocess_chat_request(
         request,
         tokenizer=_w_tokenizer,
@@ -366,6 +388,7 @@ def _preprocess_worker(
         structural_tag_mode=_w_structural_tag_mode,
         structural_tag_scope=_w_structural_tag_scope,
         structural_tag_schema=_w_structural_tag_schema,
+        context=timing,
     )
 
     n = request.get("n", 1)
@@ -397,6 +420,7 @@ def _preprocess_worker(
         force_reasoning=pre.force_reasoning,
         effective_reasoning_parser_name=effective_reasoning_parser_name,
         named_zero_arg_tool=pre.named_zero_arg_tool,
+        template_tokenize_intervals_ns=timing.intervals_ns if timing else None,
     )
 
 
@@ -618,18 +642,20 @@ async def _generator_inner(
             if self.debug_perf:
                 t0 = time.monotonic()
 
-            pre = preprocess_chat_request(
-                request,
-                tokenizer=self.tokenizer,
-                tool_call_parser_name=self.tool_call_parser_name,
-                reasoning_parser_name=self.reasoning_parser_name,
-                exclude_tools_when_tool_choice_none=self.exclude_tools_when_tool_choice_none,
-                template_force_reasoning=self.template_force_reasoning,
-                default_thinking_mode=self.default_thinking_mode,
-                structural_tag_mode=self.structural_tag_mode,
-                structural_tag_scope=self.structural_tag_scope,
-                structural_tag_schema=self.structural_tag_schema,
-            )
+            with ttft_stage(context, "python_preprocess"):
+                pre = preprocess_chat_request(
+                    request,
+                    tokenizer=self.tokenizer,
+                    tool_call_parser_name=self.tool_call_parser_name,
+                    reasoning_parser_name=self.reasoning_parser_name,
+                    exclude_tools_when_tool_choice_none=self.exclude_tools_when_tool_choice_none,
+                    template_force_reasoning=self.template_force_reasoning,
+                    default_thinking_mode=self.default_thinking_mode,
+                    structural_tag_mode=self.structural_tag_mode,
+                    structural_tag_scope=self.structural_tag_scope,
+                    structural_tag_schema=self.structural_tag_schema,
+                    context=context,
+                )
 
             if self.debug_perf:
                 t1 = time.monotonic()
@@ -702,17 +728,36 @@ async def _generator_inner_pool(
         # --- Phase 1: Preprocess (semaphore held) ---
         assert self._worker_semaphore is not None
         assert self.preprocess_pool is not None
+        record_stage = getattr(context, "record_ttft_stage_ago", None)
         try:
-            async with self._worker_semaphore:
-                future = self.preprocess_pool.submit(
-                    _preprocess_worker,
-                    request,
-                    request["model"],
-                    self.eos_token_ids,
-                )
-                preproc_result: SglangPreprocessWorkerResult = (
-                    await asyncio.wrap_future(future)
-                )
+            with ttft_stage(context, "python_preprocess"):
+                with ttft_stage(context, "preprocess_queue"):
+                    await self._worker_semaphore.acquire()
+                try:
+                    future = self.preprocess_pool.submit(
+                        _preprocess_worker,
+                        request,
+                        request["model"],
+                        self.eos_token_ids,
+                        record_stage is not None,
+                    )
+                    preproc_result: SglangPreprocessWorkerResult = (
+                        await asyncio.wrap_future(future)
+                    )
+                finally:
+                    self._worker_semaphore.release()
+                if record_stage is not None:
+                    for start_ns, end_ns in (
+                        preproc_result.template_tokenize_intervals_ns or ()
+                    ):
+                        # monotonic_ns is shared across processes on this host.
+                        # Re-anchor immediately before crossing into Rust.
+                        now_ns = time.monotonic_ns()
+                        record_stage(
+                            "template_tokenize",
+                            (now_ns - start_ns) / 1_000_000_000,
+                            (now_ns - end_ns) / 1_000_000_000,
+                        )
         except InvalidArgument:
             raise
         except PreprocessError as exc: