diff --git a/workflows/CHANGELOG.md b/workflows/CHANGELOG.md
index b818c6c7a7..8f6730d664 100644
--- a/workflows/CHANGELOG.md
+++ b/workflows/CHANGELOG.md
@@ -59,6 +59,9 @@ for contributor and maintainer responsibilities.
 
 - Root-coordinate recovery (`sv_detections_to_root_coordinates`, applied to workflow outputs by default) re-encodes `rle_mask` from the re-anchored dense masks. A block that ran on a crop and returns RLE with dense masks, such as SAM 3 v3 with `output_format: rle`, used to serialize root-coordinate boxes next to an RLE still sized and placed in the crop. RLE-only predictions (`mask=None`) are unchanged, as in `scale_sv_detections`.
 
+- Video-state blocks now accept an image from a step output as well as from a Workflow input, so they keep working inside an embedded `roboflow_core/inner_workflow@v1` whose `image` is bound to a parent step (for example a preprocessing block). Inlining replaces the child's `$inputs.image` with that step selector, and the merged definition used to fail with `Could not parse workflow definition` even though each Workflow compiled on its own. Affected blocks: `time_in_zone@v2`/`@v3`, `velocity@v1`, `line_counter@v2`, `path_deviation_analytics@v2`, `detection_event_log@v1`, `byte_tracker@v2`, `cache_get@v1`, `cache_set@v1`, `delta_filter@v1` and `rate_limiter@v1` (`video_reference_image`). The blocks read `video_metadata` from the image, which step outputs built with `WorkflowImageData.copy_and_replace` keep, so per-video state is unchanged.
+- Absolute and relative static crops (`absolute_static_crop@v1`, `relative_statoic_crop@v1`) now give a crop the same `video_identifier` on every frame (`<video> | crop: <block>:<x_min>,<y_min>,<x_max>,<y_max>`). They used to append the per-frame crop ID, so a stateful block fed by a static crop (Cache Get/Set, trackers, analytics) saw each frame as a new video: the cache missed on every frame and state kept growing with one namespace per frame. Crop `parent_id` values are still unique per frame. `WorkflowImageData.create_crop` and `create_crop_from_tensor` take an optional `video_crop_identifier` for this; other callers are unchanged.
+- Camera Calibration (`camera-calibration@v1`) keeps the input image's video metadata and root-ancestor metadata on its output. It used to drop them, so a rate limiter using the calibrated frame as `video_reference_image` saw every frame at time zero and suppressed its downstream steps after the first frame.
 
 ## `0.2.14`
 
diff --git a/workflows/roboflow_workflows/core_steps/analytics/detection_event_log/v1.py b/workflows/roboflow_workflows/core_steps/analytics/detection_event_log/v1.py
index 9450f629fb..3872dd3333 100644
--- a/workflows/roboflow_workflows/core_steps/analytics/detection_event_log/v1.py
+++ b/workflows/roboflow_workflows/core_steps/analytics/detection_event_log/v1.py
@@ -19,11 +19,11 @@
 from roboflow_workflows.execution_engine.entities.types import (
     DICTIONARY_KIND,
     FLOAT_KIND,
+    IMAGE_KIND,
     INSTANCE_SEGMENTATION_PREDICTION_KIND,
     INTEGER_KIND,
     OBJECT_DETECTION_PREDICTION_KIND,
     Selector,
-    WorkflowImageSelector,
     WorkflowParameterSelector,
 )
 from roboflow_workflows.execution_engine.entities.workload import (
@@ -94,7 +94,7 @@ class BlockManifest(WorkflowBlockManifest):
     )
     type: Literal["roboflow_core/detection_event_log@v1"]
 
-    image: WorkflowImageSelector = Field(
+    image: Selector(kind=[IMAGE_KIND]) = Field(
         description="Reference to the image for video metadata (frame number, timestamp).",
         examples=["$inputs.image"],
     )
diff --git a/workflows/roboflow_workflows/core_steps/analytics/detection_event_log/v1_tensor.py b/workflows/roboflow_workflows/core_steps/analytics/detection_event_log/v1_tensor.py
index 1448069b33..070ec65fb8 100644
--- a/workflows/roboflow_workflows/core_steps/analytics/detection_event_log/v1_tensor.py
+++ b/workflows/roboflow_workflows/core_steps/analytics/detection_event_log/v1_tensor.py
@@ -22,9 +22,9 @@
 from roboflow_workflows.execution_engine.entities.types import (
     DICTIONARY_KIND,
     FLOAT_KIND,
+    IMAGE_KIND,
     INTEGER_KIND,
     Selector,
-    WorkflowImageSelector,
     WorkflowParameterSelector,
 )
 from roboflow_workflows.execution_engine.entities.workload import (
@@ -98,7 +98,7 @@ class BlockManifest(WorkflowBlockManifest):
     )
     type: Literal["roboflow_core/detection_event_log@v1"]
 
-    image: WorkflowImageSelector = Field(
+    image: Selector(kind=[IMAGE_KIND]) = Field(
         description="Reference to the image for video metadata (frame number, timestamp).",
         examples=["$inputs.image"],
     )
diff --git a/workflows/roboflow_workflows/core_steps/analytics/line_counter/v2.py b/workflows/roboflow_workflows/core_steps/analytics/line_counter/v2.py
index 2fa78e846f..da66db1a02 100644
--- a/workflows/roboflow_workflows/core_steps/analytics/line_counter/v2.py
+++ b/workflows/roboflow_workflows/core_steps/analytics/line_counter/v2.py
@@ -10,13 +10,13 @@
     WorkflowImageData,
 )
 from roboflow_workflows.execution_engine.entities.types import (
+    IMAGE_KIND,
     INSTANCE_SEGMENTATION_PREDICTION_KIND,
     INTEGER_KIND,
     LIST_OF_VALUES_KIND,
     OBJECT_DETECTION_PREDICTION_KIND,
     STRING_KIND,
     Selector,
-    WorkflowImageSelector,
 )
 from roboflow_workflows.execution_engine.entities.workload import (
     Discovery,
@@ -136,7 +136,7 @@ class LineCounterManifest(WorkflowBlockManifest):
         }
     )
     type: Literal["roboflow_core/line_counter@v2"]
-    image: WorkflowImageSelector = Field(
+    image: Selector(kind=[IMAGE_KIND]) = Field(
         description="Image with embedded video metadata. The video_metadata contains video_identifier to maintain separate counting state for different videos. Required for persistent counting across frames.",
     )
     detections: Selector(
diff --git a/workflows/roboflow_workflows/core_steps/analytics/line_counter/v2_tensor.py b/workflows/roboflow_workflows/core_steps/analytics/line_counter/v2_tensor.py
index db0a1486c9..7cef5e72e5 100644
--- a/workflows/roboflow_workflows/core_steps/analytics/line_counter/v2_tensor.py
+++ b/workflows/roboflow_workflows/core_steps/analytics/line_counter/v2_tensor.py
@@ -20,11 +20,11 @@
     TENSOR_NATIVE_OBJECT_DETECTION_PREDICTION_KIND,
 )
 from roboflow_workflows.execution_engine.entities.types import (
+    IMAGE_KIND,
     INTEGER_KIND,
     LIST_OF_VALUES_KIND,
     STRING_KIND,
     Selector,
-    WorkflowImageSelector,
 )
 from roboflow_workflows.execution_engine.entities.workload import (
     Discovery,
@@ -147,7 +147,7 @@ class LineCounterManifest(WorkflowBlockManifest):
         }
     )
     type: Literal["roboflow_core/line_counter@v2"]
-    image: WorkflowImageSelector = Field(
+    image: Selector(kind=[IMAGE_KIND]) = Field(
         description="Image with embedded video metadata. The video_metadata contains video_identifier to maintain separate counting state for different videos. Required for persistent counting across frames.",
     )
     detections: Selector(
diff --git a/workflows/roboflow_workflows/core_steps/analytics/path_deviation/v2.py b/workflows/roboflow_workflows/core_steps/analytics/path_deviation/v2.py
index 72ca2fddfa..b277e98805 100644
--- a/workflows/roboflow_workflows/core_steps/analytics/path_deviation/v2.py
+++ b/workflows/roboflow_workflows/core_steps/analytics/path_deviation/v2.py
@@ -14,12 +14,12 @@
     WorkflowImageData,
 )
 from roboflow_workflows.execution_engine.entities.types import (
+    IMAGE_KIND,
     INSTANCE_SEGMENTATION_PREDICTION_KIND,
     LIST_OF_VALUES_KIND,
     OBJECT_DETECTION_PREDICTION_KIND,
     STRING_KIND,
     Selector,
-    WorkflowImageSelector,
 )
 from roboflow_workflows.execution_engine.entities.workload import (
     Discovery,
@@ -134,7 +134,7 @@ class PathDeviationManifest(WorkflowBlockManifest):
         }
     )
     type: Literal["roboflow_core/path_deviation_analytics@v2"]
-    image: WorkflowImageSelector = Field(
+    image: Selector(kind=[IMAGE_KIND]) = Field(
         description="Image with embedded video metadata. The video_metadata contains video_identifier to maintain separate path tracking state for different videos. Required for persistent path accumulation across frames.",
     )
     detections: Selector(
diff --git a/workflows/roboflow_workflows/core_steps/analytics/path_deviation/v2_tensor.py b/workflows/roboflow_workflows/core_steps/analytics/path_deviation/v2_tensor.py
index e08730d4bd..1fd1d8eef6 100644
--- a/workflows/roboflow_workflows/core_steps/analytics/path_deviation/v2_tensor.py
+++ b/workflows/roboflow_workflows/core_steps/analytics/path_deviation/v2_tensor.py
@@ -19,10 +19,10 @@
     TENSOR_NATIVE_OBJECT_DETECTION_PREDICTION_KIND,
 )
 from roboflow_workflows.execution_engine.entities.types import (
+    IMAGE_KIND,
     LIST_OF_VALUES_KIND,
     STRING_KIND,
     Selector,
-    WorkflowImageSelector,
 )
 from roboflow_workflows.execution_engine.entities.workload import (
     Discovery,
@@ -140,7 +140,7 @@ class PathDeviationManifest(WorkflowBlockManifest):
         }
     )
     type: Literal["roboflow_core/path_deviation_analytics@v2"]
-    image: WorkflowImageSelector = Field(
+    image: Selector(kind=[IMAGE_KIND]) = Field(
         description="Image with embedded video metadata. The video_metadata contains video_identifier to maintain separate path tracking state for different videos. Required for persistent path accumulation across frames.",
     )
     detections: Selector(
diff --git a/workflows/roboflow_workflows/core_steps/analytics/time_in_zone/v2.py b/workflows/roboflow_workflows/core_steps/analytics/time_in_zone/v2.py
index 4c3308c67d..c8fee5f2b0 100644
--- a/workflows/roboflow_workflows/core_steps/analytics/time_in_zone/v2.py
+++ b/workflows/roboflow_workflows/core_steps/analytics/time_in_zone/v2.py
@@ -15,12 +15,12 @@
 )
 from roboflow_workflows.execution_engine.entities.types import (
     BOOLEAN_KIND,
+    IMAGE_KIND,
     INSTANCE_SEGMENTATION_PREDICTION_KIND,
     LIST_OF_VALUES_KIND,
     OBJECT_DETECTION_PREDICTION_KIND,
     STRING_KIND,
     Selector,
-    WorkflowImageSelector,
 )
 from roboflow_workflows.execution_engine.entities.workload import (
     Discovery,
@@ -144,7 +144,7 @@ class TimeInZoneManifest(WorkflowBlockManifest):
         }
     )
     type: Literal["roboflow_core/time_in_zone@v2"]
-    image: Union[WorkflowImageSelector] = Field(
+    image: Selector(kind=[IMAGE_KIND]) = Field(
         title="Image",
         description="Input image for the current video frame containing embedded video metadata (fps, frame_number, frame_timestamp, video_identifier, video source) required for time calculation and state management. The block extracts video_metadata from the WorkflowImageData object. The fps and frame_number are used for video files to calculate timestamps (timestamp = frame_number / fps). For streamed video, frame_timestamp is used directly. The video_identifier is used to maintain separate tracking state and zone configurations for different videos. Used for zone visualization and reference. The image dimensions are used to validate zone coordinates. This version simplifies input by embedding metadata in the image object rather than requiring a separate metadata field.",
         examples=["$inputs.image", "$steps.cropping.crops"],
diff --git a/workflows/roboflow_workflows/core_steps/analytics/time_in_zone/v2_tensor.py b/workflows/roboflow_workflows/core_steps/analytics/time_in_zone/v2_tensor.py
index ddb601794a..acf150cf33 100644
--- a/workflows/roboflow_workflows/core_steps/analytics/time_in_zone/v2_tensor.py
+++ b/workflows/roboflow_workflows/core_steps/analytics/time_in_zone/v2_tensor.py
@@ -24,10 +24,10 @@
 )
 from roboflow_workflows.execution_engine.entities.types import (
     BOOLEAN_KIND,
+    IMAGE_KIND,
     LIST_OF_VALUES_KIND,
     STRING_KIND,
     Selector,
-    WorkflowImageSelector,
 )
 from roboflow_workflows.execution_engine.entities.workload import (
     Discovery,
@@ -154,7 +154,7 @@ class TimeInZoneManifest(WorkflowBlockManifest):
         }
     )
     type: Literal["roboflow_core/time_in_zone@v2"]
-    image: Union[WorkflowImageSelector] = Field(
+    image: Selector(kind=[IMAGE_KIND]) = Field(
         title="Image",
         description="Input image for the current video frame containing embedded video metadata (fps, frame_number, frame_timestamp, video_identifier, video source) required for time calculation and state management. The block extracts video_metadata from the WorkflowImageData object. The fps and frame_number are used for video files to calculate timestamps (timestamp = frame_number / fps). For streamed video, frame_timestamp is used directly. The video_identifier is used to maintain separate tracking state and zone configurations for different videos. Used for zone visualization and reference. The image dimensions are used to validate zone coordinates. This version simplifies input by embedding metadata in the image object rather than requiring a separate metadata field.",
         examples=["$inputs.image", "$steps.cropping.crops"],
diff --git a/workflows/roboflow_workflows/core_steps/analytics/time_in_zone/v3.py b/workflows/roboflow_workflows/core_steps/analytics/time_in_zone/v3.py
index 3c03370666..c4c72d4938 100644
--- a/workflows/roboflow_workflows/core_steps/analytics/time_in_zone/v3.py
+++ b/workflows/roboflow_workflows/core_steps/analytics/time_in_zone/v3.py
@@ -17,12 +17,12 @@
 )
 from roboflow_workflows.execution_engine.entities.types import (
     BOOLEAN_KIND,
+    IMAGE_KIND,
     INSTANCE_SEGMENTATION_PREDICTION_KIND,
     LIST_OF_VALUES_KIND,
     OBJECT_DETECTION_PREDICTION_KIND,
     STRING_KIND,
     Selector,
-    WorkflowImageSelector,
 )
 from roboflow_workflows.execution_engine.entities.workload import (
     Discovery,
@@ -167,7 +167,7 @@ class TimeInZoneManifest(WorkflowBlockManifest):
         }
     )
     type: Literal["roboflow_core/time_in_zone@v3"]
-    image: Union[WorkflowImageSelector] = Field(
+    image: Selector(kind=[IMAGE_KIND]) = Field(
         title="Image",
         description="Input image for the current video frame containing embedded video metadata (fps, frame_number, frame_timestamp, video_identifier, video source) required for time calculation and state management. The block extracts video_metadata from the WorkflowImageData object. The fps and frame_number are used for video files to calculate timestamps (timestamp = frame_number / fps). For streamed video, frame_timestamp is used directly. The video_identifier is used to maintain separate tracking state and zone configurations for different videos. Used for zone visualization and reference. The image dimensions are used to validate zone coordinates. This version supports multiple zones per video with efficient zone cache management.",
         examples=["$inputs.image", "$steps.cropping.crops"],
diff --git a/workflows/roboflow_workflows/core_steps/analytics/time_in_zone/v3_tensor.py b/workflows/roboflow_workflows/core_steps/analytics/time_in_zone/v3_tensor.py
index abcaa693f4..d29da00a53 100644
--- a/workflows/roboflow_workflows/core_steps/analytics/time_in_zone/v3_tensor.py
+++ b/workflows/roboflow_workflows/core_steps/analytics/time_in_zone/v3_tensor.py
@@ -26,10 +26,10 @@
 )
 from roboflow_workflows.execution_engine.entities.types import (
     BOOLEAN_KIND,
+    IMAGE_KIND,
     LIST_OF_VALUES_KIND,
     STRING_KIND,
     Selector,
-    WorkflowImageSelector,
 )
 from roboflow_workflows.execution_engine.entities.workload import (
     Discovery,
@@ -177,7 +177,7 @@ class TimeInZoneManifest(WorkflowBlockManifest):
         }
     )
     type: Literal["roboflow_core/time_in_zone@v3"]
-    image: Union[WorkflowImageSelector] = Field(
+    image: Selector(kind=[IMAGE_KIND]) = Field(
         title="Image",
         description="Input image for the current video frame containing embedded video metadata (fps, frame_number, frame_timestamp, video_identifier, video source) required for time calculation and state management. The block extracts video_metadata from the WorkflowImageData object. The fps and frame_number are used for video files to calculate timestamps (timestamp = frame_number / fps). For streamed video, frame_timestamp is used directly. The video_identifier is used to maintain separate tracking state and zone configurations for different videos. Used for zone visualization and reference. The image dimensions are used to validate zone coordinates. This version supports multiple zones per video with efficient zone cache management.",
         examples=["$inputs.image", "$steps.cropping.crops"],
diff --git a/workflows/roboflow_workflows/core_steps/analytics/velocity/v1.py b/workflows/roboflow_workflows/core_steps/analytics/velocity/v1.py
index 0415d727d0..b7d0cf36f0 100644
--- a/workflows/roboflow_workflows/core_steps/analytics/velocity/v1.py
+++ b/workflows/roboflow_workflows/core_steps/analytics/velocity/v1.py
@@ -18,12 +18,12 @@
 )
 from roboflow_workflows.execution_engine.entities.types import (
     FLOAT_KIND,
+    IMAGE_KIND,
     INSTANCE_SEGMENTATION_PREDICTION_KIND,
     KEYPOINT_DETECTION_PREDICTION_KIND,
     OBJECT_DETECTION_PREDICTION_KIND,
     Selector,
     StepOutputSelector,
-    WorkflowImageSelector,
 )
 from roboflow_workflows.execution_engine.entities.workload import (
     Discovery,
@@ -134,7 +134,7 @@ class VelocityManifest(WorkflowBlockManifest):
         }
     )
     type: Literal["roboflow_core/velocity@v1"]
-    image: WorkflowImageSelector = Field(
+    image: Selector(kind=[IMAGE_KIND]) = Field(
         description="Image with embedded video metadata. The video_metadata contains fps, frame_number, frame_timestamp, and video_identifier. Required for calculating time deltas and maintaining separate velocity tracking state for different videos.",
     )
     detections: StepOutputSelector(
diff --git a/workflows/roboflow_workflows/core_steps/analytics/velocity/v1_tensor.py b/workflows/roboflow_workflows/core_steps/analytics/velocity/v1_tensor.py
index dd9251a3c6..319143b54c 100644
--- a/workflows/roboflow_workflows/core_steps/analytics/velocity/v1_tensor.py
+++ b/workflows/roboflow_workflows/core_steps/analytics/velocity/v1_tensor.py
@@ -26,9 +26,9 @@
 )
 from roboflow_workflows.execution_engine.entities.types import (
     FLOAT_KIND,
+    IMAGE_KIND,
     Selector,
     StepOutputSelector,
-    WorkflowImageSelector,
 )
 from roboflow_workflows.execution_engine.entities.workload import (
     Discovery,
@@ -143,7 +143,7 @@ class VelocityManifest(WorkflowBlockManifest):
         }
     )
     type: Literal["roboflow_core/velocity@v1"]
-    image: WorkflowImageSelector = Field(
+    image: Selector(kind=[IMAGE_KIND]) = Field(
         description="Image with embedded video metadata. The video_metadata contains fps, frame_number, frame_timestamp, and video_identifier. Required for calculating time deltas and maintaining separate velocity tracking state for different videos.",
     )
     detections: StepOutputSelector(
diff --git a/workflows/roboflow_workflows/core_steps/cache/cache_get/v1.py b/workflows/roboflow_workflows/core_steps/cache/cache_get/v1.py
index 96dead28bf..37765de826 100644
--- a/workflows/roboflow_workflows/core_steps/cache/cache_get/v1.py
+++ b/workflows/roboflow_workflows/core_steps/cache/cache_get/v1.py
@@ -15,7 +15,6 @@
     STRING_KIND,
     WILDCARD_KIND,
     Selector,
-    WorkflowImageSelector,
 )
 from roboflow_workflows.execution_engine.entities.workload import (
     Discovery,
@@ -99,7 +98,7 @@ class BlockManifest(WorkflowBlockManifest):
         }
     )
     type: Literal["roboflow_core/cache_get@v1"]
-    image: WorkflowImageSelector = Field(
+    image: Selector(kind=[IMAGE_KIND]) = Field(
         description="Input image used to determine the cache namespace. The block extracts the video identifier from the image's video metadata and uses it as the cache namespace. If no video identifier is present, the block uses 'default' as the namespace. The namespace isolates cache entries so different videos or streams have separate cache storage. Use the same image (with the same video identifier) for both Cache Set and Cache Get blocks to access the same cache namespace.",
         examples=["$inputs.image", "$steps.input.output"],
     )
diff --git a/workflows/roboflow_workflows/core_steps/cache/cache_set/v1.py b/workflows/roboflow_workflows/core_steps/cache/cache_set/v1.py
index d6d37d0c8c..dcc0a5d1b3 100644
--- a/workflows/roboflow_workflows/core_steps/cache/cache_set/v1.py
+++ b/workflows/roboflow_workflows/core_steps/cache/cache_set/v1.py
@@ -16,7 +16,6 @@
     STRING_KIND,
     WILDCARD_KIND,
     Selector,
-    WorkflowImageSelector,
 )
 from roboflow_workflows.execution_engine.entities.workload import (
     Discovery,
@@ -100,7 +99,7 @@ class BlockManifest(WorkflowBlockManifest):
         }
     )
     type: Literal["roboflow_core/cache_set@v1"]
-    image: WorkflowImageSelector = Field(
+    image: Selector(kind=[IMAGE_KIND]) = Field(
         description="Input image used to determine the cache namespace. The block extracts the video identifier from the image's video metadata and uses it as the cache namespace. If no video identifier is present, the block uses 'default' as the namespace. The namespace isolates cache entries so different videos or streams have separate cache storage. Use the same image (with the same video identifier) for both Cache Set and Cache Get blocks to access the same cache namespace.",
         examples=["$inputs.image", "$steps.input.output"],
     )
diff --git a/workflows/roboflow_workflows/core_steps/flow_control/delta_filter/v1.py b/workflows/roboflow_workflows/core_steps/flow_control/delta_filter/v1.py
index 49e0b17fe7..d169ad572e 100644
--- a/workflows/roboflow_workflows/core_steps/flow_control/delta_filter/v1.py
+++ b/workflows/roboflow_workflows/core_steps/flow_control/delta_filter/v1.py
@@ -9,10 +9,10 @@
     WorkflowImageData,
 )
 from roboflow_workflows.execution_engine.entities.types import (
+    IMAGE_KIND,
     WILDCARD_KIND,
     Selector,
     StepSelector,
-    WorkflowImageSelector,
 )
 from roboflow_workflows.execution_engine.entities.workload import (
     Discovery,
@@ -92,7 +92,7 @@ class DeltaFilterManifest(WorkflowBlockManifest):
             },
         }
     )
-    image: WorkflowImageSelector
+    image: Selector(kind=[IMAGE_KIND])
     value: Selector(kind=[WILDCARD_KIND]) = Field(
         title="Input Value",
         description="Value to monitor for changes. Can be any data type (numbers, strings, detection counts, metrics, etc.) from workflow inputs or step outputs. The workflow branch continues to next_steps only when this value differs from the previously cached value for the current video stream. If the value remains the same, the branch terminates to avoid redundant processing. Example: Monitor a line counter count ($steps.line_counter.count_in) and trigger actions only when the count changes.",
diff --git a/workflows/roboflow_workflows/core_steps/flow_control/rate_limiter/v1.py b/workflows/roboflow_workflows/core_steps/flow_control/rate_limiter/v1.py
index ec925f7055..8eb1f6f82d 100644
--- a/workflows/roboflow_workflows/core_steps/flow_control/rate_limiter/v1.py
+++ b/workflows/roboflow_workflows/core_steps/flow_control/rate_limiter/v1.py
@@ -13,7 +13,6 @@
     IMAGE_KIND,
     Selector,
     StepSelector,
-    WorkflowImageSelector,
 )
 from roboflow_workflows.execution_engine.entities.workload import (
     Discovery,
@@ -113,7 +112,7 @@ class RateLimiterManifest(WorkflowBlockManifest):
         description="List of workflow steps to execute if the rate limit allows (i.e., sufficient time has passed since the last execution). These steps receive control flow only when the cooldown period has elapsed, enabling throttled downstream processing. If the cooldown period hasn't elapsed, these steps will not execute as the branch terminates. Each step selector references a block in the workflow that should execute when rate limiting permits.",
         examples=[["$steps.upload"]],
     )
-    video_reference_image: Optional[WorkflowImageSelector] = Field(
+    video_reference_image: Optional[Selector(kind=[IMAGE_KIND])] = Field(
         description="Optional reference to a video frame image to use for video-time-based timestamp generation. When provided, the rate limiter calculates timestamps based on video metadata (frame number and FPS) rather than system clock time. This is useful when processing video faster than real-time, ensuring rate limiting works correctly relative to video playback time rather than processing speed. If not provided (None), the block uses system clock time (datetime.now()). Only applicable for video processing workflows.",
         examples=["$inputs.image"],
         default=None,
diff --git a/workflows/roboflow_workflows/core_steps/transformations/absolute_static_crop/v1.py b/workflows/roboflow_workflows/core_steps/transformations/absolute_static_crop/v1.py
index b0e3cb9dc2..2cbae3a88f 100644
--- a/workflows/roboflow_workflows/core_steps/transformations/absolute_static_crop/v1.py
+++ b/workflows/roboflow_workflows/core_steps/transformations/absolute_static_crop/v1.py
@@ -186,4 +186,5 @@ def take_static_crop(
         offset_x=x_min,
         offset_y=y_min,
         preserve_video_metadata=True,
+        video_crop_identifier=f"absolute_static_crop:{x_min},{y_min},{x_max},{y_max}",
     )
diff --git a/workflows/roboflow_workflows/core_steps/transformations/absolute_static_crop/v1_tensor.py b/workflows/roboflow_workflows/core_steps/transformations/absolute_static_crop/v1_tensor.py
index 67cba931cc..c1ef8b284e 100644
--- a/workflows/roboflow_workflows/core_steps/transformations/absolute_static_crop/v1_tensor.py
+++ b/workflows/roboflow_workflows/core_steps/transformations/absolute_static_crop/v1_tensor.py
@@ -189,6 +189,7 @@ def take_static_crop(
             offset_x=x_min,
             offset_y=y_min,
             preserve_video_metadata=True,
+            video_crop_identifier=f"absolute_static_crop:{x_min},{y_min},{x_max},{y_max}",
         )
     # Only numpy is materialised — slice on the host and emit a numpy-backed
     # crop; no full-image numpy->device conversion is forced.
@@ -202,4 +203,5 @@ def take_static_crop(
         offset_x=x_min,
         offset_y=y_min,
         preserve_video_metadata=True,
+        video_crop_identifier=f"absolute_static_crop:{x_min},{y_min},{x_max},{y_max}",
     )
diff --git a/workflows/roboflow_workflows/core_steps/transformations/byte_tracker/v2.py b/workflows/roboflow_workflows/core_steps/transformations/byte_tracker/v2.py
index c7527f62b6..0ee6b3bab0 100644
--- a/workflows/roboflow_workflows/core_steps/transformations/byte_tracker/v2.py
+++ b/workflows/roboflow_workflows/core_steps/transformations/byte_tracker/v2.py
@@ -13,11 +13,11 @@
 )
 from roboflow_workflows.execution_engine.entities.types import (
     FLOAT_ZERO_TO_ONE_KIND,
+    IMAGE_KIND,
     INSTANCE_SEGMENTATION_PREDICTION_KIND,
     INTEGER_KIND,
     OBJECT_DETECTION_PREDICTION_KIND,
     Selector,
-    WorkflowImageSelector,
 )
 from roboflow_workflows.execution_engine.entities.workload import (
     Discovery,
@@ -136,7 +136,7 @@ class ByteTrackerBlockManifest(WorkflowBlockManifest):
         protected_namespaces=(),
     )
     type: Literal["roboflow_core/byte_tracker@v2"]
-    image: WorkflowImageSelector = Field(
+    image: Selector(kind=[IMAGE_KIND]) = Field(
         description="Input image containing embedded video metadata (fps and video_identifier) required for ByteTrack initialization and tracking state management. The block extracts video_metadata from the WorkflowImageData object. The fps value is used to configure the tracker, and the video_identifier is used to maintain separate tracking state for different videos. If fps is missing or invalid, the block defaults to 0 and logs a warning but continues operation. If processing multiple videos, each video should have a unique video_identifier in its metadata to maintain separate tracking states. The block maintains persistent trackers across frames for each video using the video_identifier. This version simplifies input by embedding metadata in the image object rather than requiring a separate metadata field.",
     )
     detections: Selector(
diff --git a/workflows/roboflow_workflows/core_steps/transformations/byte_tracker/v2_tensor.py b/workflows/roboflow_workflows/core_steps/transformations/byte_tracker/v2_tensor.py
index e0730e8049..fa55736efb 100644
--- a/workflows/roboflow_workflows/core_steps/transformations/byte_tracker/v2_tensor.py
+++ b/workflows/roboflow_workflows/core_steps/transformations/byte_tracker/v2_tensor.py
@@ -22,9 +22,9 @@
 )
 from roboflow_workflows.execution_engine.entities.types import (
     FLOAT_ZERO_TO_ONE_KIND,
+    IMAGE_KIND,
     INTEGER_KIND,
     Selector,
-    WorkflowImageSelector,
 )
 from roboflow_workflows.execution_engine.entities.workload import (
     Discovery,
@@ -203,7 +203,7 @@ class ByteTrackerBlockManifest(WorkflowBlockManifest):
         protected_namespaces=(),
     )
     type: Literal["roboflow_core/byte_tracker@v2"]
-    image: WorkflowImageSelector = Field(
+    image: Selector(kind=[IMAGE_KIND]) = Field(
         description="Input image containing embedded video metadata (fps and video_identifier) required for ByteTrack initialization and tracking state management. The block extracts video_metadata from the WorkflowImageData object. The fps value is used to configure the tracker, and the video_identifier is used to maintain separate tracking state for different videos. If fps is missing or invalid, the block defaults to 0 and logs a warning but continues operation. If processing multiple videos, each video should have a unique video_identifier in its metadata to maintain separate tracking states. The block maintains persistent trackers across frames for each video using the video_identifier. This version simplifies input by embedding metadata in the image object rather than requiring a separate metadata field.",
     )
     detections: Selector(
diff --git a/workflows/roboflow_workflows/core_steps/transformations/camera_calibration/v1.py b/workflows/roboflow_workflows/core_steps/transformations/camera_calibration/v1.py
index 278c730098..8748939a96 100644
--- a/workflows/roboflow_workflows/core_steps/transformations/camera_calibration/v1.py
+++ b/workflows/roboflow_workflows/core_steps/transformations/camera_calibration/v1.py
@@ -274,8 +274,8 @@ def remove_distortions(
         dst = cv.remap(
             img, map_x.astype(np.float32), map_y.astype(np.float32), cv.INTER_LINEAR
         )
-        return WorkflowImageData(
-            parent_metadata=image.parent_metadata,
+        return WorkflowImageData.copy_and_replace(
+            origin_image_data=image,
             numpy_image=dst,
         )
 
@@ -298,7 +298,7 @@ def remove_distortions(
         dst=None,
         newCameraMatrix=newcameramtx,
     )
-    return WorkflowImageData(
-        parent_metadata=image.parent_metadata,
+    return WorkflowImageData.copy_and_replace(
+        origin_image_data=image,
         numpy_image=dst,
     )
diff --git a/workflows/roboflow_workflows/core_steps/transformations/relative_static_crop/v1.py b/workflows/roboflow_workflows/core_steps/transformations/relative_static_crop/v1.py
index 0d9905ff42..6e8e3f393d 100644
--- a/workflows/roboflow_workflows/core_steps/transformations/relative_static_crop/v1.py
+++ b/workflows/roboflow_workflows/core_steps/transformations/relative_static_crop/v1.py
@@ -189,4 +189,5 @@ def take_static_crop(
         offset_x=x_min,
         offset_y=y_min,
         preserve_video_metadata=True,
+        video_crop_identifier=f"relative_static_crop:{x_min},{y_min},{x_max},{y_max}",
     )
diff --git a/workflows/roboflow_workflows/execution_engine/entities/base.py b/workflows/roboflow_workflows/execution_engine/entities/base.py
index 5ffaa03858..48be8f1ad8 100644
--- a/workflows/roboflow_workflows/execution_engine/entities/base.py
+++ b/workflows/roboflow_workflows/execution_engine/entities/base.py
@@ -410,10 +410,17 @@ def create_crop(
         offset_x: int,
         offset_y: int,
         preserve_video_metadata: bool = False,
+        video_crop_identifier: Optional[str] = None,
     ) -> "WorkflowImageData":
         """
         Creates new instance of `WorkflowImageData` being a crop of original image,
         making adjustment to all metadata.
+
+        With `preserve_video_metadata`, the crop keeps the origin's video metadata and
+        appends `video_crop_identifier` (or `crop_identifier` when it is not given) to
+        `video_identifier`. Pass a `video_crop_identifier` that is the same on every
+        frame for crops of a fixed region, so blocks keeping per-video state see one
+        video instead of a new one per frame.
         """
         parent_metadata = ImageParentMetadata(
             parent_id=crop_identifier,
@@ -439,7 +446,8 @@ def create_crop(
         if preserve_video_metadata and origin_image_data._video_metadata is not None:
             video_metadata = copy(origin_image_data._video_metadata)
             video_metadata.video_identifier = (
-                f"{video_metadata.video_identifier} | crop: {crop_identifier}"
+                f"{video_metadata.video_identifier} | crop: "
+                f"{video_crop_identifier or crop_identifier}"
             )
         return WorkflowImageData(
             parent_metadata=parent_metadata,
@@ -457,6 +465,7 @@ def create_crop_from_tensor(
         offset_x: int,
         offset_y: int,
         preserve_video_metadata: bool = False,
+        video_crop_identifier: Optional[str] = None,
     ) -> "WorkflowImageData":
         """
         Tensor-native mirror of `create_crop`. Identical metadata math;
@@ -487,7 +496,8 @@ def create_crop_from_tensor(
         if preserve_video_metadata and origin_image_data._video_metadata is not None:
             video_metadata = copy(origin_image_data._video_metadata)
             video_metadata.video_identifier = (
-                f"{video_metadata.video_identifier} | crop: {crop_identifier}"
+                f"{video_metadata.video_identifier} | crop: "
+                f"{video_crop_identifier or crop_identifier}"
             )
         return WorkflowImageData(
             parent_metadata=parent_metadata,
diff --git a/workflows/tests/unit_tests/execution_engine/entities/test_base.py b/workflows/tests/unit_tests/execution_engine/entities/test_base.py
index 637d434046..2c00f8e966 100644
--- a/workflows/tests/unit_tests/execution_engine/entities/test_base.py
+++ b/workflows/tests/unit_tests/execution_engine/entities/test_base.py
@@ -1045,6 +1045,38 @@ def test_workflow_image_build_create_crop_with_video_metadata_preservation() ->
     assert result.video_metadata.fps == 40, "Expected preserved metadata"
 
 
+def test_workflow_image_create_crop_uses_video_crop_identifier_for_video_id() -> None:
+    # given
+    metadata = VideoMetadata(
+        video_identifier="video_id",
+        frame_number=0,
+        frame_timestamp=datetime.now(),
+        fps=40,
+        comes_from_video_file=None,
+    )
+    image = WorkflowImageData(
+        parent_metadata=ImageParentMetadata(parent_id="parent"),
+        numpy_image=np.zeros((192, 168, 3), dtype=np.uint8),
+        video_metadata=metadata,
+    )
+
+    # when
+    result = WorkflowImageData.create_crop(
+        origin_image_data=image,
+        crop_identifier="per-frame-uuid",
+        cropped_image=np.zeros((64, 60, 3)),
+        offset_x=100,
+        offset_y=20,
+        preserve_video_metadata=True,
+        video_crop_identifier="region",
+    )
+
+    # then
+    assert result.parent_metadata.parent_id == "per-frame-uuid"
+    assert result.video_metadata.video_identifier == "video_id | crop: region"
+    assert metadata.video_identifier == "video_id", "Expected origin left intact"
+
+
 def test_parent_origin_from_origin_coordinates_system() -> None:
     # given
     origin_coords = OriginCoordinatesSystem(
diff --git a/workflows/tests/unit_tests/execution_engine/inner_workflow/test_embedded_video_blocks_with_step_output_image.py b/workflows/tests/unit_tests/execution_engine/inner_workflow/test_embedded_video_blocks_with_step_output_image.py
new file mode 100644
index 0000000000..c4ba963ea2
--- /dev/null
+++ b/workflows/tests/unit_tests/execution_engine/inner_workflow/test_embedded_video_blocks_with_step_output_image.py
@@ -0,0 +1,431 @@
+"""
+Embedded workflows whose ``image`` input is bound to a parent step output.
+
+Inlining replaces the child's ``$inputs.image`` with the parent binding, so a child
+that compiles on its own must still compile once its image comes from a step such as
+an image preprocessing block. Video-state blocks (trackers, analytics, cache, flow
+control) used to accept only ``$inputs.*`` image selectors, which made the merged
+definition fail to parse.
+"""
+
+import datetime
+from typing import Any, Dict
+
+import numpy as np
+import pytest
+from roboflow_workflows.core_steps.cache.memory_cache import WorkflowMemoryCache
+from roboflow_workflows.execution_engine.core import ExecutionEngine
+from roboflow_workflows.execution_engine.entities.base import (
+    ImageParentMetadata,
+    VideoMetadata,
+    WorkflowImageData,
+)
+
+ZONE = [[0, 0], [100, 0], [100, 100], [0, 100]]
+
+GRAYSCALE_STEP = {
+    "type": "roboflow_core/convert_grayscale@v1",
+    "name": "standard_image",
+    "image": "$inputs.image",
+}
+GRAYSCALE_IMAGE = "$steps.standard_image.image"
+
+ABSOLUTE_STATIC_CROP_STEP = {
+    "type": "roboflow_core/absolute_static_crop@v1",
+    "name": "standard_image",
+    "images": "$inputs.image",
+    "x_center": 16,
+    "y_center": 16,
+    "width": 16,
+    "height": 16,
+}
+RELATIVE_STATIC_CROP_STEP = {
+    "type": "roboflow_core/relative_statoic_crop@v1",
+    "name": "standard_image",
+    "images": "$inputs.image",
+    "x_center": 0.5,
+    "y_center": 0.5,
+    "width": 0.5,
+    "height": 0.5,
+}
+CROP_IMAGE = "$steps.standard_image.crops"
+
+CAMERA_CALIBRATION_STEP = {
+    "type": "roboflow_core/camera-calibration@v1",
+    "name": "standard_image",
+    "image": "$inputs.image",
+    "fx": 32.0,
+    "fy": 32.0,
+    "cx": 16.0,
+    "cy": 16.0,
+    "k1": 0.0,
+    "k2": 0.0,
+    "k3": 0.0,
+    "p1": 0.0,
+    "p2": 0.0,
+}
+CALIBRATED_IMAGE = "$steps.standard_image.calibrated_image"
+
+VIDEO_STATE_CHILD_WORKFLOW = {
+    "version": "1.0",
+    "inputs": [
+        {"type": "WorkflowImage", "name": "image"},
+        {
+            "type": "WorkflowBatchInput",
+            "name": "predictions",
+            "kind": ["object_detection_prediction"],
+        },
+    ],
+    "steps": [
+        {
+            "type": "roboflow_core/byte_tracker@v2",
+            "name": "tracker",
+            "image": "$inputs.image",
+            "detections": "$inputs.predictions",
+        },
+        {
+            "type": "roboflow_core/time_in_zone@v2",
+            "name": "time_in_zone_v2",
+            "image": "$inputs.image",
+            "detections": "$steps.tracker.tracked_detections",
+            "zone": ZONE,
+        },
+        {
+            "type": "roboflow_core/time_in_zone@v3",
+            "name": "time_in_zone_v3",
+            "image": "$inputs.image",
+            "detections": "$steps.tracker.tracked_detections",
+            "zone": ZONE,
+        },
+        {
+            "type": "roboflow_core/velocity@v1",
+            "name": "velocity",
+            "image": "$inputs.image",
+            "detections": "$steps.tracker.tracked_detections",
+        },
+        {
+            "type": "roboflow_core/line_counter@v2",
+            "name": "line_counter",
+            "image": "$inputs.image",
+            "detections": "$steps.tracker.tracked_detections",
+            "line_segment": [[0, 50], [100, 50]],
+        },
+        {
+            "type": "roboflow_core/path_deviation_analytics@v2",
+            "name": "path_deviation",
+            "image": "$inputs.image",
+            "detections": "$steps.tracker.tracked_detections",
+            "reference_path": [[0, 0], [100, 100]],
+        },
+        {
+            "type": "roboflow_core/detection_event_log@v1",
+            "name": "event_log",
+            "image": "$inputs.image",
+            "detections": "$steps.tracker.tracked_detections",
+        },
+        {
+            "type": "roboflow_core/cache_get@v1",
+            "name": "before",
+            "image": "$inputs.image",
+            "key": "k",
+        },
+        {
+            "type": "roboflow_core/rate_limiter@v1",
+            "name": "rate_limiter",
+            "depends_on": "$steps.before.output",
+            "next_steps": ["$steps.as_text"],
+            "video_reference_image": "$inputs.image",
+        },
+        {
+            "type": "roboflow_core/property_definition@v1",
+            "name": "as_text",
+            "data": "$steps.before.output",
+            "operations": [{"type": "ToString"}],
+        },
+        {
+            "type": "roboflow_core/delta_filter@v1",
+            "name": "delta_filter",
+            "image": "$inputs.image",
+            "value": "$steps.as_text.output",
+            "next_steps": ["$steps.store"],
+        },
+        {
+            "type": "roboflow_core/cache_set@v1",
+            "name": "store",
+            "image": "$inputs.image",
+            "key": "k",
+            "value": "$steps.as_text.output",
+        },
+    ],
+    "outputs": [
+        {
+            "type": "JsonField",
+            "name": "time_in_zone",
+            "selector": "$steps.time_in_zone_v3.timed_detections",
+        },
+        {
+            "type": "JsonField",
+            "name": "velocity",
+            "selector": "$steps.velocity.velocity_detections",
+        },
+    ],
+}
+
+CACHE_CHILD_WORKFLOW = {
+    "version": "1.0",
+    "inputs": [{"type": "WorkflowImage", "name": "image"}],
+    "steps": [
+        {
+            "type": "roboflow_core/cache_get@v1",
+            "name": "before",
+            "image": "$inputs.image",
+            "key": "k",
+        },
+        {
+            "type": "roboflow_core/property_definition@v1",
+            "name": "as_text",
+            "data": "$steps.before.output",
+            "operations": [{"type": "ToString"}],
+        },
+        {
+            "type": "roboflow_core/cache_set@v1",
+            "name": "store",
+            "image": "$inputs.image",
+            "key": "k",
+            "value": "$steps.as_text.output",
+        },
+    ],
+    "outputs": [
+        {"type": "JsonField", "name": "before", "selector": "$steps.before.output"},
+        {"type": "JsonField", "name": "stored", "selector": "$steps.store.output"},
+    ],
+}
+
+
+def _parent_workflow(
+    child: Dict[str, Any],
+    *,
+    outputs: list,
+    image_step: Dict[str, Any] = GRAYSCALE_STEP,
+    image_selector: str = GRAYSCALE_IMAGE,
+    extra_inputs: list = (),
+    extra_bindings: Dict[str, str] = {},
+) -> Dict[str, Any]:
+    return {
+        "version": "1.0",
+        "inputs": [{"type": "WorkflowImage", "name": "image"}, *extra_inputs],
+        "steps": [
+            image_step,
+            {
+                "type": "roboflow_core/inner_workflow@v1",
+                "name": "nested",
+                "workflow_definition": child,
+                "parameter_bindings": {"image": image_selector, **extra_bindings},
+            },
+        ],
+        "outputs": outputs,
+    }
+
+
+def _video_frame(*, video_id: str, frame_number: int) -> WorkflowImageData:
+    video_metadata = VideoMetadata(
+        video_identifier=video_id,
+        frame_number=frame_number,
+        frame_timestamp=datetime.datetime.fromtimestamp(
+            1726570875 + frame_number, tz=datetime.timezone.utc
+        ),
+        fps=1,
+        comes_from_video_file=True,
+    )
+    frame = WorkflowImageData(
+        parent_metadata=ImageParentMetadata(parent_id=video_id),
+        numpy_image=np.zeros((32, 32, 3), dtype=np.uint8),
+        video_metadata=video_metadata,
+    )
+
+    return frame
+
+
+def test_video_state_blocks_compile_when_embedded_image_comes_from_parent_step() -> (
+    None
+):
+    # given
+    workflow_definition = _parent_workflow(
+        VIDEO_STATE_CHILD_WORKFLOW,
+        extra_inputs=[
+            {
+                "type": "WorkflowBatchInput",
+                "name": "predictions",
+                "kind": ["object_detection_prediction"],
+            }
+        ],
+        extra_bindings={"predictions": "$inputs.predictions"},
+        outputs=[
+            {
+                "type": "JsonField",
+                "name": "time_in_zone",
+                "selector": "$steps.nested.time_in_zone",
+            },
+            {
+                "type": "JsonField",
+                "name": "velocity",
+                "selector": "$steps.nested.velocity",
+            },
+        ],
+    )
+
+    # when
+    execution_engine = ExecutionEngine.init(
+        workflow_definition=workflow_definition,
+        init_parameters={},
+        max_concurrent_steps=1,
+    )
+
+    # then
+    assert execution_engine is not None
+
+
+def _reset_memory_cache() -> None:
+    WorkflowMemoryCache.cache.clear()
+    WorkflowMemoryCache._retain_counts.clear()
+
+
+@pytest.fixture
+def clean_memory_cache():
+    _reset_memory_cache()
+    yield
+    _reset_memory_cache()
+
+
+@pytest.mark.parametrize(
+    "image_step, image_selector",
+    [
+        (GRAYSCALE_STEP, GRAYSCALE_IMAGE),
+        (ABSOLUTE_STATIC_CROP_STEP, CROP_IMAGE),
+        (RELATIVE_STATIC_CROP_STEP, CROP_IMAGE),
+    ],
+    ids=["grayscale", "absolute_static_crop", "relative_static_crop"],
+)
+def test_embedded_cache_keeps_one_namespace_per_video_across_frames(
+    clean_memory_cache: None,
+    image_step: Dict[str, Any],
+    image_selector: str,
+) -> None:
+    # given
+    workflow_definition = _parent_workflow(
+        CACHE_CHILD_WORKFLOW,
+        image_step=image_step,
+        image_selector=image_selector,
+        outputs=[
+            {
+                "type": "JsonField",
+                "name": "before",
+                "selector": "$steps.nested.before",
+            },
+            {
+                "type": "JsonField",
+                "name": "stored",
+                "selector": "$steps.nested.stored",
+            },
+        ],
+    )
+    execution_engine = ExecutionEngine.init(
+        workflow_definition=workflow_definition,
+        init_parameters={},
+        max_concurrent_steps=1,
+    )
+
+    # when
+    camera_a_results = [
+        execution_engine.run(
+            runtime_parameters={
+                "image": _video_frame(video_id="cam-a", frame_number=frame_number)
+            }
+        )[0]
+        for frame_number in (1, 2, 3)
+    ]
+    camera_b_result = execution_engine.run(
+        runtime_parameters={"image": _video_frame(video_id="cam-b", frame_number=1)}
+    )[0]
+
+    # then
+    assert camera_a_results == [
+        {"before": False, "stored": "False"},
+        {"before": "False", "stored": "False"},
+        {"before": "False", "stored": "False"},
+    ]
+    assert camera_b_result == {"before": False, "stored": "False"}
+    namespaces = sorted(WorkflowMemoryCache.cache.keys())
+    assert len(namespaces) == 2
+    assert namespaces[0].startswith("cam-a")
+    assert namespaces[1].startswith("cam-b")
+
+
+RATE_LIMITED_CHILD_WORKFLOW = {
+    "version": "1.0",
+    "inputs": [{"type": "WorkflowImage", "name": "image"}],
+    "steps": [
+        {
+            "type": "roboflow_core/rate_limiter@v1",
+            "name": "rate_limiter",
+            "cooldown_seconds": 1.0,
+            "depends_on": "$inputs.image",
+            "next_steps": ["$steps.passed"],
+            "video_reference_image": "$inputs.image",
+        },
+        {
+            "type": "roboflow_core/property_definition@v1",
+            "name": "passed",
+            "data": "$inputs.image",
+            "operations": [{"type": "ExtractImageProperty", "property_name": "width"}],
+        },
+    ],
+    "outputs": [
+        {"type": "JsonField", "name": "passed", "selector": "$steps.passed.output"},
+    ],
+}
+
+
+@pytest.mark.parametrize(
+    "image_step, image_selector",
+    [
+        (GRAYSCALE_STEP, GRAYSCALE_IMAGE),
+        (CAMERA_CALIBRATION_STEP, CALIBRATED_IMAGE),
+    ],
+    ids=["grayscale", "camera_calibration"],
+)
+def test_embedded_rate_limiter_follows_video_time_of_step_output_image(
+    image_step: Dict[str, Any],
+    image_selector: str,
+) -> None:
+    # given
+    workflow_definition = _parent_workflow(
+        RATE_LIMITED_CHILD_WORKFLOW,
+        image_step=image_step,
+        image_selector=image_selector,
+        outputs=[
+            {
+                "type": "JsonField",
+                "name": "passed",
+                "selector": "$steps.nested.passed",
+            },
+        ],
+    )
+    execution_engine = ExecutionEngine.init(
+        workflow_definition=workflow_definition,
+        init_parameters={},
+        max_concurrent_steps=1,
+    )
+
+    # when
+    results = [
+        execution_engine.run(
+            runtime_parameters={
+                "image": _video_frame(video_id="cam-a", frame_number=frame_number)
+            }
+        )[0]
+        for frame_number in (1, 2, 3, 5)
+    ]
+
+    # then
+    assert [result["passed"] for result in results] == [32, 32, 32, 32]