fix(workflow): restore task-block engine recording + valid TYPE_CHECKING import

Address Copilot review on #7174 (block.py split 1/8):

- BaseTaskBlock.get_engine() override was missing. execute_safe() replaced
  the old `isinstance(self, BaseTaskBlock)` engine check with a polymorphic
  get_engine() returning None on the base, but no override was added — so
  task blocks silently recorded engine=None on their workflow_run_block
  (observability regression). Add the override + a regression test.
- block_base.py TYPE_CHECKING-imported BaseTaskBlock from models.task_blocks,
  which does not exist in this PR (it's a later part of the stack). Point it
  at models.block where BaseTaskBlock currently lives so mypy resolves it.

The ~28 "UnboundLocalError" comments on the multi-context `with (... as
mock_app, patch(..., mock_app))` blocks are false positives: Python enters
each context manager (binding its `as` target) before evaluating the next
context expression. Those tests pass unchanged.

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

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,

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()
 

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