feat: implement temporal endpoint type MVP (MLI-6425)

Adds TEMPORAL as a fourth endpoint type in Launch so teams can run
Temporal activity worker pods through the standard Launch API instead
of maintaining custom raw K8s Deployments.

- domain/entities: add TEMPORAL to ModelEndpointType enum
- domain/entities: add temporal_task_queue field to ModelEndpointRecord
- common/dtos: add temporal_task_queue to Create/Update request DTOs
  with validator requiring it for temporal endpoints
- domain/use_cases: allow min_workers=0 for temporal (like async)
- domain/services + infra/services: thread temporal_task_queue through
  create/update endpoint call chain
- db/models: add temporal_task_queue column to ORM model
- db/migrations: alembic migration to add the column
- infra/repositories: persist and read temporal_task_queue
- k8s_resource_types: add DeploymentRunnableImageTemporalCpuArguments
  and DeploymentRunnableImageTemporalGpuArguments TypedDicts; add
  temporal branch in get_endpoint_resource_arguments_from_request
- k8s_endpoint_resource_delegate: add temporal resource name handling
  in _create_or_update_resources, _get_resources_from_deployment_type,
  _get_temporal_autoscaling_params (reads annotations); skip SQS for
  temporal in delete_resources
- live_endpoint_resource_gateway: skip SQS queue deletion for temporal
- service_template_config_map_circleci.yaml: add cpu+gpu temporal
  deployment templates (fixed replicas=MAX_WORKERS, no forwarder
  sidecar, temporal annotations)
- service_template_config_map.yaml (Helm): same templates for prod

MVP uses fixed replicas = max_workers; KEDA autoscaling is phase 2.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: lilyz-ai

charts/model-engine/templates/service_template_config_map.yaml

@@ -408,6 +408,146 @@ data:
   {{- end }}
   {{- end }}
   {{- end }}
+  {{- range $device := tuple "cpu" "gpu" }}
+  deployment-runnable-image-temporal-{{ $device }}.yaml: |-
+    apiVersion: apps/v1
+    kind: Deployment
+    metadata:
+      name: ${RESOURCE_NAME}
+      namespace: ${NAMESPACE}
+      labels:
+        {{- $service_template_labels | nindent 8 }}
+      annotations:
+        temporal.scaleml.io/taskQueue: "${TEMPORAL_TASK_QUEUE}"
+        temporal.scaleml.io/minWorkers: "${MIN_WORKERS}"
+        temporal.scaleml.io/maxWorkers: "${MAX_WORKERS}"
+        temporal.scaleml.io/perWorker: "${PER_WORKER}"
+    spec:
+      strategy:
+        type: RollingUpdate
+        rollingUpdate:
+          maxSurge: 1
+          maxUnavailable: 0
+      replicas: ${MAX_WORKERS}
+      selector:
+        matchLabels:
+          app: ${RESOURCE_NAME}
+          version: v1
+      template:
+        metadata:
+          labels:
+            app: ${RESOURCE_NAME}
+            {{- $service_template_labels | nindent 12 }}
+            sidecar.istio.io/inject: "false"
+            version: v1
+          annotations:
+            ad.datadoghq.com/main.logs: '[{"service": "${ENDPOINT_NAME}", "source": "python"}]'
+            kubernetes.io/change-cause: "${CHANGE_CAUSE_MESSAGE}"
+        spec:
+          affinity:
+            {{- include "modelEngine.serviceTemplateAffinity" . | nindent 12 }}
+          terminationGracePeriodSeconds: 1800
+          {{- if $service_template_service_account_name }}
+          serviceAccount: {{ $service_template_service_account_name }}
+          {{- else }}
+          serviceAccount: {{ $launch_name }}
+          {{- end }}
+          {{- with $node_selector }}
+          nodeSelector:
+            {{- toYaml . | nindent 12 }}
+          {{- end }}
+          {{- if eq $device "gpu" }}
+          {{- if empty $node_selector }}
+          nodeSelector:
+          {{- end }}
+            k8s.amazonaws.com/accelerator: ${GPU_TYPE}
+          tolerations:
+            - key: "nvidia.com/gpu"
+              operator: "Exists"
+              effect: "NoSchedule"
+            {{- range $gpu_tolerations }}
+            - key: {{ .key | quote }}
+              operator: "Exists"
+              effect: "NoSchedule"
+            {{- end }}
+          {{- end }}
+          priorityClassName: ${PRIORITY}
+          containers:
+            - name: main
+              {{- $security_context | nindent 14 }}
+              image: ${IMAGE}
+              imagePullPolicy: IfNotPresent
+              command: ${COMMAND}
+              env: ${MAIN_ENV}
+              resources:
+                requests:
+                  {{- if eq $device "gpu" }}
+                  nvidia.com/gpu: ${GPUS}
+                  {{- end }}
+                  cpu: ${CPUS}
+                  memory: ${MEMORY}
+                  ${STORAGE_DICT}
+                limits:
+                  {{- if eq $device "gpu" }}
+                  nvidia.com/gpu: ${GPUS}
+                  {{- end }}
+                  cpu: ${CPUS}
+                  memory: ${MEMORY}
+                  ${STORAGE_DICT}
+              volumeMounts:
+                {{- if $require_aws_config }}
+                - name: aws-config-volume
+                  mountPath: /opt/.aws/config
+                  subPath: config
+                {{- end }}
+                {{- if eq $device "gpu" }}
+                - mountPath: /dev/shm
+                  name: dshm
+                {{- end }}
+                - name: user-config
+                  mountPath: /app/user_config
+                  subPath: raw_data
+                - name: endpoint-config
+                  mountPath: /app/endpoint_config
+                  subPath: raw_data
+                {{- if $config_values }}
+                - name: infra-service-config-volume
+                  mountPath: ${INFRA_SERVICE_CONFIG_VOLUME_MOUNT_PATH}
+                {{- end }}
+          {{- if $security_context }}
+          securityContext:
+            fsGroup: 65534
+          {{- end }}
+          volumes:
+            {{- if $require_aws_config }}
+            - name: aws-config-volume
+              configMap:
+                {{- if $service_template_aws_config_map_name }}
+                name: {{ $service_template_aws_config_map_name }}
+                {{- else }}
+                name: {{ $aws_config_map_name }}
+                {{- end }}
+            {{- end }}
+            - name: user-config
+              configMap:
+                name: ${RESOURCE_NAME}
+            - name: endpoint-config
+              configMap:
+                name: ${RESOURCE_NAME}-endpoint-config
+            {{- if eq $device "gpu" }}
+            - name: dshm
+              emptyDir:
+                medium: Memory
+            {{- end }}
+            {{- if $config_values }}
+            - name: infra-service-config-volume
+              configMap:
+                name: {{ $launch_name }}-service-config
+                items:
+                  - key: infra_service_config
+                    path: config.yaml
+            {{- end }}
+  {{- end }}
   user-config.yaml: |-
     apiVersion: v1
     kind: ConfigMap

model-engine/model_engine_server/common/dtos/model_endpoints.py

@@ -11,7 +11,7 @@
 from typing import Any, Dict, List, Optional
 
 from model_engine_server.common.dtos.core import HttpUrlStr
-from model_engine_server.common.pydantic_types import BaseModel, ConfigDict, Field
+from model_engine_server.common.pydantic_types import BaseModel, ConfigDict, Field, model_validator
 from model_engine_server.domain.entities import (
     CallbackAuth,
     CpuSpecificationType,
@@ -84,6 +84,16 @@ class CreateModelEndpointV1Request(BaseModel):
         ge=1,
         description="For async endpoints, how long a task can wait in queue before expiring (in seconds). Default: 86400 (24 hours).",
     )
+    temporal_task_queue: Optional[str] = Field(
+        default=None,
+        description="For temporal endpoints, the Temporal task queue that workers will poll.",
+    )
+
+    @model_validator(mode="after")
+    def validate_temporal_task_queue(self) -> "CreateModelEndpointV1Request":
+        if self.endpoint_type == ModelEndpointType.TEMPORAL and not self.temporal_task_queue:
+            raise ValueError("temporal_task_queue is required for temporal endpoints")
+        return self
 
 
 class CreateModelEndpointV1Response(BaseModel):
@@ -122,6 +132,10 @@ class UpdateModelEndpointV1Request(BaseModel):
         ge=1,
         description="For async endpoints, how long a task can wait in queue before expiring (in seconds). Default: 86400 (24 hours).",
     )
+    temporal_task_queue: Optional[str] = Field(
+        default=None,
+        description="For temporal endpoints, the Temporal task queue that workers will poll.",
+    )
 
 
 class UpdateModelEndpointV1Response(BaseModel):

model-engine/model_engine_server/db/migrations/alembic/versions/2026_04_24_0000-b2c3d4e5f6g7_add_temporal_task_queue_column.py

@@ -0,0 +1,31 @@
+"""add temporal_task_queue column
+
+Revision ID: b2c3d4e5f6g7
+Revises: a1b2c3d4e5f6
+Create Date: 2026-04-24 00:00:00.000000
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = 'b2c3d4e5f6g7'
+down_revision = 'a1b2c3d4e5f6'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        'endpoints',
+        sa.Column('temporal_task_queue', sa.String, nullable=True),
+        schema='hosted_model_inference',
+    )
+
+
+def downgrade() -> None:
+    op.drop_column(
+        'endpoints',
+        'temporal_task_queue',
+        schema='hosted_model_inference',
+    )

model-engine/model_engine_server/db/models/hosted_model_inference.py

@@ -472,6 +472,8 @@ class Endpoint(Base):
     task_expires_seconds = Column(Integer, nullable=True)
     # Queue message visibility/lock timeout in seconds (SQS VisibilityTimeout / ASB lock_duration)
     queue_message_timeout_seconds = Column(Integer, nullable=True)
+    # Temporal task queue name for temporal endpoints
+    temporal_task_queue = Column(String, nullable=True)
 
     def __init__(
         self,
@@ -488,6 +490,7 @@ def __init__(
         public_inference: Optional[bool] = False,
         task_expires_seconds: Optional[int] = None,
         queue_message_timeout_seconds: Optional[int] = None,
+        temporal_task_queue: Optional[str] = None,
     ):
         self.id = f"end_{get_xid()}"
         self.name = name
@@ -502,6 +505,7 @@ def __init__(
         self.public_inference = public_inference
         self.task_expires_seconds = task_expires_seconds
         self.queue_message_timeout_seconds = queue_message_timeout_seconds
+        self.temporal_task_queue = temporal_task_queue
 
     @classmethod
     async def create(cls, session: AsyncSession, endpoint: "Endpoint") -> None:

model-engine/model_engine_server/domain/entities/model_endpoint_entity.py

@@ -22,6 +22,7 @@ class ModelEndpointType(str, Enum):
     ASYNC = "async"
     SYNC = "sync"
     STREAMING = "streaming"
+    TEMPORAL = "temporal"
 
 
 class ModelEndpointStatus(str, Enum):
@@ -136,6 +137,7 @@ class ModelEndpointRecord(OwnedEntity):
     public_inference: Optional[bool] = None
     task_expires_seconds: Optional[int] = None
     queue_message_timeout_seconds: Optional[int] = None
+    temporal_task_queue: Optional[str] = None
 
 
 class ModelEndpointInfraState(BaseModel):

model-engine/model_engine_server/domain/services/model_endpoint_service.py

@@ -94,6 +94,7 @@ async def create_model_endpoint(
         public_inference: Optional[bool] = False,
         queue_message_timeout_seconds: Optional[int] = None,
         task_expires_seconds: Optional[int] = None,
+        temporal_task_queue: Optional[str] = None,
     ) -> ModelEndpointRecord:
         """
         Creates a model endpoint.
@@ -230,6 +231,7 @@ async def update_model_endpoint(
         public_inference: Optional[bool] = None,
         queue_message_timeout_seconds: Optional[int] = None,
         task_expires_seconds: Optional[int] = None,
+        temporal_task_queue: Optional[str] = None,
     ) -> ModelEndpointRecord:
         """
         Updates a model endpoint.

model-engine/model_engine_server/domain/use_cases/model_endpoint_use_cases.py

@@ -116,7 +116,10 @@ def validate_deployment_resources(
     # TODO: we should be also validating the update request against the existing state in k8s (e.g.
     #  so min_workers <= max_workers always) maybe this occurs already in update_model_endpoint.
     min_endpoint_size = (
-        0 if endpoint_type == ModelEndpointType.ASYNC or can_scale_http_endpoint_from_zero else 1
+        0
+        if endpoint_type in {ModelEndpointType.ASYNC, ModelEndpointType.TEMPORAL}
+        or can_scale_http_endpoint_from_zero
+        else 1
     )
     if min_workers is not None and min_workers < min_endpoint_size:
         raise EndpointResourceInvalidRequestException(
@@ -393,6 +396,7 @@ async def execute(
             public_inference=request.public_inference,
             queue_message_timeout_seconds=request.queue_message_timeout_seconds,
             task_expires_seconds=request.task_expires_seconds,
+            temporal_task_queue=request.temporal_task_queue,
         )
         _handle_post_inference_hooks(
             created_by=user.user_id,
@@ -523,6 +527,7 @@ async def execute(
             public_inference=request.public_inference,
             queue_message_timeout_seconds=request.queue_message_timeout_seconds,
             task_expires_seconds=request.task_expires_seconds,
+            temporal_task_queue=request.temporal_task_queue,
         )
         _handle_post_inference_hooks(
             created_by=endpoint_record.created_by,

model-engine/model_engine_server/infra/gateways/resources/k8s_endpoint_resource_delegate.py

@@ -420,7 +420,11 @@ async def delete_resources(
         self, endpoint_id: str, deployment_name: str, endpoint_type: ModelEndpointType
     ) -> bool:
         await maybe_load_kube_config()
-        if endpoint_type in {ModelEndpointType.SYNC, ModelEndpointType.STREAMING}:
+        if endpoint_type in {
+            ModelEndpointType.SYNC,
+            ModelEndpointType.STREAMING,
+            ModelEndpointType.TEMPORAL,
+        }:
             return await self._delete_resources_sync(
                 endpoint_id=endpoint_id, deployment_name=deployment_name
             )
@@ -1872,6 +1876,7 @@ async def _create_or_update_resources(
         if model_endpoint_record.endpoint_type in {
             ModelEndpointType.SYNC,
             ModelEndpointType.STREAMING,
+            ModelEndpointType.TEMPORAL,
         }:
             return k8s_service_name
         elif model_endpoint_record.endpoint_type == ModelEndpointType.ASYNC:
@@ -1898,6 +1903,18 @@ def _get_vertical_autoscaling_params(
             max_memory=str(policy["maxAllowed"]["memory"]),
         )
 
+    @staticmethod
+    def _get_temporal_autoscaling_params(
+        deployment_config,
+    ) -> HorizontalAutoscalingEndpointParams:
+        metadata_annotations = deployment_config.metadata.annotations or {}
+        return dict(
+            min_workers=int(metadata_annotations.get("temporal.scaleml.io/minWorkers", 0)),
+            max_workers=int(metadata_annotations.get("temporal.scaleml.io/maxWorkers", 0)),
+            per_worker=int(metadata_annotations.get("temporal.scaleml.io/perWorker", 1)),
+            concurrent_requests_per_worker=1,
+        )
+
     @staticmethod
     def _get_async_autoscaling_params(
         deployment_config,
@@ -2013,6 +2030,8 @@ async def _get_resources_from_deployment_type(
         common_params = self._get_common_endpoint_params(deployment_config)
         if endpoint_type == ModelEndpointType.ASYNC:
             horizontal_autoscaling_params = self._get_async_autoscaling_params(deployment_config)
+        elif endpoint_type == ModelEndpointType.TEMPORAL:
+            horizontal_autoscaling_params = self._get_temporal_autoscaling_params(deployment_config)
         elif endpoint_type in {ModelEndpointType.SYNC, ModelEndpointType.STREAMING}:
             autoscaling_client = get_kubernetes_autoscaling_client()
             custom_object_client = get_kubernetes_custom_objects_client()