From d0b71f83fba337b55ff80d65c22f4a7fda67e292 Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Fri, 16 Jun 2023 18:16:10 -0400
Subject: [PATCH 01/29] Minor improvements to async_context_manager_in_thread()
 and its tests.

* Fix a confusing assert with an incorrect message.
* Comment and refactor test_loop_lifetime() for clarity.
* Rename loop_queue to loop_mailbox to reflect what we're using it for.
---
 .../opentrons/async_context_manager_in_thread.py     |  6 +++---
 .../test_async_context_manager_in_thread.py          | 12 +++++++++---
 2 files changed, 12 insertions(+), 6 deletions(-)

diff --git a/api/tests/opentrons/async_context_manager_in_thread.py b/api/tests/opentrons/async_context_manager_in_thread.py
index 75f2d982085..448f8f0ec10 100644
--- a/api/tests/opentrons/async_context_manager_in_thread.py
+++ b/api/tests/opentrons/async_context_manager_in_thread.py
@@ -60,14 +60,14 @@ def _run_loop_in_thread() -> typing.Generator[asyncio.AbstractEventLoop, None, N
 
     Exiting this context manager stops and cleans up the event loop, and then joins the thread.
     """
-    loop_queue: "queue.SimpleQueue[asyncio.AbstractEventLoop]" = queue.SimpleQueue()
+    loop_mailbox: "queue.SimpleQueue[asyncio.AbstractEventLoop]" = queue.SimpleQueue()
 
     def _in_thread() -> None:
         loop = asyncio.new_event_loop()
 
         # We assume that the lines above this will never fail,
         # so we will always reach this point to unblock the parent thread.
-        loop_queue.put(loop)
+        loop_mailbox.put(loop)
 
         loop.run_forever()
 
@@ -90,7 +90,7 @@ def _in_thread() -> None:
     with ThreadPoolExecutor(max_workers=1) as executor:
         executor.submit(_in_thread)
 
-        loop_in_thread = loop_queue.get()
+        loop_in_thread = loop_mailbox.get()
 
         try:
             yield loop_in_thread
diff --git a/api/tests/opentrons/test_async_context_manager_in_thread.py b/api/tests/opentrons/test_async_context_manager_in_thread.py
index 9eaf63c438c..6901a9c64c0 100644
--- a/api/tests/opentrons/test_async_context_manager_in_thread.py
+++ b/api/tests/opentrons/test_async_context_manager_in_thread.py
@@ -72,8 +72,14 @@ async def __aexit__(
             pass
 
     with async_context_manager_in_thread(NoOp()) as (_, loop_in_thread):
-        asyncio.run_coroutine_threadsafe(asyncio.sleep(0.000001), loop_in_thread)
-
+        # As a smoke test to see if the event loop is running and usable,
+        # run an arbitrary coroutine and wait for it to finish.
+        (
+            asyncio.run_coroutine_threadsafe(asyncio.sleep(0.000001), loop_in_thread)
+        ).result()
+
+    # The loop should be closed and unusable now that the context manager has exited.
+    assert loop_in_thread.is_closed
     with pytest.raises(RuntimeError, match="Event loop is closed"):
         loop_in_thread.call_soon_threadsafe(lambda: None)
 
@@ -111,4 +117,4 @@ async def __aexit__(
     context_manager = RaiseExceptionOnExit()
     with pytest.raises(RuntimeError, match="Oh the humanity"):
         with async_context_manager_in_thread(context_manager):
-            assert False, "We should not reach here."
+            pass

From 4896deabcd3f77baf79ab54dcf4db5ea232c5054 Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Fri, 16 Jun 2023 17:53:01 -0400
Subject: [PATCH 02/29] Move async_context_manager_in_thread() to
 opentrons.utils.

This promotes this test helper to production code.
---
 api/src/opentrons/util/async_helpers.py       | 108 +++++++++++++++++-
 .../async_context_manager_in_thread.py        |  98 ----------------
 .../opentrons/protocol_engine_in_thread.py    |   3 +-
 .../test_async_helpers.py}                    |  18 +--
 4 files changed, 114 insertions(+), 113 deletions(-)
 delete mode 100644 api/tests/opentrons/async_context_manager_in_thread.py
 rename api/tests/opentrons/{test_async_context_manager_in_thread.py => util/test_async_helpers.py} (87%)

diff --git a/api/src/opentrons/util/async_helpers.py b/api/src/opentrons/util/async_helpers.py
index 56606dda468..0c08e18afef 100644
--- a/api/src/opentrons/util/async_helpers.py
+++ b/api/src/opentrons/util/async_helpers.py
@@ -2,10 +2,22 @@
 util.async_helpers - various utilities for asyncio functions and tasks.
 """
 
+from concurrent.futures import ThreadPoolExecutor
 from functools import wraps
-from typing import TypeVar, Callable, Awaitable, cast, Any
+from typing import (
+    Any,
+    AsyncContextManager,
+    Awaitable,
+    Callable,
+    Generator,
+    Tuple,
+    TypeVar,
+    cast,
+)
 
 import asyncio
+import contextlib
+import queue
 
 
 async def asyncio_yield() -> None:
@@ -36,10 +48,10 @@ async def and await call() that still effectively "block" other concurrent tasks
     await asyncio.sleep(0)
 
 
-Wrapped = TypeVar("Wrapped", bound=Callable[..., Awaitable[Any]])
+_Wrapped = TypeVar("_Wrapped", bound=Callable[..., Awaitable[Any]])
 
 
-def ensure_yield(async_def_func: Wrapped) -> Wrapped:
+def ensure_yield(async_def_func: _Wrapped) -> _Wrapped:
     """
     A decorator that makes sure that asyncio_yield() is called after the decorated async
     function finishes executing.
@@ -57,4 +69,92 @@ async def _wrapper(*args: Any, **kwargs: Any) -> Any:
         await asyncio_yield()
         return ret
 
-    return cast(Wrapped, _wrapper)
+    return cast(_Wrapped, _wrapper)
+
+
+_ContextManagerResult = TypeVar("_ContextManagerResult")
+
+
+@contextlib.contextmanager
+def async_context_manager_in_thread(
+    async_context_manager: AsyncContextManager[_ContextManagerResult],
+) -> Generator[Tuple[_ContextManagerResult, asyncio.AbstractEventLoop], None, None]:
+    """Enter an async context manager in a worker thread.
+
+    When you enter this context manager, it:
+
+    1. Spawns a worker thread.
+    2. In that thread, starts an asyncio event loop.
+    3. In that event loop, enters the context manager that you passed in.
+    4. Returns: the result of entering that context manager, and the running event loop.
+       Use functions like `asyncio.run_coroutine_threadsafe()` to safely interact
+       with the returned object from your thread.
+
+    When you exit this context manager, it:
+
+    1. In the worker thread's event loop, exits the context manager that you passed in.
+    2. Stops and cleans up the worker thread's event loop.
+    3. Joins the worker thread.
+    """
+    with _run_loop_in_thread() as loop_in_thread:
+        async_object = asyncio.run_coroutine_threadsafe(
+            async_context_manager.__aenter__(),
+            loop=loop_in_thread,
+        ).result()
+
+        try:
+            yield async_object, loop_in_thread
+
+        finally:
+            exit = asyncio.run_coroutine_threadsafe(
+                async_context_manager.__aexit__(None, None, None),
+                loop=loop_in_thread,
+            )
+            exit.result()
+
+
+@contextlib.contextmanager
+def _run_loop_in_thread() -> Generator[asyncio.AbstractEventLoop, None, None]:
+    """Run an event loop in a worker thread.
+
+    Entering this context manager spawns a thread, starts an asyncio event loop in it,
+    and returns that loop.
+
+    Exiting this context manager stops and cleans up the event loop, and then joins the thread.
+    """
+    loop_mailbox: "queue.SimpleQueue[asyncio.AbstractEventLoop]" = queue.SimpleQueue()
+
+    def _in_thread() -> None:
+        loop = asyncio.new_event_loop()
+
+        # We assume that the lines above this will never fail,
+        # so we will always reach this point to unblock the parent thread.
+        loop_mailbox.put(loop)
+
+        loop.run_forever()
+
+        # If we've reached here, the loop has been stopped from outside this thread. Clean it up.
+        #
+        # This cleanup is naive because asyncio makes it difficult and confusing to get it right.
+        # Compare this with asyncio.run()'s cleanup, which:
+        #
+        # * Cancels and awaits any remaining tasks
+        #   (according to the source code--this seems undocumented)
+        # * Shuts down asynchronous generators
+        #   (see asyncio.shutdown_asyncgens())
+        # * Shuts down the default thread pool executor
+        #   (see https://bugs.python.org/issue34037 and asyncio.shutdown_default_executor())
+        #
+        # In Python >=3.11, we should rewrite this to use asyncio.Runner,
+        # which can take care of these nuances for us.
+        loop.close()
+
+    with ThreadPoolExecutor(max_workers=1) as executor:
+        executor.submit(_in_thread)
+
+        loop_in_thread = loop_mailbox.get()
+
+        try:
+            yield loop_in_thread
+        finally:
+            loop_in_thread.call_soon_threadsafe(loop_in_thread.stop)
diff --git a/api/tests/opentrons/async_context_manager_in_thread.py b/api/tests/opentrons/async_context_manager_in_thread.py
deleted file mode 100644
index 448f8f0ec10..00000000000
--- a/api/tests/opentrons/async_context_manager_in_thread.py
+++ /dev/null
@@ -1,98 +0,0 @@
-"""A test helper to enter an async context manager in a worker thread."""
-
-from __future__ import annotations
-
-import asyncio
-import contextlib
-import queue
-import typing
-
-from concurrent.futures import ThreadPoolExecutor
-
-
-_T = typing.TypeVar("_T")
-
-
-@contextlib.contextmanager
-def async_context_manager_in_thread(
-    async_context_manager: typing.AsyncContextManager[_T],
-) -> typing.Generator[typing.Tuple[_T, asyncio.AbstractEventLoop], None, None]:
-    """Enter an async context manager in a worker thread.
-
-    When you enter this context manager, it:
-
-    1. Spawns a worker thread.
-    2. In that thread, starts an asyncio event loop.
-    3. In that event loop, enters the context manager that you passed in.
-    4. Returns: the result of entering that context manager, and the running event loop.
-       Use functions like `asyncio.run_coroutine_threadsafe()` to safely interact
-       with the returned object from your thread.
-
-    When you exit this context manager, it:
-
-    1. In the worker thread's event loop, exits the context manager that you passed in.
-    2. Stops and cleans up the worker thread's event loop.
-    3. Joins the worker thread.
-    """
-    with _run_loop_in_thread() as loop_in_thread:
-        async_object = asyncio.run_coroutine_threadsafe(
-            async_context_manager.__aenter__(),
-            loop=loop_in_thread,
-        ).result()
-
-        try:
-            yield async_object, loop_in_thread
-
-        finally:
-            exit = asyncio.run_coroutine_threadsafe(
-                async_context_manager.__aexit__(None, None, None),
-                loop=loop_in_thread,
-            )
-            exit.result()
-
-
-@contextlib.contextmanager
-def _run_loop_in_thread() -> typing.Generator[asyncio.AbstractEventLoop, None, None]:
-    """Run an event loop in a worker thread.
-
-    Entering this context manager spawns a thread, starts an asyncio event loop in it,
-    and returns that loop.
-
-    Exiting this context manager stops and cleans up the event loop, and then joins the thread.
-    """
-    loop_mailbox: "queue.SimpleQueue[asyncio.AbstractEventLoop]" = queue.SimpleQueue()
-
-    def _in_thread() -> None:
-        loop = asyncio.new_event_loop()
-
-        # We assume that the lines above this will never fail,
-        # so we will always reach this point to unblock the parent thread.
-        loop_mailbox.put(loop)
-
-        loop.run_forever()
-
-        # If we've reached here, the loop has been stopped from outside this thread. Clean it up.
-        #
-        # This cleanup is naive because asyncio makes it difficult and confusing to get it right.
-        # Compare this with asyncio.run()'s cleanup, which:
-        #
-        # * Cancels and awaits any remaining tasks
-        #   (according to the source code--this seems undocumented)
-        # * Shuts down asynchronous generators
-        #   (see asyncio.shutdown_asyncgens())
-        # * Shuts down the default thread pool executor
-        #   (see https://bugs.python.org/issue34037 and asyncio.shutdown_default_executor())
-        #
-        # In Python >=3.11, we should rewrite this to use asyncio.Runner,
-        # which can take care of these nuances for us.
-        loop.close()
-
-    with ThreadPoolExecutor(max_workers=1) as executor:
-        executor.submit(_in_thread)
-
-        loop_in_thread = loop_mailbox.get()
-
-        try:
-            yield loop_in_thread
-        finally:
-            loop_in_thread.call_soon_threadsafe(loop_in_thread.stop)
diff --git a/api/tests/opentrons/protocol_engine_in_thread.py b/api/tests/opentrons/protocol_engine_in_thread.py
index ec7ca1b21f6..02eb6e2a22f 100644
--- a/api/tests/opentrons/protocol_engine_in_thread.py
+++ b/api/tests/opentrons/protocol_engine_in_thread.py
@@ -11,8 +11,7 @@
     Config,
     DeckType,
 )
-
-from .async_context_manager_in_thread import async_context_manager_in_thread
+from opentrons.util.async_helpers import async_context_manager_in_thread
 
 
 @contextlib.contextmanager
diff --git a/api/tests/opentrons/test_async_context_manager_in_thread.py b/api/tests/opentrons/util/test_async_helpers.py
similarity index 87%
rename from api/tests/opentrons/test_async_context_manager_in_thread.py
rename to api/tests/opentrons/util/test_async_helpers.py
index 6901a9c64c0..6220c23b3f8 100644
--- a/api/tests/opentrons/test_async_context_manager_in_thread.py
+++ b/api/tests/opentrons/util/test_async_helpers.py
@@ -1,11 +1,8 @@
-"""Tests for the `async_context_manager_in_thread` helper."""
-
-
 import asyncio
 
 import pytest
 
-from .async_context_manager_in_thread import async_context_manager_in_thread
+from opentrons.util import async_helpers as subject
 
 
 def test_enters_and_exits() -> None:
@@ -30,7 +27,7 @@ async def __aexit__(
     assert not context_manager.entered
     assert not context_manager.exited
 
-    with async_context_manager_in_thread(context_manager) as (result, _):
+    with subject.async_context_manager_in_thread(context_manager) as (result, _):
         assert context_manager.entered
         assert not context_manager.exited
         assert result == "Yay!"
@@ -51,7 +48,10 @@ async def __aexit__(
             pass
 
     context_manager = ContextManager()
-    with async_context_manager_in_thread(context_manager) as (result, loop_in_thread):
+    with subject.async_context_manager_in_thread(context_manager) as (
+        result,
+        loop_in_thread,
+    ):
         assert result is loop_in_thread
 
 
@@ -71,7 +71,7 @@ async def __aexit__(
         ) -> None:
             pass
 
-    with async_context_manager_in_thread(NoOp()) as (_, loop_in_thread):
+    with subject.async_context_manager_in_thread(NoOp()) as (_, loop_in_thread):
         # As a smoke test to see if the event loop is running and usable,
         # run an arbitrary coroutine and wait for it to finish.
         (
@@ -98,7 +98,7 @@ async def __aexit__(
 
     context_manager = RaiseExceptionOnEnter()
     with pytest.raises(RuntimeError, match="Oh the humanity"):
-        with async_context_manager_in_thread(context_manager):
+        with subject.async_context_manager_in_thread(context_manager):
             assert False, "We should not reach here."
 
 
@@ -116,5 +116,5 @@ async def __aexit__(
 
     context_manager = RaiseExceptionOnExit()
     with pytest.raises(RuntimeError, match="Oh the humanity"):
-        with async_context_manager_in_thread(context_manager):
+        with subject.async_context_manager_in_thread(context_manager):
             pass

From a2683bec99c6bc6a61bbc6e390b05e1a5521617b Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Fri, 16 Jun 2023 18:24:16 -0400
Subject: [PATCH 03/29] Collect async_context_manager_in_thread() tests into a
 class.

We don't really do this often, but it seems nice for keeping the test function names short while still leaving room for testing the other functions in this file.
---
 .../opentrons/util/test_async_helpers.py      | 216 +++++++++---------
 1 file changed, 111 insertions(+), 105 deletions(-)

diff --git a/api/tests/opentrons/util/test_async_helpers.py b/api/tests/opentrons/util/test_async_helpers.py
index 6220c23b3f8..14f9e1a0436 100644
--- a/api/tests/opentrons/util/test_async_helpers.py
+++ b/api/tests/opentrons/util/test_async_helpers.py
@@ -5,116 +5,122 @@
 from opentrons.util import async_helpers as subject
 
 
-def test_enters_and_exits() -> None:
-    """It should enter and exit the given context manager appropriately, and return its result."""
+class TestAsyncContextManagerInThread:
+    """Tests for `async_context_manager_in_thread()`."""
 
-    class ContextManager:
-        def __init__(self) -> None:
-            self.entered = False
-            self.exited = False
+    @staticmethod
+    def test_enters_and_exits() -> None:
+        """It should enter and exit the given context manager appropriately, and return its result."""
 
-        async def __aenter__(self) -> str:
-            self.entered = True
-            return "Yay!"
+        class ContextManager:
+            def __init__(self) -> None:
+                self.entered = False
+                self.exited = False
 
-        async def __aexit__(
-            self, exc_type: object, exc_val: object, exc_tb: object
-        ) -> None:
-            self.exited = True
+            async def __aenter__(self) -> str:
+                self.entered = True
+                return "Yay!"
 
-    context_manager = ContextManager()
+            async def __aexit__(
+                self, exc_type: object, exc_val: object, exc_tb: object
+            ) -> None:
+                self.exited = True
 
-    assert not context_manager.entered
-    assert not context_manager.exited
+        context_manager = ContextManager()
 
-    with subject.async_context_manager_in_thread(context_manager) as (result, _):
-        assert context_manager.entered
+        assert not context_manager.entered
         assert not context_manager.exited
-        assert result == "Yay!"
 
-    assert context_manager.exited
-
-
-def test_returns_matching_loop() -> None:
-    """It should return the event loop that the given context manager is running in."""
-
-    class ContextManager:
-        async def __aenter__(self) -> asyncio.AbstractEventLoop:
-            return asyncio.get_running_loop()
-
-        async def __aexit__(
-            self, exc_type: object, exc_val: object, exc_tb: object
-        ) -> None:
-            pass
-
-    context_manager = ContextManager()
-    with subject.async_context_manager_in_thread(context_manager) as (
-        result,
-        loop_in_thread,
-    ):
-        assert result is loop_in_thread
-
-
-def test_loop_lifetime() -> None:
-    """Test the lifetime of the returned event loop.
-
-    While the context manager is open, the event loop should be running and usable.
-    After the context manager closes, the event loop should be closed and unusable.
-    """
-
-    class NoOp:
-        async def __aenter__(self) -> None:
-            return None
-
-        async def __aexit__(
-            self, exc_type: object, exc_val: object, exc_tb: object
-        ) -> None:
-            pass
-
-    with subject.async_context_manager_in_thread(NoOp()) as (_, loop_in_thread):
-        # As a smoke test to see if the event loop is running and usable,
-        # run an arbitrary coroutine and wait for it to finish.
-        (
-            asyncio.run_coroutine_threadsafe(asyncio.sleep(0.000001), loop_in_thread)
-        ).result()
-
-    # The loop should be closed and unusable now that the context manager has exited.
-    assert loop_in_thread.is_closed
-    with pytest.raises(RuntimeError, match="Event loop is closed"):
-        loop_in_thread.call_soon_threadsafe(lambda: None)
-
-
-def test_propagates_exception_from_enter() -> None:
-    """If the given context manager raises an exception when it's entered, it should propagate."""
-
-    class RaiseExceptionOnEnter:
-        async def __aenter__(self) -> None:
-            raise RuntimeError("Oh the humanity.")
-
-        async def __aexit__(
-            self, exc_type: object, exc_val: object, exc_tb: object
-        ) -> None:
-            assert False, "We should not reach here."
-
-    context_manager = RaiseExceptionOnEnter()
-    with pytest.raises(RuntimeError, match="Oh the humanity"):
-        with subject.async_context_manager_in_thread(context_manager):
-            assert False, "We should not reach here."
-
-
-def test_propagates_exception_from_exit() -> None:
-    """If the given context manager raises an exception when it's exited, it should propagate."""
-
-    class RaiseExceptionOnExit:
-        async def __aenter__(self) -> None:
-            return None
-
-        async def __aexit__(
-            self, exc_type: object, exc_val: object, exc_tb: object
-        ) -> None:
-            raise RuntimeError("Oh the humanity.")
-
-    context_manager = RaiseExceptionOnExit()
-    with pytest.raises(RuntimeError, match="Oh the humanity"):
-        with subject.async_context_manager_in_thread(context_manager):
-            pass
+        with subject.async_context_manager_in_thread(context_manager) as (result, _):
+            assert context_manager.entered
+            assert not context_manager.exited
+            assert result == "Yay!"
+
+        assert context_manager.exited
+
+    @staticmethod
+    def test_returns_matching_loop() -> None:
+        """It should return the event loop that the given context manager is running in."""
+
+        class ContextManager:
+            async def __aenter__(self) -> asyncio.AbstractEventLoop:
+                return asyncio.get_running_loop()
+
+            async def __aexit__(
+                self, exc_type: object, exc_val: object, exc_tb: object
+            ) -> None:
+                pass
+
+        context_manager = ContextManager()
+        with subject.async_context_manager_in_thread(context_manager) as (
+            result,
+            loop_in_thread,
+        ):
+            assert result is loop_in_thread
+
+    @staticmethod
+    def test_loop_lifetime() -> None:
+        """Test the lifetime of the returned event loop.
+
+        While the context manager is open, the event loop should be running and usable.
+        After the context manager closes, the event loop should be closed and unusable.
+        """
+
+        class NoOp:
+            async def __aenter__(self) -> None:
+                return None
+
+            async def __aexit__(
+                self, exc_type: object, exc_val: object, exc_tb: object
+            ) -> None:
+                pass
+
+        with subject.async_context_manager_in_thread(NoOp()) as (_, loop_in_thread):
+            # As a smoke test to see if the event loop is running and usable,
+            # run an arbitrary coroutine and wait for it to finish.
+            (
+                asyncio.run_coroutine_threadsafe(
+                    asyncio.sleep(0.000001), loop_in_thread
+                )
+            ).result()
+
+        # The loop should be closed and unusable now that the context manager has exited.
+        assert loop_in_thread.is_closed
+        with pytest.raises(RuntimeError, match="Event loop is closed"):
+            loop_in_thread.call_soon_threadsafe(lambda: None)
+
+    @staticmethod
+    def test_propagates_exception_from_enter() -> None:
+        """If the given context manager raises an exception when it's entered, it should propagate."""
+
+        class RaiseExceptionOnEnter:
+            async def __aenter__(self) -> None:
+                raise RuntimeError("Oh the humanity.")
+
+            async def __aexit__(
+                self, exc_type: object, exc_val: object, exc_tb: object
+            ) -> None:
+                assert False, "We should not reach here."
+
+        context_manager = RaiseExceptionOnEnter()
+        with pytest.raises(RuntimeError, match="Oh the humanity"):
+            with subject.async_context_manager_in_thread(context_manager):
+                assert False, "We should not reach here."
+
+    @staticmethod
+    def test_propagates_exception_from_exit() -> None:
+        """If the given context manager raises an exception when it's exited, it should propagate."""
+
+        class RaiseExceptionOnExit:
+            async def __aenter__(self) -> None:
+                return None
+
+            async def __aexit__(
+                self, exc_type: object, exc_val: object, exc_tb: object
+            ) -> None:
+                raise RuntimeError("Oh the humanity.")
+
+        context_manager = RaiseExceptionOnExit()
+        with pytest.raises(RuntimeError, match="Oh the humanity"):
+            with subject.async_context_manager_in_thread(context_manager):
+                pass

From 91b7b22eb277535f78028936b9861383d01456ee Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Fri, 16 Jun 2023 18:36:52 -0400
Subject: [PATCH 04/29] Move protocol_engine_in_thread() to
 opentrons.protocol_engine.

This promotes this test helper to production code.
---
 api/src/opentrons/protocol_engine/__init__.py |  6 +-
 .../protocol_engine/create_protocol_engine.py | 66 +++++++++++++++++-
 api/tests/opentrons/conftest.py               |  5 +-
 .../opentrons/protocol_engine_in_thread.py    | 67 -------------------
 4 files changed, 70 insertions(+), 74 deletions(-)
 delete mode 100644 api/tests/opentrons/protocol_engine_in_thread.py

diff --git a/api/src/opentrons/protocol_engine/__init__.py b/api/src/opentrons/protocol_engine/__init__.py
index 37c6c0650b1..af5543a77f3 100644
--- a/api/src/opentrons/protocol_engine/__init__.py
+++ b/api/src/opentrons/protocol_engine/__init__.py
@@ -5,7 +5,10 @@
 protocol state and side-effects like robot movements.
 """
 
-from .create_protocol_engine import create_protocol_engine
+from .create_protocol_engine import (
+    create_protocol_engine,
+    create_protocol_engine_in_thread,
+)
 from .protocol_engine import ProtocolEngine
 from .errors import ProtocolEngineError, ErrorOccurrence
 from .commands import (
@@ -53,6 +56,7 @@
 __all__ = [
     # main factory and interface exports
     "create_protocol_engine",
+    "create_protocol_engine_in_thread",
     "ProtocolEngine",
     "StateSummary",
     "Config",
diff --git a/api/src/opentrons/protocol_engine/create_protocol_engine.py b/api/src/opentrons/protocol_engine/create_protocol_engine.py
index cca1669355f..7a0a4ae8a92 100644
--- a/api/src/opentrons/protocol_engine/create_protocol_engine.py
+++ b/api/src/opentrons/protocol_engine/create_protocol_engine.py
@@ -1,13 +1,20 @@
 """Main ProtocolEngine factory."""
-from opentrons.hardware_control import HardwareControlAPI
+import asyncio
+import contextlib
+import typing
+
+from opentrons.hardware_control import HardwareControlAPI, ThreadManagedHardware
 from opentrons.hardware_control.types import DoorState
-from opentrons.protocol_engine.resources.module_data_provider import ModuleDataProvider
+from opentrons.util.async_helpers import async_context_manager_in_thread
 
 from .protocol_engine import ProtocolEngine
-from .resources import DeckDataProvider
+from .resources import DeckDataProvider, ModuleDataProvider
 from .state import Config, StateStore
+from .types import DeckType
 
 
+# TODO(mm, 2023-06-16): Arguably, this not being a context manager makes us prone to forgetting to
+# clean it up properly, especially in tests. See e.g. https://opentrons.atlassian.net/browse/RSS-222
 async def create_protocol_engine(
     hardware_api: HardwareControlAPI,
     config: Config,
@@ -32,3 +39,56 @@ async def create_protocol_engine(
     )
 
     return ProtocolEngine(state_store=state_store, hardware_api=hardware_api)
+
+
+@contextlib.contextmanager
+def create_protocol_engine_in_thread(
+    hardware: ThreadManagedHardware,
+) -> typing.Generator[
+    typing.Tuple[ProtocolEngine, asyncio.AbstractEventLoop], None, None
+]:
+    """Run a `ProtocolEngine` in a worker thread.
+
+    When this context manager is entered, it:
+
+    1. Starts a worker thread.
+    2. Starts an asyncio event loop in that worker thread.
+    3. Creates and `.play()`s a `ProtocolEngine` in that event loop.
+    4. Returns the `ProtocolEngine` and the event loop.
+       Use functions like `asyncio.run_coroutine_threadsafe()` to safely interact with
+       the `ProtocolEngine` from your thread.
+
+    When this context manager is exited, it:
+
+    1. Cleans up the `ProtocolEngine`.
+    2. Stops and cleans up the event loop.
+    3. Joins the thread.
+    """
+    with async_context_manager_in_thread(_protocol_engine(hardware)) as (
+        protocol_engine,
+        loop,
+    ):
+        yield protocol_engine, loop
+
+
+@contextlib.asynccontextmanager
+async def _protocol_engine(
+    hardware: ThreadManagedHardware,
+) -> typing.AsyncGenerator[ProtocolEngine, None]:
+    protocol_engine = await create_protocol_engine(
+        hardware_api=hardware.wrapped(),
+        config=Config(
+            robot_type="OT-3 Standard",
+            deck_type=DeckType.OT3_STANDARD,
+            ignore_pause=True,
+            use_virtual_pipettes=True,
+            use_virtual_modules=True,
+            use_virtual_gripper=True,
+            block_on_door_open=False,
+        ),
+    )
+    try:
+        protocol_engine.play()
+        yield protocol_engine
+    finally:
+        await protocol_engine.finish()
diff --git a/api/tests/opentrons/conftest.py b/api/tests/opentrons/conftest.py
index 6c528cc1e8b..c9f80a4a675 100755
--- a/api/tests/opentrons/conftest.py
+++ b/api/tests/opentrons/conftest.py
@@ -50,12 +50,11 @@
 )
 from opentrons.protocol_api import ProtocolContext, Labware, create_protocol_context
 from opentrons.protocol_api.core.legacy.legacy_labware_core import LegacyLabwareCore
+from opentrons.protocol_engine import create_protocol_engine_in_thread
 from opentrons.protocols.api_support import deck_type
 from opentrons.protocols.api_support.types import APIVersion
 from opentrons.types import Location, Point
 
-from .protocol_engine_in_thread import protocol_engine_in_thread
-
 
 if TYPE_CHECKING:
     from opentrons.drivers.smoothie_drivers import SmoothieDriver as SmoothieDriverType
@@ -281,7 +280,7 @@ def _make_ot3_pe_ctx(
     deck_type: str,
 ) -> Generator[ProtocolContext, None, None]:
     """Return a ProtocolContext configured for an OT-3 and backed by Protocol Engine."""
-    with protocol_engine_in_thread(hardware=hardware) as (engine, loop):
+    with create_protocol_engine_in_thread(hardware=hardware) as (engine, loop):
         yield create_protocol_context(
             api_version=APIVersion(2, 14),
             hardware_api=hardware,
diff --git a/api/tests/opentrons/protocol_engine_in_thread.py b/api/tests/opentrons/protocol_engine_in_thread.py
deleted file mode 100644
index 02eb6e2a22f..00000000000
--- a/api/tests/opentrons/protocol_engine_in_thread.py
+++ /dev/null
@@ -1,67 +0,0 @@
-"""Run a `ProtocolEngine` in a worker thread."""
-
-import asyncio
-import contextlib
-import typing
-
-from opentrons.hardware_control import ThreadManagedHardware
-from opentrons.protocol_engine import (
-    create_protocol_engine,
-    ProtocolEngine,
-    Config,
-    DeckType,
-)
-from opentrons.util.async_helpers import async_context_manager_in_thread
-
-
-@contextlib.contextmanager
-def protocol_engine_in_thread(
-    hardware: ThreadManagedHardware,
-) -> typing.Generator[
-    typing.Tuple[ProtocolEngine, asyncio.AbstractEventLoop], None, None
-]:
-    """Run a `ProtocolEngine` in a worker thread.
-
-    When this context manager is entered, it:
-
-    1. Starts a worker thread.
-    2. Starts an asyncio event loop in that worker thread.
-    3. Creates and `.play()`s a `ProtocolEngine` in that event loop.
-    4. Returns the `ProtocolEngine` and the event loop.
-       Use functions like `asyncio.run_coroutine_threadsafe()` to safely interact with
-       the `ProtocolEngine` from your thread.
-
-    When this context manager is exited, it:
-
-    1. Cleans up the `ProtocolEngine`.
-    2. Stops and cleans up the event loop.
-    3. Joins the thread.
-    """
-    with async_context_manager_in_thread(_protocol_engine(hardware)) as (
-        protocol_engine,
-        loop,
-    ):
-        yield protocol_engine, loop
-
-
-@contextlib.asynccontextmanager
-async def _protocol_engine(
-    hardware: ThreadManagedHardware,
-) -> typing.AsyncGenerator[ProtocolEngine, None]:
-    protocol_engine = await create_protocol_engine(
-        hardware_api=hardware.wrapped(),
-        config=Config(
-            robot_type="OT-3 Standard",
-            deck_type=DeckType.OT3_STANDARD,
-            ignore_pause=True,
-            use_virtual_pipettes=True,
-            use_virtual_modules=True,
-            use_virtual_gripper=True,
-            block_on_door_open=False,
-        ),
-    )
-    try:
-        protocol_engine.play()
-        yield protocol_engine
-    finally:
-        await protocol_engine.finish()

From 685b801c43bb2226b23973a41cda3de39875af14 Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Fri, 16 Jun 2023 18:57:14 -0400
Subject: [PATCH 05/29] Consistently take HardwareControlAPI interface.

---
 .../protocol_engine/create_protocol_engine.py          | 10 +++++-----
 api/tests/opentrons/conftest.py                        |  5 ++++-
 2 files changed, 9 insertions(+), 6 deletions(-)

diff --git a/api/src/opentrons/protocol_engine/create_protocol_engine.py b/api/src/opentrons/protocol_engine/create_protocol_engine.py
index 7a0a4ae8a92..59067daed51 100644
--- a/api/src/opentrons/protocol_engine/create_protocol_engine.py
+++ b/api/src/opentrons/protocol_engine/create_protocol_engine.py
@@ -3,7 +3,7 @@
 import contextlib
 import typing
 
-from opentrons.hardware_control import HardwareControlAPI, ThreadManagedHardware
+from opentrons.hardware_control import HardwareControlAPI
 from opentrons.hardware_control.types import DoorState
 from opentrons.util.async_helpers import async_context_manager_in_thread
 
@@ -43,7 +43,7 @@ async def create_protocol_engine(
 
 @contextlib.contextmanager
 def create_protocol_engine_in_thread(
-    hardware: ThreadManagedHardware,
+    hardware_api: HardwareControlAPI,
 ) -> typing.Generator[
     typing.Tuple[ProtocolEngine, asyncio.AbstractEventLoop], None, None
 ]:
@@ -64,7 +64,7 @@ def create_protocol_engine_in_thread(
     2. Stops and cleans up the event loop.
     3. Joins the thread.
     """
-    with async_context_manager_in_thread(_protocol_engine(hardware)) as (
+    with async_context_manager_in_thread(_protocol_engine(hardware_api)) as (
         protocol_engine,
         loop,
     ):
@@ -73,10 +73,10 @@ def create_protocol_engine_in_thread(
 
 @contextlib.asynccontextmanager
 async def _protocol_engine(
-    hardware: ThreadManagedHardware,
+    hardware_api: HardwareControlAPI,
 ) -> typing.AsyncGenerator[ProtocolEngine, None]:
     protocol_engine = await create_protocol_engine(
-        hardware_api=hardware.wrapped(),
+        hardware_api=hardware_api,
         config=Config(
             robot_type="OT-3 Standard",
             deck_type=DeckType.OT3_STANDARD,
diff --git a/api/tests/opentrons/conftest.py b/api/tests/opentrons/conftest.py
index c9f80a4a675..219ec17425c 100755
--- a/api/tests/opentrons/conftest.py
+++ b/api/tests/opentrons/conftest.py
@@ -280,7 +280,10 @@ def _make_ot3_pe_ctx(
     deck_type: str,
 ) -> Generator[ProtocolContext, None, None]:
     """Return a ProtocolContext configured for an OT-3 and backed by Protocol Engine."""
-    with create_protocol_engine_in_thread(hardware=hardware) as (engine, loop):
+    with create_protocol_engine_in_thread(hardware_api=hardware.wrapped()) as (
+        engine,
+        loop,
+    ):
         yield create_protocol_context(
             api_version=APIVersion(2, 14),
             hardware_api=hardware,

From 6b5abf3039b3c6c921032d9dae2d60750649e2bd Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Fri, 16 Jun 2023 19:01:17 -0400
Subject: [PATCH 06/29] Consistently take a full Config instead of trying to
 guess what the caller wants.

---
 .../protocol_engine/create_protocol_engine.py | 16 ++++------------
 api/tests/opentrons/conftest.py               | 19 +++++++++++++++++--
 2 files changed, 21 insertions(+), 14 deletions(-)

diff --git a/api/src/opentrons/protocol_engine/create_protocol_engine.py b/api/src/opentrons/protocol_engine/create_protocol_engine.py
index 59067daed51..42fa42739d1 100644
--- a/api/src/opentrons/protocol_engine/create_protocol_engine.py
+++ b/api/src/opentrons/protocol_engine/create_protocol_engine.py
@@ -10,7 +10,6 @@
 from .protocol_engine import ProtocolEngine
 from .resources import DeckDataProvider, ModuleDataProvider
 from .state import Config, StateStore
-from .types import DeckType
 
 
 # TODO(mm, 2023-06-16): Arguably, this not being a context manager makes us prone to forgetting to
@@ -44,6 +43,7 @@ async def create_protocol_engine(
 @contextlib.contextmanager
 def create_protocol_engine_in_thread(
     hardware_api: HardwareControlAPI,
+    config: Config,
 ) -> typing.Generator[
     typing.Tuple[ProtocolEngine, asyncio.AbstractEventLoop], None, None
 ]:
@@ -64,7 +64,7 @@ def create_protocol_engine_in_thread(
     2. Stops and cleans up the event loop.
     3. Joins the thread.
     """
-    with async_context_manager_in_thread(_protocol_engine(hardware_api)) as (
+    with async_context_manager_in_thread(_protocol_engine(hardware_api, config)) as (
         protocol_engine,
         loop,
     ):
@@ -73,19 +73,11 @@ def create_protocol_engine_in_thread(
 
 @contextlib.asynccontextmanager
 async def _protocol_engine(
-    hardware_api: HardwareControlAPI,
+    hardware_api: HardwareControlAPI, config: Config
 ) -> typing.AsyncGenerator[ProtocolEngine, None]:
     protocol_engine = await create_protocol_engine(
         hardware_api=hardware_api,
-        config=Config(
-            robot_type="OT-3 Standard",
-            deck_type=DeckType.OT3_STANDARD,
-            ignore_pause=True,
-            use_virtual_pipettes=True,
-            use_virtual_modules=True,
-            use_virtual_gripper=True,
-            block_on_door_open=False,
-        ),
+        config=config,
     )
     try:
         protocol_engine.play()
diff --git a/api/tests/opentrons/conftest.py b/api/tests/opentrons/conftest.py
index 219ec17425c..cc980021f98 100755
--- a/api/tests/opentrons/conftest.py
+++ b/api/tests/opentrons/conftest.py
@@ -50,7 +50,11 @@
 )
 from opentrons.protocol_api import ProtocolContext, Labware, create_protocol_context
 from opentrons.protocol_api.core.legacy.legacy_labware_core import LegacyLabwareCore
-from opentrons.protocol_engine import create_protocol_engine_in_thread
+from opentrons.protocol_engine import (
+    create_protocol_engine_in_thread,
+    Config as ProtocolEngineConfig,
+    DeckType,
+)
 from opentrons.protocols.api_support import deck_type
 from opentrons.protocols.api_support.types import APIVersion
 from opentrons.types import Location, Point
@@ -280,7 +284,18 @@ def _make_ot3_pe_ctx(
     deck_type: str,
 ) -> Generator[ProtocolContext, None, None]:
     """Return a ProtocolContext configured for an OT-3 and backed by Protocol Engine."""
-    with create_protocol_engine_in_thread(hardware_api=hardware.wrapped()) as (
+    with create_protocol_engine_in_thread(
+        hardware_api=hardware.wrapped(),
+        config=ProtocolEngineConfig(
+            robot_type="OT-3 Standard",
+            deck_type=DeckType.OT3_STANDARD,
+            ignore_pause=True,
+            use_virtual_pipettes=True,
+            use_virtual_modules=True,
+            use_virtual_gripper=True,
+            block_on_door_open=False,
+        ),
+    ) as (
         engine,
         loop,
     ):

From d2745aa5b90619bd08c6a5be1b3e11d4c88e3c28 Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Thu, 29 Jun 2023 18:56:23 -0400
Subject: [PATCH 07/29] Add a hacky copy_file_like() helper.

---
 api/src/opentrons/util/entrypoint_util.py     | 63 ++++++++++++++-
 .../opentrons/util/test_entrypoint_utils.py   | 80 +++++++++++++++++++
 2 files changed, 142 insertions(+), 1 deletion(-)

diff --git a/api/src/opentrons/util/entrypoint_util.py b/api/src/opentrons/util/entrypoint_util.py
index 5625828f5d4..3a1b0b5e6fe 100644
--- a/api/src/opentrons/util/entrypoint_util.py
+++ b/api/src/opentrons/util/entrypoint_util.py
@@ -5,7 +5,8 @@
 import logging
 from json import JSONDecodeError
 import pathlib
-from typing import Dict, Sequence, Union, TYPE_CHECKING
+import shutil
+from typing import BinaryIO, Dict, Sequence, Optional, TextIO, Union, TYPE_CHECKING
 
 from jsonschema import ValidationError  # type: ignore
 
@@ -83,3 +84,63 @@ def datafiles_from_paths(paths: Sequence[Union[str, pathlib.Path]]) -> Dict[str,
                 else:
                     log.info(f"ignoring {child} in data path")
     return datafiles
+
+
+# TODO(mm, 2023-06-29): Remove this hack when we fix https://opentrons.atlassian.net/browse/RSS-281.
+def copy_file_like(source: Union[BinaryIO, TextIO], destination: pathlib.Path) -> None:
+    """Copy a file-like object to a path, attempting to faithfully copy it byte-for-byte.
+
+    If the source is text (not binary), this attempts to retrieve what its on-filesystem encoding
+    originally was and save the new file with the same one.
+
+    This is a hack to support this use case:
+
+    1. A user has a Python source file with an unusual encoding.
+       (https://docs.python.org/3.7/reference/lexical_analysis.html#encoding-declarations)
+    2. They `open()` that file in text mode and send the text stream to
+       `opentrons.simulate.simulate()` or `opentrons.execute.execute()`.
+    3. Because of temporary implementation cruft (https://opentrons.atlassian.net/browse/RSS-281),
+       those functions sometimes need to save the stream to the filesystem, reopen it in
+       *binary mode,* and parse it as bytes. When they do that, it's important that the new file's
+       encoding matches the Python encoding declaration, or the Python parser will raise an error.
+    """
+    # When we read from the source stream, will it give us bytes, or text?
+    source_is_text: bool
+    # If the source stream is text, how was it originally encoded, if that's known?
+    # If that's unknown or if the source stream is binary, this will be None.
+    source_encoding: Optional[str]
+
+    try:
+        source_encoding = getattr(source, "encoding")
+        source_is_text = True
+    except AttributeError:
+        source_encoding = None
+        source_is_text = False
+
+    # How should we open the destination file?
+    destination_mode: str
+    # With what encoding? (None if, and only if, we open it in binary mode.)
+    destination_encoding: Optional[str]
+
+    if source_is_text:
+        destination_mode = "wt"
+        # The encoding of a text source can be None (unknown) if it's an io.StringIO, for example.
+        # If this happens, we need to make some arbitrary guess.
+        #
+        # UTF-8, not the system default, is the best choice, because:
+        #   * It's Python's most common source encoding, and the default one when the source has
+        #     no encoding declaration.
+        #   * It's one of the encodings that `json.loads()` looks for.
+        #
+        # This will break if someone gives us an io.StringIO of a Python source that contains
+        # an encoding declaration other than UTF-8.
+        destination_encoding = source_encoding or "utf-8"
+    else:
+        destination_mode = "wb"
+        destination_encoding = None
+
+    with open(
+        destination, mode=destination_mode, encoding=destination_encoding
+    ) as destination_file:
+        # Use copyfileobj to limit memory usage.
+        shutil.copyfileobj(fsrc=source, fdst=destination_file)
diff --git a/api/tests/opentrons/util/test_entrypoint_utils.py b/api/tests/opentrons/util/test_entrypoint_utils.py
index c30351dec3b..cbae89dc156 100644
--- a/api/tests/opentrons/util/test_entrypoint_utils.py
+++ b/api/tests/opentrons/util/test_entrypoint_utils.py
@@ -1,13 +1,17 @@
+import io
 import json
 import os
 from pathlib import Path
 from typing import Callable
 
+import pytest
+
 from opentrons_shared_data.labware.dev_types import LabwareDefinition as LabwareDefDict
 from opentrons.util.entrypoint_util import (
     FoundLabware,
     labware_from_paths,
     datafiles_from_paths,
+    copy_file_like,
 )
 
 
@@ -75,3 +79,79 @@ def test_datafiles_from_paths(tmp_path: Path) -> None:
         "test1": "wait theres a second file???".encode(),
         "test-file": "this isnt even in a directory".encode(),
     }
+
+
+class TestCopyFileLike:
+    """Tests for `copy_file_like()`."""
+
+    @pytest.fixture(params=["abc", "µ", "🥸", "\n", "\r\n"])
+    def source_text(self, request: pytest.FixtureRequest) -> str:
+        return request.param  # type: ignore[attr-defined,no-any-return]
+
+    @pytest.fixture(params=["utf-8", "utf-16"])
+    def source_encoding(self, request: pytest.FixtureRequest) -> str:
+        return request.param  # type: ignore[attr-defined,no-any-return]
+
+    @pytest.fixture
+    def source_path(self, tmp_path: Path) -> Path:
+        return tmp_path / "source"
+
+    @pytest.fixture
+    def destination_path(self, tmp_path: Path) -> Path:
+        return tmp_path / "destination"
+
+    def test_from_text_file(
+        self,
+        source_text: str,
+        source_encoding: str,
+        source_path: Path,
+        destination_path: Path,
+    ) -> None:
+        """Test that it correctly copies from a text-mode `open()`."""
+        source_path.write_bytes(source_text.encode(source_encoding))
+
+        with open(
+            source_path,
+            mode="rt",
+            encoding=source_encoding,
+            newline="",  # Leave newlines in source_text alone; do not translate them.
+        ) as source_file:  # TODO: Newlines.
+            copy_file_like(source=source_file, destination=destination_path)
+
+        assert destination_path.read_bytes() == source_path.read_bytes()
+
+    def test_from_binary_file(
+        self,
+        source_text: str,
+        source_encoding: str,
+        source_path: Path,
+        destination_path: Path,
+    ) -> None:
+        """Test that it correctly copies from a binary-mode `open()`."""
+        source_path.write_bytes(source_text.encode(source_encoding))
+
+        with open(source_path, mode="rb") as source_file:
+            copy_file_like(source=source_file, destination=destination_path)
+
+        assert destination_path.read_bytes() == source_path.read_bytes()
+
+    def test_from_stringio(self, source_text: str, destination_path: Path) -> None:
+        """Test that it correctly copies from an `io.StringIO`."""
+        stringio = io.StringIO(
+            source_text,
+            newline="",  # Leave newlines in source_text alone; do not translate them.
+        )
+
+        copy_file_like(source=stringio, destination=destination_path)
+
+        assert destination_path.read_bytes() == source_text.encode("utf-8")
+
+    def test_from_bytesio(
+        self, source_text: str, source_encoding: str, destination_path: Path
+    ) -> None:
+        """Test that it correctly copies from an `io.BytesIO`."""
+        bytesio = io.BytesIO(source_text.encode(source_encoding))
+
+        copy_file_like(source=bytesio, destination=destination_path)
+
+        assert destination_path.read_bytes() == source_text.encode(source_encoding)

From ef6fe3e22c2b76fd8ea98f5c298dae47fe623459 Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Tue, 27 Jun 2023 17:14:02 -0400
Subject: [PATCH 08/29] =?UTF-8?q?=C2=AF\=5F(=E3=83=84)=5F/=C2=AF?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 api/src/opentrons/execute.py | 372 +++++++++++++++++++++++++++++------
 1 file changed, 316 insertions(+), 56 deletions(-)

diff --git a/api/src/opentrons/execute.py b/api/src/opentrons/execute.py
index c370c144c10..edb20594ec5 100644
--- a/api/src/opentrons/execute.py
+++ b/api/src/opentrons/execute.py
@@ -5,60 +5,101 @@
 regular python shells. It also provides a console entrypoint for running a
 protocol from the command line.
 """
+import asyncio
 import atexit
 import argparse
+import contextlib
 import logging
 import os
+from pathlib import Path
 import sys
+import tempfile
 from typing import (
     TYPE_CHECKING,
     BinaryIO,
     Callable,
     Dict,
+    Generator,
     List,
     Optional,
     TextIO,
     Union,
 )
 
+from opentrons_shared_data.robot.dev_types import RobotType
+
 from opentrons import protocol_api, __version__, should_use_ot3
-from opentrons.config import IS_ROBOT, JUPYTER_NOTEBOOK_LABWARE_DIR
-from opentrons.protocols.execution import execute as execute_apiv2
 
 from opentrons.commands import types as command_types
+
+from opentrons.config import IS_ROBOT, JUPYTER_NOTEBOOK_LABWARE_DIR
+
+from opentrons.hardware_control import (
+    API as OT2API,
+    HardwareControlAPI,
+    ThreadManagedHardware,
+    ThreadManager,
+)
+
 from opentrons.protocols import parse
-from opentrons.protocols.types import ApiDeprecationError
 from opentrons.protocols.api_support.deck_type import (
     guess_from_global_config as guess_deck_type_from_global_config,
 )
 from opentrons.protocols.api_support.types import APIVersion
-from opentrons.hardware_control import (
-    API as OT2API,
-    ThreadManagedHardware,
-    ThreadManager,
+from opentrons.protocols.execution import execute as execute_apiv2
+from opentrons.protocols.types import (
+    ApiDeprecationError,
+    Protocol,
+    PythonProtocol,
 )
-from opentrons_shared_data.robot.dev_types import RobotType
 
-from .util.entrypoint_util import labware_from_paths, datafiles_from_paths
+from opentrons.protocol_api.core.engine import ENGINE_CORE_API_VERSION
+from opentrons.protocol_api.protocol_context import ProtocolContext
+
+from opentrons.protocol_engine import (
+    Config,
+    DeckType,
+    create_protocol_engine,
+    create_protocol_engine_in_thread,
+)
+
+from opentrons.protocol_reader import ProtocolReader, ProtocolSource
+
+from opentrons.protocol_runner import create_protocol_runner
+
+from .util.entrypoint_util import (
+    FoundLabware,
+    labware_from_paths,
+    datafiles_from_paths,
+    copy_file_like,
+)
 
 if TYPE_CHECKING:
     from opentrons_shared_data.labware.dev_types import LabwareDefinition
 
+
 _THREAD_MANAGED_HW: Optional[ThreadManagedHardware] = None
 #: The background global cache that all protocol contexts created by
 #: :py:meth:`get_protocol_api` will share
 
 
+# When a ProtocolContext is using a ProtocolEngine to control the robot, it requires some
+# additional long-lived resources besides _THREAD_MANAGED_HARDWARE. There's a background thread,
+# an asyncio event loop in that thread, and some ProtocolEngine-controlled background tasks in that
+# event loop.
+#
+# When we're executing a protocol file beginning-to-end, we can clean up those resources after it
+# completes. However, when someone gets a live ProtocolContext through get_protocol_api(), we have
+# no way of knowing when they're done with it. So, as a hack, we keep these resources open
+# indefinitely, letting them leak.
+#
+# We keep this at module scope so that the contained context managers aren't garbage-collected.
+# If they're garbage collected, they can close their resources prematurely.
+# https://stackoverflow.com/a/69155026/497934
+_LIVE_PROTOCOL_ENGINE_CONTEXTS = contextlib.ExitStack()
+
+
 # See Jira RCORE-535.
-_PYTHON_TOO_NEW_MESSAGE = (
-    "Python protocols with apiLevels higher than 2.13"
-    " cannot currently be executed with"
-    " the opentrons_execute command-line tool,"
-    " the opentrons.execute.execute() function,"
-    " or the opentrons.execute.get_protocol_api() function."
-    " Use a lower apiLevel"
-    " or use the Opentrons App instead."
-)
 _JSON_TOO_NEW_MESSAGE = (
     "Protocols created by recent versions of Protocol Designer"
     " cannot currently be executed with"
@@ -68,6 +109,9 @@
 )
 
 
+_EmitRunlogCallable = Callable[[command_types.CommandMessage], None]
+
+
 def get_protocol_api(
     version: Union[str, APIVersion],
     bundled_labware: Optional[Dict[str, "LabwareDefinition"]] = None,
@@ -124,16 +168,9 @@ def get_protocol_api(
     else:
         checked_version = version
 
-    if (
-        extra_labware is None
-        and IS_ROBOT
-        and JUPYTER_NOTEBOOK_LABWARE_DIR.is_dir()  # type: ignore[union-attr]
-    ):
+    if extra_labware is None:
         extra_labware = {
-            uri: details.definition
-            for uri, details in labware_from_paths(
-                [str(JUPYTER_NOTEBOOK_LABWARE_DIR)]
-            ).items()
+            uri: details.definition for uri, details in _get_jupyter_labware().items()
         }
 
     robot_type = _get_robot_type()
@@ -141,8 +178,8 @@ def get_protocol_api(
 
     hardware_controller = _get_global_hardware_controller(robot_type)
 
-    try:
-        context = protocol_api.create_protocol_context(
+    if checked_version < ENGINE_CORE_API_VERSION:
+        context = _create_live_context_non_pe(
             api_version=checked_version,
             deck_type=deck_type,
             hardware_api=hardware_controller,
@@ -150,8 +187,26 @@ def get_protocol_api(
             bundled_data=bundled_data,
             extra_labware=extra_labware,
         )
-    except protocol_api.ProtocolEngineCoreRequiredError as e:
-        raise NotImplementedError(_PYTHON_TOO_NEW_MESSAGE) from e  # See Jira RCORE-535.
+    else:
+        if bundled_data is not None:
+            raise NotImplementedError(
+                f"The bundled_data argument is not currently supported for Python protocols"
+                f" with apiLevel {ENGINE_CORE_API_VERSION} or newer."
+            )
+        if bundled_labware is not None:
+            raise NotImplementedError(
+                f"The bundled_labware argument is not currently supported for Python protocols"
+                f" with apiLevel {ENGINE_CORE_API_VERSION} or newer."
+            )
+        context = _create_live_context_pe(
+            api_version=checked_version,
+            robot_type=robot_type,
+            deck_type=guess_deck_type_from_global_config(),
+            hardware_api=_THREAD_MANAGED_HW,  # type: ignore[arg-type]
+            bundled_labware=bundled_labware,
+            bundled_data=bundled_data,
+            extra_labware=extra_labware,
+        )
 
     hardware_controller.sync.cache_instruments()
     return context
@@ -235,12 +290,12 @@ def get_arguments(parser: argparse.ArgumentParser) -> argparse.ArgumentParser:
     return parser
 
 
-def execute(
+def execute(  # noqa: C901
     protocol_file: Union[BinaryIO, TextIO],
     protocol_name: str,
     propagate_logs: bool = False,
     log_level: str = "warning",
-    emit_runlog: Optional[Callable[[command_types.CommandMessage], None]] = None,
+    emit_runlog: Optional[_EmitRunlogCallable] = None,
     custom_labware_paths: Optional[List[str]] = None,
     custom_data_paths: Optional[List[str]] = None,
 ) -> None:
@@ -291,6 +346,9 @@ def execute(
                               are presented by the protocol context in
                               ``ProtocolContext.bundled_data``.
 
+    :raises Exception: If the device that you're running this on doesn't match the ``robotType``
+        that the protocol declares in its ``requirements`` dict.
+
     The format of the runlog entries is as follows:
 
     .. code-block:: python
@@ -306,20 +364,18 @@ def execute(
                   # will produce a string with information filled in
              }
         }
-
-
     """
     stack_logger = logging.getLogger("opentrons")
     stack_logger.propagate = propagate_logs
     stack_logger.setLevel(getattr(logging, log_level.upper(), logging.WARNING))
+
     contents = protocol_file.read()
+
     if custom_labware_paths:
-        extra_labware = {
-            uri: details.definition
-            for uri, details in labware_from_paths(custom_labware_paths).items()
-        }
+        extra_labware = labware_from_paths(custom_labware_paths)
     else:
         extra_labware = {}
+
     if custom_data_paths:
         extra_data = datafiles_from_paths(custom_data_paths)
     else:
@@ -327,7 +383,12 @@ def execute(
 
     try:
         protocol = parse.parse(
-            contents, protocol_name, extra_labware=extra_labware, extra_data=extra_data
+            contents,
+            protocol_name,
+            extra_labware={
+                uri: details.definition for uri, details in extra_labware.items()
+            },
+            extra_data=extra_data,
         )
     except parse.JSONSchemaVersionTooNewError as e:
         if e.attempted_schema_version == 6:
@@ -338,24 +399,40 @@ def execute(
 
     if protocol.api_level < APIVersion(2, 0):
         raise ApiDeprecationError(version=protocol.api_level)
-    else:
-        bundled_data = getattr(protocol, "bundled_data", {})
-        bundled_data.update(extra_data)
-        gpa_extras = getattr(protocol, "extra_labware", None) or None
-        context = get_protocol_api(
-            protocol.api_level,
-            bundled_labware=getattr(protocol, "bundled_labware", None),
-            bundled_data=bundled_data,
-            extra_labware=gpa_extras,
+
+    # Guard against trying to run protocols for the wrong robot type.
+    # This matches what robot-server does.
+    if protocol.robot_type != _get_robot_type():
+        raise RuntimeError(
+            f'This robot is of type "{_get_robot_type()}",'
+            f' so it can\'t execute protocols for robot type "{protocol.robot_type}"'
+        )
+
+    if protocol.api_level < ENGINE_CORE_API_VERSION:
+        _run_file_non_pe(
+            protocol=protocol,
+            emit_runlog=emit_runlog,
         )
+    else:
+        # TODO(mm, 2023-07-06): Once these NotImplementedErrors are resolved, consider removing
+        # the enclosing if-else block and running everything through _run_file_pe() for simplicity.
         if emit_runlog:
-            broker = context.broker
-            broker.subscribe(command_types.COMMAND, emit_runlog)
-        context.home()
-        try:
-            execute_apiv2.run_protocol(protocol, context)
-        finally:
-            context.cleanup()
+            raise NotImplementedError(
+                f"The emit_runlog argument is not currently supported for Python protocols"
+                f" with apiLevel {ENGINE_CORE_API_VERSION} or newer."
+            )
+        if custom_data_paths:
+            raise NotImplementedError(
+                f"The custom_data_paths argument is not currently supported for Python protocols"
+                f" with apiLevel {ENGINE_CORE_API_VERSION} or newer."
+            )
+        protocol_file.seek(0)
+        _run_file_pe(
+            protocol_file=protocol_file,
+            protocol_name=protocol_name,
+            extra_labware=extra_labware,
+            hardware_api=_get_global_hardware_controller(_get_robot_type()).wrapped(),
+        )
 
 
 def make_runlog_cb() -> Callable[[command_types.CommandMessage], None]:
@@ -418,11 +495,194 @@ def main() -> int:
     return 0
 
 
+def _create_live_context_non_pe(
+    api_version: APIVersion,
+    hardware_api: ThreadManagedHardware,
+    deck_type: str,
+    extra_labware: Optional[Dict[str, "LabwareDefinition"]],
+    bundled_labware: Optional[Dict[str, "LabwareDefinition"]],
+    bundled_data: Optional[Dict[str, bytes]],
+) -> ProtocolContext:
+    """Return a live ProtocolContext.
+
+    This controls the robot through the older infrastructure, instead of through Protocol Engine.
+    """
+    assert api_version < ENGINE_CORE_API_VERSION
+    return protocol_api.create_protocol_context(
+        api_version=api_version,
+        deck_type=deck_type,
+        hardware_api=hardware_api,
+        bundled_labware=bundled_labware,
+        bundled_data=bundled_data,
+        extra_labware=extra_labware,
+    )
+
+
+def _create_live_context_pe(
+    api_version: APIVersion,
+    hardware_api: ThreadManagedHardware,
+    robot_type: RobotType,
+    deck_type: str,
+    extra_labware: Optional[Dict[str, "LabwareDefinition"]],
+    bundled_labware: Optional[Dict[str, "LabwareDefinition"]],
+    bundled_data: Optional[Dict[str, bytes]],
+) -> ProtocolContext:
+    """Return a live ProtocolContext that controls the robot through ProtocolEngine."""
+    assert api_version >= ENGINE_CORE_API_VERSION
+
+    global _LIVE_PROTOCOL_ENGINE_CONTEXTS
+    pe, loop = _LIVE_PROTOCOL_ENGINE_CONTEXTS.enter_context(
+        create_protocol_engine_in_thread(
+            hardware_api=hardware_api.wrapped(),
+            config=_get_protocol_engine_config(),
+        )
+    )
+
+    return protocol_api.create_protocol_context(
+        api_version=api_version,
+        hardware_api=hardware_api,
+        deck_type=deck_type,
+        protocol_engine=pe,
+        protocol_engine_loop=loop,
+        bundled_data=bundled_data,
+    )
+
+
+def _run_file_non_pe(
+    protocol: Protocol,
+    emit_runlog: Optional[_EmitRunlogCallable],
+) -> None:
+    """Run a protocol file without Protocol Engine, with the older infrastructure instead."""
+    if isinstance(protocol, PythonProtocol):
+        extra_labware = protocol.extra_labware
+        if extra_labware is None:
+            # TODO: Is a None check correct here, or should this be {}?
+            extra_labware = {
+                uri: details.definition
+                for uri, details in _get_jupyter_labware().items()
+            }
+        bundled_labware = protocol.bundled_labware
+        bundled_data = protocol.bundled_data
+    else:
+        # JSON protocols do have "bundled labware" embedded in them, but those aren't represented in
+        # the parsed Protocol object and we don't need to create the ProtocolContext with them.
+        # execute_apiv2.run_protocol() will pull them out of the JSON and load them into the
+        # ProtocolContext.
+        extra_labware = {}
+        bundled_labware = {}
+        bundled_data = {}
+
+    context = _create_live_context_non_pe(
+        api_version=protocol.api_level,
+        hardware_api=_get_global_hardware_controller(_get_robot_type()),
+        deck_type=guess_deck_type_from_global_config(),
+        extra_labware=extra_labware,
+        bundled_labware=bundled_labware,
+        bundled_data=bundled_data,
+    )
+
+    if emit_runlog:
+        context.broker.subscribe(command_types.COMMAND, emit_runlog)
+
+    context.home()
+    try:
+        execute_apiv2.run_protocol(protocol, context)
+    finally:
+        context.cleanup()
+
+
+def _run_file_pe(
+    protocol_file: Union[BinaryIO, TextIO],
+    protocol_name: str,
+    extra_labware: Dict[str, FoundLabware],
+    hardware_api: HardwareControlAPI,
+) -> None:
+    """Run a protocol file with Protocol Engine."""
+
+    async def run(protocol_source: ProtocolSource) -> None:
+        protocol_engine = await create_protocol_engine(
+            hardware_api=hardware_api,
+            config=_get_protocol_engine_config(),
+        )
+
+        protocol_runner = create_protocol_runner(
+            protocol_config=protocol_source.config,
+            protocol_engine=protocol_engine,
+            hardware_api=hardware_api,
+        )
+
+        # TODO(mm, 2023-06-30): This will home and drop tips at the end, which is not how
+        # things have historically behaved with PAPIv2.13 and older or JSONv5 and older.
+        await protocol_runner.run(protocol_source)
+
+    with _adapt_protocol_source(
+        protocol_file=protocol_file,
+        protocol_name=protocol_name,
+        extra_labware=extra_labware,
+    ) as protocol_source:
+        asyncio.run(run(protocol_source))
+
+
 def _get_robot_type() -> RobotType:
     """Return what kind of robot we're currently running on."""
     return "OT-3 Standard" if should_use_ot3() else "OT-2 Standard"
 
 
+def _get_protocol_engine_config() -> Config:
+    """Return a Protocol Engine config to execute protocols on this device."""
+    return Config(
+        robot_type=_get_robot_type(),
+        deck_type=DeckType(guess_deck_type_from_global_config()),
+        # We deliberately omit ignore_pause=True because, in the current implementation of
+        # opentrons.protocol_api.core.engine, that would incorrectly make
+        # ProtocolContext.is_simulating() return True.
+    )
+
+
+def _get_jupyter_labware() -> Dict[str, FoundLabware]:
+    """Return labware files in this robot's Jupyter Notebook directory."""
+    if IS_ROBOT:
+        # JUPYTER_NOTEBOOK_LABWARE_DIR should never be None when IS_ROBOT == True.
+        assert JUPYTER_NOTEBOOK_LABWARE_DIR is not None
+        if JUPYTER_NOTEBOOK_LABWARE_DIR.is_dir():
+            return labware_from_paths([JUPYTER_NOTEBOOK_LABWARE_DIR])
+
+    return {}
+
+
+@contextlib.contextmanager
+def _adapt_protocol_source(
+    protocol_file: Union[BinaryIO, TextIO],
+    protocol_name: str,
+    extra_labware: Dict[str, FoundLabware],
+) -> Generator[ProtocolSource, None, None]:
+    """Create a `ProtocolSource` representing input protocol files."""
+    with tempfile.TemporaryDirectory() as temporary_directory:
+        # It's not well-defined in our customer-facing interfaces whether the supplied protocol_name
+        # should be just the filename part, or a path with separators. In case it contains stuff
+        # like "../", sanitize it to just the filename part so we don't save files somewhere bad.
+        safe_protocol_name = Path(protocol_name).name
+
+        temp_protocol_file = Path(temporary_directory) / safe_protocol_name
+
+        # FIXME(mm, 2023-06-26): Copying this file is pure overhead, and it introduces encoding
+        # hazards. Remove this when we can parse JSONv6+ and PAPIv2.14+ protocols without going
+        # through the filesystem. https://opentrons.atlassian.net/browse/RSS-281
+        copy_file_like(source=protocol_file, destination=temp_protocol_file)
+
+        custom_labware_files = [labware.path for labware in extra_labware.values()]
+
+        protocol_source = asyncio.run(
+            ProtocolReader().read_saved(
+                files=[temp_protocol_file] + custom_labware_files,
+                directory=None,
+                files_are_prevalidated=False,
+            )
+        )
+
+        yield protocol_source
+
+
 def _get_global_hardware_controller(robot_type: RobotType) -> ThreadManagedHardware:
     # Build a hardware controller in a worker thread, which is necessary
     # because ipython runs its notebook in asyncio but the notebook

From b8eca37c6cf87171a994472db98b1cdbb98dc3d5 Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Fri, 7 Jul 2023 16:31:28 -0400
Subject: [PATCH 09/29] Fix JSON protocols not being able to use any labware.

---
 api/src/opentrons/execute.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/api/src/opentrons/execute.py b/api/src/opentrons/execute.py
index edb20594ec5..8ace013bfb4 100644
--- a/api/src/opentrons/execute.py
+++ b/api/src/opentrons/execute.py
@@ -568,9 +568,9 @@ def _run_file_non_pe(
         # the parsed Protocol object and we don't need to create the ProtocolContext with them.
         # execute_apiv2.run_protocol() will pull them out of the JSON and load them into the
         # ProtocolContext.
-        extra_labware = {}
-        bundled_labware = {}
-        bundled_data = {}
+        extra_labware = None
+        bundled_labware = None
+        bundled_data = None
 
     context = _create_live_context_non_pe(
         api_version=protocol.api_level,

From 2dc868c599bb4af118f32e642d376cb1fed5bf9c Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Fri, 7 Jul 2023 16:53:26 -0400
Subject: [PATCH 10/29] Fix Python protocols not including Jupyter labware by
 default.

---
 api/src/opentrons/execute.py | 8 +-------
 1 file changed, 1 insertion(+), 7 deletions(-)

diff --git a/api/src/opentrons/execute.py b/api/src/opentrons/execute.py
index 8ace013bfb4..b2c84af04e2 100644
--- a/api/src/opentrons/execute.py
+++ b/api/src/opentrons/execute.py
@@ -374,7 +374,7 @@ def execute(  # noqa: C901
     if custom_labware_paths:
         extra_labware = labware_from_paths(custom_labware_paths)
     else:
-        extra_labware = {}
+        extra_labware = _get_jupyter_labware()
 
     if custom_data_paths:
         extra_data = datafiles_from_paths(custom_data_paths)
@@ -555,12 +555,6 @@ def _run_file_non_pe(
     """Run a protocol file without Protocol Engine, with the older infrastructure instead."""
     if isinstance(protocol, PythonProtocol):
         extra_labware = protocol.extra_labware
-        if extra_labware is None:
-            # TODO: Is a None check correct here, or should this be {}?
-            extra_labware = {
-                uri: details.definition
-                for uri, details in _get_jupyter_labware().items()
-            }
         bundled_labware = protocol.bundled_labware
         bundled_data = protocol.bundled_data
     else:

From 81a37f9e9169543f7ab182a87854cc471b55c409 Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Fri, 7 Jul 2023 17:05:54 -0400
Subject: [PATCH 11/29] Minor style fixups and comment clarifications.

---
 api/src/opentrons/util/entrypoint_util.py         | 7 ++++---
 api/tests/opentrons/util/test_entrypoint_utils.py | 2 +-
 2 files changed, 5 insertions(+), 4 deletions(-)

diff --git a/api/src/opentrons/util/entrypoint_util.py b/api/src/opentrons/util/entrypoint_util.py
index 3a1b0b5e6fe..8f21ff19a8b 100644
--- a/api/src/opentrons/util/entrypoint_util.py
+++ b/api/src/opentrons/util/entrypoint_util.py
@@ -6,7 +6,7 @@
 from json import JSONDecodeError
 import pathlib
 import shutil
-from typing import BinaryIO, Dict, Sequence, Optional, TextIO, Union, TYPE_CHECKING
+from typing import BinaryIO, Dict, Optional, Sequence, TextIO, Union, TYPE_CHECKING
 
 from jsonschema import ValidationError  # type: ignore
 
@@ -96,8 +96,9 @@ def copy_file_like(source: Union[BinaryIO, TextIO], destination: pathlib.Path) -
     This is a hack to support this use case:
 
     1. A user has a Python source file with an unusual encoding.
+       They have a matching encoding declaration at the top of the file.
        (https://docs.python.org/3.7/reference/lexical_analysis.html#encoding-declarations)
-    2. They `open()` that file in text mode and send the text stream to
+    2. They `open()` that file in text mode, with the correct encoding, and send the text stream to
        `opentrons.simulate.simulate()` or `opentrons.execute.execute()`.
     3. Because of temporary implementation cruft (https://opentrons.atlassian.net/browse/RSS-281),
        those functions sometimes need to save the stream to the filesystem, reopen it in
@@ -142,5 +143,5 @@ def copy_file_like(source: Union[BinaryIO, TextIO], destination: pathlib.Path) -
     with open(
         destination, mode=destination_mode, encoding=destination_encoding
     ) as destination_file:
-        # Use copyfileobj to limit memory usage.
+        # Use copyfileobj() to limit memory usage.
         shutil.copyfileobj(fsrc=source, fdst=destination_file)
diff --git a/api/tests/opentrons/util/test_entrypoint_utils.py b/api/tests/opentrons/util/test_entrypoint_utils.py
index cbae89dc156..70c8b79cf7f 100644
--- a/api/tests/opentrons/util/test_entrypoint_utils.py
+++ b/api/tests/opentrons/util/test_entrypoint_utils.py
@@ -115,7 +115,7 @@ def test_from_text_file(
             mode="rt",
             encoding=source_encoding,
             newline="",  # Leave newlines in source_text alone; do not translate them.
-        ) as source_file:  # TODO: Newlines.
+        ) as source_file:
             copy_file_like(source=source_file, destination=destination_path)
 
         assert destination_path.read_bytes() == source_path.read_bytes()

From 61b6763a82401bfe362c52b5ccf492f663c12f96 Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Fri, 7 Jul 2023 17:27:51 -0400
Subject: [PATCH 12/29] Bundled data is supported, actually.

---
 api/src/opentrons/execute.py | 5 -----
 1 file changed, 5 deletions(-)

diff --git a/api/src/opentrons/execute.py b/api/src/opentrons/execute.py
index b2c84af04e2..955dc44ef12 100644
--- a/api/src/opentrons/execute.py
+++ b/api/src/opentrons/execute.py
@@ -188,11 +188,6 @@ def get_protocol_api(
             extra_labware=extra_labware,
         )
     else:
-        if bundled_data is not None:
-            raise NotImplementedError(
-                f"The bundled_data argument is not currently supported for Python protocols"
-                f" with apiLevel {ENGINE_CORE_API_VERSION} or newer."
-            )
         if bundled_labware is not None:
             raise NotImplementedError(
                 f"The bundled_labware argument is not currently supported for Python protocols"

From 56f2c7641e73405a0ec046c059f9005a4010638d Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Mon, 10 Jul 2023 15:52:32 -0400
Subject: [PATCH 13/29] Fix confusing error message.

---
 api/src/opentrons/execute.py | 4 +++-
 1 file changed, 3 insertions(+), 1 deletion(-)

diff --git a/api/src/opentrons/execute.py b/api/src/opentrons/execute.py
index 955dc44ef12..d528438b495 100644
--- a/api/src/opentrons/execute.py
+++ b/api/src/opentrons/execute.py
@@ -413,8 +413,10 @@ def execute(  # noqa: C901
         # the enclosing if-else block and running everything through _run_file_pe() for simplicity.
         if emit_runlog:
             raise NotImplementedError(
-                f"The emit_runlog argument is not currently supported for Python protocols"
+                f"Printing the run log is not currently supported for Python protocols"
                 f" with apiLevel {ENGINE_CORE_API_VERSION} or newer."
+                f" Pass --no-print-runlog to opentrons_execute"
+                f" or emit_runlog=None to opentrons.execute.execute()."
             )
         if custom_data_paths:
             raise NotImplementedError(

From 6e14e77e3ec8078d642e28263b4a9bdaf6839078 Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Tue, 11 Jul 2023 14:07:11 -0400
Subject: [PATCH 14/29] Don't attempt to pass bundled_labware in PAPIv2.14+.

---
 api/src/opentrons/execute.py | 2 --
 1 file changed, 2 deletions(-)

diff --git a/api/src/opentrons/execute.py b/api/src/opentrons/execute.py
index f3d0770cdc3..0ecc528037b 100644
--- a/api/src/opentrons/execute.py
+++ b/api/src/opentrons/execute.py
@@ -198,7 +198,6 @@ def get_protocol_api(
             robot_type=robot_type,
             deck_type=guess_deck_type_from_global_config(),
             hardware_api=_THREAD_MANAGED_HW,  # type: ignore[arg-type]
-            bundled_labware=bundled_labware,
             bundled_data=bundled_data,
             extra_labware=extra_labware,
         )
@@ -524,7 +523,6 @@ def _create_live_context_pe(
     robot_type: RobotType,
     deck_type: str,
     extra_labware: Optional[Dict[str, "LabwareDefinition"]],
-    bundled_labware: Optional[Dict[str, "LabwareDefinition"]],
     bundled_data: Optional[Dict[str, bytes]],
 ) -> ProtocolContext:
     """Return a live ProtocolContext that controls the robot through ProtocolEngine."""

From fa47468754993c3005fa638c74bc37e431074e04 Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Tue, 11 Jul 2023 14:52:24 -0400
Subject: [PATCH 15/29] Fix: Don't forget to add custom labware to PAPIv2.14+
 live contexts.

---
 api/src/opentrons/execute.py | 27 +++++++++++++++++++++------
 1 file changed, 21 insertions(+), 6 deletions(-)

diff --git a/api/src/opentrons/execute.py b/api/src/opentrons/execute.py
index 0ecc528037b..3c6c8a501ab 100644
--- a/api/src/opentrons/execute.py
+++ b/api/src/opentrons/execute.py
@@ -26,6 +26,7 @@
     Union,
 )
 
+from opentrons_shared_data.labware.labware_definition import LabwareDefinition
 from opentrons_shared_data.robot.dev_types import RobotType
 
 from opentrons import protocol_api, __version__, should_use_ot3
@@ -75,7 +76,9 @@
 )
 
 if TYPE_CHECKING:
-    from opentrons_shared_data.labware.dev_types import LabwareDefinition
+    from opentrons_shared_data.labware.dev_types import (
+        LabwareDefinition as LabwareDefinitionDict,
+    )
 
 
 _THREAD_MANAGED_HW: Optional[ThreadManagedHardware] = None
@@ -114,9 +117,9 @@
 
 def get_protocol_api(
     version: Union[str, APIVersion],
-    bundled_labware: Optional[Dict[str, "LabwareDefinition"]] = None,
+    bundled_labware: Optional[Dict[str, "LabwareDefinitionDict"]] = None,
     bundled_data: Optional[Dict[str, bytes]] = None,
-    extra_labware: Optional[Dict[str, "LabwareDefinition"]] = None,
+    extra_labware: Optional[Dict[str, "LabwareDefinitionDict"]] = None,
 ) -> protocol_api.ProtocolContext:
     """
     Build and return a ``protocol_api.ProtocolContext``
@@ -498,8 +501,8 @@ def _create_live_context_non_pe(
     api_version: APIVersion,
     hardware_api: ThreadManagedHardware,
     deck_type: str,
-    extra_labware: Optional[Dict[str, "LabwareDefinition"]],
-    bundled_labware: Optional[Dict[str, "LabwareDefinition"]],
+    extra_labware: Optional[Dict[str, "LabwareDefinitionDict"]],
+    bundled_labware: Optional[Dict[str, "LabwareDefinitionDict"]],
     bundled_data: Optional[Dict[str, bytes]],
 ) -> ProtocolContext:
     """Return a live ProtocolContext.
@@ -522,7 +525,7 @@ def _create_live_context_pe(
     hardware_api: ThreadManagedHardware,
     robot_type: RobotType,
     deck_type: str,
-    extra_labware: Optional[Dict[str, "LabwareDefinition"]],
+    extra_labware: Dict[str, "LabwareDefinitionDict"],
     bundled_data: Optional[Dict[str, bytes]],
 ) -> ProtocolContext:
     """Return a live ProtocolContext that controls the robot through ProtocolEngine."""
@@ -536,6 +539,18 @@ def _create_live_context_pe(
         )
     )
 
+    # `async def` so we can use loop.run_coroutine_threadsafe() to wait for its completion.
+    # Non-async would use call_soon_threadsafe(), which makes the waiting harder.
+    async def add_all_extra_labware() -> None:
+        for labware_definition_dict in extra_labware.values():
+            labware_definition = LabwareDefinition.parse_obj(labware_definition_dict)
+            pe.add_labware_definition(labware_definition)
+
+    # Add extra_labware to ProtocolEngine, being careful not to modify ProtocolEngine from this
+    # thread. See concurrency notes in ProtocolEngine docstring.
+    future = asyncio.run_coroutine_threadsafe(add_all_extra_labware(), loop)
+    future.result()
+
     return protocol_api.create_protocol_context(
         api_version=api_version,
         hardware_api=hardware_api,

From d6ee2fbc29363d9884987acaabcb61467882a1d1 Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Tue, 11 Jul 2023 15:09:01 -0400
Subject: [PATCH 16/29] Fix ^D hanging in a REPL after
 get_protocol_api("2.15").

---
 api/src/opentrons/execute.py                          |  9 +++++++++
 .../protocol_engine/create_protocol_engine.py         | 11 ++++++++---
 api/tests/opentrons/conftest.py                       |  1 +
 3 files changed, 18 insertions(+), 3 deletions(-)

diff --git a/api/src/opentrons/execute.py b/api/src/opentrons/execute.py
index 3c6c8a501ab..677603b6a70 100644
--- a/api/src/opentrons/execute.py
+++ b/api/src/opentrons/execute.py
@@ -536,6 +536,7 @@ def _create_live_context_pe(
         create_protocol_engine_in_thread(
             hardware_api=hardware_api.wrapped(),
             config=_get_protocol_engine_config(),
+            drop_tips_and_home_after=False,
         )
     )
 
@@ -717,5 +718,13 @@ def _clear_cached_hardware_controller() -> None:
         _THREAD_MANAGED_HW = None
 
 
+# This atexit registration must come after _clear_cached_hardware_controller()
+# to ensure we tear things down in order from highest level to lowest level.
+@atexit.register
+def _clear_live_protocol_engine_contexts() -> None:
+    global _LIVE_PROTOCOL_ENGINE_CONTEXTS
+    _LIVE_PROTOCOL_ENGINE_CONTEXTS.close()
+
+
 if __name__ == "__main__":
     sys.exit(main())
diff --git a/api/src/opentrons/protocol_engine/create_protocol_engine.py b/api/src/opentrons/protocol_engine/create_protocol_engine.py
index 42fa42739d1..f4e70afc4e7 100644
--- a/api/src/opentrons/protocol_engine/create_protocol_engine.py
+++ b/api/src/opentrons/protocol_engine/create_protocol_engine.py
@@ -44,6 +44,7 @@ async def create_protocol_engine(
 def create_protocol_engine_in_thread(
     hardware_api: HardwareControlAPI,
     config: Config,
+    drop_tips_and_home_after: bool,
 ) -> typing.Generator[
     typing.Tuple[ProtocolEngine, asyncio.AbstractEventLoop], None, None
 ]:
@@ -64,7 +65,9 @@ def create_protocol_engine_in_thread(
     2. Stops and cleans up the event loop.
     3. Joins the thread.
     """
-    with async_context_manager_in_thread(_protocol_engine(hardware_api, config)) as (
+    with async_context_manager_in_thread(
+        _protocol_engine(hardware_api, config, drop_tips_and_home_after)
+    ) as (
         protocol_engine,
         loop,
     ):
@@ -73,7 +76,9 @@ def create_protocol_engine_in_thread(
 
 @contextlib.asynccontextmanager
 async def _protocol_engine(
-    hardware_api: HardwareControlAPI, config: Config
+    hardware_api: HardwareControlAPI,
+    config: Config,
+    drop_tips_and_home_after: bool,
 ) -> typing.AsyncGenerator[ProtocolEngine, None]:
     protocol_engine = await create_protocol_engine(
         hardware_api=hardware_api,
@@ -83,4 +88,4 @@ async def _protocol_engine(
         protocol_engine.play()
         yield protocol_engine
     finally:
-        await protocol_engine.finish()
+        await protocol_engine.finish(drop_tips_and_home=drop_tips_and_home_after)
diff --git a/api/tests/opentrons/conftest.py b/api/tests/opentrons/conftest.py
index 37ace845114..c2b520ad727 100755
--- a/api/tests/opentrons/conftest.py
+++ b/api/tests/opentrons/conftest.py
@@ -296,6 +296,7 @@ def _make_ot3_pe_ctx(
             use_virtual_gripper=True,
             block_on_door_open=False,
         ),
+        drop_tips_and_home_after=False,
     ) as (
         engine,
         loop,

From e9c59d8aa63fdf27963e64ed0ea678085d41cffd Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Tue, 11 Jul 2023 16:54:40 -0400
Subject: [PATCH 17/29] Simplify the copy_file_like() hack by trying less.

---
 api/src/opentrons/util/entrypoint_util.py     | 60 +++++++------------
 .../opentrons/util/test_entrypoint_utils.py   | 37 +++++-------
 2 files changed, 36 insertions(+), 61 deletions(-)

diff --git a/api/src/opentrons/util/entrypoint_util.py b/api/src/opentrons/util/entrypoint_util.py
index 8f21ff19a8b..332333f947c 100644
--- a/api/src/opentrons/util/entrypoint_util.py
+++ b/api/src/opentrons/util/entrypoint_util.py
@@ -86,62 +86,46 @@ def datafiles_from_paths(paths: Sequence[Union[str, pathlib.Path]]) -> Dict[str,
     return datafiles
 
 
-# TODO(mm, 2023-06-29): Remove this hack when we fix https://opentrons.atlassian.net/browse/RSS-281.
+# HACK(mm, 2023-06-29): This function is attempting to do something fundamentally wrong.
+# Remove it when we fix https://opentrons.atlassian.net/browse/RSS-281.
 def copy_file_like(source: Union[BinaryIO, TextIO], destination: pathlib.Path) -> None:
-    """Copy a file-like object to a path, attempting to faithfully copy it byte-for-byte.
-
-    If the source is text (not binary), this attempts to retrieve what its on-filesystem encoding
-    originally was and save the new file with the same one.
-
-    This is a hack to support this use case:
-
-    1. A user has a Python source file with an unusual encoding.
-       They have a matching encoding declaration at the top of the file.
-       (https://docs.python.org/3.7/reference/lexical_analysis.html#encoding-declarations)
-    2. They `open()` that file in text mode, with the correct encoding, and send the text stream to
-       `opentrons.simulate.simulate()` or `opentrons.execute.execute()`.
-    3. Because of temporary implementation cruft (https://opentrons.atlassian.net/browse/RSS-281),
-       those functions sometimes need to save the stream to the filesystem, reopen it in
-       *binary mode,* and parse it as bytes. When they do that, it's important that the new file's
-       encoding matches the Python encoding declaration, or the Python parser will raise an error.
+    """Copy a file-like object to a path.
+
+    Limitations:
+        If `source` is a Python source code file with a non-UTF-8 encoding,
+        the new file's encoding will not correctly match the original's encoding declaration
+        (https://docs.python.org/3.7/reference/lexical_analysis.html#encoding-declarations).
+        This will make the Python parser raise an error when it tries to parse it.
+
+        If `source` is text-mode, its newlines may get translated, either when they're read
+        from `source` or when they're written to `destination`.
     """
     # When we read from the source stream, will it give us bytes, or text?
-    source_is_text: bool
-    # If the source stream is text, how was it originally encoded, if that's known?
-    # If that's unknown or if the source stream is binary, this will be None.
-    source_encoding: Optional[str]
-
     try:
-        source_encoding = getattr(source, "encoding")
-        source_is_text = True
+        # Experimentally, this is present (but possibly None) on text-mode streams,
+        # and not present on binary-mode streams.
+        getattr(source, "encoding")
     except AttributeError:
-        source_encoding = None
         source_is_text = False
+    else:
+        source_is_text = True
 
     # How should we open the destination file?
-    destination_mode: str
-    # With what encoding? (None if, and only if, we open it in binary mode.)
-    destination_encoding: Optional[str]
-
     if source_is_text:
         destination_mode = "wt"
-        # The encoding of a text source can be None (unknown) if it's an io.StringIO, for example.
-        # If this happens, we need to make some arbitrary guess.
-        #
-        # UTF-8, not the system default, is the best choice, because:
+        # If we have to choose an arbitrary encoding, UTF-8 is better than the system default:
         #   * It's Python's most common source encoding, and the default one when the source has
         #     no encoding declaration.
         #   * It's one of the encodings that `json.loads()` looks for.
-        #
-        # This will break if someone gives us an io.StringIO of a Python source that contains
-        # an encoding declaration other than UTF-8.
-        destination_encoding = source_encoding or "utf-8"
+        destination_encoding: Optional[str] = "utf-8"
     else:
         destination_mode = "wb"
         destination_encoding = None
 
     with open(
-        destination, mode=destination_mode, encoding=destination_encoding
+        destination,
+        mode=destination_mode,
+        encoding=destination_encoding,
     ) as destination_file:
         # Use copyfileobj() to limit memory usage.
         shutil.copyfileobj(fsrc=source, fdst=destination_file)
diff --git a/api/tests/opentrons/util/test_entrypoint_utils.py b/api/tests/opentrons/util/test_entrypoint_utils.py
index 70c8b79cf7f..1666cdeddc2 100644
--- a/api/tests/opentrons/util/test_entrypoint_utils.py
+++ b/api/tests/opentrons/util/test_entrypoint_utils.py
@@ -84,13 +84,13 @@ def test_datafiles_from_paths(tmp_path: Path) -> None:
 class TestCopyFileLike:
     """Tests for `copy_file_like()`."""
 
-    @pytest.fixture(params=["abc", "µ", "🥸", "\n", "\r\n"])
+    @pytest.fixture(params=["abc", "µ", "🥸"])
     def source_text(self, request: pytest.FixtureRequest) -> str:
         return request.param  # type: ignore[attr-defined,no-any-return]
 
-    @pytest.fixture(params=["utf-8", "utf-16"])
-    def source_encoding(self, request: pytest.FixtureRequest) -> str:
-        return request.param  # type: ignore[attr-defined,no-any-return]
+    @pytest.fixture
+    def source_bytes(self, source_text: str) -> bytes:
+        return source_text.encode("utf-8")
 
     @pytest.fixture
     def source_path(self, tmp_path: Path) -> Path:
@@ -103,55 +103,46 @@ def destination_path(self, tmp_path: Path) -> Path:
     def test_from_text_file(
         self,
         source_text: str,
-        source_encoding: str,
         source_path: Path,
         destination_path: Path,
     ) -> None:
         """Test that it correctly copies from a text-mode `open()`."""
-        source_path.write_bytes(source_text.encode(source_encoding))
+        source_path.write_text(source_text)
 
         with open(
             source_path,
             mode="rt",
-            encoding=source_encoding,
-            newline="",  # Leave newlines in source_text alone; do not translate them.
         ) as source_file:
             copy_file_like(source=source_file, destination=destination_path)
 
-        assert destination_path.read_bytes() == source_path.read_bytes()
+        assert destination_path.read_text() == source_text
 
     def test_from_binary_file(
         self,
-        source_text: str,
-        source_encoding: str,
+        source_bytes: bytes,
         source_path: Path,
         destination_path: Path,
     ) -> None:
         """Test that it correctly copies from a binary-mode `open()`."""
-        source_path.write_bytes(source_text.encode(source_encoding))
+        source_path.write_bytes(source_bytes)
 
         with open(source_path, mode="rb") as source_file:
             copy_file_like(source=source_file, destination=destination_path)
 
-        assert destination_path.read_bytes() == source_path.read_bytes()
+        assert destination_path.read_bytes() == source_bytes
 
     def test_from_stringio(self, source_text: str, destination_path: Path) -> None:
         """Test that it correctly copies from an `io.StringIO`."""
-        stringio = io.StringIO(
-            source_text,
-            newline="",  # Leave newlines in source_text alone; do not translate them.
-        )
+        stringio = io.StringIO(source_text)
 
         copy_file_like(source=stringio, destination=destination_path)
 
-        assert destination_path.read_bytes() == source_text.encode("utf-8")
+        assert destination_path.read_text() == source_text
 
-    def test_from_bytesio(
-        self, source_text: str, source_encoding: str, destination_path: Path
-    ) -> None:
+    def test_from_bytesio(self, source_bytes: bytes, destination_path: Path) -> None:
         """Test that it correctly copies from an `io.BytesIO`."""
-        bytesio = io.BytesIO(source_text.encode(source_encoding))
+        bytesio = io.BytesIO(source_bytes)
 
         copy_file_like(source=bytesio, destination=destination_path)
 
-        assert destination_path.read_bytes() == source_text.encode(source_encoding)
+        assert destination_path.read_bytes() == source_bytes

From 8754ec72386794770a545b96ce089ed7e348b728 Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Tue, 11 Jul 2023 17:21:55 -0400
Subject: [PATCH 18/29] Ugh.

---
 api/tests/opentrons/util/test_entrypoint_utils.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/api/tests/opentrons/util/test_entrypoint_utils.py b/api/tests/opentrons/util/test_entrypoint_utils.py
index 1666cdeddc2..3a6349dbfc6 100644
--- a/api/tests/opentrons/util/test_entrypoint_utils.py
+++ b/api/tests/opentrons/util/test_entrypoint_utils.py
@@ -107,7 +107,7 @@ def test_from_text_file(
         destination_path: Path,
     ) -> None:
         """Test that it correctly copies from a text-mode `open()`."""
-        source_path.write_text(source_text)
+        source_path.write_text(source_text, encoding="utf-8")
 
         with open(
             source_path,
@@ -115,7 +115,7 @@ def test_from_text_file(
         ) as source_file:
             copy_file_like(source=source_file, destination=destination_path)
 
-        assert destination_path.read_text() == source_text
+        assert destination_path.read_text(encoding="utf-8") == source_text
 
     def test_from_binary_file(
         self,
@@ -137,7 +137,7 @@ def test_from_stringio(self, source_text: str, destination_path: Path) -> None:
 
         copy_file_like(source=stringio, destination=destination_path)
 
-        assert destination_path.read_text() == source_text
+        assert destination_path.read_text(encoding="utf-8") == source_text
 
     def test_from_bytesio(self, source_bytes: bytes, destination_path: Path) -> None:
         """Test that it correctly copies from an `io.BytesIO`."""

From fcc62c13698c2b3012996f567c1f2215dc5e4fbb Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Wed, 12 Jul 2023 09:55:15 -0400
Subject: [PATCH 19/29] Test fixup: Write encoding should match read encoding.

---
 api/tests/opentrons/util/test_entrypoint_utils.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/api/tests/opentrons/util/test_entrypoint_utils.py b/api/tests/opentrons/util/test_entrypoint_utils.py
index 3a6349dbfc6..4d61e2e92e4 100644
--- a/api/tests/opentrons/util/test_entrypoint_utils.py
+++ b/api/tests/opentrons/util/test_entrypoint_utils.py
@@ -107,7 +107,7 @@ def test_from_text_file(
         destination_path: Path,
     ) -> None:
         """Test that it correctly copies from a text-mode `open()`."""
-        source_path.write_text(source_text, encoding="utf-8")
+        source_path.write_text(source_text)
 
         with open(
             source_path,

From 6729d49e5f57239644b760fc8050d2391ce5edce Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Wed, 12 Jul 2023 09:55:44 -0400
Subject: [PATCH 20/29] Test fixup: Use dummy binary data that's more obviously
 dummy & binary.

---
 api/tests/opentrons/util/test_entrypoint_utils.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/api/tests/opentrons/util/test_entrypoint_utils.py b/api/tests/opentrons/util/test_entrypoint_utils.py
index 4d61e2e92e4..24249da38cc 100644
--- a/api/tests/opentrons/util/test_entrypoint_utils.py
+++ b/api/tests/opentrons/util/test_entrypoint_utils.py
@@ -90,7 +90,7 @@ def source_text(self, request: pytest.FixtureRequest) -> str:
 
     @pytest.fixture
     def source_bytes(self, source_text: str) -> bytes:
-        return source_text.encode("utf-8")
+        return b"\x00\x01\x02\x03\x04"
 
     @pytest.fixture
     def source_path(self, tmp_path: Path) -> Path:

From 8d6139be11a499a77821cfe425f6a9cef1fbb6f9 Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Wed, 12 Jul 2023 09:57:28 -0400
Subject: [PATCH 21/29] Test fixup: Rename test file to match source.

---
 .../util/{test_entrypoint_utils.py => test_entrypoint_util.py}    | 0
 1 file changed, 0 insertions(+), 0 deletions(-)
 rename api/tests/opentrons/util/{test_entrypoint_utils.py => test_entrypoint_util.py} (100%)

diff --git a/api/tests/opentrons/util/test_entrypoint_utils.py b/api/tests/opentrons/util/test_entrypoint_util.py
similarity index 100%
rename from api/tests/opentrons/util/test_entrypoint_utils.py
rename to api/tests/opentrons/util/test_entrypoint_util.py

From 50a74b2593bd029065b4ad13a0efe2f5fa7ef4b7 Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Wed, 12 Jul 2023 14:41:05 -0400
Subject: [PATCH 22/29] Test fixup: Remove unused param.

---
 api/tests/opentrons/test_execute.py | 2 --
 1 file changed, 2 deletions(-)

diff --git a/api/tests/opentrons/test_execute.py b/api/tests/opentrons/test_execute.py
index d9c5369f847..0111bfa2298 100644
--- a/api/tests/opentrons/test_execute.py
+++ b/api/tests/opentrons/test_execute.py
@@ -43,11 +43,9 @@ async def dummy_delay(self: Any, duration_s: float) -> None:
 def test_execute_function_apiv2(
     protocol: Protocol,
     protocol_file: str,
-    monkeypatch: pytest.MonkeyPatch,
     virtual_smoothie_env: None,
     mock_get_attached_instr: mock.AsyncMock,
 ) -> None:
-
     mock_get_attached_instr.return_value[types.Mount.LEFT] = {
         "config": load(PipetteModel("p10_single_v1.5")),
         "id": "testid",

From 8fe755c322bb1c657858890ff9ea3438e2e6ff05 Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Fri, 14 Jul 2023 13:08:37 -0400
Subject: [PATCH 23/29] Enable Protocol Engine in test parametrization.

---
 api/tests/opentrons/test_execute.py | 43 +++++++++++++++++------------
 1 file changed, 25 insertions(+), 18 deletions(-)

diff --git a/api/tests/opentrons/test_execute.py b/api/tests/opentrons/test_execute.py
index 6db1d873fb2..abec62f8aeb 100644
--- a/api/tests/opentrons/test_execute.py
+++ b/api/tests/opentrons/test_execute.py
@@ -14,9 +14,10 @@
 from opentrons_shared_data.pipette.dev_types import PipetteModel
 
 from opentrons import execute, types
+from opentrons.config.pipette_config import load
 from opentrons.hardware_control import Controller, api
+from opentrons.protocol_api.core.engine import ENGINE_CORE_API_VERSION
 from opentrons.protocols.api_support.types import APIVersion
-from opentrons.config.pipette_config import load
 
 if TYPE_CHECKING:
     from tests.opentrons.conftest import Bundle, Protocol
@@ -25,13 +26,7 @@
 HERE = Path(__file__).parent
 
 
-@pytest.fixture(
-    params=[
-        APIVersion(2, 0),
-        # TODO(mm, 2023-07-14): Enable this for https://opentrons.atlassian.net/browse/RSS-268.
-        # ENGINE_CORE_API_VERSION,
-    ]
-)
+@pytest.fixture(params=[APIVersion(2, 0), ENGINE_CORE_API_VERSION])
 def api_version(request: pytest.FixtureRequest) -> APIVersion:
     """Return an API version to test with.
 
@@ -61,12 +56,15 @@ async def dummy_delay(self: Any, duration_s: float) -> None:
 
 
 @pytest.mark.parametrize(
-    "protocol_file",
+    ("protocol_file", "expect_run_log"),
     [
-        "testosaur_v2.py",
-        # TODO(mm, 2023-07-14): Resolve this xfail. https://opentrons.atlassian.net/browse/RSS-268
+        ("testosaur_v2.py", True),
+        ("testosaur_v2_14.py", False),
+        # FIXME(mm, 2023-07-20): Support printing the run log when executing new protocols.
+        # Then, remove this expect_run_log parametrization (it should always be True).
         pytest.param(
             "testosaur_v2_14.py",
+            True,
             marks=pytest.mark.xfail(strict=True, raises=NotImplementedError),
         ),
     ],
@@ -74,6 +72,7 @@ async def dummy_delay(self: Any, duration_s: float) -> None:
 def test_execute_function_apiv2(
     protocol: Protocol,
     protocol_file: str,
+    expect_run_log: bool,
     virtual_smoothie_env: None,
     mock_get_attached_instr: mock.AsyncMock,
 ) -> None:
@@ -92,13 +91,21 @@ def emit_runlog(entry: Any) -> None:
         nonlocal entries
         entries.append(entry)
 
-    execute.execute(protocol.filelike, protocol.filename, emit_runlog=emit_runlog)
-    assert [item["payload"]["text"] for item in entries if item["$"] == "before"] == [
-        "Picking up tip from A1 of Opentrons 96 Tip Rack 1000 µL on 1",
-        "Aspirating 100.0 uL from A1 of Corning 96 Well Plate 360 µL Flat on 2 at 500.0 uL/sec",
-        "Dispensing 100.0 uL into B1 of Corning 96 Well Plate 360 µL Flat on 2 at 1000.0 uL/sec",
-        "Dropping tip into H12 of Opentrons 96 Tip Rack 1000 µL on 1",
-    ]
+    execute.execute(
+        protocol.filelike,
+        protocol.filename,
+        emit_runlog=(emit_runlog if expect_run_log else None),
+    )
+
+    if expect_run_log:
+        assert [
+            item["payload"]["text"] for item in entries if item["$"] == "before"
+        ] == [
+            "Picking up tip from A1 of Opentrons 96 Tip Rack 1000 µL on 1",
+            "Aspirating 100.0 uL from A1 of Corning 96 Well Plate 360 µL Flat on 2 at 500.0 uL/sec",
+            "Dispensing 100.0 uL into B1 of Corning 96 Well Plate 360 µL Flat on 2 at 1000.0 uL/sec",
+            "Dropping tip into H12 of Opentrons 96 Tip Rack 1000 µL on 1",
+        ]
 
 
 def test_execute_function_json_v3(

From 1b35fbf4e71a0e434943efb6f098d3026b42a851 Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Fri, 14 Jul 2023 11:58:39 -0400
Subject: [PATCH 24/29] Undo ":raises Exception:" docs addition.

This was just noise. I think it's okay to not document this explicitly, for now. We don't document other potential reasons why an execute would fail, and it's weird to treat this specially.
---
 api/src/opentrons/execute.py | 3 ---
 1 file changed, 3 deletions(-)

diff --git a/api/src/opentrons/execute.py b/api/src/opentrons/execute.py
index 37f9265ff6c..7c5ef37b361 100644
--- a/api/src/opentrons/execute.py
+++ b/api/src/opentrons/execute.py
@@ -337,9 +337,6 @@ def execute(  # noqa: C901
                               are presented by the protocol context in
                               ``ProtocolContext.bundled_data``.
 
-    :raises Exception: If the device that you're running this on doesn't match the ``robotType``
-        that the protocol declares in its ``requirements`` dict.
-
     The format of the runlog entries is as follows:
 
     .. code-block:: python

From 372c456ef2f832a88c59aa5853c31ec6116cc05e Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Fri, 14 Jul 2023 13:08:27 -0400
Subject: [PATCH 25/29] Fix errors not being reported.

---
 api/src/opentrons/execute.py | 71 ++++++++++++++++++++++++++++++------
 1 file changed, 60 insertions(+), 11 deletions(-)

diff --git a/api/src/opentrons/execute.py b/api/src/opentrons/execute.py
index 7c5ef37b361..75add87a4d6 100644
--- a/api/src/opentrons/execute.py
+++ b/api/src/opentrons/execute.py
@@ -60,6 +60,8 @@
 from opentrons.protocol_engine import (
     Config,
     DeckType,
+    EngineStatus,
+    ErrorOccurrence as ProtocolEngineErrorOccurrence,
     create_protocol_engine,
     create_protocol_engine_in_thread,
 )
@@ -474,17 +476,61 @@ def main() -> int:
         stack_logger.addHandler(logging.StreamHandler(sys.stdout))
         log_level = args.log_level
     else:
+        # TODO(mm, 2023-07-13): This default logging prints error information redundantly
+        # when executing via Protocol Engine, because Protocol Engine logs when commands fail.
         log_level = "warning"
-    # Try to migrate containers from database to v2 format
-    execute(
-        protocol_file=args.protocol,
-        protocol_name=args.protocol.name,
-        custom_labware_paths=args.custom_labware_path,
-        custom_data_paths=(args.custom_data_path + args.custom_data_file),
-        log_level=log_level,
-        emit_runlog=printer,
-    )
-    return 0
+
+    try:
+        execute(
+            protocol_file=args.protocol,
+            protocol_name=args.protocol.name,
+            custom_labware_paths=args.custom_labware_path,
+            custom_data_paths=(args.custom_data_path + args.custom_data_file),
+            log_level=log_level,
+            emit_runlog=printer,
+        )
+        return 0
+    except _ProtocolEngineExecuteError as error:
+        # _ProtocolEngineExecuteError is a wrapper that's meaningless to the CLI user.
+        # Take the actual protocol problem out of it and just print that.
+        print(error.to_stderr_string(), file=sys.stderr)
+        return 1
+    # execute() might raise other exceptions, but we don't have a nice way to print those.
+    # Just let Python show a traceback.
+
+
+class _ProtocolEngineExecuteError(Exception):
+    def __init__(self, errors: List[ProtocolEngineErrorOccurrence]) -> None:
+        """Raised when there was any fatal error running a protocol through Protocol Engine.
+
+        Protocol Engine reports errors as data, not as exceptions.
+        But the only way for `execute()` to signal problems to its caller is to raise something.
+        So we need this class to wrap them.
+
+        Params:
+            errors: The errors that Protocol Engine reported.
+        """
+        # Show the full error details if this is part of a traceback. Don't try to summarize.
+        super().__init__(errors)
+        self._error_occurrences = errors
+
+    def to_stderr_string(self) -> str:
+        """Return a string suitable as the stderr output of the `opentrons_execute` CLI.
+
+        This summarizes from the full error details.
+        """
+        # It's unclear what exactly we should extract here.
+        #
+        # First, do we print the first element, or the last, or all of them?
+        #
+        # Second, do we print the .detail? .errorCode? .errorInfo? .wrappedErrors?
+        # By contract, .detail seems like it would be insufficient, but experimentally,
+        # it includes a lot, like:
+        #
+        #     ProtocolEngineError [line 3]: Error 4000 GENERAL_ERROR (ProtocolEngineError):
+        #     UnexpectedProtocolError: Labware "fixture_12_trough" not found with version 1
+        #     in namespace "fixture".
+        return self._error_occurrences[0].detail
 
 
 def _create_live_context_non_pe(
@@ -611,7 +657,10 @@ async def run(protocol_source: ProtocolSource) -> None:
 
         # TODO(mm, 2023-06-30): This will home and drop tips at the end, which is not how
         # things have historically behaved with PAPIv2.13 and older or JSONv5 and older.
-        await protocol_runner.run(protocol_source)
+        result = await protocol_runner.run(protocol_source)
+
+        if result.state_summary.status != EngineStatus.SUCCEEDED:
+            raise _ProtocolEngineExecuteError(result.state_summary.errors)
 
     with _adapt_protocol_source(
         protocol_file=protocol_file,

From 913d4deaf40a930a2e60653e24e4efdcfa8a4762 Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Thu, 20 Jul 2023 16:56:24 -0400
Subject: [PATCH 26/29] Give up even harder on encodings, I guess.

---
 api/src/opentrons/util/entrypoint_util.py     | 19 ++++---------------
 .../opentrons/util/test_entrypoint_util.py    |  4 ++--
 2 files changed, 6 insertions(+), 17 deletions(-)

diff --git a/api/src/opentrons/util/entrypoint_util.py b/api/src/opentrons/util/entrypoint_util.py
index 332333f947c..954d837c2f3 100644
--- a/api/src/opentrons/util/entrypoint_util.py
+++ b/api/src/opentrons/util/entrypoint_util.py
@@ -6,7 +6,7 @@
 from json import JSONDecodeError
 import pathlib
 import shutil
-from typing import BinaryIO, Dict, Optional, Sequence, TextIO, Union, TYPE_CHECKING
+from typing import BinaryIO, Dict, Sequence, TextIO, Union, TYPE_CHECKING
 
 from jsonschema import ValidationError  # type: ignore
 
@@ -92,13 +92,10 @@ def copy_file_like(source: Union[BinaryIO, TextIO], destination: pathlib.Path) -
     """Copy a file-like object to a path.
 
     Limitations:
-        If `source` is a Python source code file with a non-UTF-8 encoding,
-        the new file's encoding will not correctly match the original's encoding declaration
+        If `source` is text, the new file's encoding may not correctly match its original encoding.
+        This can matter if it's a Python file and it has an encoding declaration
         (https://docs.python.org/3.7/reference/lexical_analysis.html#encoding-declarations).
-        This will make the Python parser raise an error when it tries to parse it.
-
-        If `source` is text-mode, its newlines may get translated, either when they're read
-        from `source` or when they're written to `destination`.
+        Also, its newlines may get translated.
     """
     # When we read from the source stream, will it give us bytes, or text?
     try:
@@ -110,22 +107,14 @@ def copy_file_like(source: Union[BinaryIO, TextIO], destination: pathlib.Path) -
     else:
         source_is_text = True
 
-    # How should we open the destination file?
     if source_is_text:
         destination_mode = "wt"
-        # If we have to choose an arbitrary encoding, UTF-8 is better than the system default:
-        #   * It's Python's most common source encoding, and the default one when the source has
-        #     no encoding declaration.
-        #   * It's one of the encodings that `json.loads()` looks for.
-        destination_encoding: Optional[str] = "utf-8"
     else:
         destination_mode = "wb"
-        destination_encoding = None
 
     with open(
         destination,
         mode=destination_mode,
-        encoding=destination_encoding,
     ) as destination_file:
         # Use copyfileobj() to limit memory usage.
         shutil.copyfileobj(fsrc=source, fdst=destination_file)
diff --git a/api/tests/opentrons/util/test_entrypoint_util.py b/api/tests/opentrons/util/test_entrypoint_util.py
index 24249da38cc..2b1f803f541 100644
--- a/api/tests/opentrons/util/test_entrypoint_util.py
+++ b/api/tests/opentrons/util/test_entrypoint_util.py
@@ -115,7 +115,7 @@ def test_from_text_file(
         ) as source_file:
             copy_file_like(source=source_file, destination=destination_path)
 
-        assert destination_path.read_text(encoding="utf-8") == source_text
+        assert destination_path.read_text() == source_text
 
     def test_from_binary_file(
         self,
@@ -137,7 +137,7 @@ def test_from_stringio(self, source_text: str, destination_path: Path) -> None:
 
         copy_file_like(source=stringio, destination=destination_path)
 
-        assert destination_path.read_text(encoding="utf-8") == source_text
+        assert destination_path.read_text() == source_text
 
     def test_from_bytesio(self, source_bytes: bytes, destination_path: Path) -> None:
         """Test that it correctly copies from an `io.BytesIO`."""

From 9b25d8e0dd2e7333614fc968d24315a54022c9d2 Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Mon, 24 Jul 2023 10:44:31 -0400
Subject: [PATCH 27/29] Don't test with emoji that can't be encoded by Windows'
 default encoding.

---
 api/tests/opentrons/util/test_entrypoint_util.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/api/tests/opentrons/util/test_entrypoint_util.py b/api/tests/opentrons/util/test_entrypoint_util.py
index 2b1f803f541..4e82ce0e80f 100644
--- a/api/tests/opentrons/util/test_entrypoint_util.py
+++ b/api/tests/opentrons/util/test_entrypoint_util.py
@@ -84,7 +84,7 @@ def test_datafiles_from_paths(tmp_path: Path) -> None:
 class TestCopyFileLike:
     """Tests for `copy_file_like()`."""
 
-    @pytest.fixture(params=["abc", "µ", "🥸"])
+    @pytest.fixture(params=["abc", "µ"])
     def source_text(self, request: pytest.FixtureRequest) -> str:
         return request.param  # type: ignore[attr-defined,no-any-return]
 

From 4369f9b7815b58e303f2a2dca1314c17a78f8e0c Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Mon, 24 Jul 2023 20:36:26 -0400
Subject: [PATCH 28/29] Perhaps.

---
 api/src/opentrons/util/async_helpers.py | 23 +++++++++++++----------
 1 file changed, 13 insertions(+), 10 deletions(-)

diff --git a/api/src/opentrons/util/async_helpers.py b/api/src/opentrons/util/async_helpers.py
index 0c08e18afef..f421194fbea 100644
--- a/api/src/opentrons/util/async_helpers.py
+++ b/api/src/opentrons/util/async_helpers.py
@@ -2,8 +2,8 @@
 util.async_helpers - various utilities for asyncio functions and tasks.
 """
 
-from concurrent.futures import ThreadPoolExecutor
 from functools import wraps
+from threading import Thread
 from typing import (
     Any,
     AsyncContextManager,
@@ -149,12 +149,15 @@ def _in_thread() -> None:
         # which can take care of these nuances for us.
         loop.close()
 
-    with ThreadPoolExecutor(max_workers=1) as executor:
-        executor.submit(_in_thread)
-
-        loop_in_thread = loop_mailbox.get()
-
-        try:
-            yield loop_in_thread
-        finally:
-            loop_in_thread.call_soon_threadsafe(loop_in_thread.stop)
+    thread = Thread(
+        target=_in_thread,
+        daemon=True,
+        name=f"{__name__} event loop thread",
+    )
+    thread.start()
+    loop_in_thread = loop_mailbox.get()
+    try:
+        yield loop_in_thread
+    finally:
+        loop_in_thread.call_soon_threadsafe(loop_in_thread.stop)
+        thread.join()

From 6543c2c55529196552e4f5cb93a0a843c297591d Mon Sep 17 00:00:00 2001
From: Max Marrone <max@opentrons.com>
Date: Mon, 24 Jul 2023 22:28:53 -0400
Subject: [PATCH 29/29] Explain yoself.

---
 api/src/opentrons/util/async_helpers.py | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/api/src/opentrons/util/async_helpers.py b/api/src/opentrons/util/async_helpers.py
index f421194fbea..3e44c11153c 100644
--- a/api/src/opentrons/util/async_helpers.py
+++ b/api/src/opentrons/util/async_helpers.py
@@ -151,8 +151,11 @@ def _in_thread() -> None:
 
     thread = Thread(
         target=_in_thread,
-        daemon=True,
         name=f"{__name__} event loop thread",
+        # This is a load-bearing daemon=True. It avoids @atexit-related deadlocks when this is used
+        # by opentrons.execute and cleaned up by opentrons.execute's @atexit handler.
+        # https://github.com/Opentrons/opentrons/pull/12970#issuecomment-1648243785
+        daemon=True,
     )
     thread.start()
     loop_in_thread = loop_mailbox.get()