Make roboflow_vision_events@v1 non-SIMD so scalar fields accept dynamic step outputs (#2231)

* Make roboflow_vision_events@v1 non-SIMD so scalar fields accept dynamic step outputs

The SIMD declaration caused the compile-time checker to reject any batch-oriented
selector on scalar metadata fields like item_count, location, item_type, etc. Since
the block never performed any shared per-batch computation (just a serial for-loop),
SIMD was unnecessary. This aligns with the other sink blocks (email_notification,
webhook, event_writer) which are all non-SIMD.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Add regression tests for non-SIMD compile-time batch compatibility (ENT-1126)

Adds two tests that guard against re-introducing SIMD on vision_events:
- test_manifest_is_not_simd: asserts accepts_batch_input() is False
- test_batch_selector_on_scalar_field_passes_compile_check: exercises the
  exact graph constructor check (verify_declared_batch_compatibility_against_actual_inputs)
  that previously rejected batch-oriented selectors on scalar metadata fields

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-authored-by: Rafel Bennasar <253519461+rafel-roboflow@users.noreply.github.com>

Author: 3 people

inference/core/workflows/core_steps/sinks/roboflow/vision_events/v1.py

@@ -21,7 +21,6 @@
     PREDICTION_TYPE_KEY,
 )
 from inference.core.workflows.execution_engine.entities.base import (
-    Batch,
     OutputDefinition,
     WorkflowImageData,
 )
@@ -355,10 +354,6 @@ class BlockManifest(WorkflowBlockManifest):
         examples=[False, "$inputs.disable_vision_events"],
     )
 
-    @classmethod
-    def get_parameters_accepting_batches(cls) -> List[str]:
-        return ["input_image", "output_image", "predictions"]
-
     @classmethod
     def describe_outputs(cls) -> List[OutputDefinition]:
         return [
@@ -393,9 +388,9 @@ def get_manifest(cls) -> Type[WorkflowBlockManifest]:
 
     def run(
         self,
-        input_image: Optional[Batch[WorkflowImageData]],
-        output_image: Optional[Batch[WorkflowImageData]],
-        predictions: Optional[Batch[Union[sv.Detections, dict]]],
+        input_image: Optional[WorkflowImageData],
+        output_image: Optional[WorkflowImageData],
+        predictions: Optional[Union[sv.Detections, dict]],
         event_type: str,
         solution: str,
         custom_metadata: Dict[str, Any],
@@ -420,79 +415,56 @@ def run(
                 "https://docs.roboflow.com/api-reference/authentication"
                 "#retrieve-an-api-key to learn how to retrieve one."
             )
-        # Determine batch size from whichever image input is provided
-        batch_size = 1
-        if input_image is not None:
-            batch_size = len(input_image)
-        elif output_image is not None:
-            batch_size = len(output_image)
-        elif predictions is not None:
-            batch_size = len(predictions)
 
         if disable_sink:
-            return [
-                {
-                    "error_status": False,
-                    "message": "Sink was disabled by parameter `disable_sink`",
-                }
-                for _ in range(batch_size)
-            ]
+            return {
+                "error_status": False,
+                "message": "Sink was disabled by parameter `disable_sink`",
+            }
 
-        input_images = [None] * batch_size if input_image is None else input_image
-        output_images = [None] * batch_size if output_image is None else output_image
-        predictions_list = [None] * batch_size if predictions is None else predictions
-
-        result = []
-        for img_in, img_out, pred in zip(input_images, output_images, predictions_list):
-            event_data = _build_event_data(
-                event_type=event_type,
-                external_id=external_id,
-                qc_result=qc_result,
-                location=location,
-                item_count=item_count,
-                item_type=item_type,
-                alert_type=alert_type,
-                severity=severity,
-                alert_description=alert_description,
-                custom_value=custom_value,
-                related_event_id=related_event_id,
-                feedback=feedback,
-            )
-
-            task = partial(
-                _execute_vision_event,
-                api_base_url=API_BASE_URL,
-                api_key=self._api_key,
-                input_image=img_in,
-                output_image=img_out,
-                prediction=pred,
-                event_type=event_type,
-                solution=solution,
-                event_data=event_data,
-                custom_metadata=custom_metadata,
-            )
+        event_data = _build_event_data(
+            event_type=event_type,
+            external_id=external_id,
+            qc_result=qc_result,
+            location=location,
+            item_count=item_count,
+            item_type=item_type,
+            alert_type=alert_type,
+            severity=severity,
+            alert_description=alert_description,
+            custom_value=custom_value,
+            related_event_id=related_event_id,
+            feedback=feedback,
+        )
 
-            if fire_and_forget and self._background_tasks:
-                self._background_tasks.add_task(task)
-                result.append(
-                    {
-                        "error_status": False,
-                        "message": "Vision event sent in background task",
-                    }
-                )
-            elif fire_and_forget and self._thread_pool_executor:
-                self._thread_pool_executor.submit(task)
-                result.append(
-                    {
-                        "error_status": False,
-                        "message": "Vision event sent in background task",
-                    }
-                )
-            else:
-                error_status, message = task()
-                result.append({"error_status": error_status, "message": message})
+        task = partial(
+            _execute_vision_event,
+            api_base_url=API_BASE_URL,
+            api_key=self._api_key,
+            input_image=input_image,
+            output_image=output_image,
+            prediction=predictions,
+            event_type=event_type,
+            solution=solution,
+            event_data=event_data,
+            custom_metadata=custom_metadata,
+        )
 
-        return result
+        if fire_and_forget and self._background_tasks:
+            self._background_tasks.add_task(task)
+            return {
+                "error_status": False,
+                "message": "Vision event sent in background task",
+            }
+        elif fire_and_forget and self._thread_pool_executor:
+            self._thread_pool_executor.submit(task)
+            return {
+                "error_status": False,
+                "message": "Vision event sent in background task",
+            }
+        else:
+            error_status, message = task()
+            return {"error_status": error_status, "message": message}
 
 
 def _build_event_data(

tests/workflows/unit_tests/core_steps/sinks/roboflow/vision_events/test_v1.py

@@ -453,10 +453,9 @@ def test_run_disabled() -> None:
         fire_and_forget=False,
         disable_sink=True,
     )
-    assert isinstance(result, list)
-    assert len(result) == 1
-    assert result[0]["error_status"] is False
-    assert "disabled" in result[0]["message"].lower()
+    assert isinstance(result, dict)
+    assert result["error_status"] is False
+    assert "disabled" in result["message"].lower()
 
 
 @patch(
@@ -481,8 +480,8 @@ def test_run_fire_and_forget_background_tasks(mock_execute: MagicMock) -> None:
     )
 
     background_tasks.add_task.assert_called_once()
-    assert result[0]["error_status"] is False
-    assert "background" in result[0]["message"].lower()
+    assert result["error_status"] is False
+    assert "background" in result["message"].lower()
 
 
 @patch(
@@ -507,8 +506,8 @@ def test_run_fire_and_forget_thread_pool(mock_execute: MagicMock) -> None:
     )
 
     thread_pool.submit.assert_called_once()
-    assert result[0]["error_status"] is False
-    assert "background" in result[0]["message"].lower()
+    assert result["error_status"] is False
+    assert "background" in result["message"].lower()
 
 
 @patch(
@@ -533,5 +532,78 @@ def test_run_synchronous(mock_execute: MagicMock) -> None:
     )
 
     mock_execute.assert_called_once()
-    assert result[0]["error_status"] is False
-    assert result[0]["message"] == "Vision event sent successfully"
+    assert result["error_status"] is False
+    assert result["message"] == "Vision event sent successfully"
+
+
+# === Non-SIMD / Compilation Regression Tests (ENT-1126) ===
+
+
+def test_manifest_is_not_simd() -> None:
+    """Block must be non-SIMD so the engine broadcasts scalar params per image."""
+    assert BlockManifest.accepts_batch_input() is False
+
+
+def test_batch_selector_on_scalar_field_passes_compile_check() -> None:
+    """Regression test for ENT-1126.
+
+    Exercises ``verify_declared_batch_compatibility_against_actual_inputs``
+    (graph_constructor.py:1659), the exact compile-time check that rejected
+    batch-oriented selectors on ``item_count`` when the block was SIMD.
+
+    A StepNode backed by our manifest with a batch-oriented input on
+    ``item_count`` must NOT raise ``ExecutionGraphStructureError``.
+    """
+    from inference.core.workflows.errors import ExecutionGraphStructureError
+    from inference.core.workflows.execution_engine.v1.compiler.entities import (
+        DynamicStepInputDefinition,
+        NodeInputCategory,
+        ParameterSpecification,
+        StepNode,
+    )
+    from inference.core.workflows.execution_engine.v1.compiler.graph_constructor import (
+        verify_declared_batch_compatibility_against_actual_inputs,
+    )
+
+    manifest = BlockManifest(
+        type="roboflow_core/roboflow_vision_events@v1",
+        name="vision_events",
+        event_type="inventory_count",
+        solution="test-solution",
+        item_count="$steps.counter.count",
+    )
+
+    step_node = StepNode(
+        node_category="step_node",
+        name="vision_events",
+        selector="$steps.vision_events",
+        data_lineage=[],
+        step_manifest=manifest,
+        input_data={
+            "item_count": DynamicStepInputDefinition(
+                parameter_specification=ParameterSpecification(
+                    parameter_name="item_count",
+                    nested_element_key=None,
+                    nested_element_index=None,
+                ),
+                category=NodeInputCategory.BATCH_STEP_OUTPUT,
+                data_lineage=["<workflow_input>"],
+                selector="$steps.counter.count",
+            ),
+        },
+        batch_oriented_parameters=set(),
+    )
+
+    # batch_compatibility_of_properties says item_count is NOT batch-compatible
+    batch_compat = {"item_count": {False}}
+
+    # Must not raise. Before the fix (when accepts_batch_input() was True),
+    # this exact call raised ExecutionGraphStructureError because a
+    # batch-oriented selector was plugged into a non-batch parameter.
+    result = verify_declared_batch_compatibility_against_actual_inputs(
+        node="$steps.vision_events",
+        step_node_data=step_node,
+        input_data=step_node.input_data,
+        batch_compatibility_of_properties=batch_compat,
+    )
+    assert isinstance(result, set)