Resolve Comments

Author: Bihan Rana

src/dstack/_internal/core/models/configurations.py

@@ -1052,6 +1052,23 @@ def validate_at_most_one_router_replica_group(cls, values):
                 raise ValueError("For now replica group with `router` must have `count: 1`.")
         return values
 
+    @root_validator()
+    def validate_replica_group_router_mutex(cls, values):
+        """
+        When a replica group sets `router:`, service-level `router` must be omitted.
+        (Gateway-level SGLang is rejected at service registration when a gateway is selected.)
+        """
+        replicas = values.get("replicas")
+        if not isinstance(replicas, list):
+            return values
+        if not any(g.router is not None for g in replicas):
+            return values
+        if values.get("router") is not None:
+            raise ValueError(
+                "Service-Level router configuration is not allowed together with replica-group `router`."
+            )
+        return values
+
 
 class ServiceConfigurationConfig(
     ProfileParamsConfig,

src/dstack/_internal/core/models/routers.py

@@ -31,6 +31,16 @@ class SGLangGatewayRouterConfig(CoreModel):
 
 class SGLangServiceRouterConfig(CoreModel):
     type: Annotated[Literal["sglang"], Field(description="The router type")] = "sglang"
+    managed_by: Annotated[
+        Literal["gateway", "service"],
+        Field(
+            description=(
+                "Where the router process is managed. "
+                "`gateway`: the gateway runs and manages the router process. "
+                "`service`: the router runs inside the service (replica-group `router:`)."
+            )
+        ),
+    ] = "gateway"
     policy: Annotated[
         Literal["random", "round_robin", "cache_aware", "power_of_two"],
         Field(

src/dstack/_internal/proxy/gateway/services/nginx.py

@@ -110,7 +110,11 @@ async def register(self, conf: SiteConfig, acme: ACMESettings) -> None:
             if conf.https:
                 await run_async(self.run_certbot, conf.domain, acme)
 
-            if isinstance(conf, ServiceConfig) and conf.router:
+            if (
+                isinstance(conf, ServiceConfig)
+                and conf.router
+                and getattr(conf.router, "managed_by", "gateway") == "gateway"
+            ):
                 if conf.router.type == RouterType.SGLANG:
                     # Check if router already exists for this domain
                     if conf.domain in self._domain_to_router:

src/dstack/_internal/server/background/pipeline_tasks/service_router_worker_sync.py

@@ -214,10 +214,19 @@ async def process(self, item: ServiceRouterWorkerSyncPipelineItem) -> None:
                 or run_model.status != RunStatus.RUNNING
                 or not run_model_has_router_replica_group(run_model)
             ):
-                sync_row.deleted = True
-                sync_row.lock_expires_at = None
-                sync_row.lock_token = None
-                sync_row.lock_owner = None
+                early_cleanup_update_map: _SyncRowUpdateMap = {"deleted": True}
+                set_processed_update_map_fields(early_cleanup_update_map)
+                set_unlock_update_map_fields(early_cleanup_update_map)
+                now = get_current_datetime()
+                resolve_now_placeholders(early_cleanup_update_map, now=now)
+                await session.execute(
+                    update(ServiceRouterWorkerSyncModel)
+                    .where(
+                        ServiceRouterWorkerSyncModel.id == item.id,
+                        ServiceRouterWorkerSyncModel.lock_token == item.lock_token,
+                    )
+                    .values(**early_cleanup_update_map)
+                )
                 await session.commit()
                 return
 

src/dstack/_internal/server/services/runs/router_worker_sync.py

@@ -8,7 +8,7 @@
 from typing_extensions import NotRequired
 
 from dstack._internal.core.errors import SSHError
-from dstack._internal.core.models.configurations import ServiceConfiguration
+from dstack._internal.core.models.configurations import ReplicaGroup, ServiceConfiguration
 from dstack._internal.core.models.runs import JobStatus, RunSpec, get_service_port
 from dstack._internal.server.models import JobModel, RunModel
 from dstack._internal.server.services.jobs import get_job_provisioning_data, get_job_spec
@@ -98,7 +98,7 @@ def run_model_has_router_replica_group(run_model: RunModel) -> bool:
     return run_spec_has_router_replica_group(run_spec)
 
 
-def _get_router_job(run_model: RunModel, router_group) -> Optional[JobModel]:
+def _get_router_job(run_model: RunModel, router_group: ReplicaGroup) -> Optional[JobModel]:
     group_name = router_group.name
     assert group_name is not None, "Replica group name is set by validation"
     router_jobs = [
@@ -280,7 +280,7 @@ async def _get_worker_payload(job_model: JobModel, worker_url: str) -> _WorkerPa
 async def _build_target_workers(
     run_model: RunModel,
     run_spec: RunSpec,
-    replica_groups: List,
+    replica_groups: list[ReplicaGroup],
 ) -> List[_TargetWorker]:
     payloads: List[_TargetWorker] = []
     config = run_spec.configuration

src/dstack/_internal/server/services/runs/service_router_worker_sync.py

@@ -2,7 +2,7 @@
 
 import uuid
 
-from sqlalchemy import select
+from sqlalchemy import select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 
 import dstack._internal.utils.common as common_utils
@@ -36,11 +36,21 @@ async def ensure_service_router_worker_sync_row(
     now = common_utils.get_current_datetime()
     if sync_row is not None:
         if sync_row.deleted:
-            sync_row.deleted = False
-            sync_row.lock_expires_at = None
-            sync_row.lock_token = None
-            sync_row.lock_owner = None
-            sync_row.last_processed_at = now
+            # If the router replica group is reintroduced (e.g. via re-apply), reactivate the
+            # existing sync row so the background pipeline resumes syncing router workers.
+            # Do not import pipeline_tasks.base here: loading that package runs
+            # pipeline_tasks/__init__.py -> jobs_running -> runs and causes a circular import.
+            await session.execute(
+                update(ServiceRouterWorkerSyncModel)
+                .where(ServiceRouterWorkerSyncModel.id == sync_row.id)
+                .values(
+                    deleted=False,
+                    last_processed_at=now,
+                    lock_expires_at=None,
+                    lock_token=None,
+                    lock_owner=None,
+                )
+            )
         return
     session.add(
         ServiceRouterWorkerSyncModel(

src/dstack/_internal/server/services/services/__init__.py

@@ -102,6 +102,14 @@ async def _register_service_in_gateway(
 
     gateway_configuration = get_gateway_configuration(gateway)
 
+    has_replica_group_router = any(
+        g.router is not None for g in run_spec.configuration.replica_groups
+    )
+    if has_replica_group_router and _gateway_has_sglang_router(gateway_configuration):
+        raise ServerClientError(
+            "A replica-group `router:` cannot be used with a gateway that has router configuration."
+        )
+
     # Check: service specifies SGLang router but gateway does not have it
     service_router = run_spec.configuration.router
     service_wants_sglang = service_router is not None and isinstance(
@@ -266,12 +274,24 @@ def _build_service_router_config(
     service_configuration: ServiceConfiguration,
 ) -> Optional[AnyServiceRouterConfig]:
     """
-    Build router config from gateway (type, policy) + service (pd_disaggregation, policy override).
-    Service's policy overrides gateway's if present. Keeps backward compat: SGLang enabled
-    automatically when gateway has it configured.
+    Router metadata to store on the gateway proxy for this service (`service.router`).
+
+    A replica-group `router:` does **not** depend on the gateway having its own global SGLang
+    router block—the router runs on service replicas. When the gateway has no global SGLang
+    config but a replica group does declare `router:`, we still return a default
+    `SGLangServiceRouterConfig` so nginx/proxy code can treat the service as SGLang and apply path
+    rules. When the gateway *does* have global SGLang, we merge gateway policy with service-level
+    `configuration.router` overrides as before.
     """
+    has_replica_group_router = any(
+        g.router is not None for g in service_configuration.replica_groups
+    )
     if not _gateway_has_sglang_router(gateway_configuration):
-        return None
+        if not has_replica_group_router:
+            return None
+        # In later releases we will deprecate service-level and gateway-level router
+        # configuration and return `ReplicaGroupRouterConfig` here instead.
+        return SGLangServiceRouterConfig(managed_by="service")
 
     gateway_router = gateway_configuration.router
     assert gateway_router is not None  # ensured by _gateway_has_sglang_router
@@ -287,6 +307,7 @@ def _build_service_router_config(
 
     return SGLangServiceRouterConfig(
         type=router_type,
+        managed_by="gateway",
         policy=policy,
         pd_disaggregation=pd_disaggregation,
     )

src/tests/_internal/core/models/test_configurations.py

@@ -120,6 +120,27 @@ def test_replica_group_router(self):
         assert isinstance(router_g.router, ReplicaGroupRouterConfig)
         assert router_g.router.type == "sglang"
 
+    def test_replica_group_router_forbids_service_level_router(self):
+        conf = {
+            "type": "service",
+            "port": 8000,
+            "router": {"type": "sglang"},
+            "replicas": [
+                {
+                    "name": "router",
+                    "count": 1,
+                    "commands": ["sglang serve"],
+                    "router": {"type": "sglang"},
+                },
+                {"name": "worker", "count": 2, "commands": ["worker"]},
+            ],
+        }
+        with pytest.raises(
+            ConfigurationError,
+            match="Service-Level router configuration is not allowed together with replica-group",
+        ):
+            parse_run_configuration(conf)
+
     @pytest.mark.parametrize("shell", [None, "sh", "bash", "/usr/bin/zsh"])
     def test_shell_valid(self, shell: Optional[str]):
         conf = {