feat(workers): Dynamo conversation-aware routing via nvext.session_control

docs/cli-options.md

@@ -324,6 +324,22 @@ Content type for request body serialization. By default, requests are sent as 'a
 
 HTTP header name used to carry the per-session affinity identifier. When set, replaces the default `X-Correlation-ID` header with the provided name (e.g., `--session-header X-Session-ID`).
 
+#### `--use-dynamo-conv-aware-routing`, `--use-dynamo-session-control`
+
+Emit Dynamo nvext.session_control in OpenAI-compatible request bodies so Dynamo can bind all turns from the same replayed conversation lineage to the same backend worker. This is only intended for Dynamo frontends that implement session_control.
+<br/>_Flag (no value required)_
+
+#### `--use-legacy-dynamo-session-control`
+
+Emit the legacy Dynamo nvext.session_control lifecycle that released Dynamo (v1.2.x) understands: action 'open' on the first turn, session_id only on intermediate turns, and action 'close' on the final turn. Use this when the target Dynamo predates the 'bind' action (added in v1.3.0-dev); otherwise 'bind' is rejected with an HTTP 400. Requires --use-dynamo-conv-aware-routing, and the Dynamo deployment must expose a worker session_control endpoint for 'open' to take effect.
+<br/>_Flag (no value required)_
+
+#### `--dynamo-session-timeout-seconds` `<int>`
+
+Dynamo nvext.session_control timeout in seconds when --use-dynamo-conv-aware-routing is enabled.
+<br/>_Constraints: ≥ 1_
+<br/>_Default: `300`_
+
 ### Tokenizer
 
 #### `--tokenizer` `<str>`
@@ -1721,6 +1737,22 @@ Content type for request body serialization. By default, requests are sent as 'a
 
 HTTP header name used to carry the per-session affinity identifier. When set, replaces the default `X-Correlation-ID` header with the provided name (e.g., `--session-header X-Session-ID`).
 
+#### `--use-dynamo-conv-aware-routing`, `--use-dynamo-session-control`
+
+Emit Dynamo nvext.session_control in OpenAI-compatible request bodies so Dynamo can bind all turns from the same replayed conversation lineage to the same backend worker. This is only intended for Dynamo frontends that implement session_control.
+<br/>_Flag (no value required)_
+
+#### `--use-legacy-dynamo-session-control`
+
+Emit the legacy Dynamo nvext.session_control lifecycle that released Dynamo (v1.2.x) understands: action 'open' on the first turn, session_id only on intermediate turns, and action 'close' on the final turn. Use this when the target Dynamo predates the 'bind' action (added in v1.3.0-dev); otherwise 'bind' is rejected with an HTTP 400. Requires --use-dynamo-conv-aware-routing, and the Dynamo deployment must expose a worker session_control endpoint for 'open' to take effect.
+<br/>_Flag (no value required)_
+
+#### `--dynamo-session-timeout-seconds` `<int>`
+
+Dynamo nvext.session_control timeout in seconds when --use-dynamo-conv-aware-routing is enabled.
+<br/>_Constraints: ≥ 1_
+<br/>_Default: `300`_
+
 ### Tokenizer
 
 #### `--tokenizer` `<str>`

src/aiperf/common/models/model_endpoint_info.py

@@ -139,6 +139,20 @@ def _redact_headers(self, value: list[tuple[str, str]]) -> list[tuple[str, str]]
         description="HTTP header name to use for the per-session affinity identifier. "
         "When set, replaces the default `X-Correlation-ID` header name with this value.",
     )
+    use_dynamo_conv_aware_routing: bool = Field(
+        default=EndpointDefaults.USE_DYNAMO_CONV_AWARE_ROUTING,
+        description="Emit Dynamo nvext.session_control for conversation-aware routing.",
+    )
+    use_legacy_dynamo_session_control: bool = Field(
+        default=EndpointDefaults.USE_LEGACY_DYNAMO_SESSION_CONTROL,
+        description="Emit the v1.2.x-compatible open/close session_control lifecycle "
+        "instead of the 'bind' action (which only exists in Dynamo >= v1.3.0-dev).",
+    )
+    dynamo_session_timeout_seconds: int = Field(
+        default=EndpointDefaults.DYNAMO_SESSION_TIMEOUT_SECONDS,
+        ge=1,
+        description="Timeout in seconds for Dynamo nvext.session_control sessions.",
+    )
     collect_trace_chunks: bool = Field(
         default=False,
         description="Collect per-chunk trace data (timestamps and sizes) for HTTP trace export. "
@@ -213,6 +227,15 @@ def from_run(cls, run: BenchmarkRun) -> ModelEndpointInfo:
                 collect_trace_chunks=False,
                 template=getattr(ep, "template", None),
                 session_header=getattr(ep, "session_header", None),
+                use_dynamo_conv_aware_routing=getattr(
+                    ep, "use_dynamo_conv_aware_routing", False
+                ),
+                use_legacy_dynamo_session_control=getattr(
+                    ep, "use_legacy_dynamo_session_control", False
+                ),
+                dynamo_session_timeout_seconds=getattr(
+                    ep, "dynamo_session_timeout_seconds", 300
+                ),
             ),
             transport=ep.transport,
         )

src/aiperf/config/endpoint.py

@@ -57,6 +57,9 @@ class EndpointDefaults:
     CONNECTION_REUSE_STRATEGY = ConnectionReuseStrategy.POOLED
     DOWNLOAD_VIDEO_CONTENT = False
     REQUEST_CONTENT_TYPE = None
+    USE_DYNAMO_CONV_AWARE_ROUTING = False
+    USE_LEGACY_DYNAMO_SESSION_CONTROL = False
+    DYNAMO_SESSION_TIMEOUT_SECONDS = 300
     # Readiness probe defaults. Timeout 0 disables the probe (the default);
     # any positive value enables it. Interval is only consulted when the
     # probe is enabled but is validated positive so mis-configuration
@@ -317,6 +320,48 @@ def _redact_headers(self, value: dict[str, str]) -> dict[str, str]:
         ),
     ]
 
+    use_dynamo_conv_aware_routing: Annotated[
+        bool,
+        Field(
+            default=EndpointDefaults.USE_DYNAMO_CONV_AWARE_ROUTING,
+            description=(
+                "Emit Dynamo nvext.session_control in OpenAI-compatible request "
+                "bodies so Dynamo can bind all turns from the same replayed "
+                "conversation lineage to the same backend worker. This is only "
+                "intended for Dynamo frontends that implement session_control."
+            ),
+        ),
+    ]
+
+    use_legacy_dynamo_session_control: Annotated[
+        bool,
+        Field(
+            default=EndpointDefaults.USE_LEGACY_DYNAMO_SESSION_CONTROL,
+            description=(
+                "Emit the legacy Dynamo nvext.session_control lifecycle that "
+                "released Dynamo (v1.2.x) understands: action 'open' on the first "
+                "turn, session_id only on intermediate turns, and action 'close' "
+                "on the final turn. Use this when the target Dynamo predates the "
+                "'bind' action (added in v1.3.0-dev); otherwise 'bind' is rejected "
+                "with an HTTP 400. Requires use_dynamo_conv_aware_routing, and "
+                "the Dynamo deployment must expose a worker session_control "
+                "endpoint for 'open' to take effect."
+            ),
+        ),
+    ]
+
+    dynamo_session_timeout_seconds: Annotated[
+        int,
+        Field(
+            default=EndpointDefaults.DYNAMO_SESSION_TIMEOUT_SECONDS,
+            ge=1,
+            description=(
+                "Dynamo nvext.session_control timeout in seconds when "
+                "use_dynamo_conv_aware_routing is enabled."
+            ),
+        ),
+    ]
+
     wait_for_model_timeout: Annotated[
         float,
         Field(
@@ -466,6 +511,23 @@ def _validate_template_required(self) -> Self:
             raise ValueError("template is required when endpoint type is 'template'")
         return self
 
+    @model_validator(mode="after")
+    def validate_dynamo_session_control_coherent(self) -> Self:
+        """Reject legacy Dynamo session control unless conversation-aware routing
+        is enabled, since the legacy flag only selects the wire contract for the
+        session_control that use_dynamo_conv_aware_routing emits.
+        """
+        if (
+            self.use_legacy_dynamo_session_control
+            and not self.use_dynamo_conv_aware_routing
+        ):
+            raise ValueError(
+                "--use-legacy-dynamo-session-control has no effect unless "
+                "--use-dynamo-conv-aware-routing is enabled. Enable conversation-"
+                "aware routing, or drop the legacy flag."
+            )
+        return self
+
     @model_validator(mode="after")
     def _validate_wait_for_model_coherent(self) -> Self:
         """Reject configurations where probe sub-options are set to non-default

src/aiperf/config/flags/_converter_endpoint.py

@@ -72,6 +72,9 @@ def _endpoint_template_fallback(endpoint: dict[str, Any]) -> None:
     "download_video_content": "download_video_content",
     "request_content_type": "request_content_type",
     "session_header": "session_header",
+    "use_dynamo_conv_aware_routing": "use_dynamo_conv_aware_routing",
+    "use_legacy_dynamo_session_control": "use_legacy_dynamo_session_control",
+    "dynamo_session_timeout_seconds": "dynamo_session_timeout_seconds",
 }
 
 

src/aiperf/config/flags/_section_fields.py

@@ -20,6 +20,7 @@
         "connection_reuse_strategy",
         "custom_endpoint",
         "download_video_content",
+        "dynamo_session_timeout_seconds",
         "model_names",
         "model_selection_strategy",
         "request_content_type",
@@ -30,6 +31,8 @@
         "endpoint_type",
         "url_selection_strategy",
         "urls",
+        "use_dynamo_conv_aware_routing",
+        "use_legacy_dynamo_session_control",
         "use_legacy_max_tokens",
         "use_server_token_count",
         "wait_for_model_interval",

src/aiperf/config/flags/cli_config.py

@@ -389,6 +389,60 @@ class CLIConfig(BaseConfig):
         ),
     ] = None
 
+    use_dynamo_conv_aware_routing: Annotated[
+        bool,
+        Field(
+            description=(
+                "Emit Dynamo nvext.session_control in OpenAI-compatible request "
+                "bodies so Dynamo can bind all turns from the same replayed "
+                "conversation lineage to the same backend worker. This is only "
+                "intended for Dynamo frontends that implement session_control."
+            ),
+        ),
+        CLIParameter(
+            name=(
+                "--use-dynamo-conv-aware-routing",
+                "--use-dynamo-session-control",
+            ),
+            group=Groups.ENDPOINT,
+        ),
+    ] = EndpointDefaults.USE_DYNAMO_CONV_AWARE_ROUTING
+
+    use_legacy_dynamo_session_control: Annotated[
+        bool,
+        Field(
+            description=(
+                "Emit the legacy Dynamo nvext.session_control lifecycle that "
+                "released Dynamo (v1.2.x) understands: action 'open' on the first "
+                "turn, session_id only on intermediate turns, and action 'close' "
+                "on the final turn. Use this when the target Dynamo predates the "
+                "'bind' action (added in v1.3.0-dev); otherwise 'bind' is rejected "
+                "with an HTTP 400. Requires --use-dynamo-conv-aware-routing, and "
+                "the Dynamo deployment must expose a worker session_control "
+                "endpoint for 'open' to take effect."
+            ),
+        ),
+        CLIParameter(
+            name=("--use-legacy-dynamo-session-control",),
+            group=Groups.ENDPOINT,
+        ),
+    ] = EndpointDefaults.USE_LEGACY_DYNAMO_SESSION_CONTROL
+
+    dynamo_session_timeout_seconds: Annotated[
+        int,
+        Field(
+            ge=1,
+            description=(
+                "Dynamo nvext.session_control timeout in seconds when "
+                "--use-dynamo-conv-aware-routing is enabled."
+            ),
+        ),
+        CLIParameter(
+            name=("--dynamo-session-timeout-seconds",),
+            group=Groups.ENDPOINT,
+        ),
+    ] = EndpointDefaults.DYNAMO_SESSION_TIMEOUT_SECONDS
+
     @property
     def url(self) -> str:
         """Return the first URL for backward compatibility."""

src/aiperf/config/schema/aiperf-config.schema.json

@@ -7214,6 +7214,43 @@
           "description": "HTTP header name used to carry the per-session affinity identifier. When set, replaces the default `X-Correlation-ID` header. Useful when the inference server expects a custom session-affinity header (e.g. `--session-header X-Session-ID`).",
           "title": "Sessionheader"
         },
+        "useDynamoConvAwareRouting": {
+          "default": false,
+          "description": "Emit Dynamo nvext.session_control in OpenAI-compatible request bodies so Dynamo can bind all turns from the same replayed conversation lineage to the same backend worker. This is only intended for Dynamo frontends that implement session_control.",
+          "title": "Usedynamoconvawarerouting",
+          "type": "boolean"
+        },
+        "useLegacyDynamoSessionControl": {
+          "default": false,
+          "description": "Emit the legacy Dynamo nvext.session_control lifecycle that released Dynamo (v1.2.x) understands: action 'open' on the first turn, session_id only on intermediate turns, and action 'close' on the final turn. Use this when the target Dynamo predates the 'bind' action (added in v1.3.0-dev); otherwise 'bind' is rejected with an HTTP 400. Requires use_dynamo_conv_aware_routing, and the Dynamo deployment must expose a worker session_control endpoint for 'open' to take effect.",
+          "title": "Uselegacydynamosessioncontrol",
+          "type": "boolean"
+        },
+        "dynamoSessionTimeoutSeconds": {
+          "default": 300,
+          "description": "Dynamo nvext.session_control timeout in seconds when use_dynamo_conv_aware_routing is enabled.",
+          "title": "Dynamosessiontimeoutseconds",
+          "oneOf": [
+            {
+              "default": 300,
+              "description": "Dynamo nvext.session_control timeout in seconds when use_dynamo_conv_aware_routing is enabled.",
+              "minimum": 1,
+              "title": "Dynamosessiontimeoutseconds",
+              "type": "integer"
+            },
+            {
+              "type": "string",
+              "pattern": ".*\\{\\{.*\\}\\}.*",
+              "description": "Jinja2 template (e.g., '{{ variable }}')."
+            },
+            {
+              "type": "string",
+              "pattern": ".*\\$\\{[A-Za-z_][A-Za-z0-9_]*(?::[^}]*)?\\}.*",
+              "description": "Environment variable (e.g., '${VAR}' or '${VAR:default}')."
+            }
+          ],
+          "x-jinja2-supported": true
+        },
         "waitForModelTimeout": {
           "default": 0.0,
           "description": "Enable a pre-flight readiness probe by setting this to a positive value (seconds). aiperf applies this timeout to each URL/model probe before starting the benchmark, aborting with a non-zero exit if any probe times out. For multiple URLs or models, worst-case wall-clock time can be roughly this timeout multiplied by the number of URL/model probes. The probe strategy is controlled by `--wait-for-model-mode`, which defaults to sending a 1-token inference request. 0 (default) disables the probe. Eliminates the need for external shell-based readiness loops in containers and Kubernetes recipes.",

src/aiperf/exporters/console_api_error_exporter.py

@@ -77,11 +77,64 @@ def detect(error_summary: list[ErrorDetailsCount]) -> ErrorInsight | None:
         return None
 
 
+class DynamoSessionControlDetector:
+    @staticmethod
+    def detect(error_summary: list[ErrorDetailsCount]) -> ErrorInsight | None:
+        if not error_summary or not isinstance(error_summary, list):
+            return None
+
+        for item in error_summary:
+            err = getattr(item, "error_details", None)
+            if err is None:
+                continue
+
+            raw_msg = err.message or ""
+            parsed = None
+            with contextlib.suppress(Exception):
+                parsed = orjson.loads(raw_msg)
+
+            backend_msg = None
+            if isinstance(parsed, dict):
+                backend_msg = parsed.get("message")
+
+            error_blob = str(backend_msg or raw_msg).lower()
+
+            # serde rejects the unknown enum value with e.g.
+            # `unknown variant `bind`, expected `open` or `close``.
+            if "unknown variant" in error_blob and "bind" in error_blob:
+                return ErrorInsight(
+                    title="Unsupported Dynamo session_control action: bind",
+                    problem=(
+                        "The Dynamo frontend rejected nvext.session_control with "
+                        "action='bind'. This Dynamo build's SessionAction only "
+                        "accepts 'open' and 'close' -- the 'bind' action was added "
+                        "after the v1.2.x release line (first available in "
+                        "v1.3.0-dev / upstream commit d97c889ba)."
+                    ),
+                    causes=[
+                        "--use-dynamo-conv-aware-routing emits action='bind' on every non-final turn.",
+                        "The target Dynamo server predates the 'bind' action (e.g. v1.2.1).",
+                    ],
+                    investigation=[
+                        "Check the Dynamo frontend version and its supported SessionAction values.",
+                        "Inspect request payloads in profile_export.jsonl -> nvext.session_control.",
+                    ],
+                    fixes=[
+                        "Upgrade Dynamo to a build that supports action='bind' (>= v1.3.0-dev, upstream commit d97c889ba).",
+                        "Or run with --use-legacy-dynamo-session-control to emit the v1.2.x-compatible open/close lifecycle (requires the worker to expose a session_control endpoint).",
+                        "Or disable --use-dynamo-conv-aware-routing.",
+                    ],
+                )
+
+        return None
+
+
 class ConsoleApiErrorExporter(AIPerfLoggerMixin):
     """Displays helpful diagnostic panels for known API error patterns."""
 
     DETECTORS: ClassVar[list] = [
         MaxCompletionTokensDetector,
+        DynamoSessionControlDetector,
     ]
 
     def __init__(self, exporter_config: ExporterConfig, **kwargs):