feat(api): Do not enqueue json commands on protocol load (#14759)

# Overview

closes https://opentrons.atlassian.net/browse/EXEC-352.
first step towards fixit commands. do not enqueue json protocol
commands.

# Test Plan

Tested with Json protocols and Postman:
- Make sure loading a protocol and executing it are happening within
order.
- Make sure get run `/commands` returning the list properly with
successful commands.
- Make sure loading a failed protocol should fail the run and fail the
command.
- Make sure get run `/commands` for failed runs the list of commands,
last command being the failed command.
- Fixed e2e test to comply with these changes


# Changelog

- Do no enqueue commands in PE for Json command upon load. 
- Execute commands one by one when run get started - same way we do for
python protocols.


# Review requests

Changes make sense? 
GET run` /commands` will not return the full list of commands if the run
did not start - its a change we are doing to make json protocols run
like python protocols. are we ok with this?

# Risk assessment

Medium. need to do smoke tests for Json protocols and make sure these
changes do not affect anything.

---------

Co-authored-by: Max Marrone <[email protected]>

api/src/opentrons/protocol_runner/protocol_runner.py

@@ -36,6 +36,7 @@
     LegacyExecutor,
     LegacyLoadInfo,
 )
+from ..protocol_engine.errors import ProtocolCommandFailedError
 from ..protocol_engine.types import (
     PostRunHardwareState,
     DeckConfigurationType,
@@ -283,6 +284,7 @@ def __init__(
         )
 
         self._hardware_api.should_taskify_movement_execution(taskify=False)
+        self._queued_commands: List[pe_commands.CommandCreate] = []
 
     async def load(self, protocol_source: ProtocolSource) -> None:
         """Load a JSONv6+ ProtocolSource into managed ProtocolEngine."""
@@ -324,17 +326,16 @@ async def load(self, protocol_source: ProtocolSource) -> None:
                 color=liquid.displayColor,
             )
             await _yield()
+
         initial_home_command = pe_commands.HomeCreate(
             params=pe_commands.HomeParams(axes=None)
         )
         # this command homes all axes, including pipette plugner and gripper jaw
         self._protocol_engine.add_command(request=initial_home_command)
 
-        for command in commands:
-            self._protocol_engine.add_command(request=command)
-            await _yield()
+        self._queued_commands = commands
 
-        self._task_queue.set_run_func(func=self._protocol_engine.wait_until_complete)
+        self._task_queue.set_run_func(func=self._add_command_and_execute)
 
     async def run(  # noqa: D102
         self,
@@ -355,6 +356,15 @@ async def run(  # noqa: D102
         commands = self._protocol_engine.state_view.commands.get_all()
         return RunResult(commands=commands, state_summary=run_data, parameters=[])
 
+    async def _add_command_and_execute(self) -> None:
+        for command in self._queued_commands:
+            result = await self._protocol_engine.add_and_execute_command(command)
+            if result and result.error:
+                raise ProtocolCommandFailedError(
+                    original_error=result.error,
+                    message=f"{result.error.errorType}: {result.error.detail}",
+                )
+
 
 class LiveRunner(AbstractRunner):
     """Protocol runner implementation for live http protocols."""

api/tests/opentrons/protocol_runner/test_protocol_runner.py

@@ -1,4 +1,6 @@
 """Tests for the PythonAndLegacyRunner, JsonRunner & LiveRunner classes."""
+from datetime import datetime
+
 import pytest
 from pytest_lazyfixture import lazy_fixture  # type: ignore[import-untyped]
 from decoy import Decoy, matchers
@@ -18,7 +20,12 @@
 from opentrons.util.broker import Broker
 
 from opentrons import protocol_reader
-from opentrons.protocol_engine import ProtocolEngine, Liquid, commands as pe_commands
+from opentrons.protocol_engine import (
+    ProtocolEngine,
+    Liquid,
+    commands as pe_commands,
+    errors as pe_errors,
+)
 from opentrons.protocol_reader import (
     ProtocolSource,
     JsonProtocolConfig,
@@ -328,6 +335,96 @@ async def test_run_json_runner(
     )
 
 
+async def test_run_json_runner_stop_requested_stops_enquqing(
+    decoy: Decoy,
+    hardware_api: HardwareAPI,
+    protocol_engine: ProtocolEngine,
+    task_queue: TaskQueue,
+    json_runner_subject: JsonRunner,
+    json_file_reader: JsonFileReader,
+    json_translator: JsonTranslator,
+) -> None:
+    """It should run a protocol to completion."""
+    labware_definition = LabwareDefinition.construct()  # type: ignore[call-arg]
+    json_protocol_source = ProtocolSource(
+        directory=Path("/dev/null"),
+        main_file=Path("/dev/null/abc.json"),
+        files=[],
+        metadata={},
+        robot_type="OT-2 Standard",
+        config=JsonProtocolConfig(schema_version=6),
+        content_hash="abc123",
+    )
+
+    commands: List[pe_commands.CommandCreate] = [
+        pe_commands.HomeCreate(params=pe_commands.HomeParams()),
+        pe_commands.WaitForDurationCreate(
+            params=pe_commands.WaitForDurationParams(seconds=10)
+        ),
+        pe_commands.LoadLiquidCreate(
+            params=pe_commands.LoadLiquidParams(
+                liquidId="water-id", labwareId="labware-id", volumeByWell={"A1": 30}
+            )
+        ),
+    ]
+
+    liquids: List[Liquid] = [
+        Liquid(id="water-id", displayName="water", description="water desc")
+    ]
+
+    json_protocol = ProtocolSchemaV6.construct()  # type: ignore[call-arg]
+
+    decoy.when(
+        await protocol_reader.extract_labware_definitions(json_protocol_source)
+    ).then_return([labware_definition])
+    decoy.when(json_file_reader.read(json_protocol_source)).then_return(json_protocol)
+    decoy.when(json_translator.translate_commands(json_protocol)).then_return(commands)
+    decoy.when(json_translator.translate_liquids(json_protocol)).then_return(liquids)
+    decoy.when(
+        await protocol_engine.add_and_execute_command(
+            pe_commands.HomeCreate(params=pe_commands.HomeParams()),
+        )
+    ).then_return(
+        pe_commands.Home.construct(status=pe_commands.CommandStatus.SUCCEEDED)  # type: ignore[call-arg]
+    )
+    decoy.when(
+        await protocol_engine.add_and_execute_command(
+            pe_commands.WaitForDurationCreate(
+                params=pe_commands.WaitForDurationParams(seconds=10)
+            ),
+        )
+    ).then_return(
+        pe_commands.WaitForDuration.construct(  # type: ignore[call-arg]
+            error=pe_errors.ErrorOccurrence.from_failed(
+                id="some-id",
+                createdAt=datetime(year=2021, month=1, day=1),
+                error=pe_errors.ProtocolEngineError(),
+            )
+        )
+    )
+
+    await json_runner_subject.load(json_protocol_source)
+
+    run_func_captor = matchers.Captor()
+
+    decoy.verify(
+        protocol_engine.add_labware_definition(labware_definition),
+        protocol_engine.add_liquid(
+            id="water-id", name="water", description="water desc", color=None
+        ),
+        protocol_engine.add_command(
+            request=pe_commands.HomeCreate(params=pe_commands.HomeParams(axes=None))
+        ),
+        task_queue.set_run_func(func=run_func_captor),
+    )
+
+    # Verify that the run func calls the right things:
+    run_func = run_func_captor.value
+
+    with pytest.raises(pe_errors.ProtocolEngineError):
+        await run_func()
+
+
 @pytest.mark.parametrize(
     "schema_version, json_protocol",
     [
@@ -385,6 +482,8 @@ async def test_load_json_runner(
 
     await json_runner_subject.load(json_protocol_source)
 
+    run_func_captor = matchers.Captor()
+
     decoy.verify(
         protocol_engine.add_labware_definition(labware_definition),
         protocol_engine.add_liquid(
@@ -393,24 +492,30 @@ async def test_load_json_runner(
         protocol_engine.add_command(
             request=pe_commands.HomeCreate(params=pe_commands.HomeParams(axes=None))
         ),
-        protocol_engine.add_command(
+        task_queue.set_run_func(func=run_func_captor),
+    )
+
+    # Verify that the run func calls the right things:
+    run_func = run_func_captor.value
+    await run_func()
+    decoy.verify(
+        await protocol_engine.add_and_execute_command(
             request=pe_commands.WaitForResumeCreate(
                 params=pe_commands.WaitForResumeParams(message="hello")
-            )
+            ),
         ),
-        protocol_engine.add_command(
+        await protocol_engine.add_and_execute_command(
             request=pe_commands.WaitForResumeCreate(
                 params=pe_commands.WaitForResumeParams(message="goodbye")
-            )
+            ),
         ),
-        protocol_engine.add_command(
+        await protocol_engine.add_and_execute_command(
             request=pe_commands.LoadLiquidCreate(
                 params=pe_commands.LoadLiquidParams(
                     liquidId="water-id", labwareId="labware-id", volumeByWell={"A1": 30}
                 )
             ),
         ),
-        task_queue.set_run_func(func=protocol_engine.wait_until_complete),
     )