diff --git a/packages/apps/src/microsoft_teams/apps/state/__init__.py b/packages/apps/src/microsoft_teams/apps/state/__init__.py
new file mode 100644
index 00000000..11b1305b
--- /dev/null
+++ b/packages/apps/src/microsoft_teams/apps/state/__init__.py
@@ -0,0 +1,17 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+"""
+
+from .container import TurnStateContainer
+from .loader import TurnStateLoader
+from .options import StateOptions
+from .turn_state import TurnState, TurnStateSealedError
+
+__all__ = [
+    "TurnState",
+    "TurnStateSealedError",
+    "TurnStateContainer",
+    "TurnStateLoader",
+    "StateOptions",
+]
diff --git a/packages/apps/src/microsoft_teams/apps/state/container.py b/packages/apps/src/microsoft_teams/apps/state/container.py
new file mode 100644
index 00000000..52c4c392
--- /dev/null
+++ b/packages/apps/src/microsoft_teams/apps/state/container.py
@@ -0,0 +1,58 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Awaitable, Callable, Optional
+
+from .turn_state import TurnState
+
+_Deleter = Callable[[], Awaitable[None]]
+
+
+@dataclass(kw_only=True)
+class TurnStateContainer:
+    """The state scopes loaded for one turn, together with the identity they
+    were loaded for.
+
+    ``conversation`` is always present. ``user`` is ``None`` when the activity has
+    no ``from`` identity, so there is no per-user scope to load or persist.
+
+    ``conversation_id``/``user_id`` record the identity this container was loaded
+    for. The loader reads them back off the container when saving, so a save can
+    never be told to persist under a different key than it was loaded from.
+
+    Fields are keyword-only so the public constructor is not tied to positional
+    order and can evolve without breaking callers.
+    """
+
+    conversation: TurnState
+    conversation_id: str
+    user: Optional[TurnState] = None
+    user_id: Optional[str] = None
+    _deleter: Optional[_Deleter] = field(default=None, repr=False, compare=False)
+
+    def seal(self) -> None:
+        """Seal every scope so post-turn access raises."""
+        self.conversation.seal()
+        if self.user is not None:
+            self.user.seal()
+
+    async def delete(self) -> None:
+        """Clear both scopes and remove them from the backing store.
+
+        The injected deleter removes the keys immediately, then in-memory scopes
+        are cleared so state reflects the deletion during the current turn.
+        """
+        if self._deleter is None:
+            raise RuntimeError("State deletion is not available. Call UseState() during service registration.")
+
+        await self._deleter()
+        self.conversation.clear()
+        self.conversation.mark_clean()
+        if self.user is not None:
+            self.user.clear()
+            self.user.mark_clean()
diff --git a/packages/apps/src/microsoft_teams/apps/state/loader.py b/packages/apps/src/microsoft_teams/apps/state/loader.py
new file mode 100644
index 00000000..393c2ca5
--- /dev/null
+++ b/packages/apps/src/microsoft_teams/apps/state/loader.py
@@ -0,0 +1,162 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import Any, Dict, Optional, cast
+from urllib.parse import quote
+
+from microsoft_teams.common import Storage
+
+from .container import TurnStateContainer
+from .options import StateOptions
+from .turn_state import TurnState
+
+logger = logging.getLogger(__name__)
+
+
+class TurnStateLoader:
+    """Loads and persists :class:`TurnState` scopes over a ``Storage`` backend.
+
+    Values are stored as JSON **strings** so any ``Storage`` implementation works
+    regardless of how it serializes values. Expiry and other write behavior are
+    configured directly on the selected storage provider.
+    """
+
+    def __init__(self, storage: Optional[Storage[str, Any]] = None, options: Optional[StateOptions] = None) -> None:
+        self._options = options or StateOptions()
+        resolved = storage if storage is not None else self._options.storage
+        if resolved is None:
+            raise ValueError("TurnStateLoader requires a Storage backend (pass one explicitly or via StateOptions).")
+        self._storage: Storage[str, Any] = resolved
+
+    @property
+    def options(self) -> StateOptions:
+        return self._options
+
+    def conversation_key(self, conversation_id: str) -> str:
+        """Key for the conversation-scoped blob."""
+        return f"{self._options.key_prefix}:conv:{quote(conversation_id, safe='')}"
+
+    def user_key(self, conversation_id: str, user_id: str) -> str:
+        """Key for the user-scoped blob."""
+        return f"{self._options.key_prefix}:user:{quote(conversation_id, safe='')}:{quote(user_id, safe='')}"
+
+    async def load(self, conversation_id: str, user_id: Optional[str] = None) -> TurnStateContainer:
+        """Load both scopes for the turn. ``user`` is ``None`` when ``user_id`` is."""
+        conversation = await self._load_scope(self.conversation_key(conversation_id))
+
+        user: Optional[TurnState] = None
+        if user_id is not None:
+            user = await self._load_scope(self.user_key(conversation_id, user_id))
+
+        async def _delete() -> None:
+            await self.delete(conversation_id, user_id)
+
+        return TurnStateContainer(
+            conversation=conversation,
+            user=user,
+            conversation_id=conversation_id,
+            user_id=user_id,
+            _deleter=_delete,
+        )
+
+    async def save(self, container: TurnStateContainer) -> None:
+        """Persist dirty scopes under the identity the container was loaded for.
+
+        Identity is read off the container (``conversation_id``/``user_id``), so a
+        save always targets the same keys the container was loaded from.
+        Empty-but-dirty scopes are deleted.
+        """
+        if not container.conversation_id:
+            raise ValueError("TurnStateContainer.conversation_id must be set to save state.")
+        if container.user is not None and not container.user_id:
+            raise ValueError("TurnStateContainer.user_id must be set to save user state.")
+
+        pending_deletes: list[str] = []
+        pending_sets: list[tuple[str, str]] = []
+        pending_clean: list[TurnState] = []
+        self._prepare_scope_save(
+            self.conversation_key(container.conversation_id),
+            container.conversation,
+            pending_deletes,
+            pending_sets,
+            pending_clean,
+        )
+        if container.user is not None and container.user_id is not None:
+            self._prepare_scope_save(
+                self.user_key(container.conversation_id, container.user_id),
+                container.user,
+                pending_deletes,
+                pending_sets,
+                pending_clean,
+            )
+
+        for key in pending_deletes:
+            await self._storage.async_delete(key)
+        for key, value in pending_sets:
+            await self._storage.async_set(key, value)
+        for scope in pending_clean:
+            scope.mark_clean()
+
+    async def delete(self, conversation_id: str, user_id: Optional[str] = None) -> None:
+        """Delete both scope blobs for the turn's identity."""
+        await self._storage.async_delete(self.conversation_key(conversation_id))
+        if user_id is not None:
+            await self._storage.async_delete(self.user_key(conversation_id, user_id))
+
+    async def _load_scope(self, key: str) -> TurnState:
+        raw = await self._storage.async_get(key)
+        if raw is None:
+            return TurnState()
+
+        data = self._deserialize(raw)
+        if data is None:
+            await self._storage.async_delete(key)
+            return TurnState()
+
+        return TurnState(data)
+
+    def _prepare_scope_save(
+        self,
+        key: str,
+        scope: TurnState,
+        pending_deletes: list[str],
+        pending_sets: list[tuple[str, str]],
+        pending_clean: list[TurnState],
+    ) -> None:
+        if not scope.is_dirty:
+            return
+        if scope.is_empty:
+            pending_deletes.append(key)
+            pending_clean.append(scope)
+            return
+        pending_sets.append((key, json.dumps(scope.to_dict())))
+        pending_clean.append(scope)
+
+    def _deserialize(self, raw: Any) -> Optional[Dict[str, Any]]:
+        """Parse a stored blob.
+
+        Never raises: unreadable or malformed blobs are treated as missing.
+        """
+        if isinstance(raw, dict):
+            parsed: Any = cast(Dict[Any, Any], raw)
+        elif isinstance(raw, str):
+            try:
+                parsed = json.loads(raw)
+            except ValueError:
+                logger.debug("Discarding unreadable state blob at load")
+                return None
+        else:
+            return None
+
+        if not isinstance(parsed, dict):
+            return None
+        mapping = cast(Dict[object, Any], parsed)
+        if not all(isinstance(key, str) for key in mapping):
+            return None
+        return cast(Dict[str, Any], mapping)
diff --git a/packages/apps/src/microsoft_teams/apps/state/options.py b/packages/apps/src/microsoft_teams/apps/state/options.py
new file mode 100644
index 00000000..a19d3a3a
--- /dev/null
+++ b/packages/apps/src/microsoft_teams/apps/state/options.py
@@ -0,0 +1,26 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Optional
+
+from microsoft_teams.common import Storage
+
+
+@dataclass(frozen=True)
+class StateOptions:
+    """Configuration for the per-turn state layer.
+
+    Scope keys are namespaced under ``key_prefix``. Storage-specific behavior,
+    including expiry, is configured directly on the selected storage provider.
+    """
+
+    storage: Optional[Storage[str, Any]] = None
+    """Backing store for state blobs. When ``None`` the loader must be given one."""
+
+    key_prefix: str = "ts"
+    """Namespace prefix for scope keys (``{prefix}:conv:...`` / ``{prefix}:user:...``)."""
diff --git a/packages/apps/src/microsoft_teams/apps/state/turn_state.py b/packages/apps/src/microsoft_teams/apps/state/turn_state.py
new file mode 100644
index 00000000..ab9fe7dc
--- /dev/null
+++ b/packages/apps/src/microsoft_teams/apps/state/turn_state.py
@@ -0,0 +1,114 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from collections.abc import Iterator, Mapping, MutableMapping
+from typing import Any, Dict, Optional
+
+
+class TurnStateSealedError(RuntimeError):
+    """Raised when a sealed :class:`TurnState` is accessed after its turn ends."""
+
+
+class TurnState(MutableMapping[str, Any]):
+    """One state scope for a single turn.
+
+    Behaves like a ``dict`` but adds two things the loader relies on:
+
+    * **Dirty tracking** — the loader compares the current contents to the
+      loaded snapshot, so nested mutations are persisted without dirtying reads.
+    * **Sealing** — at the end of a turn the scope is sealed; any later access
+      raises :class:`TurnStateSealedError`.
+
+    **Values must be JSON-serializable.** Each scope is encoded with
+    ``json.dumps`` when it is saved, so store only JSON-native types (``str``,
+    ``int``, ``float``, ``bool``, ``None``, ``list``, ``dict``). A non-serializable
+    value (e.g. a ``datetime`` or a custom object) is accepted on assignment but
+    raises ``TypeError`` later, when the turn is saved.
+    """
+
+    def __init__(self, data: Optional[Mapping[str, Any]] = None) -> None:
+        self._data: Dict[str, Any] = dict(data) if data else {}
+        self._baseline = self._try_fingerprint(self._data)
+        self._sealed = False
+
+    @property
+    def is_dirty(self) -> bool:
+        """Whether the scope has been mutated since it was loaded."""
+        fingerprint = self._try_fingerprint(self._data)
+        if fingerprint is None:
+            return True
+        return fingerprint != self._baseline
+
+    @property
+    def is_empty(self) -> bool:
+        """Whether the scope currently holds no keys."""
+        return not self._data
+
+    @property
+    def is_sealed(self) -> bool:
+        """Whether the scope has been sealed for the turn."""
+        return self._sealed
+
+    def seal(self) -> None:
+        """Seal the scope; subsequent access raises :class:`TurnStateSealedError`."""
+        self._sealed = True
+
+    def mark_clean(self) -> None:
+        """Mark the current contents as clean after a successful save."""
+        fingerprint = self._try_fingerprint(self._data)
+        if fingerprint is not None:
+            self._baseline = fingerprint
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Return a shallow copy of the raw contents (used for serialization).
+
+        Intentionally does not check the seal: the loader serializes a scope just
+        before sealing it, and callers should not reach for this directly.
+        """
+        return dict(self._data)
+
+    def _ensure_active(self) -> None:
+        if self._sealed:
+            raise TurnStateSealedError("TurnState has been sealed and can no longer be accessed.")
+
+    @staticmethod
+    def _try_fingerprint(data: Mapping[str, Any]) -> Optional[str]:
+        try:
+            canonical = json.dumps(data, sort_keys=True, default=repr, separators=(",", ":"))
+        except (TypeError, ValueError):
+            return None
+        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
+
+    def __getitem__(self, key: str) -> Any:
+        self._ensure_active()
+        return self._data[key]
+
+    def __setitem__(self, key: str, value: Any) -> None:
+        self._ensure_active()
+        self._data[key] = value
+
+    def __delitem__(self, key: str) -> None:
+        self._ensure_active()
+        del self._data[key]
+
+    def __iter__(self) -> Iterator[str]:
+        self._ensure_active()
+        # Snapshot so callers can mutate the scope while iterating (e.g. clear()).
+        return iter(list(self._data))
+
+    def __len__(self) -> int:
+        return len(self._data)
+
+    def __contains__(self, key: object) -> bool:
+        self._ensure_active()
+        return key in self._data
+
+    def __repr__(self) -> str:
+        status = "sealed" if self._sealed else ("dirty" if self.is_dirty else "clean")
+        return f"TurnState({self._data!r}, {status})"
diff --git a/packages/apps/tests/test_state.py b/packages/apps/tests/test_state.py
new file mode 100644
index 00000000..544744fa
--- /dev/null
+++ b/packages/apps/tests/test_state.py
@@ -0,0 +1,371 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+"""
+
+import json
+from typing import Any
+
+import pytest
+from microsoft_teams.apps.state import (
+    StateOptions,
+    TurnState,
+    TurnStateContainer,
+    TurnStateLoader,
+    TurnStateSealedError,
+)
+from microsoft_teams.common import LocalStorage
+
+# ---------------------------------------------------------------------------
+# TurnState
+# ---------------------------------------------------------------------------
+
+
+class TestTurnState:
+    def test_starts_clean_and_empty(self):
+        state = TurnState()
+        assert state.is_dirty is False
+        assert state.is_empty is True
+        assert len(state) == 0
+
+    def test_seeded_data_is_clean(self):
+        state = TurnState({"a": 1})
+        assert state.is_dirty is False
+        assert state.is_empty is False
+        assert state["a"] == 1
+
+    def test_set_marks_dirty(self):
+        state = TurnState()
+        state["x"] = 1
+        assert state.is_dirty is True
+        assert state.is_empty is False
+
+    def test_delete_marks_dirty(self):
+        state = TurnState({"x": 1})
+        del state["x"]
+        assert state.is_dirty is True
+        assert state.is_empty is True
+
+    def test_read_does_not_mark_dirty(self):
+        state = TurnState({"x": 1})
+        _ = state["x"]
+        _ = "x" in state
+        _ = list(state)
+        _ = len(state)
+        assert state.is_dirty is False
+
+    def test_mapping_protocol(self):
+        state = TurnState()
+        state.update({"a": 1, "b": 2})
+        assert dict(state) == {"a": 1, "b": 2}
+        assert sorted(state) == ["a", "b"]
+        assert state.get("missing") is None
+        assert state.pop("a") == 1
+        assert "a" not in state
+
+    def test_to_dict_returns_copy(self):
+        state = TurnState({"a": 1})
+        snapshot = state.to_dict()
+        snapshot["a"] = 999
+        assert state["a"] == 1  # original untouched
+
+    def test_nested_dict_mutation_marks_dirty(self):
+        state = TurnState({"oauth": {"github": {"pending": False}}})
+        state["oauth"]["github"]["pending"] = True
+        assert state.is_dirty is True
+
+    def test_nested_list_mutation_marks_dirty(self):
+        state = TurnState({"items": [1, 2]})
+        state["items"].append(3)
+        assert state.is_dirty is True
+
+    def test_mutate_then_revert_is_clean(self):
+        state = TurnState({"x": 1})
+        state["x"] = 2
+        assert state.is_dirty is True
+        state["x"] = 1
+        assert state.is_dirty is False
+
+    def test_circular_value_is_dirty_without_raising(self):
+        state = TurnState()
+        value: dict[str, object] = {}
+        value["self"] = value
+
+        state["value"] = value
+
+        assert state.is_dirty is True
+
+    def test_seal_blocks_access(self):
+        state = TurnState({"a": 1})
+        state.seal()
+        assert state.is_sealed is True
+        with pytest.raises(TurnStateSealedError):
+            _ = state["a"]
+        with pytest.raises(TurnStateSealedError):
+            state["b"] = 2
+        with pytest.raises(TurnStateSealedError):
+            del state["a"]
+        with pytest.raises(TurnStateSealedError):
+            _ = "a" in state
+        with pytest.raises(TurnStateSealedError):
+            _ = list(state)
+
+    def test_seal_still_allows_metadata(self):
+        state = TurnState({"a": 1})
+        state["b"] = 2
+        state.seal()
+        # Diagnostics remain readable after sealing.
+        assert state.is_sealed is True
+        assert state.is_dirty is True
+        assert state.is_empty is False
+        assert len(state) == 2
+
+
+# ---------------------------------------------------------------------------
+# TurnStateContainer
+# ---------------------------------------------------------------------------
+
+
+class TestTurnStateContainer:
+    def test_seal_seals_both_scopes(self):
+        container = TurnStateContainer(conversation=TurnState(), conversation_id="c1", user=TurnState())
+        container.seal()
+        assert container.conversation.is_sealed
+        assert container.user is not None and container.user.is_sealed
+
+    def test_seal_tolerates_missing_user(self):
+        container = TurnStateContainer(conversation=TurnState(), conversation_id="c1", user=None)
+        container.seal()  # must not raise
+        assert container.conversation.is_sealed
+
+    async def test_delete_clears_scopes_and_calls_deleter(self):
+        calls = []
+
+        async def deleter():
+            calls.append(True)
+
+        container = TurnStateContainer(
+            conversation=TurnState({"a": 1}),
+            conversation_id="c1",
+            user=TurnState({"b": 2}),
+            _deleter=deleter,
+        )
+        await container.delete()
+        assert container.conversation.is_empty
+        assert container.conversation.is_dirty is False
+        assert container.user is not None and container.user.is_empty
+        assert container.user.is_dirty is False
+        assert calls == [True]
+
+    async def test_delete_without_deleter_raises(self):
+        container = TurnStateContainer(
+            conversation=TurnState({"a": 1}),
+            conversation_id="c1",
+            user=TurnState({"b": 2}),
+        )
+
+        with pytest.raises(RuntimeError, match="State deletion is not available"):
+            await container.delete()
+
+        assert container.conversation["a"] == 1
+        assert container.user is not None and container.user["b"] == 2
+
+
+# ---------------------------------------------------------------------------
+# TurnStateLoader
+# ---------------------------------------------------------------------------
+
+
+class TestTurnStateLoader:
+    def test_requires_a_storage_backend(self):
+        with pytest.raises(ValueError):
+            TurnStateLoader()
+
+    def test_key_layout_matches_csharp(self):
+        loader = TurnStateLoader(LocalStorage())
+        assert loader.conversation_key("c1") == "ts:conv:c1"
+        assert loader.user_key("c1", "u1") == "ts:user:c1:u1"
+
+    def test_key_segments_are_escaped(self):
+        loader = TurnStateLoader(LocalStorage())
+        assert loader.conversation_key("c:1;tenant=a") == "ts:conv:c%3A1%3Btenant%3Da"
+        assert loader.user_key("c:1", "u;1=a/b") == "ts:user:c%3A1:u%3B1%3Da%2Fb"
+
+    def test_key_prefix_is_configurable(self):
+        loader = TurnStateLoader(LocalStorage(), StateOptions(key_prefix="mybot"))
+        assert loader.conversation_key("c1") == "mybot:conv:c1"
+
+    async def test_load_missing_returns_empty_scopes(self):
+        loader = TurnStateLoader(LocalStorage())
+        container = await loader.load("c1", "u1")
+        assert container.conversation.is_empty
+        assert container.user is not None and container.user.is_empty
+
+    async def test_load_without_user_id_has_no_user_scope(self):
+        loader = TurnStateLoader(LocalStorage())
+        container = await loader.load("c1")
+        assert container.user is None
+
+    async def test_round_trip(self):
+        storage = LocalStorage()
+        loader = TurnStateLoader(storage)
+        container = await loader.load("c1", "u1")
+        container.conversation["greeted"] = True
+        assert container.user is not None
+        container.user["step"] = 3
+        await loader.save(container)
+
+        reloaded = await loader.load("c1", "u1")
+        assert reloaded.conversation["greeted"] is True
+        assert reloaded.user is not None and reloaded.user["step"] == 3
+
+    async def test_save_marks_saved_scopes_clean(self):
+        storage = LocalStorage()
+        loader = TurnStateLoader(storage)
+        container = await loader.load("c1", "u1")
+        container.conversation["saved"] = True
+        assert container.user is not None
+        container.user["saved"] = True
+
+        await loader.save(container)
+
+        assert container.conversation.is_dirty is False
+        assert container.user.is_dirty is False
+
+    async def test_save_surfaces_circular_value_during_serialization(self):
+        storage = LocalStorage()
+        loader = TurnStateLoader(storage)
+        container = await loader.load("c1")
+        value: dict[str, object] = {}
+        value["self"] = value
+        container.conversation["value"] = value
+
+        with pytest.raises(ValueError):
+            await loader.save(container)
+
+        assert storage.get("ts:conv:c1") is None
+        assert container.conversation.is_dirty is True
+
+    async def test_nested_mutation_persists(self):
+        storage = LocalStorage()
+        loader = TurnStateLoader(storage)
+        container = await loader.load("c1")
+        container.conversation["oauth"] = {"github": {"pending": False}}
+        await loader.save(container)
+
+        again = await loader.load("c1")
+        again.conversation["oauth"]["github"]["pending"] = True
+        await loader.save(again)
+
+        reloaded = await loader.load("c1")
+        assert reloaded.conversation["oauth"]["github"]["pending"] is True
+
+    async def test_save_persists_json_string(self):
+        storage = LocalStorage()
+        loader = TurnStateLoader(storage)
+        container = await loader.load("c1")
+        container.conversation["k"] = "v"
+        await loader.save(container)
+
+        stored = storage.get("ts:conv:c1")
+        assert isinstance(stored, str)  # design §13.1: always a str
+        parsed = json.loads(stored)
+        assert parsed == {"k": "v"}
+
+    async def test_clean_scope_is_not_written(self):
+        storage = LocalStorage()
+        loader = TurnStateLoader(storage)
+        container = await loader.load("c1")
+        # never mutated -> nothing written
+        await loader.save(container)
+        assert storage.get("ts:conv:c1") is None
+
+    async def test_save_serializes_all_scopes_before_writing(self):
+        storage = LocalStorage()
+        loader = TurnStateLoader(storage)
+        container = await loader.load("c1", "u1")
+        container.conversation["value"] = "old"
+        assert container.user is not None
+        container.user["value"] = "old"
+        await loader.save(container)
+
+        again = await loader.load("c1", "u1")
+        again.conversation["value"] = "new"
+        assert again.user is not None
+        again.user["bad"] = object()
+
+        with pytest.raises(TypeError):
+            await loader.save(again)
+
+        reloaded = await loader.load("c1", "u1")
+        assert reloaded.conversation["value"] == "old"
+        assert reloaded.user is not None and reloaded.user["value"] == "old"
+
+    async def test_emptied_scope_is_deleted(self):
+        storage = LocalStorage()
+        loader = TurnStateLoader(storage)
+        # seed an existing blob
+        container = await loader.load("c1")
+        container.conversation["k"] = "v"
+        await loader.save(container)
+        assert storage.get("ts:conv:c1") is not None
+
+        # now empty it and save -> key removed
+        again = await loader.load("c1")
+        del again.conversation["k"]
+        await loader.save(again)
+        assert storage.get("ts:conv:c1") is None
+
+    async def test_delete_removes_both_keys(self):
+        storage = LocalStorage()
+        loader = TurnStateLoader(storage)
+        container = await loader.load("c1", "u1")
+        container.conversation["a"] = 1
+        assert container.user is not None
+        container.user["b"] = 2
+        await loader.save(container)
+        assert storage.get("ts:conv:c1") is not None
+        assert storage.get("ts:user:c1:u1") is not None
+
+        await loader.delete("c1", "u1")
+        assert storage.get("ts:conv:c1") is None
+        assert storage.get("ts:user:c1:u1") is None
+
+    async def test_corrupt_blob_loads_as_empty(self):
+        storage = LocalStorage()
+        await storage.async_set("ts:conv:c1", "not-json{{{")
+        loader = TurnStateLoader(storage)
+        container = await loader.load("c1")
+        assert container.conversation.is_empty
+        assert storage.get("ts:conv:c1") is None
+
+    async def test_non_mapping_blob_loads_as_empty_and_is_deleted(self):
+        storage = LocalStorage()
+        await storage.async_set("ts:conv:c1", json.dumps(["not", "state"]))
+        loader = TurnStateLoader(storage)
+        container = await loader.load("c1")
+        assert container.conversation.is_empty
+        assert storage.get("ts:conv:c1") is None
+
+    async def test_already_deserialized_dict_loads_normally(self):
+        storage: LocalStorage[Any] = LocalStorage()
+        await storage.async_set("ts:conv:c1", {"a": 1})
+        loader = TurnStateLoader(storage)
+        container = await loader.load("c1")
+        assert container.conversation["a"] == 1
+
+    async def test_dict_with_non_string_key_is_deleted(self):
+        storage: LocalStorage[Any] = LocalStorage()
+        await storage.async_set("ts:conv:c1", {1: "not state"})
+        loader = TurnStateLoader(storage)
+        container = await loader.load("c1")
+        assert container.conversation.is_empty
+        assert storage.get("ts:conv:c1") is None
+
+    async def test_storage_from_options_is_used(self):
+        storage = LocalStorage()
+        loader = TurnStateLoader(options=StateOptions(storage=storage))
+        container = await loader.load("c1")
+        container.conversation["k"] = "v"
+        await loader.save(container)
+        assert storage.get("ts:conv:c1") is not None