diff --git a/.github/workflows/inngest_remote_state.yml b/.github/workflows/inngest_remote_state.yml
new file mode 100644
index 00000000..9e5928a4
--- /dev/null
+++ b/.github/workflows/inngest_remote_state.yml
@@ -0,0 +1,120 @@
+name: "inngest_remote_state"
+
+on:
+  push:
+    branches:
+      - "main"
+    tags:
+      - "*"
+  pull_request:
+    paths:
+      - "!examples/**"
+      - "**/*.py"
+      - ".github/**"
+      - "Makefile"
+      - "constraints.txt"
+      - "pyproject.toml"
+      - "pytest.ini"
+      - "ruff.toml"
+
+jobs:
+  itest:
+    runs-on: "ubuntu-latest"
+    strategy:
+      matrix:
+        python-version: ["3.9", "3.12"]
+    timeout-minutes: 2
+    steps:
+      - uses: "actions/checkout@v2"
+      - name: "Set up Python ${{ matrix.python-version }}"
+        uses: "actions/setup-python@v2"
+        with:
+          python-version: "${{ matrix.python-version }}"
+      - name: "Install"
+        run: "make install"
+      - name: "Integration test"
+        run: "make itest"
+        working-directory: "./pkg/inngest_remote_state"
+
+  lint:
+    runs-on: "ubuntu-latest"
+    strategy:
+      matrix:
+        python-version: ["3.9", "3.12"]
+    steps:
+      - uses: "actions/checkout@v2"
+      - name: "Set up Python ${{ matrix.python-version }}"
+        uses: "actions/setup-python@v2"
+        with:
+          python-version: "${{ matrix.python-version }}"
+      - name: "Install"
+        run: "make install"
+      - name: "Lint"
+        run: "make lint"
+        working-directory: "./pkg/inngest_remote_state"
+
+  publish-pypi:
+    runs-on: "ubuntu-latest"
+    needs:
+      - "itest"
+      - "lint"
+      - "type-check"
+      - "utest"
+
+    # Only publish tagged versions.
+    # TODO: Add a check to ensure that the git tag matches the version.
+    if: "startsWith(github.ref, 'refs/tags/inngest_remote_state@')"
+    permissions:
+      id-token: write
+    strategy:
+      matrix:
+        python-version: ["3.9"]
+    steps:
+      - uses: "actions/checkout@v2"
+      - name: "Set up Python ${{ matrix.python-version }}"
+        uses: "actions/setup-python@v2"
+        with:
+          python-version: "${{ matrix.python-version }}"
+      - name: "Install"
+        run: "make install"
+      - name: "Build"
+        run: "make build"
+        working-directory: "./pkg/inngest_remote_state"
+      - name: "Upload package to PyPI"
+        uses: "pypa/gh-action-pypi-publish@release/v1"
+        with:
+          packages-dir: "./pkg/inngest_remote_state/dist"
+
+  type-check:
+    runs-on: "ubuntu-latest"
+    strategy:
+      matrix:
+        python-version: ["3.9", "3.12"]
+    steps:
+      - uses: "actions/checkout@v2"
+      - name: "Set up Python ${{ matrix.python-version }}"
+        uses: "actions/setup-python@v2"
+        with:
+          python-version: "${{ matrix.python-version }}"
+      - name: "Install"
+        run: "make install"
+      - name: "Type check"
+        run: "make type-check"
+        working-directory: "./pkg/inngest_remote_state"
+
+  utest:
+    runs-on: "ubuntu-latest"
+    strategy:
+      matrix:
+        python-version: ["3.9", "3.12"]
+    steps:
+      - uses: "actions/checkout@v2"
+      - name: "Set up Python ${{ matrix.python-version }}"
+        uses: "actions/setup-python@v2"
+        with:
+          python-version: "${{ matrix.python-version }}"
+      - name: "Install"
+        run: "make install"
+      - name: "Unit test"
+        run: "make utest"
+        working-directory: "./pkg/inngest_remote_state"
diff --git a/.vscode/settings.json b/.vscode/settings.json
index 998ec3f3..e474cf47 100644
--- a/.vscode/settings.json
+++ b/.vscode/settings.json
@@ -11,6 +11,7 @@
   "python.analysis.extraPaths": [
     "pkg/inngest",
     "pkg/inngest_encryption",
+    "pkg/inngest_remote_state",
     "pkg/test_core"
   ],
   "python.analysis.typeCheckingMode": "basic",
diff --git a/Makefile b/Makefile
index 8aa781ab..d23da7d2 100644
--- a/Makefile
+++ b/Makefile
@@ -11,11 +11,17 @@ format-check: check-venv
 	@ruff format --check .
 
 install: check-venv
-	@pip install -e '.[extra]' -e ./pkg/inngest -e ./pkg/inngest_encryption -e ./pkg/test_core -c constraints.txt
+	@pip install \
+		-e '.[extra]' -c constraints.txt \
+		-e ./pkg/inngest \
+		-e ./pkg/inngest_encryption \
+		-e ./pkg/inngest_remote_state \
+		-e ./pkg/test_core
 
 itest: check-venv
 	@cd pkg/inngest && make itest
 	@cd pkg/inngest_encryption && make itest
+	@cd pkg/inngest_remote_state && make itest
 
 pre-commit: format-check lint type-check utest
 
@@ -23,14 +29,17 @@ lint: check-venv
 	@cd examples && make lint
 	@cd pkg/inngest && make lint
 	@cd pkg/inngest_encryption && make lint
+	@cd pkg/inngest_remote_state && make lint
 	@cd pkg/test_core && make lint
 
 type-check: check-venv
 	@cd examples && make type-check
 	@cd pkg/inngest && make type-check
 	@cd pkg/inngest_encryption && make type-check
+	@cd pkg/inngest_remote_state && make type-check
 	@cd pkg/test_core && make type-check
 
 utest: check-venv
 	@cd pkg/inngest && make utest
 	@cd pkg/inngest_encryption && make utest
+	@cd pkg/inngest_remote_state && make utest
diff --git a/mypy.ini b/mypy.ini
index 17d55d02..50e4dd16 100644
--- a/mypy.ini
+++ b/mypy.ini
@@ -1,7 +1,7 @@
 [mypy]
 enable_error_code = possibly-undefined, redundant-expr, truthy-bool
 incremental = false
-mypy_path = ./pkg/inngest, ./pkg/inngest_encryption, ./pkg/test_core
+mypy_path = ./pkg/inngest, ./pkg/inngest_encryption, ./pkg/inngest_remote_state, ./pkg/test_core
 strict = true
 warn_unreachable = true
 
diff --git a/pkg/inngest_remote_state/Makefile b/pkg/inngest_remote_state/Makefile
new file mode 100644
index 00000000..5e871c4b
--- /dev/null
+++ b/pkg/inngest_remote_state/Makefile
@@ -0,0 +1,32 @@
+export MYPYPATH=../inngest
+
+.PHONY: build
+build:
+	@if [ -d "dist" ]; then rm -rf dist; fi
+	@python -m build
+
+.PHONY: check-venv
+check-venv:
+	@if [ -z "$${CI}" ] && [ -z "$${VIRTUAL_ENV}" ]; then \
+		echo "virtual environment is not activated"; \
+		exit 1; \
+	fi
+
+.PHONY: itest
+itest: check-venv
+	@cd ../../tests/test_inngest_remote_state && pytest -n 4 -v .
+
+.PHONY: lint
+lint: check-venv
+	@ruff check .
+
+release:
+	@grep "version = \"$${VERSION}\"" pyproject.toml && git tag inngest_remote_state@$${VERSION} && git push origin inngest_remote_state@$${VERSION} || echo "pyproject.toml version does not match"
+
+.PHONY: type-check
+type-check: check-venv
+	@mypy --config-file=../../mypy.ini .
+
+.PHONY: utest
+utest: check-venv
+	@echo "inngest_remote_state unit tests not implemented"
diff --git a/pkg/inngest_remote_state/README.md b/pkg/inngest_remote_state/README.md
new file mode 100644
index 00000000..236d9f1e
--- /dev/null
+++ b/pkg/inngest_remote_state/README.md
@@ -0,0 +1,38 @@
+# Inngest Python SDK: Remote state
+
+This package provides the tools for storing step output in a custom store (e.g. AWS S3). This can drastically reduce bandwidth to/from the Inngest server, since step output is stored within your infrastructure rather than Inngest's.
+
+## Usage
+
+Setting remote state middleware on the client will turn on remote state for steps in all functions:
+
+```py
+import inngest
+from inngest_remote_state import RemoteStateMiddleware
+from inngest_remote_state.s3 import S3Driver
+
+inngest.Inngest(
+    app_id="my-app",
+    middleware=[
+        RemoteStateMiddleware.factory(
+            S3Driver(
+                bucket="inngest-remote-state",
+                client=boto3.client("s3"),
+            )
+        )
+    ],
+)
+```
+
+The entire `step.run` output is stored in the remote store:
+
+```py
+def _my_step() -> dict[str, object]:
+    # Stored in the remote store.
+    return {"msg": "hello"}
+
+output = await step.run("my-step", _my_step)
+
+# Available within this function (it's automatically loaded by the middleware).
+print(output)
+```
diff --git a/pkg/inngest_remote_state/inngest_remote_state/__init__.py b/pkg/inngest_remote_state/inngest_remote_state/__init__.py
new file mode 100644
index 00000000..04837349
--- /dev/null
+++ b/pkg/inngest_remote_state/inngest_remote_state/__init__.py
@@ -0,0 +1,5 @@
+"""Public entrypoint for the Inngest SDK encryption package."""
+
+from ._internal import RemoteStateMiddleware, StateDriver
+
+__all__ = ["RemoteStateMiddleware", "StateDriver"]
diff --git a/pkg/inngest_remote_state/inngest_remote_state/_internal/__init__.py b/pkg/inngest_remote_state/inngest_remote_state/_internal/__init__.py
new file mode 100644
index 00000000..053a7787
--- /dev/null
+++ b/pkg/inngest_remote_state/inngest_remote_state/_internal/__init__.py
@@ -0,0 +1,10 @@
+"""
+Remote state middleware for Inngest. This middleware allows you to store state
+where you want, rather than in Inngest's infrastructure. This is useful for:
+- Reducing bandwidth to/from the Inngest server.
+- Avoiding step output size limits.
+"""
+
+from .middleware import RemoteStateMiddleware, StateDriver
+
+__all__ = ["RemoteStateMiddleware", "StateDriver"]
diff --git a/pkg/inngest_remote_state/inngest_remote_state/_internal/middleware.py b/pkg/inngest_remote_state/inngest_remote_state/_internal/middleware.py
new file mode 100644
index 00000000..1cf22a13
--- /dev/null
+++ b/pkg/inngest_remote_state/inngest_remote_state/_internal/middleware.py
@@ -0,0 +1,129 @@
+from __future__ import annotations
+
+import typing
+
+import inngest
+
+
+class StateDriver(typing.Protocol):
+    """
+    Protocol for the state driver.
+    """
+
+    def load_steps(self, steps: inngest.StepMemos) -> None:
+        """
+        Retrieve the value associated with the key.
+
+        Args:
+        ----
+            steps: Steps whose output may need to be loaded from the remote store.
+        """
+
+        ...
+
+    def save_step(
+        self,
+        run_id: str,
+        value: object,
+    ) -> dict[str, object]:
+        """
+        Store the value and return a key to retrieve it later.
+
+        Args:
+        ----
+            run_id: Run ID.
+            value: Output for an ended step.
+        """
+
+        ...
+
+
+class RemoteStateMiddleware(inngest.MiddlewareSync):
+    """
+    Middleware that reads/writes step output in a custom store (e.g. AWS S3).
+    This can drastically reduce bandwidth to/from the Inngest server, since step
+    output is stored within your infrastructure rather than Inngest's.
+    """
+
+    _run_id: typing.Optional[str] = None
+
+    def __init__(
+        self,
+        client: inngest.Inngest,
+        raw_request: object,
+        driver: StateDriver,
+    ) -> None:
+        """
+        Args:
+        ----
+            client: Inngest client.
+            raw_request: Framework/platform specific request object.
+            driver: State driver.
+        """
+
+        super().__init__(client, raw_request)
+
+        self._driver = driver
+
+    @classmethod
+    def factory(
+        cls,
+        driver: StateDriver,
+    ) -> typing.Callable[[inngest.Inngest, object], RemoteStateMiddleware]:
+        """
+        Create a remote state middleware that can be passed to an Inngest client
+        or function.
+
+        Args:
+        ----
+            driver: State driver.
+        """
+
+        def _factory(
+            client: inngest.Inngest,
+            raw_request: object,
+        ) -> RemoteStateMiddleware:
+            return cls(
+                client,
+                raw_request,
+                driver,
+            )
+
+        return _factory
+
+    def transform_input(
+        self,
+        ctx: inngest.Context,
+        function: inngest.Function,
+        steps: inngest.StepMemos,
+    ) -> None:
+        """
+        Inject remote state.
+        """
+
+        self._driver.load_steps(steps)
+        self._run_id = ctx.run_id
+
+    def transform_output(self, result: inngest.TransformOutputResult) -> None:
+        """
+        Store step output externally and replace with a marker and key.
+        """
+
+        if result.step is None:
+            return None
+
+        # Only support step.run, but that may change in the future.
+        if result.step.op.value != "StepRun":
+            return None
+
+        if result.has_output() is False:
+            return None
+
+        if self._run_id is None:
+            # Unreachable
+            raise Exception("missing run ID")
+
+        result.output = self._driver.save_step(
+            self._run_id,
+            result.output,
+        )
diff --git a/pkg/inngest_remote_state/inngest_remote_state/in_memory.py b/pkg/inngest_remote_state/inngest_remote_state/in_memory.py
new file mode 100644
index 00000000..f1b6b2e6
--- /dev/null
+++ b/pkg/inngest_remote_state/inngest_remote_state/in_memory.py
@@ -0,0 +1,74 @@
+import secrets
+import string
+import typing
+
+import inngest
+import pydantic
+
+from ._internal import StateDriver
+
+
+class _StatePlaceholder(pydantic.BaseModel):
+    key: str
+
+
+class InMemoryDriver(StateDriver):
+    """
+    In-memory driver for remote state middleware. This probably doesn't have any
+    use besides being a reference.
+    """
+
+    # Marker to indicate that the data is stored remotely.
+    _marker: typing.Final = "__REMOTE_STATE__"
+
+    # Marker to indicate which strategy was used. This is useful for knowing
+    # whether the official S3 driver was used.
+    _strategy_marker: typing.Final = "__STRATEGY__"
+
+    _strategy_identifier: typing.Final = "inngest/memory"
+
+    def __init__(self) -> None:  # noqa: D107
+        self._data: dict[str, object] = {}
+
+    def _create_key(self) -> str:
+        chars = string.ascii_letters + string.digits
+        return "".join(secrets.choice(chars) for _ in range(32))
+
+    def load_steps(self, steps: inngest.StepMemos) -> None:
+        """
+        Hydrate steps with remote state if necessary.
+        """
+
+        for step in steps.values():
+            if not isinstance(step.data, dict):
+                continue
+            if self._marker not in step.data:
+                continue
+            if self._strategy_marker not in step.data:
+                continue
+            if step.data[self._strategy_marker] != self._strategy_identifier:
+                continue
+
+            placeholder = _StatePlaceholder.model_validate(step.data)
+
+            step.data = self._data[placeholder.key]
+
+    def save_step(
+        self,
+        run_id: str,
+        value: object,
+    ) -> dict[str, object]:
+        """
+        Save a step's output to the remote store and return a placeholder.
+        """
+
+        key = self._create_key()
+        self._data[key] = value
+
+        placeholder: dict[str, object] = {
+            self._marker: True,
+            self._strategy_marker: self._strategy_identifier,
+            **_StatePlaceholder(key=key).model_dump(),
+        }
+
+        return placeholder
diff --git a/pkg/inngest_remote_state/inngest_remote_state/py.typed b/pkg/inngest_remote_state/inngest_remote_state/py.typed
new file mode 100644
index 00000000..e69de29b
diff --git a/pkg/inngest_remote_state/inngest_remote_state/s3.py b/pkg/inngest_remote_state/inngest_remote_state/s3.py
new file mode 100644
index 00000000..50df1bd6
--- /dev/null
+++ b/pkg/inngest_remote_state/inngest_remote_state/s3.py
@@ -0,0 +1,123 @@
+from __future__ import annotations
+
+import json
+import secrets
+import string
+import typing
+
+import inngest
+import pydantic
+import typing_extensions
+
+from ._internal import StateDriver
+
+if typing.TYPE_CHECKING:
+    from mypy_boto3_s3 import S3Client
+
+
+class _StateSurrogate(pydantic.BaseModel):
+    """
+    Replaces step output sent back to Inngest. Its data is sufficient to
+    retrieve the actual state.
+    """
+
+    bucket: str
+    key: str
+
+
+class S3Driver(StateDriver):
+    """
+    S3 driver for remote state middleware.
+    """
+
+    # Marker to indicate that the data is stored remotely.
+    _marker: typing.Final = "__REMOTE_STATE__"
+
+    # Marker to indicate which strategy was used. This is useful for knowing
+    # whether the official S3 driver was used.
+    _strategy_marker: typing.Final = "__STRATEGY__"
+
+    _strategy_identifier: typing.Final = "inngest/s3"
+
+    def __init__(
+        self,
+        *,
+        bucket: str,
+        client: S3Client,
+    ) -> None:
+        """
+        Args:
+        ----
+            bucket: Bucket name to store remote state.
+            client: Boto3 S3 client.
+        """
+
+        self._bucket = bucket
+        self._client = client
+
+    def _create_key(self) -> str:
+        chars = string.ascii_letters + string.digits
+        return "".join(secrets.choice(chars) for _ in range(32))
+
+    def _is_remote(
+        self, data: object
+    ) -> typing_extensions.TypeGuard[dict[str, object]]:
+        return (
+            isinstance(data, dict)
+            and self._marker in data
+            and self._strategy_marker in data
+            and data[self._strategy_marker] == self._strategy_identifier
+        )
+
+    def load_steps(self, steps: inngest.StepMemos) -> None:
+        """
+        Hydrate steps with remote state if necessary.
+
+        Args:
+        ----
+            steps: Steps that may need hydration.
+        """
+
+        for step in steps.values():
+            if not self._is_remote(step.data):
+                continue
+
+            surrogate = _StateSurrogate.model_validate(step.data)
+
+            step.data = json.loads(
+                self._client.get_object(
+                    Bucket=surrogate.bucket,
+                    Key=surrogate.key,
+                )["Body"]
+                .read()
+                .decode()
+            )
+
+    def save_step(
+        self,
+        run_id: str,
+        value: object,
+    ) -> dict[str, object]:
+        """
+        Save a step's output to the remote store and return a placeholder.
+
+        Args:
+        ----
+            run_id: Run ID.
+            value: Step output.
+        """
+
+        key = f"inngest/remote_state/{run_id}/{self._create_key()}"
+        self._client.put_object(
+            Body=json.dumps(value),
+            Bucket=self._bucket,
+            Key=key,
+        )
+
+        surrogate = {
+            self._marker: True,
+            self._strategy_marker: self._strategy_identifier,
+            **_StateSurrogate(bucket=self._bucket, key=key).model_dump(),
+        }
+
+        return surrogate
diff --git a/pkg/inngest_remote_state/pyproject.toml b/pkg/inngest_remote_state/pyproject.toml
new file mode 100644
index 00000000..c5a8590b
--- /dev/null
+++ b/pkg/inngest_remote_state/pyproject.toml
@@ -0,0 +1,25 @@
+[project]
+name = "inngest_remote_state"
+version = "0.0.1"
+authors = [{ name = "Inngest Inc.", email = "hello@inngest.com" }]
+description = "Remote state for the Inngest SDK"
+readme = "README.md"
+classifiers = [
+    "Development Status :: 3 - Alpha",
+    "Intended Audience :: Developers",
+    "Programming Language :: Python :: 3.9",
+    "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.11",
+    "Programming Language :: Python :: 3.12",
+]
+requires-python = ">=3.9"
+
+dependencies = ["inngest>=0.4.20"]
+
+[project.urls]
+"Homepage" = "https://github.com/inngest/inngest-py"
+"Bug Tracker" = "https://github.com/inngest/inngest-py/issues"
+
+[tool.setuptools.package-data]
+# Makes py.typed appear when users install the inngest package.
+inngest_encryption = ["py.typed"]
diff --git a/tests/conftest.py b/tests/conftest.py
index 09b9ef28..a0100781 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -4,3 +4,4 @@
 # `assert` failures just show "AssertionError" with no helpful diff.
 pytest.register_assert_rewrite("test_inngest")
 pytest.register_assert_rewrite("test_inngest_encryption")
+pytest.register_assert_rewrite("test_inngest_remote_state")
diff --git a/tests/test_inngest_remote_state/__init__.py b/tests/test_inngest_remote_state/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/test_inngest_remote_state/cases/__init__.py b/tests/test_inngest_remote_state/cases/__init__.py
new file mode 100644
index 00000000..bf5ef55e
--- /dev/null
+++ b/tests/test_inngest_remote_state/cases/__init__.py
@@ -0,0 +1,36 @@
+import inngest
+from inngest._internal import server_lib
+
+from . import base, step_failed, step_output_in_memory, step_output_s3
+
+_modules = (
+    step_failed,
+    step_output_in_memory,
+    step_output_s3,
+)
+
+
+def create_async_cases(
+    client: inngest.Inngest,
+    framework: server_lib.Framework,
+) -> list[base.Case]:
+    return [
+        module.create(client, framework, is_sync=False) for module in _modules
+    ]
+
+
+def create_sync_cases(
+    client: inngest.Inngest,
+    framework: server_lib.Framework,
+) -> list[base.Case]:
+    cases = []
+    for module in _modules:
+        case = module.create(client, framework, is_sync=True)
+        if isinstance(case.fn, list) and len(case.fn) == 0:
+            continue
+        cases.append(case)
+
+    return cases
+
+
+__all__ = ["create_async_cases", "create_sync_cases"]
diff --git a/tests/test_inngest_remote_state/cases/base.py b/tests/test_inngest_remote_state/cases/base.py
new file mode 100644
index 00000000..5668b0aa
--- /dev/null
+++ b/tests/test_inngest_remote_state/cases/base.py
@@ -0,0 +1,74 @@
+import dataclasses
+import json
+import os
+import typing
+
+import inngest
+import nacl.encoding
+import nacl.secret
+from inngest._internal import server_lib
+from test_core import base
+
+BaseState = base.BaseState
+create_test_name = base.create_test_name
+wait_for = base.wait_for
+
+
+class TestClass(typing.Protocol):
+    client: inngest.Inngest
+
+
+@dataclasses.dataclass
+class Case:
+    fn: typing.Union[inngest.Function, list[inngest.Function]]
+    name: str
+    run_test: typing.Callable[[TestClass], typing.Awaitable[None]]
+
+
+def create_event_name(framework: server_lib.Framework, test_name: str) -> str:
+    suffix = ""
+    worker_id = os.getenv("PYTEST_XDIST_WORKER")
+    if worker_id:
+        suffix += f"-{worker_id}"
+
+    return f"{framework.value}/{test_name}{suffix}"
+
+
+def create_fn_id(test_name: str) -> str:
+    suffix = ""
+    worker_id = os.getenv("PYTEST_XDIST_WORKER")
+    if worker_id:
+        suffix += f"-{worker_id}"
+
+    return test_name + suffix
+
+
+class Encryptor:
+    def __init__(self, secret_key: bytes) -> None:
+        self._box = nacl.secret.SecretBox(
+            secret_key, encoder=nacl.encoding.HexEncoder
+        )
+
+    def encrypt(self, data: object) -> dict[str, typing.Union[bool, str]]:
+        """
+        Encrypt data the way middleware would.
+        """
+
+        byt = json.dumps(data).encode()
+        ciphertext = self._box.encrypt(
+            byt,
+            encoder=nacl.encoding.Base64Encoder,
+        )
+        return {
+            "__ENCRYPTED__": True,
+            "__STRATEGY__": "inngest/libsodium",
+            "data": ciphertext.decode(),
+        }
+
+    def decrypt(self, data: bytes) -> object:
+        return json.loads(
+            self._box.decrypt(
+                data,
+                encoder=nacl.encoding.Base64Encoder,
+            ).decode()
+        )
diff --git a/tests/test_inngest_remote_state/cases/step_failed.py b/tests/test_inngest_remote_state/cases/step_failed.py
new file mode 100644
index 00000000..491a14b7
--- /dev/null
+++ b/tests/test_inngest_remote_state/cases/step_failed.py
@@ -0,0 +1,115 @@
+"""
+Ensure step and function output is encrypted and decrypted correctly
+"""
+
+import json
+
+import inngest
+import inngest_remote_state
+import inngest_remote_state.in_memory
+import test_core.helper
+from inngest._internal import server_lib
+
+from . import base
+
+
+class _State(base.BaseState):
+    event: inngest.Event
+    events: list[inngest.Event]
+
+
+def create(
+    client: inngest.Inngest,
+    framework: server_lib.Framework,
+    is_sync: bool,
+) -> base.Case:
+    test_name = base.create_test_name(__file__)
+    event_name = base.create_event_name(framework, test_name)
+    fn_id = base.create_fn_id(test_name)
+    state = _State()
+    driver = inngest_remote_state.in_memory.InMemoryDriver()
+    mw = inngest_remote_state.RemoteStateMiddleware.factory(driver)
+
+    @client.create_function(
+        fn_id=fn_id,
+        middleware=[mw],
+        retries=0,
+        trigger=inngest.TriggerEvent(event=event_name),
+    )
+    def fn_sync(
+        ctx: inngest.Context,
+        step: inngest.StepSync,
+    ) -> str:
+        state.run_id = ctx.run_id
+
+        def _step() -> str:
+            raise Exception("oh no")
+
+        try:
+            step.run("step_1", _step)
+        except Exception as e:
+            return str(e)
+
+        return "unreachable"
+
+    @client.create_function(
+        fn_id=fn_id,
+        middleware=[inngest_remote_state.RemoteStateMiddleware.factory(driver)],
+        retries=0,
+        trigger=inngest.TriggerEvent(event=event_name),
+    )
+    async def fn_async(
+        ctx: inngest.Context,
+        step: inngest.Step,
+    ) -> str:
+        state.run_id = ctx.run_id
+
+        state.run_id = ctx.run_id
+
+        def _step() -> str:
+            raise Exception("oh no")
+
+        try:
+            await step.run("step_1", _step)
+        except Exception as e:
+            return str(e)
+
+        return "unreachable"
+
+    async def run_test(self: base.TestClass) -> None:
+        self.client.send_sync(inngest.Event(name=event_name))
+
+        run_id = await state.wait_for_run_id()
+        run = await test_core.helper.client.wait_for_run_status(
+            run_id,
+            test_core.helper.RunStatus.COMPLETED,
+        )
+
+        output = json.loads(
+            await test_core.helper.client.get_step_output(
+                run_id=run_id,
+                step_id="step_1",
+            )
+        )
+        assert isinstance(output, dict)
+        assert output.get("data") is None
+        error = output.get("error")
+        assert isinstance(error, dict)
+
+        # Ensure that the error data was not remotely stored.
+        assert driver._marker not in error
+        assert error.get("message") == "oh no"
+
+        assert run.output is not None
+        assert json.loads(run.output) == "oh no"
+
+    if is_sync:
+        fn = fn_sync
+    else:
+        fn = fn_async
+
+    return base.Case(
+        fn=fn,
+        run_test=run_test,
+        name=test_name,
+    )
diff --git a/tests/test_inngest_remote_state/cases/step_output_in_memory.py b/tests/test_inngest_remote_state/cases/step_output_in_memory.py
new file mode 100644
index 00000000..906a5649
--- /dev/null
+++ b/tests/test_inngest_remote_state/cases/step_output_in_memory.py
@@ -0,0 +1,146 @@
+"""
+Ensure step and function output is encrypted and decrypted correctly
+"""
+
+import datetime
+import json
+
+import inngest
+import inngest_remote_state
+import inngest_remote_state.in_memory
+import test_core.helper
+from inngest._internal import server_lib
+
+from . import base
+
+
+class _State(base.BaseState):
+    event: inngest.Event
+    events: list[inngest.Event]
+
+
+def create(
+    client: inngest.Inngest,
+    framework: server_lib.Framework,
+    is_sync: bool,
+) -> base.Case:
+    test_name = base.create_test_name(__file__)
+    event_name = base.create_event_name(framework, test_name)
+    fn_id = base.create_fn_id(test_name)
+    state = _State()
+    mw = inngest_remote_state.RemoteStateMiddleware.factory(
+        inngest_remote_state.in_memory.InMemoryDriver()
+    )
+
+    @client.create_function(
+        fn_id=fn_id,
+        middleware=[mw],
+        retries=0,
+        trigger=inngest.TriggerEvent(event=event_name),
+    )
+    def fn_sync(
+        ctx: inngest.Context,
+        step: inngest.StepSync,
+    ) -> str:
+        state.run_id = ctx.run_id
+
+        def _step_1() -> str:
+            return "test string"
+
+        step_1_output = step.run("step_1", _step_1)
+        assert step_1_output == "test string"
+
+        def _step_2() -> list[inngest.JSON]:
+            return [{"a": {"b": 1}}]
+
+        step_2_output = step.run("step_2", _step_2)
+        assert step_2_output == [{"a": {"b": 1}}]
+
+        step.sleep("zzz", datetime.timedelta(seconds=1))
+
+        step.wait_for_event(
+            "wait",
+            event="never",
+            timeout=datetime.timedelta(seconds=1),
+        )
+
+        return "function output"
+
+    @client.create_function(
+        fn_id=fn_id,
+        middleware=[mw],
+        retries=0,
+        trigger=inngest.TriggerEvent(event=event_name),
+    )
+    async def fn_async(
+        ctx: inngest.Context,
+        step: inngest.Step,
+    ) -> str:
+        state.run_id = ctx.run_id
+
+        def _step_1() -> str:
+            return "test string"
+
+        step_1_output = await step.run("step_1", _step_1)
+        assert step_1_output == "test string"
+
+        def _step_2() -> list[inngest.JSON]:
+            return [{"a": {"b": 1}}]
+
+        step_2_output = await step.run("step_2", _step_2)
+        assert step_2_output == [{"a": {"b": 1}}]
+
+        await step.sleep("zzz", datetime.timedelta(seconds=1))
+
+        await step.wait_for_event(
+            "wait",
+            event="never",
+            timeout=datetime.timedelta(seconds=1),
+        )
+
+        return "function output"
+
+    async def run_test(self: base.TestClass) -> None:
+        self.client.send_sync(inngest.Event(name=event_name))
+
+        run_id = await state.wait_for_run_id()
+        run = await test_core.helper.client.wait_for_run_status(
+            run_id,
+            test_core.helper.RunStatus.COMPLETED,
+        )
+
+        # Ensure that step_1 output is remotely stored.
+        output = json.loads(
+            await test_core.helper.client.get_step_output(
+                run_id=run_id,
+                step_id="step_1",
+            )
+        )
+        assert isinstance(output, dict)
+        data = output.get("data")
+        assert isinstance(data, dict)
+
+        # Ensure that step_2 output is remotely stored.
+        output = json.loads(
+            await test_core.helper.client.get_step_output(
+                run_id=run_id,
+                step_id="step_2",
+            )
+        )
+        assert isinstance(output, dict)
+        data = output.get("data")
+        assert isinstance(data, dict)
+
+        assert run.output is not None
+        assert json.loads(run.output) == "function output"
+
+    if is_sync:
+        fn = fn_sync
+    else:
+        fn = fn_async
+
+    return base.Case(
+        fn=fn,
+        run_test=run_test,
+        name=test_name,
+    )
diff --git a/tests/test_inngest_remote_state/cases/step_output_s3.py b/tests/test_inngest_remote_state/cases/step_output_s3.py
new file mode 100644
index 00000000..7fac44d8
--- /dev/null
+++ b/tests/test_inngest_remote_state/cases/step_output_s3.py
@@ -0,0 +1,178 @@
+"""
+Ensure step and function output is encrypted and decrypted correctly
+"""
+
+import datetime
+import json
+
+import boto3
+import inngest
+import inngest_remote_state
+import inngest_remote_state.s3
+import moto
+import moto.server
+import test_core.helper
+from inngest._internal import server_lib
+from test_core import net
+
+from . import base
+
+
+class _State(base.BaseState):
+    event: inngest.Event
+    events: list[inngest.Event]
+
+
+def create(
+    client: inngest.Inngest,
+    framework: server_lib.Framework,
+    is_sync: bool,
+) -> base.Case:
+    test_name = base.create_test_name(__file__)
+    event_name = base.create_event_name(framework, test_name)
+    fn_id = base.create_fn_id(test_name)
+    state = _State()
+
+    aws_port = net.get_available_port()
+    aws_url = f"http://localhost:{aws_port}"
+    aws_access_key_id = "test"
+    aws_secret_access_key = "test"
+    aws_region = "us-east-1"
+    s3_bucket = "inngest"
+
+    s3_client = boto3.client(
+        "s3",
+        endpoint_url=aws_url,
+        region_name=aws_region,
+        aws_access_key_id=aws_access_key_id,
+        aws_secret_access_key=aws_secret_access_key,
+    )
+
+    driver = inngest_remote_state.s3.S3Driver(
+        bucket=s3_bucket,
+        client=s3_client,
+    )
+
+    mw = inngest_remote_state.RemoteStateMiddleware.factory(driver)
+
+    @client.create_function(
+        fn_id=fn_id,
+        middleware=[mw],
+        retries=0,
+        trigger=inngest.TriggerEvent(event=event_name),
+    )
+    def fn_sync(
+        ctx: inngest.Context,
+        step: inngest.StepSync,
+    ) -> str:
+        state.run_id = ctx.run_id
+
+        def _step_1() -> str:
+            return "test string"
+
+        step_1_output = step.run("step_1", _step_1)
+        assert step_1_output == "test string"
+
+        def _step_2() -> list[inngest.JSON]:
+            return [{"a": {"b": 1}}]
+
+        step_2_output = step.run("step_2", _step_2)
+        assert step_2_output == [{"a": {"b": 1}}]
+
+        step.sleep("zzz", datetime.timedelta(seconds=1))
+
+        step.wait_for_event(
+            "wait",
+            event="never",
+            timeout=datetime.timedelta(seconds=1),
+        )
+
+        return "function output"
+
+    @client.create_function(
+        fn_id=fn_id,
+        middleware=[mw],
+        retries=0,
+        trigger=inngest.TriggerEvent(event=event_name),
+    )
+    async def fn_async(
+        ctx: inngest.Context,
+        step: inngest.Step,
+    ) -> str:
+        state.run_id = ctx.run_id
+
+        def _step_1() -> str:
+            return "test string"
+
+        step_1_output = await step.run("step_1", _step_1)
+        assert step_1_output == "test string"
+
+        def _step_2() -> list[inngest.JSON]:
+            return [{"a": {"b": 1}}]
+
+        step_2_output = await step.run("step_2", _step_2)
+        assert step_2_output == [{"a": {"b": 1}}]
+
+        await step.sleep("zzz", datetime.timedelta(seconds=1))
+
+        await step.wait_for_event(
+            "wait",
+            event="never",
+            timeout=datetime.timedelta(seconds=1),
+        )
+
+        return "function output"
+
+    async def run_test(self: base.TestClass) -> None:
+        aws_server = moto.server.ThreadedMotoServer(port=aws_port)
+        aws_server.start()
+
+        s3_client.create_bucket(Bucket=s3_bucket)
+
+        self.client.send_sync(inngest.Event(name=event_name))
+        run_id = await state.wait_for_run_id()
+        run = await test_core.helper.client.wait_for_run_status(
+            run_id,
+            test_core.helper.RunStatus.COMPLETED,
+        )
+
+        output = json.loads(
+            await test_core.helper.client.get_step_output(
+                run_id=run_id,
+                step_id="step_1",
+            )
+        )
+
+        assert isinstance(output, dict)
+        data = output.get("data")
+        assert isinstance(data, dict)
+
+        # Ensure the step output is remotely stored.
+        assert driver._marker in data
+
+        output = json.loads(
+            await test_core.helper.client.get_step_output(
+                run_id=run_id,
+                step_id="step_2",
+            )
+        )
+        assert isinstance(output, dict)
+        data = output.get("data")
+        assert isinstance(data, dict)
+
+        # Ensure the step output is remotely stored.
+        assert driver._marker in data
+
+        assert run.output is not None
+        assert json.loads(run.output) == "function output"
+
+    if is_sync:
+        fn = fn_sync
+    else:
+        fn = fn_async
+
+    return base.Case(
+        fn=fn,
+        run_test=run_test,
+        name=test_name,
+    )
diff --git a/tests/test_inngest_remote_state/conftest.py b/tests/test_inngest_remote_state/conftest.py
new file mode 100644
index 00000000..bd340d12
--- /dev/null
+++ b/tests/test_inngest_remote_state/conftest.py
@@ -0,0 +1,10 @@
+import pytest
+from inngest.experimental import dev_server
+
+
+def pytest_configure(config: pytest.Config) -> None:
+    dev_server.server.start()
+
+
+def pytest_unconfigure(config: pytest.Config) -> None:
+    dev_server.server.stop()
diff --git a/tests/test_inngest_remote_state/test_fast_api.py b/tests/test_inngest_remote_state/test_fast_api.py
new file mode 100644
index 00000000..d3b14056
--- /dev/null
+++ b/tests/test_inngest_remote_state/test_fast_api.py
@@ -0,0 +1,73 @@
+import threading
+import unittest
+
+import fastapi
+import fastapi.testclient
+import inngest
+import inngest.fast_api
+import uvicorn
+from inngest._internal import server_lib
+from inngest.experimental import dev_server
+from test_core import base, net
+
+from . import cases
+
+_framework = server_lib.Framework.FAST_API
+_app_id = f"{_framework.value}-remote-state-middleware"
+
+_client = inngest.Inngest(
+    api_base_url=dev_server.server.origin,
+    app_id=_app_id,
+    event_api_base_url=dev_server.server.origin,
+    is_production=False,
+)
+
+_cases = cases.create_sync_cases(_client, _framework)
+_fns: list[inngest.Function] = []
+for case in _cases:
+    if isinstance(case.fn, list):
+        _fns.extend(case.fn)
+    else:
+        _fns.append(case.fn)
+
+
+class TestRemoteStateMiddleware(unittest.IsolatedAsyncioTestCase):
+    client = _client
+    app_thread: threading.Thread
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        super().setUpClass()
+
+        port = net.get_available_port()
+
+        def start_app() -> None:
+            app = fastapi.FastAPI()
+            inngest.fast_api.serve(
+                app,
+                _client,
+                _fns,
+            )
+            uvicorn.run(app, host="0.0.0.0", port=port, log_level="warning")
+
+        # Start FastAPI in a thread instead of using their test client, since
+        # their test client doesn't seem to actually run requests in parallel
+        # (this is evident in the flakiness of our asyncio race test). If we fix
+        # this issue, we can go back to their test client
+        cls.app_thread = threading.Thread(daemon=True, target=start_app)
+        cls.app_thread.start()
+        base.register(port)
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        super().tearDownClass()
+        cls.app_thread.join(timeout=1)
+
+
+for case in _cases:
+    test_name = f"test_{case.name}"
+    setattr(TestRemoteStateMiddleware, test_name, case.run_test)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/tests/test_inngest_remote_state/test_flask.py b/tests/test_inngest_remote_state/test_flask.py
new file mode 100644
index 00000000..cfb1a685
--- /dev/null
+++ b/tests/test_inngest_remote_state/test_flask.py
@@ -0,0 +1,84 @@
+import typing
+import unittest
+
+import flask
+import flask.logging
+import flask.testing
+import inngest
+import inngest.flask
+from inngest._internal import server_lib
+from inngest.experimental import dev_server
+from test_core import base, http_proxy
+
+from . import cases
+
+_framework = server_lib.Framework.FLASK
+_app_id = f"{_framework.value}-remote-state-middleware"
+
+_client = inngest.Inngest(
+    api_base_url=dev_server.server.origin,
+    app_id=_app_id,
+    event_api_base_url=dev_server.server.origin,
+    is_production=False,
+)
+
+_cases = cases.create_sync_cases(_client, _framework)
+_fns: list[inngest.Function] = []
+for case in _cases:
+    if isinstance(case.fn, list):
+        _fns.extend(case.fn)
+    else:
+        _fns.append(case.fn)
+
+
+class TestRemoteStateMiddleware(unittest.IsolatedAsyncioTestCase):
+    app: flask.testing.FlaskClient
+    client: inngest.Inngest
+    dev_server_port: int
+    proxy: http_proxy.Proxy
+
+    @classmethod
+    def setUpClass(cls) -> None:
+        super().setUpClass()
+        app = flask.Flask(__name__)
+        cls.client = _client
+
+        inngest.flask.serve(
+            app,
+            cls.client,
+            _fns,
+        )
+        cls.app = app.test_client()
+        cls.proxy = http_proxy.Proxy(cls.on_proxy_request).start()
+        base.register(cls.proxy.port)
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        super().tearDownClass()
+        cls.proxy.stop()
+
+    @classmethod
+    def on_proxy_request(
+        cls,
+        *,
+        body: typing.Optional[bytes],
+        headers: dict[str, list[str]],
+        method: str,
+        path: str,
+    ) -> http_proxy.Response:
+        return http_proxy.on_proxy_flask_request(
+            cls.app,
+            body=body,
+            headers=headers,
+            method=method,
+            path=path,
+        )
+
+
+for case in _cases:
+    test_name = f"test_{case.name}"
+    setattr(TestRemoteStateMiddleware, test_name, case.run_test)
+
+
+if __name__ == "__main__":
+    unittest.main()