feat(observability): persist forward pass metrics traces

components/src/dynamo/common/configuration/groups/runtime_args.py

@@ -4,6 +4,8 @@
 """Dynamo runtime configuration ArgGroup."""
 
 import argparse
+import logging
+import os
 from typing import List, Optional
 
 from dynamo._core import get_reasoning_parser_names, get_tool_parser_names
@@ -13,6 +15,10 @@
 from dynamo.common.utils.namespace import get_worker_namespace
 from dynamo.common.utils.output_modalities import OutputModality
 
+logger = logging.getLogger(__name__)
+_FPM_TRACE_VALUES = {"1", "0", "true", "false", "on", "off", "yes", "no"}
+_fpm_trace_invalid_warning_emitted = False
+
 
 class DynamoRuntimeConfig(ConfigBase):
     """Configuration for Dynamo runtime (common across all backends)."""
@@ -22,6 +28,7 @@ class DynamoRuntimeConfig(ConfigBase):
     discovery_backend: str
     request_plane: str
     event_plane: Optional[str] = None
+    fpm_trace: bool = False
     connector: list[str]
     enable_local_indexer: bool
     durable_kv_events: bool
@@ -52,6 +59,29 @@ class DynamoRuntimeConfig(ConfigBase):
     def validate(self) -> None:
         self.namespace = get_worker_namespace(self.namespace)
 
+        # The Rust FPM sink reads this setting from the process environment.
+        # Canonicalize the resolved CLI/env value before the runtime or backend
+        # child processes are created so --fpm-trace and --no-fpm-trace apply to
+        # both the Python instrumentation and the Rust persistence layer.
+        if self.fpm_trace or "DYN_FPM_TRACE" in os.environ:
+            raw_fpm_trace = os.environ.get("DYN_FPM_TRACE")
+            if (
+                raw_fpm_trace is not None
+                and raw_fpm_trace.strip().lower() not in _FPM_TRACE_VALUES
+                and not self.fpm_trace
+                and "DYN_FORWARDPASS_METRIC_PORT" not in os.environ
+            ):
+                global _fpm_trace_invalid_warning_emitted
+                if not _fpm_trace_invalid_warning_emitted:
+                    _fpm_trace_invalid_warning_emitted = True
+                    logger.warning(
+                        "Invalid DYN_FPM_TRACE value %r; expected one of 1/0, "
+                        "true/false, on/off, or yes/no. FPM tracing is disabled "
+                        "for this worker.",
+                        raw_fpm_trace,
+                    )
+            os.environ["DYN_FPM_TRACE"] = "1" if self.fpm_trace else "0"
+
         # TODO  get a better way for spot fixes like this.
         self.enable_local_indexer = not self.durable_kv_events
         self._validate_output_modalities()
@@ -126,6 +156,13 @@ def add_arguments(self, parser: argparse.ArgumentParser) -> None:
             "all discovery backends. Set to 'nats' to use a NATS-based event plane.",
             choices=["nats", "zmq"],
         )
+        add_negatable_bool_argument(
+            g,
+            flag_name="--fpm-trace",
+            env_var="DYN_FPM_TRACE",
+            default=False,
+            help="Persist backend forward-pass metrics to rotating gzip JSONL trace files. Also enables the backend FPM instrumentation required to produce those records.",
+        )
         add_argument(
             g,
             flag_name="--connector",

components/src/dynamo/common/configuration/utils.py

@@ -42,7 +42,7 @@ def env_or_default(
     target_type = value_type if value_type is not None else type(default)
 
     if target_type is bool:
-        return value.lower() in ("true", "1", "yes", "on")  # type: ignore
+        return value.strip().lower() in ("true", "1", "yes", "on")  # type: ignore
     if target_type is int:
         return int(value)  # type: ignore
     if target_type is float:

components/src/dynamo/common/tests/configuration/test_runtime_args.py

@@ -0,0 +1,113 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Tests for shared Dynamo runtime arguments."""
+
+import argparse
+import logging
+import os
+
+import pytest
+
+import dynamo.common.configuration.groups.runtime_args as runtime_args
+from dynamo.common.configuration.groups.runtime_args import (
+    DynamoRuntimeArgGroup,
+    DynamoRuntimeConfig,
+)
+
+pytestmark = [
+    pytest.mark.unit,
+    pytest.mark.gpu_0,
+    pytest.mark.pre_merge,
+]
+
+
+def _parse_runtime_args(argv: list[str]) -> tuple[DynamoRuntimeConfig, str]:
+    parser = argparse.ArgumentParser()
+    DynamoRuntimeArgGroup().add_arguments(parser)
+    args = parser.parse_args(argv)
+    config = DynamoRuntimeConfig.from_cli_args(args)
+    config.validate()
+    return config, parser.format_help()
+
+
+def test_fpm_trace_defaults_disabled(monkeypatch):
+    monkeypatch.delenv("DYN_FPM_TRACE", raising=False)
+
+    config, _ = _parse_runtime_args([])
+
+    assert config.fpm_trace is False
+    assert "DYN_FPM_TRACE" not in os.environ
+
+
+def test_fpm_trace_env_enables_and_is_canonicalized(monkeypatch):
+    monkeypatch.setenv("DYN_FPM_TRACE", "on")
+
+    config, _ = _parse_runtime_args([])
+
+    assert config.fpm_trace is True
+    assert os.environ["DYN_FPM_TRACE"] == "1"
+
+
+def test_fpm_trace_env_is_trimmed(monkeypatch):
+    monkeypatch.setenv("DYN_FPM_TRACE", " true ")
+
+    config, _ = _parse_runtime_args([])
+
+    assert config.fpm_trace is True
+    assert os.environ["DYN_FPM_TRACE"] == "1"
+
+
+def test_invalid_fpm_trace_warns_once_and_is_disabled(monkeypatch, caplog):
+    monkeypatch.setenv("DYN_FPM_TRACE", "sometimes")
+    monkeypatch.setattr(runtime_args, "_fpm_trace_invalid_warning_emitted", False)
+
+    with caplog.at_level(logging.WARNING, logger=runtime_args.__name__):
+        config, _ = _parse_runtime_args([])
+        monkeypatch.setenv("DYN_FPM_TRACE", "still-invalid")
+        _parse_runtime_args([])
+
+    assert config.fpm_trace is False
+    assert os.environ["DYN_FPM_TRACE"] == "0"
+    assert caplog.text.count("Invalid DYN_FPM_TRACE value") == 1
+
+
+def test_explicit_fpm_port_preserves_precedence_over_invalid_trace(monkeypatch, caplog):
+    monkeypatch.setenv("DYN_FORWARDPASS_METRIC_PORT", "23456")
+    monkeypatch.setenv("DYN_FPM_TRACE", "sometimes")
+    monkeypatch.setattr(runtime_args, "_fpm_trace_invalid_warning_emitted", False)
+
+    with caplog.at_level(logging.WARNING, logger=runtime_args.__name__):
+        config, _ = _parse_runtime_args([])
+
+    assert config.fpm_trace is False
+    assert os.environ["DYN_FPM_TRACE"] == "0"
+    assert "Invalid DYN_FPM_TRACE value" not in caplog.text
+
+
+def test_fpm_trace_cli_enables_and_is_exported(monkeypatch):
+    monkeypatch.delenv("DYN_FPM_TRACE", raising=False)
+
+    config, _ = _parse_runtime_args(["--fpm-trace"])
+
+    assert config.fpm_trace is True
+    assert os.environ["DYN_FPM_TRACE"] == "1"
+
+
+def test_no_fpm_trace_cli_overrides_enabled_env(monkeypatch):
+    monkeypatch.setenv("DYN_FPM_TRACE", "true")
+
+    config, _ = _parse_runtime_args(["--no-fpm-trace"])
+
+    assert config.fpm_trace is False
+    assert os.environ["DYN_FPM_TRACE"] == "0"
+
+
+def test_fpm_trace_help_lists_flag_and_env(monkeypatch):
+    monkeypatch.delenv("DYN_FPM_TRACE", raising=False)
+
+    _, help_text = _parse_runtime_args([])
+
+    assert "--fpm-trace" in help_text
+    assert "--no-fpm-trace" in help_text
+    assert "DYN_FPM_TRACE" in help_text

components/src/dynamo/common/tests/configuration/test_utils.py

@@ -39,7 +39,7 @@ def test_returns_env_when_set(self, monkeypatch):
 
     def test_bool_conversion_true(self, monkeypatch):
         """Test bool conversion for true values."""
-        test_cases = ["true", "True", "1", "yes", "YES", "on", "ON"]
+        test_cases = ["true", "True", "1", "yes", "YES", "on", "ON", " true "]
 
         for value in test_cases:
             monkeypatch.setenv("TEST_BOOL", value)
@@ -48,7 +48,7 @@ def test_bool_conversion_true(self, monkeypatch):
 
     def test_bool_conversion_false(self, monkeypatch):
         """Test bool conversion for false values."""
-        test_cases = ["false", "False", "0", "no", "NO", "off", "OFF"]
+        test_cases = ["false", "False", "0", "no", "NO", "off", "OFF", " off "]
 
         for value in test_cases:
             monkeypatch.setenv("TEST_BOOL", value)

components/src/dynamo/common/utils/tests/test_env.py

@@ -30,7 +30,7 @@ def test_truthy_values(self, monkeypatch, value):
         monkeypatch.setenv("FOO", value)
         assert env_bool("FOO") is True
 
-    @pytest.mark.parametrize("value", ["false", "0", "no", "anything"])
+    @pytest.mark.parametrize("value", ["false", "0", "no", "on", "off", "anything"])
     def test_falsy_values(self, monkeypatch, value):
         monkeypatch.setenv("FOO", value)
         assert env_bool("FOO") is False

components/src/dynamo/sglang/args.py

@@ -66,6 +66,48 @@ def _set_serving_strategy(self):
             return DisaggregationMode.AGGREGATED
 
 
+def _unsupported_fpm_trace_role(dynamo_config: DynamoConfig) -> Optional[str]:
+    """Return the worker role when the selected path does not create an FPM relay."""
+    if is_snapshot_enabled():
+        return "snapshot"
+    if dynamo_config.embedding_worker:
+        return "embedding"
+    if (
+        dynamo_config.multimodal_encode_worker
+        or dynamo_config.multimodal_worker
+        or dynamo_config.dedicated_mm_encoder
+    ):
+        return "dedicated multimodal"
+    if dynamo_config.image_diffusion_worker:
+        return "image diffusion"
+    if dynamo_config.video_generation_worker:
+        return "video generation"
+    return None
+
+
+def _forward_pass_metrics_source(
+    dynamo_config: DynamoConfig, *, fpm_trace_relay_supported: bool = True
+) -> Optional[str]:
+    """Resolve the FPM opt-in source while preserving the legacy port switch."""
+    if os.environ.get("DYN_FORWARDPASS_METRIC_PORT"):
+        return "DYN_FORWARDPASS_METRIC_PORT"
+    if not dynamo_config.fpm_trace:
+        return None
+
+    unsupported_role = _unsupported_fpm_trace_role(dynamo_config)
+    if unsupported_role is None and not fpm_trace_relay_supported:
+        unsupported_role = "unified backend"
+    if unsupported_role is None:
+        return "--fpm-trace/DYN_FPM_TRACE"
+
+    logging.warning(
+        "--fpm-trace/DYN_FPM_TRACE is enabled, but SGLang %s workers do not create a Dynamo "
+        "FPM relay. Trace-based FPM activation is disabled for this worker.",
+        unsupported_role,
+    )
+    return None
+
+
 def use_modelexpress_remote_instance(args: Any) -> bool:
     return (
         getattr(args, "load_format", None) == "remote_instance"
@@ -264,12 +306,16 @@ def _dump_disagg_config_section(disagg_config: dict[str, Any]) -> str:
     return temp_path
 
 
-async def parse_args(args: list[str]) -> Config:
+async def parse_args(
+    args: list[str], *, fpm_trace_relay_supported: bool = True
+) -> Config:
     """Parse CLI arguments and return combined configuration.
     Download the model if necessary.
 
     Args:
         args: Command-line argument strings.
+        fpm_trace_relay_supported: Whether this entry point constructs the
+            Dynamo relay required for trace-based FPM activation.
 
     Returns:
         Config object with server_args and dynamo_args.
@@ -526,11 +572,13 @@ async def parse_args(args: list[str]) -> Config:
     )
 
     # Enable forward pass metrics from dynamo env var if configured
-    if os.environ.get("DYN_FORWARDPASS_METRIC_PORT") and not getattr(
-        server_args, "enable_forward_pass_metrics", False
-    ):
+    fpm_source = _forward_pass_metrics_source(
+        dynamo_config,
+        fpm_trace_relay_supported=fpm_trace_relay_supported,
+    )
+    if fpm_source and not getattr(server_args, "enable_forward_pass_metrics", False):
         server_args.enable_forward_pass_metrics = True
-        logging.info("Enabled forward_pass_metrics from DYN_FORWARDPASS_METRIC_PORT")
+        logging.info("Enabled forward_pass_metrics from %s", fpm_source)
 
     # Auto-detect diffusion worker mode if dllm_algorithm
     diffusion_worker = server_args.dllm_algorithm is not None

components/src/dynamo/sglang/llm_engine.py

@@ -149,7 +149,10 @@ def __init__(self, server_args, dynamo_args, serving_mode: DisaggregationMode):
     async def from_args(
         cls, argv: list[str] | None = None
     ) -> tuple[SglangLLMEngine, WorkerConfig]:
-        config = await parse_args(argv if argv is not None else sys.argv[1:])
+        config = await parse_args(
+            argv if argv is not None else sys.argv[1:],
+            fpm_trace_relay_supported=False,
+        )
         server_args = config.server_args
         dynamo_args = config.dynamo_args