roboflow
diff --git a/‎docs/workflows/execution_engine_changelog.md‎
Lines changed: 31 additions & 0 deletions b/‎docs/workflows/execution_engine_changelog.md‎
Lines changed: 31 additions & 0 deletions
diff --git a/‎docs/workflows/inner_workflow_design.md‎
Lines changed: 159 additions & 0 deletions b/‎docs/workflows/inner_workflow_design.md‎
Lines changed: 159 additions & 0 deletions
diff --git a/‎docs/workflows/workflows_compiler.md‎
Lines changed: 2 additions & 0 deletions b/‎docs/workflows/workflows_compiler.md‎
Lines changed: 2 additions & 0 deletions
diff --git a/‎inference/core/env.py‎
Lines changed: 6 additions & 0 deletions b/‎inference/core/env.py‎
Lines changed: 6 additions & 0 deletions
diff --git a/‎inference/core/workflows/core_steps/flow_control/inner_workflow/__init__.py‎ b/‎inference/core/workflows/core_steps/flow_control/inner_workflow/__init__.py‎
diff --git a/‎inference/core/workflows/core_steps/flow_control/inner_workflow/v1.py‎
Lines changed: 141 additions & 0 deletions b/‎inference/core/workflows/core_steps/flow_control/inner_workflow/v1.py‎
Lines changed: 141 additions & 0 deletions
diff --git a/‎inference/core/workflows/core_steps/loader.py‎
Lines changed: 4 additions & 0 deletions b/‎inference/core/workflows/core_steps/loader.py‎
Lines changed: 4 additions & 0 deletions
@@ -2,6 +2,37 @@
 
 Below you can find the changelog for Execution Engine.
 
+## Execution Engine `v1.9.0` | inference `v1.2.0`
+
+!!! Note "New feature: nested workflows via compile-time inlining"
+
+    This release adds the **`roboflow_core/inner_workflow@v1`** block so a workflow can embed another
+    workflow definition (inline JSON or resolved from `workflow_workspace_id` / `workflow_id` /
+    optional `workflow_version_id`). Child inputs are wired from the parent with
+    **`parameter_bindings`** (child `inputs[].name` → parent selectors). At compile time the
+    engine **inlines** nested steps into the parent graph; execution uses the same path as ordinary
+    steps (no separate nested runtime).
+
+**What changed**
+
+* **Inner workflow block** — New flow-control block type `roboflow_core/inner_workflow@v1` registered
+  in `roboflow_core`. Parent outputs may reference child workflow JsonField outputs as
+  `$steps.<inner_step_name>.<child_output_name>` until inlining rewrites selectors.
+
+* **Compile pipeline** — Before parsing the root definition, the compiler: (1) **normalizes**
+  references (default: Roboflow API + `workflows_core.api_key`, or custom
+  `workflows_core.inner_workflow_spec_resolver`), (2) **validates composition** (acyclicity, max
+  nesting depth, max inner-workflow count), (3) **inlines** all inner workflow steps into ordinary
+  steps, then continues with parse, workflow specification validation, and execution graph
+  construction.
+
+* **Limits (environment variables)** — `WORKFLOWS_MAX_INNER_WORKFLOW_DEPTH` (default `4`) caps
+  containment depth from the root; `WORKFLOWS_MAX_INNER_WORKFLOW_COUNT` (default `32`) caps the
+  total number of `inner_workflow` steps in the nested definition.
+
+* **Documentation** — See [Inner workflows (nested definitions)](./inner_workflow_design.md) for usage,
+  bindings, limits, and an example.
+
 ## Execution Engine `v1.8.0` | inference `v1.1.1`
 
 !!! Note "Additive change + one breaking change due to bug fix with minimal expected impact"
 
@@ -0,0 +1,159 @@
+# Inner workflows (nested definitions)
+
+The **Inner workflow** block (`roboflow_core/inner_workflow@v1`) lets you embed one workflow definition inside another. At **compile time**, the engine resolves any saved-workflow references, validates **composition** (nesting limits and cycles), validates **parameter bindings**, then **inlines** the child’s steps into the parent. After compilation there is no separate “nested run”: the graph is the same as if you had written those steps at the parent level.
+
+This page describes the block, the compile-time pipeline, limits, and a minimal Python example. For general compilation stages, see [Compilation of Workflow Definition](./workflows_compiler.md).
+
+!!! Note
+
+    This feature is implemented in Execution Engine **v1**. The inner workflow block’s `run()` method is never used at runtime; the step is removed during compilation.
+
+## Inner workflow block
+
+Each inner workflow step is a JSON object in the parent’s `steps` list with `type: "roboflow_core/inner_workflow@v1"`.
+
+### How you supply the child definition
+
+You must provide **either**:
+
+- **`workflow_definition`**: a full nested workflow JSON object (same shape as a root workflow: `version`, `inputs`, `steps`, `outputs`), **or**
+- **`workflow_workspace_id`** and **`workflow_id`**, with optional **`workflow_version_id`**, to load a saved workflow spec at compile time.
+
+You must **not** set both an inline `workflow_definition` and the reference fields on the same step.
+
+### `parameter_bindings`
+
+`parameter_bindings` is an object whose **keys** are the **names** of the **child workflow’s** entries in its `inputs` array. Each **value** is a **selector** (or value the engine can coerce) from the **parent** scope, typically:
+
+- `$inputs.<parent_input_name>` for parent workflow inputs, or  
+- `$steps.<parent_step_name>.<output_property>` for data produced by earlier parent steps.
+
+**Rules:**
+
+- Every child input that **requires** a value from the parent must appear in `parameter_bindings`, **except** inputs of type `WorkflowParameter` / `InferenceParameter` that declare a **non-null** `default_value` in the child definition. Those may be omitted; the child’s default is applied when the definition is inlined.
+- Keys that are not child input names are rejected at compile time.
+- Child steps should consume parent data through **`$inputs.<child_input_name>`** in the nested definition; the compiler replaces those references with the bound parent selectors (or injected defaults) during inlining.
+
+### Referencing child outputs from the parent
+
+The nested workflow’s `outputs` array defines **JsonField** entries with a `name` and `selector`. After compilation, the parent treats the inner step like a logical block with outputs named after those JsonField **`name`** values.
+
+From the parent you reference them as:
+
+```text
+$steps.<inner_step_name>.<child_output_name>
+```
+
+where `<child_output_name>` is the `name` field of a JsonField in the child’s `outputs`, not necessarily the last step’s name.
+
+## Compile-time pipeline (Execution Engine v1)
+
+When `compile_workflow_graph` runs, inner workflows go through the following **before** the main “parse workflow definition” step:
+
+1. **Reference resolution (normalization)**  
+   Any step that uses `workflow_workspace_id` / `workflow_id` (and optional `workflow_version_id`) is resolved to an inline `workflow_definition`. This happens recursively inside nested definitions.  
+   - Default resolver uses the Roboflow API and **`workflows_core.api_key`** in workflow init parameters (unless the workspace is `"local"` or you supply a custom resolver).  
+   - Override with init parameter **`workflows_core.inner_workflow_spec_resolver`**: a callable `(workspace_id, workflow_id, workflow_version_id, init_parameters) -> dict` returning the child workflow JSON.
+
+2. **Composition validation**  
+   The engine builds a **composition graph**: one edge per `inner_workflow` step from the parent workflow’s fingerprint to the child definition’s fingerprint. It then checks:  
+   - the graph is **acyclic** (no A → B → … → A),  
+   - **nesting depth** from the root is within **`WORKFLOWS_MAX_INNER_WORKFLOW_DEPTH`**,  
+   - the **total number** of inner-workflow steps (edges) is within **`WORKFLOWS_MAX_INNER_WORKFLOW_COUNT`**.  
+   See [Limits and environment variables](#limits-and-environment-variables) below.
+
+3. **Inlining**  
+   Each `inner_workflow` step is expanded into ordinary steps: child step names become **`{inner_step_name}__{child_step_name}`** (with collision handling), selectors are rewritten (`$inputs` / `$steps` in the child, and parent references to `$steps.<inner_step_name>…`), then the inner step is removed. The rest of compilation (parse, workflow specification validation, execution graph construction, step initialization) sees only a flat workflow.
+
+4. **Parsing and validation**  
+   The flattened JSON is parsed with block manifests, `validate_workflow_specification` runs, and the execution graph is built like any other workflow.
+
+## Example (Python)
+
+The following pattern matches the `examples/workflows/inner_workflows/main.py` example in this repository: resolve a saved workflow by id, bind the parent image into the child’s expected input name, then consume a child output from a downstream parent step.
+
+```python
+import json
+import os
+
+from inference.core.managers.base import ModelManager
+from inference.core.registries.roboflow import RoboflowModelRegistry
+from inference.core.workflows.core_steps.common.entities import StepExecutionMode
+from inference.core.workflows.execution_engine.core import ExecutionEngine
+from inference.models.utils import ROBOFLOW_MODEL_TYPES
+
+WORKFLOW_DEFINITION = {
+    "version": "1.0",
+    "inputs": [
+        {"type": "WorkflowImage", "name": "image"},
+    ],
+    "steps": [
+        {
+            "type": "roboflow_core/inner_workflow@v1",
+            "name": "inner",
+            "workflow_workspace_id": "your-workspace",
+            "workflow_id": "your-workflow-id",
+            "workflow_version_id": "optional-version-id",
+            "parameter_bindings": {
+                "image": "$inputs.image",
+            },
+        },
+        {
+            "type": "roboflow_core/roboflow_classification_model@v2",
+            "name": "classification",
+            "images": "$steps.inner.dynamic_crop_output",
+            "model_id": "resnet50",
+        },
+    ],
+    "outputs": [
+        {
+            "type": "JsonField",
+            "name": "predictions",
+            "selector": "$steps.classification.predictions",
+        },
+    ],
+}
+
+if __name__ == "__main__":
+    model_registry = RoboflowModelRegistry(ROBOFLOW_MODEL_TYPES)
+    model_manager = ModelManager(model_registry=model_registry)
+
+    execution_engine = ExecutionEngine.init(
+        workflow_definition=WORKFLOW_DEFINITION,
+        init_parameters={
+            "workflows_core.model_manager": model_manager,
+            "workflows_core.api_key": os.getenv("ROBOFLOW_API_KEY"),
+            "workflows_core.step_execution_mode": StepExecutionMode.LOCAL,
+        },
+    )
+
+    result = execution_engine.run(
+        runtime_parameters={
+            "image": {
+                "type": "file",
+                "value": "/path/to/your/image.jpg",
+            },
+        },
+    )
+
+    print(json.dumps(result, indent=2, default=str))
+```
+
+Replace workspace, workflow, version, model, output selectors, and image path with values that match your saved workflow and parent graph. The key requirement is that **`parameter_bindings`** keys match the **child** workflow’s `inputs[].name` fields.
+
+## Limits and environment variables
+
+| Variable | Default | Meaning |
+|----------|---------|--------|
+| `WORKFLOWS_MAX_INNER_WORKFLOW_DEPTH` | `4` | Maximum **depth** of the composition graph from the root workflow: each direct `inner_workflow` child counts as one level along a path. |
+| `WORKFLOWS_MAX_INNER_WORKFLOW_COUNT` | `32` | Maximum **number** of `inner_workflow` steps across the whole nested definition (each inner step is one edge in the composition graph). |
+
+**Cycles:** The composition graph must be a **DAG**. You cannot have a cycle of nested references (for example, workflow A embedding B embedding A), even if the per-step execution graph of each workflow is acyclic.
+
+Violations raise compile-time errors (`InnerWorkflowNestingDepthError`, `InnerWorkflowTotalCountError`, `InnerWorkflowCompositionCycleError`, etc.) with messages describing depth, count, or cycle involvement.
+
+## Related reading
+
+- [Workflows definitions](./definitions.md) — JSON shape, inputs, steps, outputs  
+- [Compiler](./workflows_compiler.md) — overall compilation stages  
+- [Workflow execution](./workflow_execution.md) — runtime behavior after compilation  
@@ -35,6 +35,8 @@ during execution and verifying Workflow integrity
 5. Initializing Workflow steps from blocks: Setting up the individual workflow steps based on the available blocks, 
 steps definitions and configuration of execution environment.
 
+If the definition contains [`roboflow_core/inner_workflow@v1`](./inner_workflow_design.md) steps, the compiler **first** resolves saved-workflow references, validates nested **composition** (depth, total count, acyclicity), and **inlines** child steps into the parent **before** parsing and building the execution graph. See [Inner workflows (nested definitions)](./inner_workflow_design.md) for the full pipeline, `parameter_bindings`, and environment limits.
+
 Let's take a closer look at each of the workflow compilation steps.
 
 ### Workflows blocks loading
 
@@ -639,6 +639,12 @@
 WORKFLOWS_STEP_EXECUTION_MODE = os.getenv("WORKFLOWS_STEP_EXECUTION_MODE", "local")
 WORKFLOWS_REMOTE_API_TARGET = os.getenv("WORKFLOWS_REMOTE_API_TARGET", "hosted")
 WORKFLOWS_MAX_CONCURRENT_STEPS = int(os.getenv("WORKFLOWS_MAX_CONCURRENT_STEPS", "8"))
+WORKFLOWS_MAX_INNER_WORKFLOW_DEPTH = int(
+    os.getenv("WORKFLOWS_MAX_INNER_WORKFLOW_DEPTH", "4")
+)
+WORKFLOWS_MAX_INNER_WORKFLOW_COUNT = int(
+    os.getenv("WORKFLOWS_MAX_INNER_WORKFLOW_COUNT", "32")
+)
 WORKFLOWS_REMOTE_EXECUTION_MAX_STEP_BATCH_SIZE = int(
     os.getenv("WORKFLOWS_REMOTE_EXECUTION_MAX_STEP_BATCH_SIZE", "1")
 )
 
@@ -0,0 +1,141 @@
+from typing import Any, Dict, List, Literal, Optional, Type
+
+from pydantic import ConfigDict, Field, model_validator
+
+from inference.core.workflows.execution_engine.entities.base import OutputDefinition
+from inference.core.workflows.execution_engine.entities.types import (
+    WILDCARD_KIND,
+    Selector,
+)
+from inference.core.workflows.execution_engine.v1.inner_workflow.errors import (
+    InnerWorkflowRunNotSupportedError,
+)
+from inference.core.workflows.prototypes.block import (
+    BlockResult,
+    WorkflowBlock,
+    WorkflowBlockManifest,
+)
+
+SHORT_DESCRIPTION = (
+    "Run a nested workflow definition with parameters mapped from the parent workflow."
+)
+
+LONG_DESCRIPTION = """
+Execute a **nested workflow** while mapping parent data into the child's inputs via `parameter_bindings`.
+
+Provide either a full inline definition in `workflow_definition`, or resolve a saved workflow using
+`workflow_workspace_id` and `workflow_id` (optional `workflow_version_id`).
+Reference fields are expanded at compile time via `workflows_core.inner_workflow_spec_resolver`
+(default: Roboflow API using `workflows_core.api_key`, or local definitions when workspace is
+`"local"`).
+
+At compile time the engine validates composition (acyclicity, max depth) and `parameter_bindings`,
+then **inlines** the child's steps into the parent workflow graph (same execution path as ordinary
+steps).
+
+The block's `run()` method is not used at runtime; do not call it directly.
+"""
+
+
+class BlockManifest(WorkflowBlockManifest):
+    model_config = ConfigDict(
+        json_schema_extra={
+            "name": "Inner Workflow",
+            "version": "v1",
+            "short_description": SHORT_DESCRIPTION,
+            "long_description": LONG_DESCRIPTION,
+            "license": "Apache-2.0",
+            "block_type": "flow_control",
+            "ui_manifest": {
+                "section": "flow_control",
+                "icon": "fak fa-diagram-nested",
+                "blockPriority": 2,
+            },
+        }
+    )
+    type: Literal["roboflow_core/inner_workflow@v1"]
+    workflow_definition: Optional[Dict[str, Any]] = Field(
+        default=None,
+        description=(
+            "Full nested workflow definition (same JSON shape as a root workflow: version, inputs, "
+            "steps, outputs). Required unless `workflow_workspace_id` and `workflow_id` are set; "
+            "mutually exclusive with those reference fields."
+        ),
+    )
+    workflow_workspace_id: Optional[str] = Field(
+        default=None,
+        description=(
+            'Workspace id for a saved workflow to load (Roboflow slug or `"local"` for on-disk '
+            "definitions). Use with `workflow_id`; mutually exclusive with a non-empty "
+            "`workflow_definition`."
+        ),
+    )
+    workflow_id: Optional[str] = Field(
+        default=None,
+        description="Saved workflow id to fetch. Use with `workflow_workspace_id`.",
+    )
+    workflow_version_id: Optional[str] = Field(
+        default=None,
+        description="Optional pinned workflow version when resolving by id.",
+    )
+    parameter_bindings: Dict[str, Selector()] = Field(
+        description=(
+            "Maps **child** workflow input names to a selector (or literal coerced by the engine) "
+            "from the parent. Required for every child input except `WorkflowParameter` / "
+            "`InferenceParameter` entries that declare a non-null `default_value` (those may be "
+            "omitted and the child's default is used during compilation inlining)."
+        ),
+        json_schema_extra={
+            "keys_bound_in": "parameter_bindings",
+        },
+    )
+
+    @model_validator(mode="after")
+    def validate_workflow_or_reference(self) -> "BlockManifest":
+        has_inline = (
+            isinstance(self.workflow_definition, dict)
+            and len(self.workflow_definition) > 0
+        )
+
+        workspace_id = (self.workflow_workspace_id or "").strip()
+        workflow_id = (self.workflow_id or "").strip()
+        has_ref = bool(workspace_id and workflow_id)
+
+        if has_inline and has_ref:
+            raise ValueError(
+                "Provide either `workflow_definition` or workflow reference fields "
+                "(`workflow_workspace_id` and `workflow_id`), not both."
+            )
+
+        if has_inline or has_ref:
+            return self
+
+        raise ValueError(
+            "inner_workflow requires a non-empty `workflow_definition` object or both "
+            "`workflow_workspace_id` and `workflow_id`."
+        )
+
+    @classmethod
+    def describe_outputs(cls) -> List[OutputDefinition]:
+        return [OutputDefinition(name="*", kind=[WILDCARD_KIND])]
+
+    @classmethod
+    def accepts_batch_input(cls) -> bool:
+        return False
+
+    @classmethod
+    def get_execution_engine_compatibility(cls) -> Optional[str]:
+        return ">=1.4.0,<2.0.0"
+
+
+class InnerWorkflowBlockV1(WorkflowBlock):
+    """Placeholder block; inner workflows are expanded at compile time and never executed as a unit."""
+
+    @classmethod
+    def get_manifest(cls) -> Type[WorkflowBlockManifest]:
+        return BlockManifest
+
+    def run(self, *args, **kwargs) -> BlockResult:
+        raise InnerWorkflowRunNotSupportedError(
+            "inner_workflow steps are compiled away into ordinary steps; block.run() must not be called."
+        )
@@ -137,6 +137,9 @@
 from inference.core.workflows.core_steps.flow_control.delta_filter.v1 import (
     DeltaFilterBlockV1,
 )
+from inference.core.workflows.core_steps.flow_control.inner_workflow.v1 import (
+    InnerWorkflowBlockV1,
+)
 from inference.core.workflows.core_steps.flow_control.rate_limiter.v1 import (
     RateLimiterBlockV1,
 )
@@ -714,6 +717,7 @@ def load_blocks() -> List[Type[WorkflowBlock]]:
         DetectionsTransformationBlockV1,
         RoboflowDatasetUploadBlockV1,
         ContinueIfBlockV1,
+        InnerWorkflowBlockV1,
         RateLimiterBlockV1,
         PerspectiveCorrectionBlockV1,
         DeltaFilterBlockV1,