diff --git a/skyvern/forge/sdk/workflow/models/block.py b/skyvern/forge/sdk/workflow/models/block.py
index 896f3ef15..bb561e4d0 100644
--- a/skyvern/forge/sdk/workflow/models/block.py
+++ b/skyvern/forge/sdk/workflow/models/block.py
@@ -336,6 +336,9 @@ class BaseTaskBlock(Block):
     download_timeout: float | None = None  # minutes
     include_extracted_text: bool = True
 
+    def get_engine(self) -> RunEngine | None:
+        return self.engine
+
     def get_all_parameters(
         self,
         workflow_run_id: str,
diff --git a/skyvern/forge/sdk/workflow/models/block_base.py b/skyvern/forge/sdk/workflow/models/block_base.py
index 00a91a472..771cee7d0 100644
--- a/skyvern/forge/sdk/workflow/models/block_base.py
+++ b/skyvern/forge/sdk/workflow/models/block_base.py
@@ -44,7 +44,7 @@ from skyvern.webeye.browser_factory import rebind_download_dir
 from skyvern.webeye.browser_state import BrowserState
 
 if TYPE_CHECKING:
-    from skyvern.forge.sdk.workflow.models.task_blocks import BaseTaskBlock
+    from skyvern.forge.sdk.workflow.models.block import BaseTaskBlock
 
 LOG = structlog.get_logger()
 
diff --git a/tests/unit/test_block_description_caching.py b/tests/unit/test_block_description_caching.py
index cc69cf9f0..321e00a40 100644
--- a/tests/unit/test_block_description_caching.py
+++ b/tests/unit/test_block_description_caching.py
@@ -12,6 +12,7 @@ import pytest
 
 from skyvern.forge.sdk.workflow.models.block import Block, TaskBlock
 from skyvern.forge.sdk.workflow.models.parameter import OutputParameter
+from skyvern.schemas.runs import RunEngine
 from skyvern.schemas.workflows import BlockResult, BlockStatus
 
 
@@ -122,3 +123,28 @@ class TestDescriptionSkippedOnLoopIterations:
                 await block.execute_safe(workflow_run_id="wr_1", current_index=i)
 
             assert mock_gen_desc.call_count == 1
+
+
+class TestEngineRecordedForTaskBlock:
+    """Regression guard for the block.py split: task blocks must record their RunEngine
+    on the workflow_run_block. The split replaced an ``isinstance(self, BaseTaskBlock)``
+    check in execute_safe with a polymorphic ``get_engine()``; if BaseTaskBlock's override
+    is dropped, execute_safe silently records ``engine=None`` (observability regression).
+    """
+
+    @pytest.mark.asyncio
+    async def test_task_block_records_its_engine(self) -> None:
+        block = _make_block()  # _make_block defaults to RunEngine.skyvern_v1
+
+        with (
+            patch("skyvern.forge.sdk.workflow.models.block.app") as mock_app,
+            patch("skyvern.forge.sdk.workflow.models.block_base.app", mock_app),
+            patch.object(TaskBlock, "execute", new_callable=AsyncMock, return_value=_block_result()),
+            patch.object(Block, "_generate_workflow_run_block_description", new_callable=AsyncMock),
+        ):
+            _setup_mocks(mock_app)
+
+            await block.execute_safe(workflow_run_id="wr_1", current_index=None)
+
+            _, kwargs = mock_app.DATABASE.observer.create_workflow_run_block.call_args
+            assert kwargs["engine"] == RunEngine.skyvern_v1