Add a minimal if pipeline is compatible mode

.github/workflows/pipeline.yaml

@@ -286,7 +286,7 @@ jobs:
       # associated with roxanne+test_edge account since for some reason it was failing with
       # prod biggies
       GROUNDLIGHT_API_TOKEN: ${{ secrets.K3S_TEST_GROUNDLIGHT_API_TOKEN }}
-      USE_MINIMAL_IMAGE: true
+      INFERENCE_IMAGE_MODE: fully_minimal
     steps:
       - name: Check out code
         uses: actions/checkout@v4

app/api/routes/image_queries.py

@@ -14,6 +14,7 @@
     refresh_detector_metadata_if_needed,
 )
 from app.core.edge_config_manager import EdgeConfigManager
+from app.core.inference_image import detector_uses_minimal_image
 from app.core.naming import get_edge_inference_model_name
 from app.core.utils import create_iq, generate_iq_id, generate_metadata_dict, generate_request_id
 from app.escalation_queue.models import SubmitImageQueryParams
@@ -288,7 +289,7 @@ async def post_image_query(  # noqa: PLR0913, PLR0915, PLR0912
             }
         )
 
-        if app_state.separate_oodd_inference:
+        if not detector_uses_minimal_image(detector_id, app_state.db_manager):
             oodd_model_name = get_edge_inference_model_name(detector_id=detector_id, is_oodd=True)
             app_state.db_manager.create_or_update_inference_deployment_record(
                 deployment={

app/core/app_state.py

@@ -1,5 +1,4 @@
 import logging
-import os
 import time
 from functools import lru_cache
 
@@ -22,8 +21,6 @@
 MAX_DETECTOR_IDS_CACHE_SIZE = 1000
 STALE_METADATA_THRESHOLD_SEC = 60  # 60 seconds
 
-USE_MINIMAL_IMAGE = os.environ.get("USE_MINIMAL_IMAGE", "false") == "true"
-
 
 @lru_cache(maxsize=MAX_SDK_INSTANCES_CACHE_SIZE)
 @trace_span
@@ -101,12 +98,8 @@ def get_detector_metadata(detector_id: str, gl: Groundlight) -> Detector:
 
 class AppState:
     def __init__(self):
-        # We only launch a separate OODD inference pod if we are not using the minimal image.
-        # Pipelines used in the minimal image include OODD inference and confidence adjustment,
-        # so they do not need to be adjusted separately.
-        self.separate_oodd_inference = not USE_MINIMAL_IMAGE
-        self.edge_inference_manager = EdgeInferenceManager(separate_oodd_inference=self.separate_oodd_inference)
         self.db_manager = DatabaseManager()
+        self.edge_inference_manager = EdgeInferenceManager(db_manager=self.db_manager)
         self.is_ready = False
         self.queue_writer = QueueWriter()
 

app/core/database.py

@@ -9,6 +9,7 @@
 
 from app.core.file_paths import DATABASE_FILEPATH, DATABASE_ORM_LOG_FILE, DATABASE_ORM_LOG_FILE_SIZE
 from app.core.models import Base, InferenceDeployment
+from app.core.naming import get_edge_inference_model_name
 
 logger = logging.getLogger(__name__)
 
@@ -102,6 +103,13 @@ def update_inference_deployment_record(self, model_name: str, fields_to_update:
                     setattr(detector_record, field, value)
             session.commit()
 
+    def get_inference_deployment_record(self, detector_id: str, is_oodd: bool = False) -> InferenceDeployment | None:
+        """Return the primary or OODD record for a detector, or None if it doesn't exist."""
+        model_name = get_edge_inference_model_name(detector_id, is_oodd=is_oodd)
+        with self.session_maker() as session:
+            query = select(InferenceDeployment).filter_by(model_name=model_name)
+            return session.execute(query).scalar_one_or_none()
+
     def get_inference_deployment_records(self, **kwargs) -> Sequence[InferenceDeployment]:
         """
         Query the database table for detectors based on a given query predicate.

app/core/edge_inference.py

@@ -15,8 +15,15 @@
 from jinja2 import Template
 from model import ModeEnum
 
+from app.core.database import DatabaseManager
 from app.core.edge_config_manager import EdgeConfigManager
 from app.core.file_paths import MODEL_REPOSITORY_PATH
+from app.core.inference_image import (
+    INFERENCE_IMAGE_MODE,
+    MODE_FULLY_MINIMAL,
+    MODE_STANDARD,
+    detector_uses_minimal_image,
+)
 from app.core.naming import (
     get_detector_models_dir,
     get_edge_inference_service_name,
@@ -33,6 +40,16 @@
 # This will be process-specific, so each edge-endpoint worker will have its own cache instance.
 ttl_cache = TTLCache(maxsize=128, ttl=5)
 
+# Per-worker TTL cache for the "run separate OODD?" decision. Keyed by detector_id only (db_manager dropped from key)
+# so repeated inference calls skip the DB read. TTL matches is_edge_inference_ready so staleness windows align;
+# a minimal_compatible flip triggers a pod redeployment that takes far longer than 5s to complete.
+_separate_oodd_cache: TTLCache = TTLCache(maxsize=128, ttl=5)
+
+
+@cached(_separate_oodd_cache, key=lambda detector_id, db_manager: detector_id)
+def _uses_separate_oodd_cached(detector_id: str, db_manager) -> bool:
+    return not detector_uses_minimal_image(detector_id, db_manager)
+
 
 @cached(ttl_cache)
 def is_edge_inference_ready(inference_client_url: str) -> bool:
@@ -258,27 +275,26 @@ class EdgeInferenceManager:
 
     def __init__(
         self,
+        db_manager: DatabaseManager,
         verbose: bool = False,
-        separate_oodd_inference: bool = True,
     ) -> None:
         self.verbose = verbose
+        self.db_manager = db_manager
         self.speedmon = SpeedMonitor()
-        self.separate_oodd_inference = separate_oodd_inference
         self.last_escalation_times: dict[str, float | None] = {}
 
+    def uses_separate_oodd(self, detector_id: str) -> bool:
+        """Whether this detector runs OODD as a separate pod (full image) vs folded into primary (minimal image)."""
+        return _uses_separate_oodd_cached(detector_id, self.db_manager)
+
     @trace_span
     def inference_is_available(self, detector_id: str) -> bool:
         """Check whether inference pods for this detector are ready to serve."""
+        separate_oodd = self.uses_separate_oodd(detector_id)
         primary_url = get_edge_inference_service_name(detector_id) + ":8000"
-        oodd_url = (
-            get_edge_inference_service_name(detector_id, is_oodd=True) + ":8000"
-            if self.separate_oodd_inference
-            else None
-        )
+        oodd_url = get_edge_inference_service_name(detector_id, is_oodd=True) + ":8000" if separate_oodd else None
 
-        ready = is_edge_inference_ready(primary_url) and (
-            not self.separate_oodd_inference or is_edge_inference_ready(oodd_url)
-        )
+        ready = is_edge_inference_ready(primary_url) and (not separate_oodd or is_edge_inference_ready(oodd_url))
         if not ready:
             logger.debug(
                 f"Edge inference server and/or OODD inference server is not ready. {primary_url=}, {oodd_url=}"
@@ -308,8 +324,9 @@ def run_inference(self, detector_id: str, image_bytes: bytes, content_type: str,
         logger.info(f"Submitting image to edge inference service. {detector_id=}")
         start_time = time.perf_counter()
 
+        separate_oodd = self.uses_separate_oodd(detector_id)
         primary_url = get_edge_inference_service_name(detector_id) + ":8000"
-        if self.separate_oodd_inference:
+        if separate_oodd:
             oodd_url = get_edge_inference_service_name(detector_id, is_oodd=True) + ":8000"
             with ThreadPoolExecutor(max_workers=2) as executor:
                 if get_current_tracer() is not None:
@@ -340,7 +357,7 @@ def run_inference(self, detector_id: str, image_bytes: bytes, content_type: str,
             primary_dir = get_primary_edge_model_dir(self.MODEL_REPOSITORY, detector_id)
             if mlb_key := get_current_model_ksuid(primary_dir, primary_version):
                 output_dict["mlb_key"] = mlb_key
-            if self.separate_oodd_inference and oodd_response is not None:
+            if separate_oodd and oodd_response is not None:
                 oodd_version = get_current_model_version(self.MODEL_REPOSITORY, detector_id, is_oodd=True)
                 oodd_dir = get_oodd_model_dir(self.MODEL_REPOSITORY, detector_id)
                 if oodd_mlb_key := get_current_model_ksuid(oodd_dir, oodd_version):
@@ -356,12 +373,13 @@ def run_inference(self, detector_id: str, image_bytes: bytes, content_type: str,
         logger.info(f"Recent-average FPS for {detector_id=}: {fps:.2f}")
         return output_dict
 
-    def update_models_if_available(self, detector_id: str) -> bool:
+    def sync_models_from_cloud(self, detector_id: str) -> tuple[bool, bool]:
         """
-        Request a new model from Groundlight. If there is a new model available for primary or OODD
-        inference, download it and write it to the model repository as a new version.
+        Fetch model metadata from Groundlight and save new primary/OODD binaries when available.
 
-        Returns True if a new model was downloaded and saved, False otherwise.
+        Returns ``(models_saved, minimal_compatible)`` where ``models_saved`` is whether anything
+        was downloaded and written to the model repository this cycle and ``minimal_compatible``
+        is the cloud's latest assertion for the primary pipeline (always from this fetch).
         """
         logger.debug(f"Checking if there are new models available for {detector_id}")
 
@@ -373,11 +391,19 @@ def update_models_if_available(self, detector_id: str) -> bool:
 
         edge_model_info, oodd_model_info = fetch_model_info(detector_id, api_token=api_token)
 
+        # Decide OODD update from the just-fetched flag rather than the (possibly stale) DB row.
+        if INFERENCE_IMAGE_MODE == MODE_FULLY_MINIMAL:
+            run_separate_oodd = False
+        elif INFERENCE_IMAGE_MODE == MODE_STANDARD:
+            run_separate_oodd = True
+        else:
+            run_separate_oodd = not edge_model_info.minimal_compatible
+
         primary_version = get_current_model_version(self.MODEL_REPOSITORY, detector_id)
         primary_edge_model_dir = get_primary_edge_model_dir(self.MODEL_REPOSITORY, detector_id)
         update_primary_model = should_update(edge_model_info, primary_edge_model_dir, primary_version)
 
-        if self.separate_oodd_inference:
+        if run_separate_oodd:
             oodd_version = get_current_model_version(self.MODEL_REPOSITORY, detector_id, is_oodd=True)
             oodd_model_dir = get_oodd_model_dir(self.MODEL_REPOSITORY, detector_id)
             update_oodd_model = should_update(oodd_model_info, oodd_model_dir, oodd_version)
@@ -386,7 +412,7 @@ def update_models_if_available(self, detector_id: str) -> bool:
 
         if not update_primary_model and not update_oodd_model:
             logger.debug(f"No new models available for {detector_id}")
-            return False
+            return False, edge_model_info.minimal_compatible
 
         logger.info(f"At least one new model is available for {detector_id}, saving models to repository.")
         save_models_to_repository(
@@ -397,7 +423,7 @@ def update_models_if_available(self, detector_id: str) -> bool:
             oodd_model_info=oodd_model_info if update_oodd_model else None,
             repository_root=self.MODEL_REPOSITORY,
         )
-        return True
+        return True, edge_model_info.minimal_compatible
 
     @trace_span
     def escalation_cooldown_complete(self, detector_id: str, edge_config: EdgeEndpointConfig) -> bool:

app/core/inference_image.py

@@ -0,0 +1,61 @@
+"""Per-detector inference image flavor selection.
+
+Three modes set globally via ``INFERENCE_IMAGE_MODE``. The concrete container image URIs
+for each flavor are ``FULL_INFERENCE_IMAGE_URI`` and ``MINIMAL_INFERENCE_IMAGE_URI``
+(set by Helm from registry + tag):
+
+- ``standard``: every detector runs on the full image.
+- ``minimal_if_compatible``: per-detector — minimal when the cloud reported
+  ``minimal_compatible=True``, full otherwise.
+- ``fully_minimal``: every detector runs on the minimal image.
+
+``minimal_compatible`` is persisted on the primary ``InferenceDeployment`` row by
+the model-updater after each fetch-model-urls cycle. The same selection is the
+single signal for both the deployment image (minimal vs full) and whether to run
+OODD as a separate pod (full → yes, minimal → folded into primary), so all
+three call sites — model-updater, ``EdgeInferenceManager.run_inference``, and
+deletion path — derive from this module and stay in agreement.
+"""
+
+import logging
+import os
+
+from app.core.database import DatabaseManager
+
+logger = logging.getLogger(__name__)
+
+MODE_STANDARD = "standard"
+MODE_MINIMAL_IF_COMPATIBLE = "minimal_if_compatible"
+MODE_FULLY_MINIMAL = "fully_minimal"
+
+VALID_MODES = {MODE_STANDARD, MODE_MINIMAL_IF_COMPATIBLE, MODE_FULLY_MINIMAL}
+
+INFERENCE_IMAGE_MODE = os.environ.get("INFERENCE_IMAGE_MODE", MODE_STANDARD)
+if INFERENCE_IMAGE_MODE not in VALID_MODES:
+    raise ValueError(
+        f"INFERENCE_IMAGE_MODE={INFERENCE_IMAGE_MODE!r} is not one of {sorted(VALID_MODES)}. "
+        "The helm chart enforces this via values.schema.json; an invalid value here means the "
+        "env var was set manually outside of helm."
+    )
+
+FULL_INFERENCE_IMAGE_URI = os.environ.get("FULL_INFERENCE_IMAGE_URI", "")
+MINIMAL_INFERENCE_IMAGE_URI = os.environ.get("MINIMAL_INFERENCE_IMAGE_URI", "")
+
+
+def detector_uses_minimal_image(detector_id: str, db_manager: DatabaseManager) -> bool:
+    """Whether ``detector_id`` should run on the minimal inference image."""
+    if INFERENCE_IMAGE_MODE == MODE_FULLY_MINIMAL:
+        return True
+    if INFERENCE_IMAGE_MODE == MODE_STANDARD:
+        return False
+    record = db_manager.get_inference_deployment_record(detector_id, is_oodd=False)
+    return bool(record and record.minimal_compatible)
+
+
+def detector_image(detector_id: str, db_manager: DatabaseManager) -> str:
+    """The fully-qualified inference image (incl. tag) for ``detector_id``."""
+    return (
+        MINIMAL_INFERENCE_IMAGE_URI
+        if detector_uses_minimal_image(detector_id, db_manager)
+        else FULL_INFERENCE_IMAGE_URI
+    )

app/core/kubernetes_management.py

@@ -8,15 +8,18 @@
 from kubernetes import config
 from kubernetes.client import V1Deployment
 
+from .database import DatabaseManager
 from .edge_inference import get_current_model_version
 from .file_paths import INFERENCE_DEPLOYMENT_TEMPLATE_PATH, KUBERNETES_NAMESPACE_PATH, MODEL_REPOSITORY_PATH
+from .inference_image import detector_image
 from .naming import get_edge_inference_deployment_name, get_edge_inference_model_name, get_edge_inference_service_name
 
 logger = logging.getLogger(__name__)
 
 
 class InferenceDeploymentManager:
-    def __init__(self) -> None:
+    def __init__(self, db_manager: DatabaseManager) -> None:
+        self._db_manager = db_manager
         self._setup_kube_client()
         self._inference_deployment_template = self._load_inference_deployment_template()
 
@@ -82,7 +85,7 @@ def _create_from_kube_manifest(self, namespace: str, manifest: str) -> None:
                 else:
                     raise e
 
-    def _substitute_placeholders(self, service_name: str, deployment_name: str, model_name: str) -> str:
+    def _substitute_placeholders(self, service_name: str, deployment_name: str, model_name: str, image: str) -> str:
         inference_deployment = self._inference_deployment_template
         inference_deployment = inference_deployment.replace("placeholder-inference-service-name", service_name)
         inference_deployment = inference_deployment.replace("placeholder-inference-deployment-name", deployment_name)
@@ -92,6 +95,7 @@ def _substitute_placeholders(self, service_name: str, deployment_name: str, mode
         )
 
         inference_deployment = inference_deployment.replace("placeholder-model-name", model_name)
+        inference_deployment = inference_deployment.replace("placeholder-inference-image", image)
         return inference_deployment.strip()
 
     def create_inference_deployment(self, detector_id: str, is_oodd: bool = False) -> None:
@@ -110,8 +114,9 @@ def create_inference_deployment(self, detector_id: str, is_oodd: bool = False) -
         deployment_name = get_edge_inference_deployment_name(detector_id, is_oodd)
         service_name = get_edge_inference_service_name(detector_id, is_oodd)
         model_name = get_edge_inference_model_name(detector_id, is_oodd)
+        image = detector_image(detector_id, self._db_manager)
         inference_deployment = self._substitute_placeholders(
-            service_name=service_name, deployment_name=deployment_name, model_name=model_name
+            service_name=service_name, deployment_name=deployment_name, model_name=model_name, image=image
         )
 
         model_version = get_current_model_version(MODEL_REPOSITORY_PATH, detector_id, is_oodd=is_oodd)
@@ -145,6 +150,17 @@ def get_inference_deployment(self, deployment_name: str) -> V1Deployment | None:
                 return None
             raise e
 
+    def get_deployment_image(self, deployment_name: str) -> str | None:
+        """Return the image (including tag) on the running Deployment, or None if the Deployment is absent."""
+        deployment = self.get_inference_deployment(deployment_name)
+        if deployment is None:
+            return None
+        containers = deployment.spec.template.spec.containers
+        for c in containers:
+            if c.name == "inference-server":
+                return c.image
+        return containers[0].image if containers else None
+
     def get_or_create_inference_deployment(self, detector_id: str, is_oodd: bool = False) -> V1Deployment | None:
         """
         Retrieves an existing inference deployment for the specified detector ID, or creates a new

app/core/models.py

@@ -47,6 +47,15 @@ class InferenceDeployment(Base):
         nullable=False,
         comment="When True, the model-updater will delete this detector's K8s resources and then remove this row.",
     )
+    minimal_compatible = Column(
+        Boolean,
+        nullable=True,
+        comment=(
+            "Cloud's assertion (from fetch-model-urls) that this detector's primary pipeline "
+            "can run on the minimal inference image. Only meaningful on the primary row "
+            "(is_oodd=False); NULL until the model-updater first writes it."
+        ),
+    )
 
     created_at = Column(
         DateTime, nullable=True, default=datetime.datetime.utcnow, comment="Timestamp of record creation"

app/core/utils.py

@@ -375,6 +375,7 @@ class ModelInfoBase(BaseModel):
 
     pipeline_config: str
     predictor_metadata: str
+    minimal_compatible: bool = False
 
 
 class ModelInfoNoBinary(ModelInfoBase):