diff --git a/CHANGELOG.md b/CHANGELOG.md
index 8b4f160..f0e4efd 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,12 +8,23 @@ no leading `v`.
 ## [Unreleased]
 
 ### Added
+- `JobSLORequirement` routes. The index watches both kinds, and a route
+  resolves across them; a job route's body has `queue` (`maxDepth`, `promql`)
+  in place of `ttft` / `otps`. A cluster without the JobSLO CRD serves the LLM
+  routes as before. The RBAC in `config/rbac/` grants `jobslorequirements`.
+- The `pdengine` bound type on `LLMSLORequirement` (helm-charts#95).
+- Bound values of 0 (scale-to-zero), which the CRD has allowed since
+  helm-charts#20; they were refused with `400` before.
 - CI on every pull request: pytest and the end-to-end smoke run on Python 3.10
   and 3.12, a docker build, and a gate for the license text and committed
   credentials.
 - Dependabot for the Python dependencies and the GitHub Actions.
 - `NOTICE`.
 
+### Changed
+- A write the CRD's rules reject (API server `422`) is `400 invalid_request`
+  with the API server's message, not `502`.
+
 ### Removed
 - The internal GitLab pipeline (`.gitlab-ci.yml`).
 
diff --git a/README.md b/README.md
index d539ee6..4fec8a9 100644
--- a/README.md
+++ b/README.md
@@ -3,7 +3,8 @@
 A small HTTP API for reading and changing the SLO of an LLM inference service,
 without handing the caller access to Kubernetes.
 
-The SLO itself lives in an `LLMSLORequirement` custom resource, which
+The SLO itself lives in an `LLMSLORequirement` custom resource (or a
+`JobSLORequirement`, for a queue-driven job service), which
 [slo-scaler-decision-gen](https://github.com/modelsphere/slo-scaler-decision-gen)
 reads to decide replica counts. slo-api is a front door onto those resources:
 callers address a service by its route name, send plain JSON, and never see a
@@ -21,11 +22,11 @@ namespace, a CR name or a kubeconfig.
   the service. The RBAC in `config/rbac/` grants exactly
   `get / list / watch / patch / update` and nothing else.
 - **It only touches the SLO fields** — `priority`, `minimumDeployment`,
-  `maximumDeployment`, `ttft`, `otps`. Anything else in the spec, written by
-  other controllers, survives a write.
-- **A route resolves through `spec.serviceId`.** If two namespaces have a CR with
-  the same `serviceId`, writes are refused with `409` rather than guessing; reads
-  return the first one seen.
+  `maximumDeployment`, `ttft`, `otps` (`queue` on a job route). Anything else in
+  the spec, written by other controllers, survives a write.
+- **A route resolves through `spec.serviceId`**, across both kinds. If two CRs
+  have the same `serviceId` (two namespaces, or one of each kind), reads and
+  writes are refused with `409` rather than guessing.
 - Writes merge onto the **live** CR, read fresh from the API server rather than
   from the watch cache, so a concurrent change is not silently reverted.
 
@@ -56,8 +57,25 @@ boolean:
 }
 ```
 
-`highPriority: true` writes `priority: 10`; `false` writes `0`. Metric types are
-`avg`, `p50`, `p80`, `p90`, `p95`, `p99`. Errors come back as
+`highPriority: true` writes `priority: 10`; `false` writes `0`. A bound's
+`value` may be 0: `{min: 0, max: 0}` turns the service off (scale-to-zero). Metric types are
+`avg`, `p50`, `p80`, `p90`, `p95`, `p99`. A bound's `type` is `replica`,
+`concurrency` (minimum only) or `pdengine` (a PD service's bounds counted in
+prefill + decode engines; both bounds must then be `pdengine`).
+
+A route backed by a `JobSLORequirement` has `queue` in place of `ttft` and
+`otps`, and its bounds are `replica` only (`concurrency` for the minimum):
+
+```json
+{"route": "img-gen", "queue": {"maxDepth": 8, "promql": "sum(jobs_pending{service=\"img-gen\"})"}}
+```
+
+`queue.promql` may be left out of a `PUT` to keep the CR's own. `DELETE` keeps
+`queue` (the CRD requires it) and resets the rest.
+
+The CRD's own cross-field rules (minimum at most maximum, matching bound types,
+...) are checked by the API server against the merged spec; a write that breaks
+one is `400` with the API server's message. Errors come back as
 `{"error": {"code", "message", "field"}}`, and an `X-Request-Id` header on the
 request is echoed on the response.
 
diff --git a/config/rbac/rbac.yaml b/config/rbac/rbac.yaml
index 741ad23..8243ab6 100644
--- a/config/rbac/rbac.yaml
+++ b/config/rbac/rbac.yaml
@@ -16,7 +16,7 @@ rules:
   # Both groups: the CRDs moved from inference.x-k8s.io, and slo-api picks
   # whichever the cluster serves at startup (slo_api.resolve_group).
   - apiGroups: ["inference.modelsphere.dev", "inference.x-k8s.io"]
-    resources: ["llmslorequirements"]
+    resources: ["llmslorequirements", "jobslorequirements"]
     verbs: ["get", "list", "watch", "patch", "update"]
 ---
 apiVersion: rbac.authorization.k8s.io/v1
diff --git a/docs/openapi.yaml b/docs/openapi.yaml
index bb8064c..dff46db 100644
--- a/docs/openapi.yaml
+++ b/docs/openapi.yaml
@@ -4,25 +4,27 @@ info:
   version: 0.2.1
   summary: Read and change the SLO of an LLM inference service by its route.
   description: |
-    slo-api is a front door onto `LLMSLORequirement` custom resources, which
+    slo-api is a front door onto `LLMSLORequirement` and `JobSLORequirement`
+    custom resources, which
     slo-scaler-decision-gen reads to size each service. Callers address a
     service by its route and send plain JSON; they never see a namespace, a CR
     name or a kubeconfig.
 
     ```
-    client -- Bearer token --> slo-api --> LLMSLORequirement CRs --> decision-gen
+    client -- Bearer token --> slo-api --> {LLM,Job}SLORequirement CRs --> decision-gen
                                (patch only; never creates or deletes a CR)
     ```
 
     ## Routes
 
-    A route is a CR's `spec.serviceId`. The CR's lifecycle belongs to whatever
+    A route is a CR's `spec.serviceId`, of either kind. The CR's lifecycle belongs to whatever
     deployed the service (for swiss, the chart at install, with the release
     name as the serviceId). Here a route is only read and patched:
 
     - An unknown route is `404`, for writes too. Nothing here creates a CR.
-    - A serviceId found in more than one namespace is `409` for reads and
-      writes alike, with the candidates in `error.details`.
+    - A serviceId found on more than one CR (two namespaces, or one of each
+      kind) is `409` for reads and writes alike, with the candidates in
+      `error.details`.
 
     ## Readiness
 
@@ -33,9 +35,11 @@ info:
     ## Fields
 
     slo-api owns five fields of the CR spec: `priority`, `minimumDeployment`,
-    `maximumDeployment`, `ttft` and `otps`. Every other field, written by other
-    controllers, survives a write. `priority` (0-10 in the CR) is exposed as
-    the boolean `highPriority`.
+    `maximumDeployment`, `ttft` and `otps` -- on a `JobSLORequirement` route,
+    `queue` in place of `ttft` and `otps`, and bounds of type `replica` (or
+    `concurrency` for the minimum) only. A section of the other kind is `400`.
+    Every other field, written by other controllers, survives a write.
+    `priority` (0-10 in the CR) is exposed as the boolean `highPriority`.
 
     ## Errors
 
@@ -68,9 +72,9 @@ paths:
       summary: List routes
       description: |
         Routes in name order, read from the watch index rather than the API
-        server. Items leave out `ttft` and `otps`; read one route for those.
+        server. Items leave out `ttft`, `otps` and `queue`; read one route for those.
 
-        A route found in more than one namespace counts in `total` but is left
+        A route found on more than one CR counts in `total` but is left
         out of `items`, so a page can be shorter than `pageSize`.
       operationId: listConfigs
       parameters:
@@ -138,9 +142,12 @@ paths:
         the write, so a concurrent change to another field is not reverted.
         `spec.serviceId` is kept.
 
+        A `queue.promql` left out is kept, the same way.
+
         The CRD's own rules, such as `minimumDeployment.value` at most
-        `maximumDeployment.value`, are checked by the API server. A body that
-        breaks one comes back as `502`, not `400`.
+        `maximumDeployment.value`, or both bounds of the same `type`, are
+        checked by the API server on the merged spec. A body that breaks one
+        comes back as `400` with the API server's message.
       operationId: putConfig
       requestBody:
         required: true
@@ -176,7 +183,8 @@ paths:
       description: |
         Resets the five fields to the CRD defaults; the CR stays. `priority`
         becomes 0 and `minimumDeployment` `{type: replica, value: 1}`;
-        `maximumDeployment`, `ttft` and `otps` are removed.
+        `maximumDeployment`, `ttft` and `otps` are removed. A
+        `JobSLORequirement` keeps its `queue`, which its CRD requires.
 
         decision-gen treats a requirement with no `maximumDeployment` as opted
         out of autoscaling, so after a reset it stops sizing the service.
@@ -297,14 +305,14 @@ components:
           example:
             error: { code: not_found, message: "route 'ghost' not registered" }
     Ambiguous:
-      description: The route is the serviceId of CRs in more than one namespace.
+      description: The route is the serviceId of more than one CR.
       content:
         application/json:
           schema: { $ref: "#/components/schemas/Error" }
           example:
             error:
               code: conflict
-              message: "route 'my-model' maps to multiple LLMSLORequirement CRs; cannot auto-select a target"
+              message: "route 'my-model' maps to multiple SLO requirement CRs; cannot auto-select a target"
               details:
                 - { namespace: team-a, name: my-model }
                 - { namespace: team-b, name: my-model }
@@ -332,10 +340,11 @@ components:
           description: Absent means decision-gen does not autoscale the service.
         ttft: { $ref: "#/components/schemas/TTFT" }
         otps: { $ref: "#/components/schemas/OTPS" }
+        queue: { $ref: "#/components/schemas/Queue" }
 
     SLOSummary:
       type: object
-      description: A list item. `ttft` and `otps` are left out.
+      description: A list item. `ttft`, `otps` and `queue` are left out.
       required: [route, highPriority]
       properties:
         route: { type: string }
@@ -369,24 +378,40 @@ components:
           type: object
           required: [value]
           properties:
-            type: { type: string, enum: [replica, concurrency], default: replica }
-            value: { type: integer, minimum: 1 }
+            type: { type: string, enum: [replica, concurrency, pdengine], default: replica }
+            value: { type: integer, minimum: 0 }
         maximumDeployment:
           type: object
           required: [value]
           properties:
-            type: { type: string, enum: [replica], default: replica }
-            value: { type: integer, minimum: 1 }
+            type: { type: string, enum: [replica, pdengine], default: replica }
+            value: { type: integer, minimum: 0 }
         ttft: { $ref: "#/components/schemas/TTFT" }
         otps: { $ref: "#/components/schemas/OTPS" }
+        queue:
+          $ref: "#/components/schemas/Queue"
+          description: JobSLORequirement routes only. `promql` may be left out to keep the stored one.
 
     Bound:
       type: object
-      description: A replica bound as stored. The CRD allows 0 (scale to zero), which this API does not write.
+      description: |
+        A bound as stored. 0 is scale-to-zero: a minimum of 0 needs a maximum of
+        0, which turns the service off (a CRD rule, so breaking it is `400`).
+        `pdengine` counts a PD service's prefill plus decode engines
+        rather than replicas, and needs both bounds to be `pdengine`.
       properties:
-        type: { type: string, enum: [replica, concurrency] }
+        type: { type: string, enum: [replica, concurrency, pdengine] }
         value: { type: integer, minimum: 0 }
 
+    Queue:
+      type: object
+      description: Queue SLO of a `JobSLORequirement`.
+      additionalProperties: false
+      required: [maxDepth]
+      properties:
+        maxDepth: { type: integer, minimum: 0, description: Target pending jobs per replica. }
+        promql: { type: string, minLength: 1, description: PromQL that decision-gen evaluates to the queue depth. }
+
     TTFT:
       description: |
         Time to first token, a ceiling in seconds: a metric is violated when the
diff --git a/src/slo_api/__init__.py b/src/slo_api/__init__.py
index fc617ed..053896b 100644
--- a/src/slo_api/__init__.py
+++ b/src/slo_api/__init__.py
@@ -1,4 +1,4 @@
-"""slo-api: HTTP facade over LLMSLORequirement CRs.
+"""slo-api: HTTP facade over LLMSLORequirement and JobSLORequirement CRs.
 
 Exposes a route-keyed REST API and translates to/from the underlying
 custom resource. Never creates new CRs; only get/patch/replace/delete
@@ -9,7 +9,10 @@
 _log = _logging.getLogger("slo-api")
 
 VERSION = "v1alpha1"
-PLURAL = "llmslorequirements"
+LLM_PLURAL = "llmslorequirements"
+JOB_PLURAL = "jobslorequirements"
+# Both kinds share one route space, as decision-gen sizes them side by side.
+PLURALS = (LLM_PLURAL, JOB_PLURAL)
 
 # The CRDs moved from inference.x-k8s.io to inference.modelsphere.dev. Both are
 # listed so this runs against a cluster on either, in the order to prefer them:
diff --git a/src/slo_api/cr_index.py b/src/slo_api/cr_index.py
index 2354621..ff58abf 100644
--- a/src/slo_api/cr_index.py
+++ b/src/slo_api/cr_index.py
@@ -1,15 +1,15 @@
-"""In-memory index of LLMSLORequirement CRs, keyed by serviceId.
+"""In-memory index of SLO requirement CRs (every kind in PLURALS), keyed by serviceId.
 
 The public API routes by the opaque `route` string, which we resolve
 against `spec.serviceId`. A CR is an addressable target iff it has a
 non-empty serviceId and at least one namespace hosting it.
 
-Two CRs in different namespaces may share a serviceId. Writes against
-such routes are rejected 409 (we refuse to guess); reads return the
-first-seen CR and log a warning.
+Two CRs may share a serviceId (different namespaces, or one of each
+kind). Such routes are rejected 409 (we refuse to guess).
 
-The index is maintained by a background watch with capped exponential
-backoff reconnect, mirroring decision_gen.SLOStore's pattern.
+The index is maintained by one background watch per kind with capped
+exponential backoff reconnect, mirroring decision_gen.SLOStore's pattern.
+A kind whose CRD is not installed is indexed as empty.
 """
 
 import logging
@@ -17,8 +17,9 @@
 import time
 
 from kubernetes import client, watch
+from kubernetes.client import ApiException
 
-from slo_api import VERSION, PLURAL, group
+from slo_api import VERSION, PLURALS, group
 
 log = logging.getLogger(__name__)
 
@@ -29,16 +30,16 @@
 class CRIndex:
     def __init__(self, api=None, watch_factory=None):
         self._lock = threading.Lock()
-        # (ns, name) -> full CR dict  (authoritative record)
+        # (plural, ns, name) -> full CR dict  (authoritative record)
         self._by_key = {}
-        # service_id -> [(ns, name), ...]  in first-seen order
+        # service_id -> [(plural, ns, name), ...]  in first-seen order
         self._by_service = {}
 
         self._api = api
         self._watch_factory = watch_factory
         self._stop = threading.Event()
-        self._synced = threading.Event()
-        self._thread = None
+        self._synced = {p: threading.Event() for p in PLURALS}
+        self._threads = []
         self._started = False
 
     # ---------- lifecycle ----------
@@ -47,21 +48,26 @@ def start(self):
         if self._started:
             return
         self._started = True
-        self._thread = threading.Thread(
-            target=self._run, daemon=True, name="cr-index",
-        )
-        self._thread.start()
+        for plural in PLURALS:
+            t = threading.Thread(
+                target=self._run, args=(plural,), daemon=True,
+                name=f"cr-index-{plural}",
+            )
+            t.start()
+            self._threads.append(t)
 
     def stop(self):
         self._stop.set()
-        if self._thread:
-            self._thread.join(timeout=5)
+        for t in self._threads:
+            t.join(timeout=5)
 
     def synced(self):
-        return self._synced.is_set()
+        return all(e.is_set() for e in self._synced.values())
 
     def wait_synced(self, timeout):
-        return self._synced.wait(timeout)
+        deadline = time.monotonic() + timeout
+        return all(e.wait(max(0.0, deadline - time.monotonic()))
+                   for e in self._synced.values())
 
     # ---------- lookups ----------
 
@@ -69,8 +75,8 @@ def resolve(self, route):
         """Return (status, payload) for `route`.
 
         status:
-          "ok"        — payload = (ns, name, cr)
-          "ambiguous" — payload = [(ns, name), ...]
+          "ok"        — payload = (plural, ns, name, cr)
+          "ambiguous" — payload = [(plural, ns, name), ...]
           "missing"   — payload = None
         """
         with self._lock:
@@ -79,12 +85,11 @@ def resolve(self, route):
                 return "missing", None
             if len(keys) > 1:
                 return "ambiguous", list(keys)
-            ns, name = keys[0]
-            cr = self._by_key.get((ns, name))
+            cr = self._by_key.get(keys[0])
             if cr is None:
                 # index said it existed but the record is gone — treat as missing
                 return "missing", None
-            return "ok", (ns, name, cr)
+            return "ok", (*keys[0], cr)
 
     def list_routes(self):
         """All currently indexed routes (serviceIds), sorted."""
@@ -92,7 +97,7 @@ def list_routes(self):
             return sorted(self._by_service.keys())
 
     def all_crs(self):
-        """Snapshot of indexed CRs: {(ns, name): cr}."""
+        """Snapshot of indexed CRs: {(plural, ns, name): cr}."""
         with self._lock:
             return dict(self._by_key)
 
@@ -103,28 +108,39 @@ def _api_client(self):
             self._api = client.CustomObjectsApi()
         return self._api
 
-    def _run(self):
+    def _run(self, plural):
         backoff = _BACKOFF_INITIAL
         while not self._stop.is_set():
             try:
-                self._watch_once()
+                self._watch_once(plural)
                 backoff = _BACKOFF_INITIAL
             except Exception:
                 if self._stop.is_set():
                     break
-                log.exception("cr-index watch error; retry in %.1fs", backoff)
+                log.exception("cr-index %s watch error; retry in %.1fs", plural, backoff)
                 self._stop.wait(backoff)
                 backoff = min(backoff * 2, _BACKOFF_MAX)
 
-    def _watch_once(self):
+    def _watch_once(self, plural):
         api = self._api_client()
         # List first, then watch from the list's resourceVersion. A bare watch only marks the index
         # synced on its first event, and with no CRs in the cluster there is none: the API server
         # sends no ADDED event and its first BOOKMARK can take minutes, so readiness stayed 503 and
         # the liveness probe killed the pod before it ever served. An empty list is a complete one.
-        listing = api.list_cluster_custom_object(group(), VERSION, PLURAL)
-        self._replace((listing or {}).get("items") or [])
-        self._synced.set()
+        try:
+            listing = api.list_cluster_custom_object(group(), VERSION, plural)
+        except ApiException as e:
+            if e.status != 404:
+                raise
+            # CRD not installed (JobSLORequirement is optional): no CRs of this kind is a complete
+            # answer. Look again later, in case it gets installed.
+            log.debug("cr-index: %s not served; retry in %.0fs", plural, _BACKOFF_MAX)
+            self._replace(plural, [])
+            self._synced[plural].set()
+            self._stop.wait(_BACKOFF_MAX)
+            return
+        self._replace(plural, (listing or {}).get("items") or [])
+        self._synced[plural].set()
         rv = ((listing or {}).get("metadata") or {}).get("resourceVersion")
 
         wf = self._watch_factory or watch.Watch
@@ -135,7 +151,7 @@ def _watch_once(self):
         try:
             for event in w.stream(
                     api.list_cluster_custom_object,
-                    group(), VERSION, PLURAL,
+                    group(), VERSION, plural,
                     **kwargs):
                 if self._stop.is_set():
                     return
@@ -144,35 +160,37 @@ def _watch_once(self):
                 if etype == "ERROR":
                     raise RuntimeError(f"watch ERROR event: {cr!r}")
                 if etype == "BOOKMARK":
-                    self._synced.set()
+                    self._synced[plural].set()
                     continue
-                self._apply(etype, cr)
-                self._synced.set()
+                self._apply(plural, etype, cr)
+                self._synced[plural].set()
         finally:
             stop = getattr(w, "stop", None)
             if callable(stop):
                 stop()
 
-    def _replace(self, items):
-        """Rebuild the index from a full list: drops CRs deleted while no watch was running.
+    def _replace(self, plural, items):
+        """Rebuild one kind's part of the index from a full list: drops CRs deleted while no watch
+        was running. Other kinds' entries are kept.
 
         Built aside and swapped in under the lock, so a reader sees the old index or the new one,
         never the empty or half-built one in between -- this runs on every watch reconnect."""
-        by_key, by_service = {}, {}
+        fresh = {}
         for cr in items:
             meta = cr.get("metadata") or {}
             ns, name = meta.get("namespace"), meta.get("name")
-            if not ns or not name:
-                continue
-            sid = (cr.get("spec") or {}).get("serviceId") or ""
-            by_key[(ns, name)] = cr
-            lst = by_service.setdefault(sid, [])
-            if (ns, name) not in lst:
-                lst.append((ns, name))
+            if ns and name:
+                fresh[(plural, ns, name)] = cr
         with self._lock:
+            by_key = {k: v for k, v in self._by_key.items() if k[0] != plural}
+            by_key.update(fresh)
+            by_service = {}
+            for key, cr in by_key.items():
+                sid = (cr.get("spec") or {}).get("serviceId") or ""
+                by_service.setdefault(sid, []).append(key)
             self._by_key, self._by_service = by_key, by_service
 
-    def _apply(self, etype, cr):
+    def _apply(self, plural, etype, cr):
         meta = cr.get("metadata") or {}
         spec = cr.get("spec") or {}
         ns = meta.get("namespace")
@@ -180,29 +198,30 @@ def _apply(self, etype, cr):
         if not ns or not name:
             return
         service_id = spec.get("serviceId") or ""
+        key = (plural, ns, name)
 
         with self._lock:
             if etype == "DELETED":
-                self._by_key.pop((ns, name), None)
+                self._by_key.pop(key, None)
                 lst = self._by_service.get(service_id)
-                if lst and (ns, name) in lst:
-                    lst.remove((ns, name))
+                if lst and key in lst:
+                    lst.remove(key)
                     if not lst:
                         self._by_service.pop(service_id, None)
                 return
 
             # ADDED / MODIFIED
-            prev = self._by_key.get((ns, name))
+            prev = self._by_key.get(key)
             prev_sid = (prev or {}).get("spec", {}).get("serviceId") or ""
             if prev_sid and prev_sid != service_id:
                 # serviceId changed: detach from old key's index
                 lst = self._by_service.get(prev_sid)
-                if lst and (ns, name) in lst:
-                    lst.remove((ns, name))
+                if lst and key in lst:
+                    lst.remove(key)
                     if not lst:
                         self._by_service.pop(prev_sid, None)
 
-            self._by_key[(ns, name)] = cr
+            self._by_key[key] = cr
             lst = self._by_service.setdefault(service_id, [])
-            if (ns, name) not in lst:
-                lst.append((ns, name))
+            if key not in lst:
+                lst.append(key)
diff --git a/src/slo_api/server.py b/src/slo_api/server.py
index 975b5de..bfec745 100644
--- a/src/slo_api/server.py
+++ b/src/slo_api/server.py
@@ -20,7 +20,7 @@
 
 from kubernetes.client import ApiException
 
-from slo_api import PLURAL, VERSION, group
+from slo_api import VERSION, group
 from slo_api.translate import (
     ValidationError, api_to_spec, merge_spec, spec_to_api,
 )
@@ -48,6 +48,21 @@ def _error(status, code, message, field=None, details=None):
     return status, body
 
 
+def _invalid(e):
+    """400 for a 422 from the API server. Built from the Status causes, not
+    its message, which leads with the CR's kind and name; a lone cause's
+    field (`spec.x.y`) becomes the API's own path (`x.y`)."""
+    try:
+        causes = json.loads(e.body)["details"]["causes"]
+        message = "; ".join(c["message"] for c in causes)
+    except (TypeError, ValueError, KeyError):
+        return _error(400, "invalid_request",
+                      f"kubernetes rejected the update: {e.reason}")
+    field = causes[0].get("field", "") if len(causes) == 1 else ""
+    return _error(400, "invalid_request", message,
+                  field=field.removeprefix("spec.") if field != "spec" else None)
+
+
 def make_handler(index, api, auth_token):
     """Factory. All three args dependency-injectable for tests."""
 
@@ -170,10 +185,10 @@ def _resolve(self, route):
             if status == "ambiguous":
                 return _error(
                     409, "conflict",
-                    f"route '{route}' maps to multiple LLMSLORequirement CRs; "
+                    f"route '{route}' maps to multiple SLO requirement CRs; "
                     "cannot auto-select a target",
                     details=[{"namespace": ns, "name": n}
-                             for ns, n in payload]), None
+                             for _, ns, n in payload]), None
             return None, payload
 
         # ---------- handlers ----------
@@ -181,19 +196,19 @@ def _resolve(self, route):
         def _handle_get(self, route):
             err, target = self._resolve(route)
             if err: return self._respond(*err)
-            ns, name, _ = target
-            cr, gerr = self._fresh_get(ns, name, route)
+            plural, ns, name, _ = target
+            cr, gerr = self._fresh_get(plural, ns, name, route)
             if gerr: return self._respond(*gerr)
             return self._respond(200, spec_to_api(route, cr.get("spec") or {}))
 
-        def _fresh_get(self, ns, name, route):
+        def _fresh_get(self, plural, ns, name, route):
             """Read the CR from the apiserver, bypassing the watch cache.
             Guarantees read-after-write consistency: callers that just
             modified the resource see their own writes. Returns
             (cr_dict, None) on success, (None, error_tuple) on failure."""
             try:
                 cr = api.get_namespaced_custom_object(
-                    group(), VERSION, ns, PLURAL, name)
+                    group(), VERSION, ns, plural, name)
                 return cr, None
             except ApiException as e:
                 if e.status == 404:
@@ -206,7 +221,7 @@ def _fresh_get(self, ns, name, route):
         def _handle_put(self, route):
             err, target = self._resolve(route)
             if err: return self._respond(*err)
-            ns, name, _ = target
+            plural, ns, name, _ = target
 
             body, perr = self._read_json()
             if perr: return self._respond(*perr)
@@ -217,14 +232,14 @@ def _handle_put(self, route):
                     400, "invalid_request",
                     "body route does not match URL", field="route"))
             try:
-                update = api_to_spec(body)
+                update = api_to_spec(body, plural)
             except ValidationError as e:
                 return self._respond(*_error(
                     400, "invalid_request", e.message, field=e.field))
 
             # Read the live CR (not the watch cache) so we merge over the
             # freshest state and never silently revert a concurrent write.
-            cr, gerr = self._fresh_get(ns, name, route)
+            cr, gerr = self._fresh_get(plural, ns, name, route)
             if gerr: return self._respond(*gerr)
             existing = cr.get("spec") or {}
             merged = merge_spec(existing, update)
@@ -232,7 +247,7 @@ def _handle_put(self, route):
 
             try:
                 api.patch_namespaced_custom_object(
-                    group(), VERSION, ns, PLURAL, name,
+                    group(), VERSION, ns, plural, name,
                     {"spec": merged},
                 )
             except ApiException as e:
@@ -241,6 +256,10 @@ def _handle_put(self, route):
                     return self._respond(*_error(
                         404, "not_found",
                         f"CR for route '{route}' disappeared during update"))
+                if e.status == 422:
+                    # The CRD's own rules (CEL) rejected the merged spec, e.g.
+                    # min/max bound types that differ: the caller's to fix.
+                    return self._respond(*_invalid(e))
                 return self._respond(*_error(
                     502, "upstream_error",
                     f"kubernetes patch failed: {e.reason}"))
@@ -252,17 +271,19 @@ def _handle_put(self, route):
         def _handle_delete(self, route):
             err, target = self._resolve(route)
             if err: return self._respond(*err)
-            ns, name, _ = target
+            plural, ns, name, _ = target
 
             # Read the live CR (not the watch cache) so we know exactly
             # which SLO fields exist and need nulling out.
-            cr, gerr = self._fresh_get(ns, name, route)
+            cr, gerr = self._fresh_get(plural, ns, name, route)
             if gerr: return self._respond(*gerr)
 
             # Reset SLO-owned fields to their CRD defaults. priority and
             # minimumDeployment have explicit CRD defaults; the other
             # three have none, so "reset" = absent. JSON merge-patch only
-            # removes keys sent explicitly as null.
+            # removes keys sent explicitly as null, and nulls a key the CR
+            # does not have as a no-op: on a JobSLORequirement ttft/otps are
+            # nothing, and its queue (required, no default) is kept.
             patch = {"spec": {
                 "priority": 0,
                 "minimumDeployment": {"type": "replica", "value": 1},
@@ -273,7 +294,7 @@ def _handle_delete(self, route):
 
             try:
                 api.patch_namespaced_custom_object(
-                    group(), VERSION, ns, PLURAL, name, patch,
+                    group(), VERSION, ns, plural, name, patch,
                 )
             except ApiException as e:
                 log.warning("reset %s/%s failed: %s %s", ns, name, e.status, e.reason)
@@ -310,10 +331,10 @@ def _handle_list(self, params):
                     # not 500'd — list offers a "what exists" view, not
                     # target resolution
                     continue
-                _, _, cr = payload
+                *_, cr = payload
                 summary = spec_to_api(r, cr.get("spec") or {})
-                summary.pop("ttft", None)
-                summary.pop("otps", None)
+                for key in ("ttft", "otps", "queue"):
+                    summary.pop(key, None)
                 items.append(summary)
 
             return self._respond(200, {
diff --git a/src/slo_api/static/index.html b/src/slo_api/static/index.html
index 6eb493b..e9c8118 100644
--- a/src/slo_api/static/index.html
+++ b/src/slo_api/static/index.html
@@ -113,15 +113,15 @@ <h3>Deployment bounds</h3>
     <div>
       <label>minimumDeployment</label>
       <div class="row">
-        <select id="minType"><option value="replica">replica</option><option value="concurrency">concurrency</option></select>
-        <input type="number" id="minValue" min="1" step="1">
+        <select id="minType"><option value="replica">replica</option><option value="concurrency">concurrency</option><option value="pdengine">pdengine</option></select>
+        <input type="number" id="minValue" min="0" step="1">
       </div>
     </div>
     <div>
       <label>maximumDeployment</label>
       <div class="row">
-        <select id="maxType"><option value="replica">replica</option></select>
-        <input type="number" id="maxValue" min="1" step="1">
+        <select id="maxType"><option value="replica">replica</option><option value="pdengine">pdengine</option></select>
+        <input type="number" id="maxValue" min="0" step="1">
       </div>
     </div>
   </div>
diff --git a/src/slo_api/translate.py b/src/slo_api/translate.py
index e859fdd..d15d01b 100644
--- a/src/slo_api/translate.py
+++ b/src/slo_api/translate.py
@@ -1,4 +1,4 @@
-"""Translate between the public API shape and the LLMSLORequirement CR spec.
+"""Translate between the public API shape and the SLO requirement CR spec.
 
 API shape (what clients see):
   {
@@ -11,6 +11,9 @@
     "otps": {...}
   }
 
+A JobSLORequirement route has `queue: {maxDepth, promql}` in place of
+`ttft` / `otps`.
+
 CRD shape uses `priority: int 0..10` and adds `serviceId`. Everything
 else is structurally identical. `highPriority=true`  ⇆ priority==10
 `highPriority=false` ⇆ priority in [0..9] (we write 0 on false).
@@ -18,7 +21,9 @@
 
 import copy
 
-from slo_api import HIGH_PRIORITY_VALUE, LOW_PRIORITY_VALUE
+from slo_api import (
+    HIGH_PRIORITY_VALUE, JOB_PLURAL, LLM_PLURAL, LOW_PRIORITY_VALUE,
+)
 
 # Fields the API is authoritative for — everything else in the CR spec
 # is left untouched on write so out-of-band fields (owned by other
@@ -29,11 +34,19 @@
     "maximumDeployment",
     "ttft",
     "otps",
+    "queue",
 )
 
+# Per kind, as its CRD schema has them: the SLO sections of the spec and the
+# bound types. Cross-field rules (min/max types match, pdengine max != 1, ...)
+# are CEL on the CRD, checked by the API server against the merged spec.
+SECTIONS = {LLM_PLURAL: ("ttft", "otps"), JOB_PLURAL: ("queue",)}
+ALLOWED_DEPLOY_TYPES_MIN = {LLM_PLURAL: {"replica", "concurrency", "pdengine"},
+                            JOB_PLURAL: {"replica", "concurrency"}}
+ALLOWED_DEPLOY_TYPES_MAX = {LLM_PLURAL: {"replica", "pdengine"},
+                            JOB_PLURAL: {"replica"}}
+
 ALLOWED_METRIC_TYPES = {"avg", "p50", "p80", "p90", "p95", "p99"}
-ALLOWED_DEPLOY_TYPES_MIN = {"replica", "concurrency"}
-ALLOWED_DEPLOY_TYPES_MAX = {"replica"}  # CRD schema: maximumDeployment only supports replica
 
 
 class ValidationError(ValueError):
@@ -45,10 +58,11 @@ def __init__(self, message, field=None):
 
 # ---------- API → CRD ----------
 
-def api_to_spec(body):
-    """Validate the API request body and return the SLO-owned slice of
-    the CR spec. The caller is responsible for merging onto the
-    existing spec (preserving foreign fields) or replacing wholesale.
+def api_to_spec(body, plural=LLM_PLURAL):
+    """Validate the API request body against the CR kind `plural` and
+    return the SLO-owned slice of the CR spec. The caller is responsible
+    for merging onto the existing spec (preserving foreign fields) or
+    replacing wholesale.
     """
     if not isinstance(body, dict):
         raise ValidationError("request body must be a JSON object")
@@ -64,20 +78,19 @@ def api_to_spec(body):
     if "minimumDeployment" in body:
         spec["minimumDeployment"] = _validate_deployment(
             body["minimumDeployment"], "minimumDeployment",
-            allowed_types=ALLOWED_DEPLOY_TYPES_MIN,
+            allowed_types=ALLOWED_DEPLOY_TYPES_MIN[plural],
         )
     if "maximumDeployment" in body:
         spec["maximumDeployment"] = _validate_deployment(
             body["maximumDeployment"], "maximumDeployment",
-            allowed_types=ALLOWED_DEPLOY_TYPES_MAX,
+            allowed_types=ALLOWED_DEPLOY_TYPES_MAX[plural],
         )
 
-    if "ttft" in body:
-        spec["ttft"] = _validate_slo_section(body["ttft"], "ttft")
-    if "otps" in body:
-        spec["otps"] = _validate_slo_section(body["otps"], "otps")
-
-    _reject_unknown_top_level(body)
+    _reject_unknown_top_level(body, SECTIONS[plural])
+    for key in SECTIONS[plural]:
+        if key in body:
+            validate = _validate_queue if key == "queue" else _validate_slo_section
+            spec[key] = validate(body[key], key)
     return spec
 
 
@@ -105,7 +118,7 @@ def spec_to_api(route, spec):
     priority = spec.get("priority", LOW_PRIORITY_VALUE)
     out["highPriority"] = (priority == HIGH_PRIORITY_VALUE)
 
-    for key in ("minimumDeployment", "maximumDeployment", "ttft", "otps"):
+    for key in SLO_FIELDS[1:]:  # all but priority, translated above
         if key in spec:
             out[key] = copy.deepcopy(spec[key])
     return out
@@ -113,9 +126,9 @@ def spec_to_api(route, spec):
 
 # ---------- validators ----------
 
-def _reject_unknown_top_level(body):
+def _reject_unknown_top_level(body, sections):
     allowed = {"route", "highPriority", "minimumDeployment",
-               "maximumDeployment", "ttft", "otps"}
+               "maximumDeployment", *sections}
     unknown = set(body) - allowed
     if unknown:
         raise ValidationError(
@@ -131,12 +144,37 @@ def _validate_deployment(value, field, allowed_types):
             f"{field}.type must be one of {sorted(allowed_types)}",
             f"{field}.type")
     dep_value = value.get("value")
-    if not isinstance(dep_value, int) or isinstance(dep_value, bool) or dep_value < 1:
+    # 0 is scale-to-zero; the CRD's CEL ties it to a maximum of 0 too.
+    if not isinstance(dep_value, int) or isinstance(dep_value, bool) or dep_value < 0:
         raise ValidationError(
-            f"{field}.value must be an integer >= 1", f"{field}.value")
+            f"{field}.value must be an integer >= 0", f"{field}.value")
     return {"type": dep_type, "value": dep_value}
 
 
+def _validate_queue(value, field):
+    if not isinstance(value, dict):
+        raise ValidationError(f"{field} must be an object", field)
+    unknown = set(value) - {"maxDepth", "promql"}
+    if unknown:
+        raise ValidationError(
+            f"{field}: unknown field(s): {', '.join(sorted(unknown))}",
+            field)
+    depth = value.get("maxDepth")
+    if not isinstance(depth, int) or isinstance(depth, bool) or depth < 0:
+        raise ValidationError(
+            f"{field}.maxDepth must be an integer >= 0", f"{field}.maxDepth")
+    out = {"maxDepth": depth}
+    # Optional: the write is a merge patch, so a promql left out keeps the
+    # CR's own (which the CRD requires).
+    if "promql" in value:
+        promql = value["promql"]
+        if not isinstance(promql, str) or not promql.strip():
+            raise ValidationError(
+                f"{field}.promql must be a non-empty string", f"{field}.promql")
+        out["promql"] = promql
+    return out
+
+
 def _validate_slo_section(value, field):
     if not isinstance(value, dict):
         raise ValidationError(f"{field} must be an object", field)
diff --git a/tests/conftest.py b/tests/conftest.py
index a12018a..6e5fe81 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -3,38 +3,45 @@
 
 import pytest
 
+from slo_api import JOB_PLURAL, LLM_PLURAL
 from slo_api.cr_index import CRIndex
 
 
-CRD_TEMPLATE = lambda ns, name, sid, extra_spec=None: {
+PLURAL_OF = {"LLMSLORequirement": LLM_PLURAL, "JobSLORequirement": JOB_PLURAL}
+
+CRD_TEMPLATE = lambda ns, name, sid, extra_spec=None, kind="LLMSLORequirement": {
     "apiVersion": "inference.modelsphere.dev/v1alpha1",
-    "kind": "LLMSLORequirement",
+    "kind": kind,
     "metadata": {"namespace": ns, "name": name, "resourceVersion": "1"},
     "spec": {"serviceId": sid, **(extra_spec or {})},
 }
 
 
 class FakeCustomObjectsApi:
-    """In-memory CustomObjectsApi for tests."""
+    """In-memory CustomObjectsApi for tests. A plural in `missing` has no CRD: 404."""
 
-    def __init__(self):
+    def __init__(self, missing=()):
         self._lock = threading.Lock()
-        self._store = {}  # (ns, name) -> CR
+        self._store = {}  # (plural, ns, name) -> CR
+        self.missing = set(missing)
 
     def seed(self, cr):
         ns = cr["metadata"]["namespace"]
         name = cr["metadata"]["name"]
         with self._lock:
-            self._store[(ns, name)] = copy.deepcopy(cr)
+            self._store[(PLURAL_OF[cr["kind"]], ns, name)] = copy.deepcopy(cr)
 
     def list_cluster_custom_object(self, group, version, plural, **kw):
+        if plural in self.missing:
+            from kubernetes.client import ApiException
+            raise ApiException(status=404, reason="Not Found")
         with self._lock:
-            items = [copy.deepcopy(c) for c in self._store.values()]
+            items = [copy.deepcopy(c) for k, c in self._store.items() if k[0] == plural]
         return {"items": items}
 
     def get_namespaced_custom_object(self, group, version, ns, plural, name):
         with self._lock:
-            cr = self._store.get((ns, name))
+            cr = self._store.get((plural, ns, name))
             if cr is None:
                 from kubernetes.client import ApiException
                 raise ApiException(status=404, reason="Not Found")
@@ -52,7 +59,7 @@ def merge(dst, src):
                     dst[k] = v
 
         with self._lock:
-            cr = self._store.get((ns, name))
+            cr = self._store.get((plural, ns, name))
             if cr is None:
                 from kubernetes.client import ApiException
                 raise ApiException(status=404, reason="Not Found")
@@ -70,9 +77,9 @@ def __init__(self, api, outer_stop):
         self._api = api
         self._outer_stop = outer_stop
 
-    def stream(self, fn, *args, **kwargs):
+    def stream(self, fn, group, version, plural, **kwargs):
         import time
-        for cr in self._api.list_cluster_custom_object("", "", "")["items"]:
+        for cr in self._api.list_cluster_custom_object(group, version, plural)["items"]:
             yield {"type": "ADDED", "object": cr}
         # BOOKMARK signals end-of-initial-list; index uses it to mark synced
         yield {"type": "BOOKMARK", "object": {}}
diff --git a/tests/test_cr_index_sync.py b/tests/test_cr_index_sync.py
index d9240b2..1892c3a 100644
--- a/tests/test_cr_index_sync.py
+++ b/tests/test_cr_index_sync.py
@@ -1,5 +1,6 @@
 import threading
 
+from slo_api import JOB_PLURAL, LLM_PLURAL
 from slo_api.cr_index import CRIndex
 
 from conftest import CRD_TEMPLATE, FakeCustomObjectsApi
@@ -57,7 +58,7 @@ def test_relist_never_shows_an_empty_index():
     api = FakeCustomObjectsApi()
     old = [CRD_TEMPLATE("ns1", "a", "route-a"), CRD_TEMPLATE("ns2", "b", "route-b")]
     idx = _index(api)
-    idx._replace(old)
+    idx._replace(LLM_PLURAL, old)
     seen = []
 
     def relisted():
@@ -65,6 +66,28 @@ def relisted():
             seen.append(idx.list_routes())   # what a reader sees while the rebuild is under way
             yield cr
 
-    idx._replace(relisted())
+    idx._replace(LLM_PLURAL, relisted())
     assert all(s == ["route-a", "route-b"] for s in seen), seen
     assert idx.list_routes() == ["route-a", "route-b", "route-c"]
+
+
+def test_missing_job_crd_is_synced_and_serves_llm_routes():
+    """JobSLORequirement is optional: a cluster without its CRD must not hold readiness at 503."""
+    api = FakeCustomObjectsApi(missing={JOB_PLURAL})
+    api.seed(CRD_TEMPLATE("ns1", "a", "route-a"))
+    idx = _index(api)
+    idx.start()
+    try:
+        assert idx.wait_synced(timeout=2)
+        assert idx.list_routes() == ["route-a"]
+    finally:
+        idx.stop()
+
+
+def test_relist_of_one_kind_keeps_the_other():
+    api = FakeCustomObjectsApi()
+    idx = _index(api)
+    idx._replace(LLM_PLURAL, [CRD_TEMPLATE("ns1", "a", "route-a")])
+    idx._replace(JOB_PLURAL, [CRD_TEMPLATE("ns1", "j", "route-j", kind="JobSLORequirement")])
+    idx._replace(LLM_PLURAL, [CRD_TEMPLATE("ns1", "b", "route-b")])
+    assert idx.list_routes() == ["route-b", "route-j"]
diff --git a/tests/test_server.py b/tests/test_server.py
index feea0c6..30a8488 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -8,6 +8,7 @@
 import pytest
 
 from slo_api.server import serve
+from slo_api import JOB_PLURAL, LLM_PLURAL
 from tests.conftest import CRD_TEMPLATE, FakeWatch
 from slo_api.cr_index import CRIndex
 
@@ -120,7 +121,7 @@ def test_get_returns_full_config(running_server):
     spec = {k: v for k, v in BASE_CR_SPEC.items() if k != "serviceId"}
     cr = CRD_TEMPLATE("kimi", "kimi-k25", "kimi-k2.5", extra_spec=spec)
     api.seed(cr)
-    idx._apply("ADDED", cr)
+    idx._apply(LLM_PLURAL, "ADDED", cr)
 
     st, body = _req("GET", f"{url}/config/kimi-k2.5")
     assert st == 200
@@ -152,7 +153,7 @@ def test_put_validation_error_returns_400(running_server):
     url, api, idx = running_server
     cr = CRD_TEMPLATE("kimi", "kimi-k25", "kimi-k2.5")
     api.seed(cr)
-    idx._apply("ADDED", cr)
+    idx._apply(LLM_PLURAL, "ADDED", cr)
 
     st, body = _req("PUT", f"{url}/config/kimi-k2.5",
                     body={"highPriority": "yes"})
@@ -167,7 +168,7 @@ def test_put_updates_priority_and_preserves_foreign_fields(running_server):
         "someForeignField": {"keep": "me"},
     })
     api.seed(cr)
-    idx._apply("ADDED", cr)
+    idx._apply(LLM_PLURAL, "ADDED", cr)
 
     body = {
         "highPriority": True,
@@ -193,7 +194,7 @@ def test_put_body_route_mismatch_returns_400(running_server):
     url, api, idx = running_server
     cr = CRD_TEMPLATE("kimi", "kimi-k25", "kimi-k2.5")
     api.seed(cr)
-    idx._apply("ADDED", cr)
+    idx._apply(LLM_PLURAL, "ADDED", cr)
 
     st, body = _req("PUT", f"{url}/config/kimi-k2.5",
                     body={"route": "other", "highPriority": True})
@@ -218,7 +219,7 @@ def test_delete_resets_slo_fields_keeps_cr(running_server):
         "someForeignField": {"keep": "me"},
     })
     api.seed(cr)
-    idx._apply("ADDED", cr)
+    idx._apply(LLM_PLURAL, "ADDED", cr)
 
     st, _ = _req("DELETE", f"{url}/config/kimi-k2.5")
     assert st == 204
@@ -261,7 +262,7 @@ def test_list_pagination_and_summary_shape(running_server):
             "maximumDeployment": {"type": "replica", "value": i + 5},
         })
         api.seed(cr)
-        idx._apply("ADDED", cr)
+        idx._apply(LLM_PLURAL, "ADDED", cr)
 
     st, body = _req("GET", f"{url}/config?page=1&pageSize=2")
     assert st == 200
@@ -297,7 +298,7 @@ def test_ambiguous_route_returns_409(running_server):
     cr1 = CRD_TEMPLATE("ns-a", "k25", "kimi-k2.5")
     cr2 = CRD_TEMPLATE("ns-b", "k25", "kimi-k2.5")
     api.seed(cr1); api.seed(cr2)
-    idx._apply("ADDED", cr1); idx._apply("ADDED", cr2)
+    idx._apply(LLM_PLURAL, "ADDED", cr1); idx._apply(LLM_PLURAL, "ADDED", cr2)
 
     st, body = _req("GET", f"{url}/config/kimi-k2.5")
     assert st == 409
@@ -306,3 +307,122 @@ def test_ambiguous_route_returns_409(running_server):
     st, _ = _req("PUT", f"{url}/config/kimi-k2.5",
                  body={"highPriority": True})
     assert st == 409
+
+
+# ---------- JobSLORequirement ----------
+
+JOB_SPEC = {
+    "priority": 0,
+    "minimumDeployment": {"type": "replica", "value": 1},
+    "maximumDeployment": {"type": "replica", "value": 4},
+    "queue": {"maxDepth": 8, "promql": 'sum(jobs_pending{service="img"})'},
+}
+
+
+def _seed_job(api, idx, sid="img-gen"):
+    cr = CRD_TEMPLATE("jobs", "img", sid, extra_spec=JOB_SPEC, kind="JobSLORequirement")
+    api.seed(cr)
+    idx._apply(JOB_PLURAL, "ADDED", cr)
+
+
+def _job_cr(api):
+    return api.get_namespaced_custom_object(
+        "inference.modelsphere.dev", "v1alpha1", "jobs", JOB_PLURAL, "img")
+
+
+def test_job_route_get_put_keeps_promql(running_server):
+    url, api, idx = running_server
+    _seed_job(api, idx)
+
+    st, body = _req("GET", f"{url}/config/img-gen")
+    assert st == 200
+    assert body["queue"] == JOB_SPEC["queue"]
+
+    st, body = _req("PUT", f"{url}/config/img-gen", body={"queue": {"maxDepth": 2}})
+    assert st == 200, body
+    assert _job_cr(api)["spec"]["queue"] == {
+        "maxDepth": 2, "promql": JOB_SPEC["queue"]["promql"]}
+
+
+def test_job_route_rejects_llm_fields(running_server):
+    url, api, idx = running_server
+    _seed_job(api, idx)
+    st, body = _req("PUT", f"{url}/config/img-gen", body={
+        "otps": {"default": {"metrics": [{"type": "p80", "threshold": 1}]}}})
+    assert st == 400
+    st, body = _req("PUT", f"{url}/config/img-gen", body={
+        "maximumDeployment": {"type": "pdengine", "value": 4}})
+    assert st == 400
+    assert body["error"]["field"] == "maximumDeployment.type"
+
+
+def test_job_route_delete_keeps_queue(running_server):
+    url, api, idx = running_server
+    _seed_job(api, idx)
+    st, _ = _req("DELETE", f"{url}/config/img-gen")
+    assert st == 204
+    spec = _job_cr(api)["spec"]
+    assert spec["queue"] == JOB_SPEC["queue"]
+    assert "maximumDeployment" not in spec
+
+
+def test_list_includes_job_routes_without_queue(running_server):
+    url, api, idx = running_server
+    _seed_job(api, idx)
+    st, body = _req("GET", f"{url}/config")
+    assert st == 200
+    assert [it["route"] for it in body["items"]] == ["img-gen"]
+    assert "queue" not in body["items"][0]
+
+
+def test_route_shared_by_both_kinds_is_ambiguous(running_server):
+    url, api, idx = running_server
+    _seed_job(api, idx, sid="kimi-k2.5")
+    cr = CRD_TEMPLATE("kimi", "kimi-k25", "kimi-k2.5")
+    api.seed(cr)
+    idx._apply(LLM_PLURAL, "ADDED", cr)
+    st, body = _req("GET", f"{url}/config/kimi-k2.5")
+    assert st == 409
+    assert len(body["error"]["details"]) == 2
+
+
+# ---------- CRD rules ----------
+
+def _reject_with(api, causes):
+    """Make the fake API server answer patches the way it does when the CRD rejects the spec."""
+    from kubernetes.client import ApiException
+    def reject(*a, **kw):
+        e = ApiException(status=422, reason="Unprocessable Entity")
+        e.body = json.dumps({
+            "kind": "Status", "code": 422, "reason": "Invalid",
+            "message": 'LLMSLORequirement.inference.modelsphere.dev "kimi-k25" is invalid: ...',
+            "details": {"name": "kimi-k25", "causes": causes},
+        })
+        raise e
+    api.patch_namespaced_custom_object = reject
+
+
+@pytest.mark.parametrize("cause,field", [
+    # CEL rule on the merged spec: no single field to blame.
+    ({"reason": "FieldValueInvalid", "field": "spec",
+      "message": 'Invalid value: "object": minimumDeployment.type and '
+                 'maximumDeployment.type must match'}, None),
+    # Schema enum, e.g. pdengine before the CRD has it: the field, in API terms.
+    ({"reason": "FieldValueNotSupported", "field": "spec.maximumDeployment.type",
+      "message": 'Unsupported value: "pdengine": supported values: "replica"'},
+     "maximumDeployment.type"),
+])
+def test_crd_rule_rejection_is_400_without_cr_name(running_server, cause, field):
+    url, api, idx = running_server
+    cr = CRD_TEMPLATE("kimi", "kimi-k25", "kimi-k2.5")
+    api.seed(cr)
+    idx._apply(LLM_PLURAL, "ADDED", cr)
+    _reject_with(api, [cause])
+
+    st, body = _req("PUT", f"{url}/config/kimi-k2.5", body={
+        "maximumDeployment": {"type": "pdengine", "value": 4}})
+    assert st == 400
+    assert body["error"]["code"] == "invalid_request"
+    assert body["error"]["message"] == cause["message"]
+    assert body["error"].get("field") == field
+    assert "kimi-k25" not in json.dumps(body)
diff --git a/tests/test_translate.py b/tests/test_translate.py
index 2800c8b..2297574 100644
--- a/tests/test_translate.py
+++ b/tests/test_translate.py
@@ -1,5 +1,6 @@
 import pytest
 
+from slo_api import JOB_PLURAL
 from slo_api.translate import (
     ValidationError, api_to_spec, merge_spec, spec_to_api,
 )
@@ -36,8 +37,15 @@ def test_maximum_deployment_rejects_concurrency():
     assert e.value.field == "maximumDeployment.type"
 
 
-def test_deployment_value_must_be_positive_int():
-    for bad in (0, -1, "two", 1.5, True):
+def test_deployment_value_zero_is_scale_to_zero():
+    spec = api_to_spec({"minimumDeployment": {"type": "replica", "value": 0},
+                        "maximumDeployment": {"type": "replica", "value": 0}})
+    assert spec["minimumDeployment"]["value"] == 0
+    assert spec["maximumDeployment"]["value"] == 0
+
+
+def test_deployment_value_must_be_non_negative_int():
+    for bad in (-1, "two", 1.5, True):
         with pytest.raises(ValidationError):
             api_to_spec({"minimumDeployment": {"type": "replica", "value": bad}})
 
@@ -162,3 +170,50 @@ def test_spec_to_api_roundtrip_full_body():
     spec["serviceId"] = "kimi-k2.5"
     out = spec_to_api("kimi-k2.5", spec)
     assert out == {"route": "kimi-k2.5", **body}
+
+
+# ---------- per kind ----------
+
+def test_pdengine_bounds_accepted_on_llm():
+    spec = api_to_spec({"minimumDeployment": {"type": "pdengine", "value": 2},
+                        "maximumDeployment": {"type": "pdengine", "value": 8}})
+    assert spec["maximumDeployment"] == {"type": "pdengine", "value": 8}
+
+
+def test_job_rejects_pdengine_and_llm_sections():
+    with pytest.raises(ValidationError) as e:
+        api_to_spec({"maximumDeployment": {"type": "pdengine", "value": 4}}, JOB_PLURAL)
+    assert e.value.field == "maximumDeployment.type"
+    with pytest.raises(ValidationError) as e:
+        api_to_spec({"ttft": {"default": {"metrics": [{"type": "p80", "threshold": 1}]}}},
+                    JOB_PLURAL)
+    assert "ttft" in e.value.message
+
+
+def test_llm_rejects_queue():
+    with pytest.raises(ValidationError):
+        api_to_spec({"queue": {"maxDepth": 4}})
+
+
+def test_job_queue_promql_optional():
+    assert api_to_spec({"queue": {"maxDepth": 0}}, JOB_PLURAL) == {"queue": {"maxDepth": 0}}
+    body = {"queue": {"maxDepth": 4, "promql": 'sum(jobs_pending{service="x"})'}}
+    assert api_to_spec(body, JOB_PLURAL) == body
+
+
+@pytest.mark.parametrize("queue,field", [
+    ({}, "queue.maxDepth"),
+    ({"maxDepth": -1}, "queue.maxDepth"),
+    ({"maxDepth": True}, "queue.maxDepth"),
+    ({"maxDepth": 1, "promql": " "}, "queue.promql"),
+    ({"maxDepth": 1, "extra": 1}, "queue"),
+])
+def test_job_queue_invalid(queue, field):
+    with pytest.raises(ValidationError) as e:
+        api_to_spec({"queue": queue}, JOB_PLURAL)
+    assert e.value.field == field
+
+
+def test_spec_to_api_includes_queue():
+    out = spec_to_api("j", {"serviceId": "j", "queue": {"maxDepth": 3, "promql": "q"}})
+    assert out["queue"] == {"maxDepth": 3, "promql": "q"}