[MLI-6876] feat(api): add forwarder_max_concurrency to model endpoint create/update

[diazagasatya]

Note: this PR supersedes #834 — same code, moved from my fork (diazagasatya/llm-engine) to an upstream branch now that I have write access, so CircleCI can run. All review comments + Lily's approval still apply to the same 3 commits below.

Summary

Adds an optional forwarder_max_concurrency: Optional[int] field to the Launch model-endpoint create and update API. When set, this propagates as a FORWARDER_MAX_CONCURRENCY env var on the generated K8s deployment; the downstream deployer uses it to set LIRA's ForwarderDeployOptions.forwarder_max_concurrency, overriding the HTTP forwarder container's --concurrency flag independently of pod.concurrency.

Default None preserves existing behavior end-to-end.

Why

Followup to Scale-internal MLI-5366 (LIRA layer) and the public-SDK chain (MLI-6875, MLI-6874). MLI-5366 added the field to LIRA's deploy schema; this PR exposes it at the API boundary so the SDK chain can plumb it from public callers all the way to the rendered K8s manifest.

Changes

File	Change
dtos/model_endpoints.py	Add field to CreateModelEndpointV1Request + UpdateModelEndpointV1Request (Optional[int], gt=0, le=20 to match LIRA's FORWARDER_MAX_CONCURRENCY_LIMIT)
dtos/endpoint_builder.py	Add to BuildEndpointRequest so it flows through the celery task to the service builder
domain/services/model_endpoint_service.py	Add to abstract service interface for both create and update methods
domain/use_cases/model_endpoint_use_cases.py	Extract from the request in both create and update use cases, pass to service
infra/services/live_model_endpoint_service.py	Thread kwarg through to the infra gateway calls (create + update)
infra/gateways/model_endpoint_infra_gateway.py + live_model_endpoint_infra_gateway.py	Accept the kwarg and set it on BuildEndpointRequest
infra/gateways/resources/k8s_resource_types.py	Declare on _DeploymentArguments TypedDict; inject as env var at all 11 template-substitution sites that already set CONCURRENT_REQUESTS_PER_WORKER
infra/gateways/resources/k8s_endpoint_resource_delegate.py	Strip None-valued kwargs before Template.substitute() so Optional fields don't render as the literal string "None" (review fix from #834)
tests/unit/conftest.py	Update fake gateway and fake service signatures so existing tests still pass under the new abstract interface
specs/openapi-3.0.json, specs/openapi.json, specs/metadata.json	Regenerated via python model-engine/scripts/generate_openapi_schemas.py model-engine/specs/

NOT in this PR (downstream coordination)

The deployer that reads the new FORWARDER_MAX_CONCURRENCY env var off the K8s deployment and sets ForwarderDeployOptions.forwarder_max_concurrency lives outside this repo (likely in scaleapi/models/model-engine-internal/ or a dedicated deployer service). Worth confirming that bridge works as expected once this lands and the rest of the chain (MLI-6875, MLI-6874) is live.

Test plan

• CI: existing unit tests pass (fake gateway/service signatures updated to match the new abstract interface)
• After follow-up PRs ship: create a test endpoint with forwarder_max_concurrency=5 and confirm via kubectl get deploy <name> -o yaml | grep -A2 -- '--concurrency' that the forwarder containers reflect the override while the user_service container is unchanged

Greptile Summary

This PR adds an optional forwarder_max_concurrency field to the model endpoint create and update APIs, propagating it as a FORWARDER_MAX_CONCURRENCY env var on the generated K8s deployment so LIRA's HTTP forwarder container concurrency can be tuned independently of per_worker. The field is threaded end-to-end through DTOs, domain entities, service/gateway interfaces, and the K8s resource-type substitution layer.

• API layer: forwarder_max_concurrency: Optional[int] (gt=0, le=20) added to both CreateModelEndpointV1Request and UpdateModelEndpointV1Request; constraints match LIRA's FORWARDER_MAX_CONCURRENCY_LIMIT and default None preserves existing behavior.
• K8s resource layer: FORWARDER_MAX_CONCURRENCY declared on _BaseDeploymentArguments and injected at all 11 template-substitution sites; load_k8s_yaml now uses Template.safe_substitute after stripping None entries.
• Update fallback: The if forwarder_max_concurrency is None guard in update_model_endpoint_infra is a no-op \u2014 forwarder_max_concurrency is never read back from the running K8s deployment, so the fallback always resolves to None and the value is silently dropped on any update that omits it.

Confidence Score: 3/5

The create path is safe, but the update path silently loses forwarder_max_concurrency on any update that omits the field because the delegate never reads it back from the running deployment.

The fallback added to preserve forwarder_max_concurrency across updates always resolves to None because _get_resources_from_deployment_type and _get_resources_from_lws_type never populate the field when reconstructing ModelEndpointDeploymentState from K8s. Once downstream templates inject ${FORWARDER_MAX_CONCURRENCY}, any update call that omits the parameter will silently strip the env var from the deployment.

live_model_endpoint_infra_gateway.py (update fallback) and k8s_endpoint_resource_delegate.py (_get_resources_from_deployment_type / _get_resources_from_lws_type)

Important Files Changed

Filename	Overview
model-engine/model_engine_server/infra/gateways/live_model_endpoint_infra_gateway.py	Adds forwarder_max_concurrency to create/update paths; the update fallback reads from a field never populated by the K8s delegate, making it a silent no-op.
model-engine/model_engine_server/infra/gateways/resources/k8s_endpoint_resource_delegate.py	Switches to safe_substitute with None-filtering; ModelEndpointDeploymentState construction does not yet read forwarder_max_concurrency back from K8s.
model-engine/model_engine_server/common/dtos/model_endpoints.py	Adds forwarder_max_concurrency: Optional[int] (gt=0, le=20) to create and update request DTOs.
model-engine/model_engine_server/infra/gateways/resources/k8s_resource_types.py	Declares FORWARDER_MAX_CONCURRENCY: Optional[int] and injects at all 11 substitution sites.
model-engine/model_engine_server/domain/entities/model_endpoint_entity.py	Adds forwarder_max_concurrency: Optional[int] to ModelEndpointDeploymentState.
model-engine/tests/unit/conftest.py	Updates fake gateway/service signatures and all fixture deployment states.

Comments Outside Diff (1)

1. model-engine/model_engine_server/infra/gateways/live_model_endpoint_infra_gateway.py, line 150-215 (link)

   forwarder_max_concurrency not preserved across updates

   All comparable fields in update_model_endpoint_infra fall back to infra_state when None is passed (e.g., concurrent_requests_per_worker at line 156-159), but forwarder_max_concurrency has no such fallback. A caller that sends PUT /endpoint/{id} without specifying forwarder_max_concurrency (the default) will silently reset the value to None in BuildEndpointRequest, even when the endpoint was originally created with a non-None value. This differs from the established contract for every other comparable field, and will produce incorrect K8s manifests once the downstream templates start injecting ${FORWARDER_MAX_CONCURRENCY}. The root cause is that forwarder_max_concurrency was not added to ModelEndpointDeploymentState (domain entity), so infra_state.deployment_state has no forwarder_max_concurrency to fall back to.

   Prompt To Fix With AI

   This is a comment left during a code review.
   Path: model-engine/model_engine_server/infra/gateways/live_model_endpoint_infra_gateway.py
   Line: 150-215
   
   Comment:
   **`forwarder_max_concurrency` not preserved across updates**
   
   All comparable fields in `update_model_endpoint_infra` fall back to `infra_state` when `None` is passed (e.g., `concurrent_requests_per_worker` at line 156-159), but `forwarder_max_concurrency` has no such fallback. A caller that sends `PUT /endpoint/{id}` without specifying `forwarder_max_concurrency` (the default) will silently reset the value to `None` in `BuildEndpointRequest`, even when the endpoint was originally created with a non-`None` value. This differs from the established contract for every other comparable field, and will produce incorrect K8s manifests once the downstream templates start injecting `${FORWARDER_MAX_CONCURRENCY}`. The root cause is that `forwarder_max_concurrency` was not added to `ModelEndpointDeploymentState` (domain entity), so `infra_state.deployment_state` has no `forwarder_max_concurrency` to fall back to.
   
   How can I resolve this? If you propose a fix, please make it concise.

   

Prompt To Fix All With AI

Fix the following 1 code review issue. Work through them one at a time, proposing concise fixes.

---

### Issue 1 of 1
model-engine/model_engine_server/infra/gateways/live_model_endpoint_infra_gateway.py:160-161
**Fallback reads from a field never populated by the K8s delegate**

A fallback was added here to preserve `forwarder_max_concurrency` across updates when the caller omits it, but `infra_state.deployment_state.forwarder_max_concurrency` is always `None` in practice. Neither `_get_resources_from_deployment_type` (line 2282 of `k8s_endpoint_resource_delegate.py`) nor `_get_resources_from_lws_type` sets `forwarder_max_concurrency` when constructing `ModelEndpointDeploymentState` — the field just keeps its default `None`. As a result the fallback is a no-op: every update that omits `forwarder_max_concurrency` will silently reset it to `None`, stripping the env var from the deployment once downstream templates reference `${FORWARDER_MAX_CONCURRENCY}`.

_{Reviews (4): Last reviewed commit: "fix: use Template.safe_substitute in loa..." | Re-trigger Greptile}