diff --git a/burr/common/async_utils.py b/burr/common/async_utils.py
index b1f60881f..56ce75ff2 100644
--- a/burr/common/async_utils.py
+++ b/burr/common/async_utils.py
@@ -16,7 +16,7 @@
 # under the License.
 
 import inspect
-from typing import AsyncGenerator, AsyncIterable, Generator, List, TypeVar, Union
+from typing import Any, AsyncGenerator, AsyncIterable, Coroutine, Generator, List, TypeVar, Union
 
 T = TypeVar("T")
 
@@ -27,6 +27,46 @@
 SyncOrAsyncGeneratorOrItemOrList = Union[SyncOrAsyncGenerator[GenType], List[GenType], GenType]
 
 
+class _AsyncPersisterContextManager:
+    """Wraps an async coroutine that returns a persister so it can be used
+    directly with ``async with``::
+
+        async with AsyncSQLitePersister.from_values(...) as persister:
+            ...
+
+    The wrapper awaits the coroutine on ``__aenter__`` and delegates
+    ``__aexit__`` to the persister's own ``__aexit__``.
+
+    .. note::
+        Each instance wraps a single coroutine and can only be consumed once,
+        either via ``await`` or ``async with``. A second use will raise
+        ``RuntimeError``.
+    """
+
+    def __init__(self, coro: Coroutine[Any, Any, Any]):
+        self._coro = coro
+        self._persister = None
+        self._consumed = False
+
+    def __await__(self):
+        if self._consumed:
+            raise RuntimeError("This factory result has already been consumed")
+        self._consumed = True
+        return self._coro.__await__()
+
+    async def __aenter__(self):
+        if self._consumed:
+            raise RuntimeError("This factory result has already been consumed")
+        self._consumed = True
+        self._persister = await self._coro
+        return await self._persister.__aenter__()
+
+    async def __aexit__(self, exc_type, exc_value, traceback):
+        if self._persister is None:
+            return False
+        return await self._persister.__aexit__(exc_type, exc_value, traceback)
+
+
 async def asyncify_generator(
     generator: SyncOrAsyncGenerator[GenType],
 ) -> AsyncGenerator[GenType, None]:
diff --git a/burr/core/__init__.py b/burr/core/__init__.py
index c4da5a48e..aa2f75a46 100644
--- a/burr/core/__init__.py
+++ b/burr/core/__init__.py
@@ -15,7 +15,7 @@
 # specific language governing permissions and limitations
 # under the License.
 
-from burr.core.action import Action, Condition, Result, action, default, expr, when
+from burr.core.action import Action, Condition, Result, action, default, expr, type_eraser, when
 from burr.core.application import (
     Application,
     ApplicationBuilder,
@@ -35,6 +35,7 @@
     "Condition",
     "default",
     "expr",
+    "type_eraser",
     "Result",
     "State",
     "when",
diff --git a/burr/core/action.py b/burr/core/action.py
index df15b7f0d..eb6fceff5 100644
--- a/burr/core/action.py
+++ b/burr/core/action.py
@@ -100,8 +100,75 @@ def visit_Subscript(self, node):
         )
 
 
+from functools import wraps
+
 from burr.core.typing import ActionSchema
 
+
+def type_eraser(func: Callable[..., Any]) -> Callable[..., Any]:
+    """Decorator for ``run``, ``stream_run``, and ``run_and_update`` overrides
+    that declare explicit parameters instead of ``**run_kwargs``.
+
+    Applying this decorator prevents mypy ``[override]`` errors caused by
+    narrowing the base-class signature (which uses ``**run_kwargs``).
+
+    Example usage::
+
+        from burr.core import Action, State, type_eraser
+
+        class Counter(Action):
+            @property
+            def reads(self) -> list[str]:
+                return ["counter"]
+
+            @type_eraser
+            def run(self, state: State, increment_by: int) -> dict:
+                return {"counter": state["counter"] + increment_by}
+
+            @property
+            def writes(self) -> list[str]:
+                return ["counter"]
+
+            def update(self, result: dict, state: State) -> State:
+                return state.update(**result)
+
+            @property
+            def inputs(self) -> list[str]:
+                return ["increment_by"]
+    """
+
+    if inspect.iscoroutinefunction(func):
+
+        @wraps(func)
+        async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
+            return await func(*args, **kwargs)
+
+        return async_wrapper
+
+    if inspect.isasyncgenfunction(func):
+
+        @wraps(func)
+        async def async_gen_wrapper(*args: Any, **kwargs: Any) -> Any:
+            async for item in func(*args, **kwargs):
+                yield item
+
+        return async_gen_wrapper
+
+    if inspect.isgeneratorfunction(func):
+
+        @wraps(func)
+        def gen_wrapper(*args: Any, **kwargs: Any) -> Any:
+            yield from func(*args, **kwargs)
+
+        return gen_wrapper
+
+    @wraps(func)
+    def wrapper(*args: Any, **kwargs: Any) -> Any:
+        return func(*args, **kwargs)
+
+    return wrapper
+
+
 # This is here to make accessing the pydantic actions easier
 # we just attach them to action so you can call `@action.pyddantic...`
 # The IDE will like it better and thus be able to auto-complete/type-check
diff --git a/burr/core/application.py b/burr/core/application.py
index 84e3f7cbd..7cbe96149 100644
--- a/burr/core/application.py
+++ b/burr/core/application.py
@@ -338,31 +338,44 @@ def _run_single_step_streaming_action(
     result = None
     state_update = None
     count = 0
-    for item in generator:
-        if not isinstance(item, tuple):
-            # TODO -- consider adding support for just returning a result.
-            raise ValueError(
-                f"Action {action.name} must yield a tuple of (result, state_update). "
-                f"For all non-final results (intermediate),"
-                f"the state update must be None"
-            )
-        result, state_update = item
-        count += 1
+    try:
+        for item in generator:
+            if not isinstance(item, tuple):
+                # TODO -- consider adding support for just returning a result.
+                raise ValueError(
+                    f"Action {action.name} must yield a tuple of (result, state_update). "
+                    f"For all non-final results (intermediate),"
+                    f"the state update must be None"
+                )
+            result, state_update = item
+            if state_update is None:
+                count += 1
+                if first_stream_start_time is None:
+                    first_stream_start_time = system.now()
+                lifecycle_adapters.call_all_lifecycle_hooks_sync(
+                    "post_stream_item",
+                    item=result,
+                    item_index=count,
+                    stream_initialize_time=stream_initialize_time,
+                    first_stream_item_start_time=first_stream_start_time,
+                    action=action.name,
+                    app_id=app_id,
+                    partition_key=partition_key,
+                    sequence_id=sequence_id,
+                )
+                yield result, None
+    except Exception as e:
         if state_update is None:
-            if first_stream_start_time is None:
-                first_stream_start_time = system.now()
-            lifecycle_adapters.call_all_lifecycle_hooks_sync(
-                "post_stream_item",
-                item=result,
-                item_index=count,
-                stream_initialize_time=stream_initialize_time,
-                first_stream_item_start_time=first_stream_start_time,
-                action=action.name,
-                app_id=app_id,
-                partition_key=partition_key,
-                sequence_id=sequence_id,
-            )
-            yield result, None
+            raise
+        logger.warning(
+            "Streaming action '%s' raised %s after yielding %d items. "
+            "Proceeding with final state from generator cleanup. Original error: %s",
+            action.name,
+            type(e).__name__,
+            count,
+            e,
+            exc_info=True,
+        )
 
     if state_update is None:
         raise ValueError(
@@ -391,31 +404,45 @@ async def _arun_single_step_streaming_action(
     result = None
     state_update = None
     count = 0
-    async for item in generator:
-        if not isinstance(item, tuple):
-            # TODO -- consider adding support for just returning a result.
-            raise ValueError(
-                f"Action {action.name} must yield a tuple of (result, state_update). "
-                f"For all non-final results (intermediate),"
-                f"the state update must be None"
-            )
-        result, state_update = item
+    try:
+        async for item in generator:
+            if not isinstance(item, tuple):
+                # TODO -- consider adding support for just returning a result.
+                raise ValueError(
+                    f"Action {action.name} must yield a tuple of (result, state_update). "
+                    f"For all non-final results (intermediate),"
+                    f"the state update must be None"
+                )
+            result, state_update = item
+            if state_update is None:
+                count += 1
+                if first_stream_start_time is None:
+                    first_stream_start_time = system.now()
+                await lifecycle_adapters.call_all_lifecycle_hooks_sync_and_async(
+                    "post_stream_item",
+                    item=result,
+                    item_index=count,
+                    stream_initialize_time=stream_initialize_time,
+                    first_stream_item_start_time=first_stream_start_time,
+                    action=action.name,
+                    app_id=app_id,
+                    partition_key=partition_key,
+                    sequence_id=sequence_id,
+                )
+                yield result, None
+    except Exception as e:
         if state_update is None:
-            if first_stream_start_time is None:
-                first_stream_start_time = system.now()
-            await lifecycle_adapters.call_all_lifecycle_hooks_sync_and_async(
-                "post_stream_item",
-                item=result,
-                item_index=count,
-                stream_initialize_time=stream_initialize_time,
-                first_stream_item_start_time=first_stream_start_time,
-                action=action.name,
-                app_id=app_id,
-                partition_key=partition_key,
-                sequence_id=sequence_id,
-            )
-            count += 1
-            yield result, None
+            raise
+        logger.warning(
+            "Streaming action '%s' raised %s after yielding %d items. "
+            "Proceeding with final state from generator cleanup. Original error: %s",
+            action.name,
+            type(e).__name__,
+            count,
+            e,
+            exc_info=True,
+        )
+
     if state_update is None:
         raise ValueError(
             f"Action {action.name} did not return a state update. For async actions, the last yield "
@@ -450,28 +477,42 @@ def _run_multi_step_streaming_action(
     result = None
     first_stream_start_time = None
     count = 0
-    for item in generator:
-        # We want to peek ahead so we can return the last one
-        # This is slightly eager, but only in the case in which we
-        # are using a multi-step streaming action
-        next_result = result
-        result = item
-        if next_result is not None:
-            if first_stream_start_time is None:
-                first_stream_start_time = system.now()
-            lifecycle_adapters.call_all_lifecycle_hooks_sync(
-                "post_stream_item",
-                item=next_result,
-                item_index=count,
-                stream_initialize_time=stream_initialize_time,
-                first_stream_item_start_time=first_stream_start_time,
-                action=action.name,
-                app_id=app_id,
-                partition_key=partition_key,
-                sequence_id=sequence_id,
-            )
-            count += 1
-            yield next_result, None
+    try:
+        for item in generator:
+            # We want to peek ahead so we can return the last one
+            # This is slightly eager, but only in the case in which we
+            # are using a multi-step streaming action
+            next_result = result
+            result = item
+            if next_result is not None:
+                if first_stream_start_time is None:
+                    first_stream_start_time = system.now()
+                lifecycle_adapters.call_all_lifecycle_hooks_sync(
+                    "post_stream_item",
+                    item=next_result,
+                    item_index=count,
+                    stream_initialize_time=stream_initialize_time,
+                    first_stream_item_start_time=first_stream_start_time,
+                    action=action.name,
+                    app_id=app_id,
+                    partition_key=partition_key,
+                    sequence_id=sequence_id,
+                )
+                count += 1
+                yield next_result, None
+    except Exception as e:
+        if result is None:
+            raise
+        logger.warning(
+            "Streaming action '%s' raised %s after yielding %d items. "
+            "Proceeding with last yielded result for reducer. "
+            "Note: the reducer will run on potentially partial data. Original error: %s",
+            action.name,
+            type(e).__name__,
+            count,
+            e,
+            exc_info=True,
+        )
     state_update = _run_reducer(action, state, result, action.name)
     _validate_result(result, action.name, action.schema)
     _validate_reducer_writes(action, state_update, action.name)
@@ -494,28 +535,42 @@ async def _arun_multi_step_streaming_action(
     result = None
     first_stream_start_time = None
     count = 0
-    async for item in generator:
-        # We want to peek ahead so we can return the last one
-        # This is slightly eager, but only in the case in which we
-        # are using a multi-step streaming action
-        next_result = result
-        result = item
-        if next_result is not None:
-            if first_stream_start_time is None:
-                first_stream_start_time = system.now()
-            await lifecycle_adapters.call_all_lifecycle_hooks_sync_and_async(
-                "post_stream_item",
-                item=next_result,
-                stream_initialize_time=stream_initialize_time,
-                item_index=count,
-                first_stream_item_start_time=first_stream_start_time,
-                action=action.name,
-                app_id=app_id,
-                partition_key=partition_key,
-                sequence_id=sequence_id,
-            )
-            count += 1
-            yield next_result, None
+    try:
+        async for item in generator:
+            # We want to peek ahead so we can return the last one
+            # This is slightly eager, but only in the case in which we
+            # are using a multi-step streaming action
+            next_result = result
+            result = item
+            if next_result is not None:
+                if first_stream_start_time is None:
+                    first_stream_start_time = system.now()
+                await lifecycle_adapters.call_all_lifecycle_hooks_sync_and_async(
+                    "post_stream_item",
+                    item=next_result,
+                    stream_initialize_time=stream_initialize_time,
+                    item_index=count,
+                    first_stream_item_start_time=first_stream_start_time,
+                    action=action.name,
+                    app_id=app_id,
+                    partition_key=partition_key,
+                    sequence_id=sequence_id,
+                )
+                count += 1
+                yield next_result, None
+    except Exception as e:
+        if result is None:
+            raise
+        logger.warning(
+            "Streaming action '%s' raised %s after yielding %d items. "
+            "Proceeding with last yielded result for reducer. "
+            "Note: the reducer will run on potentially partial data. Original error: %s",
+            action.name,
+            type(e).__name__,
+            count,
+            e,
+            exc_info=True,
+        )
     state_update = _run_reducer(action, state, result, action.name)
     _validate_result(result, action.name, action.schema)
     _validate_reducer_writes(action, state_update, action.name)
@@ -1862,7 +1917,9 @@ def stream_iterate(
         halt_before: Optional[Union[str, List[str]]] = None,
         inputs: Optional[Dict[str, Any]] = None,
     ) -> Generator[
-        Tuple[Action, StreamingResultContainer[ApplicationStateType, Union[dict, Any]]], None, None
+        Tuple[Action, StreamingResultContainer[ApplicationStateType, Union[dict, Any]]],
+        None,
+        None,
     ]:
         """Produces an iterator that iterates through intermediate streams. You may want
         to use this in something like deep research mode in which:
@@ -1905,7 +1962,11 @@ async def astream_iterate(
         halt_before: Optional[Union[str, List[str]]] = None,
         inputs: Optional[Dict[str, Any]] = None,
     ) -> AsyncGenerator[
-        Tuple[Action, AsyncStreamingResultContainer[ApplicationStateType, Union[dict, Any]]], None
+        Tuple[
+            Action,
+            AsyncStreamingResultContainer[ApplicationStateType, Union[dict, Any]],
+        ],
+        None,
     ]:
         """Async version of stream_iterate. Produces an async generator that iterates
         through intermediate streams. See stream_iterate for more details.
diff --git a/burr/integrations/persisters/b_aiosqlite.py b/burr/integrations/persisters/b_aiosqlite.py
index 9ce3c4a5d..a75eb682c 100644
--- a/burr/integrations/persisters/b_aiosqlite.py
+++ b/burr/integrations/persisters/b_aiosqlite.py
@@ -21,6 +21,7 @@
 
 import aiosqlite
 
+from burr.common.async_utils import _AsyncPersisterContextManager
 from burr.common.types import BaseCopyable
 from burr.core import State
 from burr.core.persistence import AsyncBaseStatePersister, PersistedStateData
@@ -60,27 +61,41 @@ def copy(self) -> "Self":
     PARTITION_KEY_DEFAULT = ""
 
     @classmethod
-    async def from_config(cls, config: dict) -> "AsyncSQLitePersister":
+    def from_config(cls, config: dict) -> "_AsyncPersisterContextManager":
         """Creates a new instance of the AsyncSQLitePersister from a configuration dictionary.
 
+        Can be used with ``await`` or as an async context manager::
+
+            persister = await AsyncSQLitePersister.from_config(config)
+            # or
+            async with AsyncSQLitePersister.from_config(config) as persister:
+                ...
+
         The config key:value pair needed are:
         db_path: str,
         table_name: str,
         serde_kwargs: dict,
         connect_kwargs: dict,
         """
-        return await cls.from_values(**config)
+        return cls.from_values(**config)
 
     @classmethod
-    async def from_values(
+    def from_values(
         cls,
         db_path: str,
         table_name: str = "burr_state",
         serde_kwargs: dict = None,
         connect_kwargs: dict = None,
-    ) -> "AsyncSQLitePersister":
+    ) -> "_AsyncPersisterContextManager":
         """Creates a new instance of the AsyncSQLitePersister from passed in values.
 
+        Can be used with ``await`` or as an async context manager::
+
+            persister = await AsyncSQLitePersister.from_values(db_path="test.db")
+            # or
+            async with AsyncSQLitePersister.from_values(db_path="test.db") as persister:
+                ...
+
         :param db_path: the path the DB will be stored.
         :param table_name: the table name to store things under.
         :param serde_kwargs: kwargs for state serialization/deserialization.
@@ -88,10 +103,14 @@ async def from_values(
         :return: async sqlite persister instance with an open connection. You are responsible
             for closing the connection yourself.
         """
-        connection = await aiosqlite.connect(
-            db_path, **connect_kwargs if connect_kwargs is not None else {}
-        )
-        return cls(connection, table_name, serde_kwargs)
+
+        async def _create():
+            connection = await aiosqlite.connect(
+                db_path, **connect_kwargs if connect_kwargs is not None else {}
+            )
+            return cls(connection, table_name, serde_kwargs)
+
+        return _AsyncPersisterContextManager(_create())
 
     def __init__(
         self,
diff --git a/burr/integrations/persisters/b_asyncpg.py b/burr/integrations/persisters/b_asyncpg.py
index 66f91f206..c694350d5 100644
--- a/burr/integrations/persisters/b_asyncpg.py
+++ b/burr/integrations/persisters/b_asyncpg.py
@@ -19,6 +19,7 @@
 import logging
 from typing import Any, ClassVar, Literal, Optional
 
+from burr.common.async_utils import _AsyncPersisterContextManager
 from burr.common.types import BaseCopyable
 from burr.core import persistence, state
 from burr.integrations import base
@@ -106,12 +107,20 @@ async def create_pool(
         return cls._pool
 
     @classmethod
-    async def from_config(cls, config: dict) -> "AsyncPostgreSQLPersister":
-        """Creates a new instance of the PostgreSQLPersister from a configuration dictionary."""
-        return await cls.from_values(**config)
+    def from_config(cls, config: dict) -> "_AsyncPersisterContextManager":
+        """Creates a new instance of the PostgreSQLPersister from a configuration dictionary.
+
+        Can be used with ``await`` or as an async context manager::
+
+            persister = await AsyncPostgreSQLPersister.from_config(config)
+            # or
+            async with AsyncPostgreSQLPersister.from_config(config) as persister:
+                ...
+        """
+        return cls.from_values(**config)
 
     @classmethod
-    async def from_values(
+    def from_values(
         cls,
         db_name: str,
         user: str,
@@ -121,9 +130,16 @@ async def from_values(
         table_name: str = "burr_state",
         use_pool: bool = False,
         **pool_kwargs,
-    ) -> "AsyncPostgreSQLPersister":
+    ) -> "_AsyncPersisterContextManager":
         """Builds a new instance of the PostgreSQLPersister from the provided values.
 
+        Can be used with ``await`` or as an async context manager::
+
+            persister = await AsyncPostgreSQLPersister.from_values(...)
+            # or
+            async with AsyncPostgreSQLPersister.from_values(...) as persister:
+                ...
+
         :param db_name: the name of the PostgreSQL database.
         :param user: the username to connect to the PostgreSQL database.
         :param password: the password to connect to the PostgreSQL database.
@@ -133,22 +149,25 @@ async def from_values(
         :param use_pool: whether to use a connection pool (True) or a direct connection (False)
         :param pool_kwargs: additional kwargs to pass to the pool creation
         """
-        if use_pool:
-            pool = await cls.create_pool(
-                user=user,
-                password=password,
-                database=db_name,
-                host=host,
-                port=port,
-                **pool_kwargs,
-            )
-            return cls(connection=None, pool=pool, table_name=table_name)
-        else:
-            # Original behavior - direct connection
-            connection = await asyncpg.connect(
-                user=user, password=password, database=db_name, host=host, port=port
-            )
-            return cls(connection=connection, table_name=table_name)
+
+        async def _create():
+            if use_pool:
+                pool = await cls.create_pool(
+                    user=user,
+                    password=password,
+                    database=db_name,
+                    host=host,
+                    port=port,
+                    **pool_kwargs,
+                )
+                return cls(connection=None, pool=pool, table_name=table_name)
+            else:
+                connection = await asyncpg.connect(
+                    user=user, password=password, database=db_name, host=host, port=port
+                )
+                return cls(connection=connection, table_name=table_name)
+
+        return _AsyncPersisterContextManager(_create())
 
     def __init__(
         self,
diff --git a/docs/concepts/parallelism.rst b/docs/concepts/parallelism.rst
index 0ced6bcc7..c875c5f83 100644
--- a/docs/concepts/parallelism.rst
+++ b/docs/concepts/parallelism.rst
@@ -698,7 +698,7 @@ When using state persistence with async parallelism, make sure to use the async
     from burr.integrations.persisters.b_asyncpg import AsyncPGPersister
 
     # Create an async persister with a connection pool
-    persister = AsyncPGPersister.from_values(
+    persister = await AsyncPGPersister.from_values(
         host="localhost",
         port=5432,
         user="postgres",
@@ -707,7 +707,7 @@ When using state persistence with async parallelism, make sure to use the async
         use_pool=True  # Important for parallelism!
     )
 
-    app = (
+    app = await (
         ApplicationBuilder()
         .with_state_persister(persister)
         .with_action(
@@ -722,12 +722,12 @@ Remember to properly clean up your async persisters when you're done with them:
 
 .. code-block:: python
 
-    # Using as a context manager
+    # Using as a context manager (recommended)
     async with AsyncPGPersister.from_values(..., use_pool=True) as persister:
         # Use persister here
 
     # Or manual cleanup
-    persister = AsyncPGPersister.from_values(..., use_pool=True)
+    persister = await AsyncPGPersister.from_values(..., use_pool=True)
     try:
         # Use persister here
     finally:
diff --git a/tests/core/test_action.py b/tests/core/test_action.py
index f65a06831..c58a2989e 100644
--- a/tests/core/test_action.py
+++ b/tests/core/test_action.py
@@ -38,6 +38,7 @@
     default,
     derive_inputs_from_fn,
     streaming_action,
+    type_eraser,
 )
 
 
@@ -1043,3 +1044,143 @@ def good_action(state: MyState):
     from burr.core.action import create_action
 
     create_action(good_action, name="test")
+
+
+class TestTypeEraser:
+    def test_sync_run(self):
+        class MyAction(Action):
+            @property
+            def reads(self) -> list[str]:
+                return ["counter"]
+
+            @type_eraser
+            def run(self, state: State, increment_by: int) -> dict:
+                return {"counter": state["counter"] + increment_by}
+
+            @property
+            def writes(self) -> list[str]:
+                return ["counter"]
+
+            def update(self, result: dict, state: State) -> State:
+                return state.update(**result)
+
+            @property
+            def inputs(self) -> list[str]:
+                return ["increment_by"]
+
+        a = MyAction()
+        result = a.run(State({"counter": 0}), increment_by=5)
+        assert result == {"counter": 5}
+        assert a.is_async() is False
+
+    def test_async_run(self):
+        class MyAsyncAction(Action):
+            @property
+            def reads(self) -> list[str]:
+                return ["counter"]
+
+            @type_eraser
+            async def run(self, state: State, increment_by: int) -> dict:
+                return {"counter": state["counter"] + increment_by}
+
+            @property
+            def writes(self) -> list[str]:
+                return ["counter"]
+
+            def update(self, result: dict, state: State) -> State:
+                return state.update(**result)
+
+            @property
+            def inputs(self) -> list[str]:
+                return ["increment_by"]
+
+        a = MyAsyncAction()
+        assert a.is_async() is True
+        result = asyncio.run(a.run(State({"counter": 0}), increment_by=5))
+        assert result == {"counter": 5}
+
+    def test_sync_stream_run(self):
+        class MyStreamingAction(StreamingAction):
+            @property
+            def reads(self) -> list[str]:
+                return ["items"]
+
+            @type_eraser
+            def stream_run(self, state: State, prefix: str) -> Generator[dict, None, None]:
+                for item in state["items"]:
+                    yield {"val": f"{prefix}_{item}"}
+
+            @property
+            def writes(self) -> list[str]:
+                return ["result"]
+
+            def update(self, result: dict, state: State) -> State:
+                return state.update(**result)
+
+            @property
+            def inputs(self) -> list[str]:
+                return ["prefix"]
+
+        a = MyStreamingAction()
+        results = list(a.stream_run(State({"items": ["a", "b"]}), prefix="x"))
+        assert results == [{"val": "x_a"}, {"val": "x_b"}]
+
+    def test_async_stream_run(self):
+        class MyAsyncStreamingAction(AsyncStreamingAction):
+            @property
+            def reads(self) -> list[str]:
+                return ["items"]
+
+            @type_eraser
+            async def stream_run(self, state: State, prefix: str) -> AsyncGenerator[dict, None]:
+                for item in state["items"]:
+                    yield {"val": f"{prefix}_{item}"}
+
+            @property
+            def writes(self) -> list[str]:
+                return ["result"]
+
+            def update(self, result: dict, state: State) -> State:
+                return state.update(**result)
+
+            @property
+            def inputs(self) -> list[str]:
+                return ["prefix"]
+
+        a = MyAsyncStreamingAction()
+        assert a.is_async() is True
+
+        async def collect():
+            return [item async for item in a.stream_run(State({"items": ["a", "b"]}), prefix="x")]
+
+        results = asyncio.run(collect())
+        assert results == [{"val": "x_a"}, {"val": "x_b"}]
+
+    def test_preserves_wrapped_name(self):
+        class MyAction(Action):
+            @property
+            def reads(self) -> list[str]:
+                return []
+
+            @type_eraser
+            def run(self, state: State, custom_param: str) -> dict:
+                return {}
+
+            @property
+            def writes(self) -> list[str]:
+                return []
+
+            def update(self, result: dict, state: State) -> State:
+                return state
+
+            @property
+            def inputs(self) -> list[str]:
+                return []
+
+        assert MyAction().run.__name__ == "run"
+        assert MyAction().run.__wrapped__.__name__ == "run"
+
+    def test_exported_from_burr_core(self):
+        from burr.core import type_eraser as te
+
+        assert te is type_eraser
diff --git a/tests/core/test_application.py b/tests/core/test_application.py
index c90c40676..7383583f0 100644
--- a/tests/core/test_application.py
+++ b/tests/core/test_application.py
@@ -630,7 +630,12 @@ def writes(self) -> list[str]:
     state = State()
     with pytest.raises(ValueError, match="missing_value"):
         gen = _run_single_step_streaming_action(
-            action, state, inputs={}, sequence_id=0, partition_key="partition_key", app_id="app_id"
+            action,
+            state,
+            inputs={},
+            sequence_id=0,
+            partition_key="partition_key",
+            app_id="app_id",
         )
         collections.deque(gen, maxlen=0)  # exhaust the generator
 
@@ -687,7 +692,12 @@ def writes(self) -> list[str]:
     state = State()
     with pytest.raises(ValueError, match="missing_value"):
         gen = _run_multi_step_streaming_action(
-            action, state, inputs={}, sequence_id=0, partition_key="partition_key", app_id="app_id"
+            action,
+            state,
+            inputs={},
+            sequence_id=0,
+            partition_key="partition_key",
+            app_id="app_id",
         )
         collections.deque(gen, maxlen=0)  # exhaust the generator
 
@@ -1005,7 +1015,12 @@ def test__run_multistep_streaming_action():
     action = base_streaming_counter.with_name("counter")
     state = State({"count": 0, "tracker": []})
     generator = _run_multi_step_streaming_action(
-        action, state, inputs={}, sequence_id=0, partition_key="partition_key", app_id="app_id"
+        action,
+        state,
+        inputs={},
+        sequence_id=0,
+        partition_key="partition_key",
+        app_id="app_id",
     )
     last_result = -1
     result = None
@@ -1111,7 +1126,12 @@ def test__run_streaming_action_incorrect_result_type():
     state = State()
     with pytest.raises(ValueError, match="returned a non-dict"):
         gen = _run_multi_step_streaming_action(
-            action, state, inputs={}, sequence_id=0, partition_key="partition_key", app_id="app_id"
+            action,
+            state,
+            inputs={},
+            sequence_id=0,
+            partition_key="partition_key",
+            app_id="app_id",
         )
         collections.deque(gen, maxlen=0)  # exhaust the generator
 
@@ -1168,7 +1188,12 @@ def test__run_single_step_streaming_action():
     action = base_streaming_single_step_counter.with_name("counter")
     state = State({"count": 0, "tracker": []})
     generator = _run_single_step_streaming_action(
-        action, state, inputs={}, sequence_id=0, partition_key="partition_key", app_id="app_id"
+        action,
+        state,
+        inputs={},
+        sequence_id=0,
+        partition_key="partition_key",
+        app_id="app_id",
     )
     last_result = -1
     result, state = None, None
@@ -1275,6 +1300,328 @@ async def post_stream_item(self, item: Any, **future_kwargs: Any):
     assert len(hook.items) == 10  # one for each streaming callback
 
 
+class SingleStepStreamingCounterWithException(SingleStepStreamingAction):
+    """Yields intermediate items, raises, then yields final state in finally block."""
+
+    def stream_run_and_update(
+        self, state: State, **run_kwargs
+    ) -> Generator[Tuple[dict, Optional[State]], None, None]:
+        count = state["count"]
+        try:
+            for i in range(3):
+                yield {"count": count + ((i + 1) / 10)}, None
+            raise RuntimeError("simulated failure")
+        finally:
+            yield {"count": count + 1}, state.update(count=count + 1).append(tracker=count + 1)
+
+    @property
+    def reads(self) -> list[str]:
+        return ["count"]
+
+    @property
+    def writes(self) -> list[str]:
+        return ["count", "tracker"]
+
+
+class SingleStepStreamingCounterWithExceptionNoState(SingleStepStreamingAction):
+    """Raises without ever yielding a final state update."""
+
+    def stream_run_and_update(
+        self, state: State, **run_kwargs
+    ) -> Generator[Tuple[dict, Optional[State]], None, None]:
+        count = state["count"]
+        for i in range(3):
+            yield {"count": count + ((i + 1) / 10)}, None
+        raise RuntimeError("simulated failure with no state")
+
+    @property
+    def reads(self) -> list[str]:
+        return ["count"]
+
+    @property
+    def writes(self) -> list[str]:
+        return ["count", "tracker"]
+
+
+class SingleStepStreamingCounterWithExceptionAsync(SingleStepStreamingAction):
+    """Async variant: yields intermediate items, raises, then yields final state in finally."""
+
+    async def stream_run_and_update(
+        self, state: State, **run_kwargs
+    ) -> AsyncGenerator[Tuple[dict, Optional[State]], None]:
+        count = state["count"]
+        try:
+            for i in range(3):
+                yield {"count": count + ((i + 1) / 10)}, None
+            raise RuntimeError("simulated failure")
+        finally:
+            yield {"count": count + 1}, state.update(count=count + 1).append(tracker=count + 1)
+
+    @property
+    def reads(self) -> list[str]:
+        return ["count"]
+
+    @property
+    def writes(self) -> list[str]:
+        return ["count", "tracker"]
+
+
+class SingleStepStreamingCounterWithExceptionNoStateAsync(SingleStepStreamingAction):
+    """Async variant: raises without ever yielding a final state update."""
+
+    async def stream_run_and_update(
+        self, state: State, **run_kwargs
+    ) -> AsyncGenerator[Tuple[dict, Optional[State]], None]:
+        count = state["count"]
+        for i in range(3):
+            yield {"count": count + ((i + 1) / 10)}, None
+        raise RuntimeError("simulated failure with no state")
+
+    @property
+    def reads(self) -> list[str]:
+        return ["count"]
+
+    @property
+    def writes(self) -> list[str]:
+        return ["count", "tracker"]
+
+
+class MultiStepStreamingCounterWithException(StreamingAction):
+    """Yields intermediate items, raises, then yields final result in finally block."""
+
+    def stream_run(self, state: State, **run_kwargs) -> Generator[dict, None, None]:
+        count = state["count"]
+        try:
+            for i in range(3):
+                yield {"count": count + ((i + 1) / 10)}
+            raise RuntimeError("simulated failure")
+        finally:
+            yield {"count": count + 1}
+
+    @property
+    def reads(self) -> list[str]:
+        return ["count"]
+
+    @property
+    def writes(self) -> list[str]:
+        return ["count", "tracker"]
+
+    def update(self, result: dict, state: State) -> State:
+        return state.update(**result).append(tracker=result["count"])
+
+
+class MultiStepStreamingCounterWithExceptionNoResult(StreamingAction):
+    """Raises without ever yielding any item."""
+
+    def stream_run(self, state: State, **run_kwargs) -> Generator[dict, None, None]:
+        raise RuntimeError("simulated failure with no result")
+        yield  # make this a generator function
+
+    @property
+    def reads(self) -> list[str]:
+        return ["count"]
+
+    @property
+    def writes(self) -> list[str]:
+        return ["count", "tracker"]
+
+    def update(self, result: dict, state: State) -> State:
+        return state.update(**result).append(tracker=result["count"])
+
+
+class MultiStepStreamingCounterWithExceptionAsync(AsyncStreamingAction):
+    """Async variant: yields intermediate items, raises, then yields final result in finally."""
+
+    async def stream_run(self, state: State, **run_kwargs) -> AsyncGenerator[dict, None]:
+        count = state["count"]
+        try:
+            for i in range(3):
+                yield {"count": count + ((i + 1) / 10)}
+            raise RuntimeError("simulated failure")
+        finally:
+            yield {"count": count + 1}
+
+    @property
+    def reads(self) -> list[str]:
+        return ["count"]
+
+    @property
+    def writes(self) -> list[str]:
+        return ["count", "tracker"]
+
+    def update(self, result: dict, state: State) -> State:
+        return state.update(**result).append(tracker=result["count"])
+
+
+class MultiStepStreamingCounterWithExceptionNoResultAsync(AsyncStreamingAction):
+    """Async variant: raises without ever yielding any item."""
+
+    async def stream_run(self, state: State, **run_kwargs) -> AsyncGenerator[dict, None]:
+        raise RuntimeError("simulated failure with no result")
+        yield  # make this an async generator
+
+    @property
+    def reads(self) -> list[str]:
+        return ["count"]
+
+    @property
+    def writes(self) -> list[str]:
+        return ["count", "tracker"]
+
+    def update(self, result: dict, state: State) -> State:
+        return state.update(**result).append(tracker=result["count"])
+
+
+def test__run_single_step_streaming_action_graceful_exception():
+    """When the generator raises but yields a final state in finally, stream completes gracefully."""
+    action = SingleStepStreamingCounterWithException().with_name("counter")
+    state = State({"count": 0, "tracker": []})
+    generator = _run_single_step_streaming_action(
+        action, state, inputs={}, sequence_id=0, partition_key="pk", app_id="app"
+    )
+    results = list(generator)
+    intermediate = [(r, s) for r, s in results if s is None]
+    final = [(r, s) for r, s in results if s is not None]
+    assert len(intermediate) == 3
+    assert len(final) == 1
+    assert final[0][0] == {"count": 1}
+    assert final[0][1].subset("count", "tracker").get_all() == {
+        "count": 1,
+        "tracker": [1],
+    }
+
+
+def test__run_single_step_streaming_action_exception_propagates():
+    """When the generator raises without yielding a final state, exception propagates."""
+    action = SingleStepStreamingCounterWithExceptionNoState().with_name("counter")
+    state = State({"count": 0, "tracker": []})
+    generator = _run_single_step_streaming_action(
+        action, state, inputs={}, sequence_id=0, partition_key="pk", app_id="app"
+    )
+    with pytest.raises(RuntimeError, match="simulated failure with no state"):
+        list(generator)
+
+
+async def test__run_single_step_streaming_action_graceful_exception_async():
+    """Async: when the generator raises but yields a final state in finally, stream completes."""
+    action = SingleStepStreamingCounterWithExceptionAsync().with_name("counter")
+    state = State({"count": 0, "tracker": []})
+    generator = _arun_single_step_streaming_action(
+        action=action,
+        state=state,
+        inputs={},
+        sequence_id=0,
+        app_id="app",
+        partition_key="pk",
+        lifecycle_adapters=LifecycleAdapterSet(),
+    )
+    results = []
+    async for item in generator:
+        results.append(item)
+    intermediate = [(r, s) for r, s in results if s is None]
+    final = [(r, s) for r, s in results if s is not None]
+    assert len(intermediate) == 3
+    assert len(final) == 1
+    assert final[0][0] == {"count": 1}
+    assert final[0][1].subset("count", "tracker").get_all() == {
+        "count": 1,
+        "tracker": [1],
+    }
+
+
+async def test__run_single_step_streaming_action_exception_propagates_async():
+    """Async: when the generator raises without yielding a final state, exception propagates."""
+    action = SingleStepStreamingCounterWithExceptionNoStateAsync().with_name("counter")
+    state = State({"count": 0, "tracker": []})
+    generator = _arun_single_step_streaming_action(
+        action=action,
+        state=state,
+        inputs={},
+        sequence_id=0,
+        app_id="app",
+        partition_key="pk",
+        lifecycle_adapters=LifecycleAdapterSet(),
+    )
+    with pytest.raises(RuntimeError, match="simulated failure with no state"):
+        async for _ in generator:
+            pass
+
+
+def test__run_multi_step_streaming_action_graceful_exception():
+    """When the generator raises but yields a final result in finally, stream completes."""
+    action = MultiStepStreamingCounterWithException().with_name("counter")
+    state = State({"count": 0, "tracker": []})
+    generator = _run_multi_step_streaming_action(
+        action, state, inputs={}, sequence_id=0, partition_key="pk", app_id="app"
+    )
+    results = list(generator)
+    intermediate = [(r, s) for r, s in results if s is None]
+    final = [(r, s) for r, s in results if s is not None]
+    assert len(intermediate) == 3
+    assert len(final) == 1
+    assert final[0][0] == {"count": 1}
+    assert final[0][1].subset("count", "tracker").get_all() == {
+        "count": 1,
+        "tracker": [1],
+    }
+
+
+def test__run_multi_step_streaming_action_exception_propagates():
+    """When the generator raises without yielding any result, exception propagates."""
+    action = MultiStepStreamingCounterWithExceptionNoResult().with_name("counter")
+    state = State({"count": 0, "tracker": []})
+    generator = _run_multi_step_streaming_action(
+        action, state, inputs={}, sequence_id=0, partition_key="pk", app_id="app"
+    )
+    with pytest.raises(RuntimeError, match="simulated failure with no result"):
+        list(generator)
+
+
+async def test__run_multi_step_streaming_action_graceful_exception_async():
+    """Async: when the generator raises but yields a final result in finally, stream completes."""
+    action = MultiStepStreamingCounterWithExceptionAsync().with_name("counter")
+    state = State({"count": 0, "tracker": []})
+    generator = _arun_multi_step_streaming_action(
+        action=action,
+        state=state,
+        inputs={},
+        sequence_id=0,
+        app_id="app",
+        partition_key="pk",
+        lifecycle_adapters=LifecycleAdapterSet(),
+    )
+    results = []
+    async for item in generator:
+        results.append(item)
+    intermediate = [(r, s) for r, s in results if s is None]
+    final = [(r, s) for r, s in results if s is not None]
+    assert len(intermediate) == 3
+    assert len(final) == 1
+    assert final[0][0] == {"count": 1}
+    assert final[0][1].subset("count", "tracker").get_all() == {
+        "count": 1,
+        "tracker": [1],
+    }
+
+
+async def test__run_multi_step_streaming_action_exception_propagates_async():
+    """Async: when the generator raises without yielding any result, exception propagates."""
+    action = MultiStepStreamingCounterWithExceptionNoResultAsync().with_name("counter")
+    state = State({"count": 0, "tracker": []})
+    generator = _arun_multi_step_streaming_action(
+        action=action,
+        state=state,
+        inputs={},
+        sequence_id=0,
+        app_id="app",
+        partition_key="pk",
+        lifecycle_adapters=LifecycleAdapterSet(),
+    )
+    with pytest.raises(RuntimeError, match="simulated failure with no result"):
+        async for _ in generator:
+            pass
+
+
 class SingleStepActionWithDeletionAsync(SingleStepActionWithDeletion):
     async def run_and_update(self, state: State, **run_kwargs) -> Tuple[dict, State]:
         return {}, state.wipe(delete=["to_delete"])
@@ -1935,7 +2282,11 @@ async def test_app_a_run_async_and_sync():
         graph=Graph(
             actions=[counter_action_sync, counter_action_async, result_action],
             transitions=[
-                Transition(counter_action_sync, counter_action_async, Condition.expr("count < 20")),
+                Transition(
+                    counter_action_sync,
+                    counter_action_async,
+                    Condition.expr("count < 20"),
+                ),
                 Transition(counter_action_async, counter_action_sync, default),
                 Transition(counter_action_sync, result_action, default),
             ],
@@ -2060,7 +2411,8 @@ async def test_astream_result_halt_after_unique_ordered_sequence_id():
 
 def test_stream_result_halt_after_run_through_streaming():
     """Tests that we can pass through streaming results,
-    fully realize them, then get to the streaming results at the end and return the stream"""
+    fully realize them, then get to the streaming results at the end and return the stream
+    """
     action_tracker = CallCaptureTracker()
     stream_event_tracker = StreamEventCaptureTracker()
     counter_action = base_streaming_single_step_counter.with_name("counter")
@@ -2665,7 +3017,10 @@ def test__adjust_single_step_output_result_and_state():
 
 def test__adjust_single_step_output_just_state():
     state = State({"count": 1})
-    assert _adjust_single_step_output(state, "test_action", DEFAULT_SCHEMA) == ({}, state)
+    assert _adjust_single_step_output(state, "test_action", DEFAULT_SCHEMA) == (
+        {},
+        state,
+    )
 
 
 def test__adjust_single_step_output_errors_incorrect_type():
@@ -2912,7 +3267,11 @@ class BrokenPersister(BaseStatePersister):
     """Broken persistor."""
 
     def load(
-        self, partition_key: str, app_id: Optional[str], sequence_id: Optional[int] = None, **kwargs
+        self,
+        partition_key: str,
+        app_id: Optional[str],
+        sequence_id: Optional[int] = None,
+        **kwargs,
     ) -> Optional[PersistedStateData]:
         return dict(
             partition_key="key",
@@ -2968,7 +3327,8 @@ def test_load_from_sync_cannot_have_async_persistor_error():
         default_entrypoint="foo",
     )
     with pytest.raises(
-        ValueError, match="are building the sync application, but have used an async initializer."
+        ValueError,
+        match="are building the sync application, but have used an async initializer.",
     ):
         # we have not initialized
         builder._load_from_sync_persister()
@@ -2984,7 +3344,8 @@ async def test_load_from_async_cannot_have_sync_persistor_error():
         default_entrypoint="foo",
     )
     with pytest.raises(
-        ValueError, match="are building the async application, but have used an sync initializer."
+        ValueError,
+        match="are building the async application, but have used an sync initializer.",
     ):
         # we have not initialized
         await builder._load_from_async_persister()
@@ -3055,7 +3416,11 @@ class DummyPersister(BaseStatePersister):
     """Dummy persistor."""
 
     def load(
-        self, partition_key: str, app_id: Optional[str], sequence_id: Optional[int] = None, **kwargs
+        self,
+        partition_key: str,
+        app_id: Optional[str],
+        sequence_id: Optional[int] = None,
+        **kwargs,
     ) -> Optional[PersistedStateData]:
         return PersistedStateData(
             partition_key="user123",
@@ -3416,7 +3781,10 @@ def post_run_execute_call(
     hook = TestingHook()
     foo = []
 
-    @action(reads=["recursion_count", "total_count"], writes=["recursion_count", "total_count"])
+    @action(
+        reads=["recursion_count", "total_count"],
+        writes=["recursion_count", "total_count"],
+    )
     def recursive_action(state: State) -> State:
         foo.append(1)
         recursion_count = state["recursion_count"]
@@ -3498,7 +3866,8 @@ def test_set_sync_state_persister_cannot_have_async_error():
     persister = AsyncDevNullPersister()
     builder.with_state_persister(persister)
     with pytest.raises(
-        ValueError, match="are building the sync application, but have used an async persister."
+        ValueError,
+        match="are building the sync application, but have used an async persister.",
     ):
         # we have not initialized
         builder._set_sync_state_persister()
@@ -3519,7 +3888,8 @@ async def test_set_async_state_persister_cannot_have_sync_error():
     persister = DevNullPersister()
     builder.with_state_persister(persister)
     with pytest.raises(
-        ValueError, match="are building the async application, but have used an sync persister."
+        ValueError,
+        match="are building the async application, but have used an sync persister.",
     ):
         # we have not initialized
         await builder._set_async_state_persister()
@@ -3620,15 +3990,27 @@ def inputs(self) -> Union[list[str], tuple[list[str], list[str]]]:
 def test_remap_context_variable_with_mangled_context_kwargs():
     _action = ActionWithKwargs()
 
-    inputs = {"__context": "context_value", "other_key": "other_value", "foo": "foo_value"}
-    expected = {"__context": "context_value", "other_key": "other_value", "foo": "foo_value"}
+    inputs = {
+        "__context": "context_value",
+        "other_key": "other_value",
+        "foo": "foo_value",
+    }
+    expected = {
+        "__context": "context_value",
+        "other_key": "other_value",
+        "foo": "foo_value",
+    }
     assert _remap_dunder_parameters(_action.run, inputs, ["__context", "__tracer"]) == expected
 
 
 def test_remap_context_variable_with_mangled_context():
     _action = ActionWithContext()
 
-    inputs = {"__context": "context_value", "other_key": "other_value", "foo": "foo_value"}
+    inputs = {
+        "__context": "context_value",
+        "other_key": "other_value",
+        "foo": "foo_value",
+    }
     expected = {
         f"_{ActionWithContext.__name__}__context": "context_value",
         "other_key": "other_value",
@@ -3657,8 +4039,16 @@ def test_remap_context_variable_with_mangled_contexttracer():
 
 def test_remap_context_variable_without_mangled_context():
     _action = ActionWithoutContext()
-    inputs = {"__context": "context_value", "other_key": "other_value", "foo": "foo_value"}
-    expected = {"__context": "context_value", "other_key": "other_value", "foo": "foo_value"}
+    inputs = {
+        "__context": "context_value",
+        "other_key": "other_value",
+        "foo": "foo_value",
+    }
+    expected = {
+        "__context": "context_value",
+        "other_key": "other_value",
+        "foo": "foo_value",
+    }
     assert _remap_dunder_parameters(_action.run, inputs, ["__context", "__tracer"]) == expected
 
 
diff --git a/tests/core/test_persistence.py b/tests/core/test_persistence.py
index b362cd96d..18c65a0d1 100644
--- a/tests/core/test_persistence.py
+++ b/tests/core/test_persistence.py
@@ -110,7 +110,12 @@ def test_sqlite_persister_save_without_initialize_raises_runtime_error():
     try:
         with pytest.raises(RuntimeError, match="Uninitialized persister"):
             persister.save(
-                "partition_key", "app_id", 1, "position", State({"key": "value"}), "completed"
+                "partition_key",
+                "app_id",
+                1,
+                "position",
+                State({"key": "value"}),
+                "completed",
             )
     finally:
         persister.cleanup()
@@ -168,17 +173,6 @@ def test_persister_methods_none_partition_key(persistence, method_name: str, kwa
 """Asyncio integration for sqlite persister + """
 
 
-class AsyncSQLiteContextManager:
-    def __init__(self, sqlite_object):
-        self.client = sqlite_object
-
-    async def __aenter__(self):
-        return self.client
-
-    async def __aexit__(self, exc_type, exc, tb):
-        await self.client.close()
-
-
 @pytest.fixture()
 async def async_persistence(request):
     yield AsyncInMemoryPersister()
@@ -276,15 +270,15 @@ async def test_AsyncSQLitePersister_connection_shutdown():
 
 @pytest.fixture()
 async def initializing_async_persistence():
-    sqlite_persister = await AsyncSQLitePersister.from_values(
+    async with AsyncSQLitePersister.from_values(
         db_path=":memory:", table_name="test_table"
-    )
-    async_context_manager = AsyncSQLiteContextManager(sqlite_persister)
-    async with async_context_manager as client:
+    ) as client:
         yield client
 
 
-async def test_async_persistence_initialization_creates_table(initializing_async_persistence):
+async def test_async_persistence_initialization_creates_table(
+    initializing_async_persistence,
+):
     await asyncio.sleep(0.00001)
     await initializing_async_persistence.initialize()
     assert await initializing_async_persistence.list_app_ids("partition_key") == []
diff --git a/tests/integrations/persisters/test_b_aiosqlite.py b/tests/integrations/persisters/test_b_aiosqlite.py
index 00c98677a..adb97532c 100644
--- a/tests/integrations/persisters/test_b_aiosqlite.py
+++ b/tests/integrations/persisters/test_b_aiosqlite.py
@@ -25,17 +25,6 @@
 from burr.integrations.persisters.b_aiosqlite import AsyncSQLitePersister
 
 
-class AsyncSQLiteContextManager:
-    def __init__(self, sqlite_object):
-        self.client = sqlite_object
-
-    async def __aenter__(self):
-        return self.client
-
-    async def __aexit__(self, exc_type, exc, tb):
-        await self.client.cleanup()
-
-
 async def test_copy_persister(async_persistence: AsyncSQLitePersister):
     copy = async_persistence.copy()
     assert copy.table_name == async_persistence.table_name
@@ -45,11 +34,9 @@ async def test_copy_persister(async_persistence: AsyncSQLitePersister):
 
 @pytest.fixture()
 async def async_persistence(request):
-    sqlite_persister = await AsyncSQLitePersister.from_values(
+    async with AsyncSQLitePersister.from_values(
         db_path=":memory:", table_name="test_table"
-    )
-    async_context_manager = AsyncSQLiteContextManager(sqlite_persister)
-    async with async_context_manager as client:
+    ) as client:
         yield client
 
 
@@ -118,6 +105,50 @@ async def test_async_persister_methods_none_partition_key(
     # these operations are stateful (i.e., read/write to a db)
 
 
+async def test_async_sqlite_from_values_as_context_manager(tmp_path):
+    """Test that from_values works directly with async with (issue #546)."""
+    db_path = str(tmp_path / "test.db")
+    async with AsyncSQLitePersister.from_values(db_path=db_path) as persister:
+        await persister.initialize()
+        await persister.save("pk", "app1", 1, "pos", State({"k": "v"}), "completed")
+        loaded = await persister.load("pk", "app1")
+        assert loaded is not None
+        assert loaded["state"] == State({"k": "v"})
+
+
+async def test_async_sqlite_from_config_as_context_manager(tmp_path):
+    """Test that from_config works directly with async with (issue #546)."""
+    db_path = str(tmp_path / "test.db")
+    config = {"db_path": db_path, "table_name": "burr_state"}
+    async with AsyncSQLitePersister.from_config(config) as persister:
+        await persister.initialize()
+        await persister.save("pk", "app1", 1, "pos", State({"k": "v"}), "completed")
+        loaded = await persister.load("pk", "app1")
+        assert loaded is not None
+
+
+async def test_async_sqlite_from_values_cannot_be_consumed_twice():
+    """Test that the factory wrapper raises on double consumption."""
+    wrapper = AsyncSQLitePersister.from_values(db_path=":memory:")
+    persister = await wrapper
+    with pytest.raises(RuntimeError, match="already been consumed"):
+        await wrapper
+    await persister.cleanup()
+
+
+async def test_async_sqlite_context_manager_aexit_safe_on_failed_aenter(tmp_path):
+    """Test that __aexit__ doesn't crash if __aenter__ never completed."""
+    from burr.common.async_utils import _AsyncPersisterContextManager
+
+    async def _failing_create():
+        raise ConnectionError("simulated connection failure")
+
+    mgr = _AsyncPersisterContextManager(_failing_create())
+    with pytest.raises(ConnectionError, match="simulated connection failure"):
+        async with mgr:
+            pass  # should never reach here
+
+
 async def test_AsyncSQLitePersister_from_values():
     await asyncio.sleep(0.00001)
     connection = await aiosqlite.connect(":memory:")
@@ -145,11 +176,9 @@ async def test_AsyncSQLitePersister_connection_shutdown():
 
 @pytest.fixture()
 async def initializing_async_persistence():
-    sqlite_persister = await AsyncSQLitePersister.from_values(
+    async with AsyncSQLitePersister.from_values(
         db_path=":memory:", table_name="test_table"
-    )
-    async_context_manager = AsyncSQLiteContextManager(sqlite_persister)
-    async with async_context_manager as client:
+    ) as client:
         yield client