From 9fb4537f5ce98a511da91a613059ec8584454d76 Mon Sep 17 00:00:00 2001
From: lilyydu <54044854+lilyydu@users.noreply.github.com>
Date: Mon, 10 Aug 2026 13:54:19 -0700
Subject: [PATCH 1/6] add state layer

---
 .../microsoft_teams/apps/state/__init__.py    |  17 ++
 .../microsoft_teams/apps/state/container.py   |  51 ++++
 .../src/microsoft_teams/apps/state/loader.py  | 125 ++++++++
 .../src/microsoft_teams/apps/state/options.py |  30 ++
 .../microsoft_teams/apps/state/turn_state.py  |  97 +++++++
 packages/apps/tests/test_state.py             | 266 ++++++++++++++++++
 6 files changed, 586 insertions(+)
 create mode 100644 packages/apps/src/microsoft_teams/apps/state/__init__.py
 create mode 100644 packages/apps/src/microsoft_teams/apps/state/container.py
 create mode 100644 packages/apps/src/microsoft_teams/apps/state/loader.py
 create mode 100644 packages/apps/src/microsoft_teams/apps/state/options.py
 create mode 100644 packages/apps/src/microsoft_teams/apps/state/turn_state.py
 create mode 100644 packages/apps/tests/test_state.py

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..31eccd2e
--- /dev/null
+++ b/packages/apps/src/microsoft_teams/apps/state/container.py
@@ -0,0 +1,51 @@
+"""
+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
+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.
+    """
+
+    conversation: TurnState
+    user: Optional[TurnState] = None
+    conversation_id: str = ""
+    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.
+
+        Clearing marks the scopes empty so a later save is a no-op, and the
+        injected deleter removes the keys immediately.
+        """
+        self.conversation.clear()
+        if self.user is not None:
+            self.user.clear()
+        if self._deleter is not None:
+            await self._deleter()
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..d93b6fcf
--- /dev/null
+++ b/packages/apps/src/microsoft_teams/apps/state/loader.py
@@ -0,0 +1,125 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import time
+from typing import Any, Dict, Optional, cast
+
+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. Each blob carries a save timestamp
+    that powers the loader-applied TTL, since ``Storage`` has no native expiry.
+
+    """
+
+    def __init__(self, storage: Optional[Storage[str, str]] = 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, str] = resolved
+
+    @property
+    def options(self) -> StateOptions:
+        return self._options
+
+    def conversation_key(self, conversation_id: str) -> str:
+        """Key for the conversation-scoped blob (mirrors C#'s ``ts:conv:{id}``)."""
+        return f"{self._options.key_prefix}:conv:{conversation_id}"
+
+    def user_key(self, conversation_id: str, user_id: str) -> str:
+        """Key for the user-scoped blob (mirrors C#'s ``ts:user:{convId}:{userId}``)."""
+        return f"{self._options.key_prefix}:user:{conversation_id}:{user_id}"
+
+    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.
+        """
+        await self._save_scope(self.conversation_key(container.conversation_id), container.conversation)
+        if container.user is not None and container.user_id is not None:
+            await self._save_scope(self.user_key(container.conversation_id, container.user_id), container.user)
+
+    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)
+        return TurnState(self._deserialize(raw))
+
+    async def _save_scope(self, key: str, scope: TurnState) -> None:
+        if not scope.is_dirty:
+            return
+        if scope.is_empty:
+            await self._storage.async_delete(key)
+            return
+        blob = {"ts": time.time(), "data": scope.to_dict()}
+        await self._storage.async_set(key, json.dumps(blob))
+
+    def _deserialize(self, raw: Optional[Any]) -> Dict[str, Any]:
+        """Turn a stored blob back into a scope dict.
+
+        Never raises: an absent, unreadable, or expired blob is treated as an
+        empty scope.
+        """
+        if not raw:
+            return {}
+        try:
+            parsed: Any = json.loads(raw)
+        except (ValueError, TypeError):
+            logger.debug("Discarding unreadable state blob at load")
+            return {}
+
+        if not isinstance(parsed, dict):
+            return {}
+        blob = cast(Dict[str, Any], parsed)
+
+        if self._options.ttl is not None:
+            saved_at = blob.get("ts")
+            if isinstance(saved_at, (int, float)) and (time.time() - saved_at) > self._options.ttl:
+                return {}
+
+        data = blob.get("data")
+        if isinstance(data, dict):
+            return cast(Dict[str, Any], data)
+        return {}
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..400e61d4
--- /dev/null
+++ b/packages/apps/src/microsoft_teams/apps/state/options.py
@@ -0,0 +1,30 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import 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`` and expiry is applied by the loader, since ``Storage``
+    has no native TTL concept.
+    """
+
+    storage: Optional[Storage[str, str]] = 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:...``)."""
+
+    ttl: Optional[int] = None
+    """Optional time-to-live, in **seconds**, applied by the loader on load."""
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..911001a0
--- /dev/null
+++ b/packages/apps/src/microsoft_teams/apps/state/turn_state.py
@@ -0,0 +1,97 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+"""
+
+from __future__ import annotations
+
+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 only writes a scope back when it was
+      mutated, so an untouched scope costs nothing to "save".
+    * **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._dirty = False
+        self._sealed = False
+
+    @property
+    def is_dirty(self) -> bool:
+        """Whether the scope has been mutated since it was loaded."""
+        return self._dirty
+
+    @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 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.")
+
+    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
+        self._dirty = True
+
+    def __delitem__(self, key: str) -> None:
+        self._ensure_active()
+        del self._data[key]
+        self._dirty = True
+
+    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._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..2cd047e9
--- /dev/null
+++ b/packages/apps/tests/test_state.py
@@ -0,0 +1,266 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+"""
+
+import json
+import time
+
+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_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(), 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(), 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}),
+            user=TurnState({"b": 2}),
+            _deleter=deleter,
+        )
+        await container.delete()
+        assert container.conversation.is_empty
+        assert container.user is not None and container.user.is_empty
+        assert calls == [True]
+
+
+# ---------------------------------------------------------------------------
+# 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_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_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["data"] == {"k": "v"}
+        assert "ts" in parsed
+
+    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_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
+
+    async def test_blob_without_data_loads_as_empty(self):
+        storage = LocalStorage()
+        await storage.async_set("ts:conv:c1", json.dumps({"ts": time.time()}))
+        loader = TurnStateLoader(storage)
+        container = await loader.load("c1")
+        assert container.conversation.is_empty
+
+    async def test_expired_blob_loads_as_empty(self):
+        storage = LocalStorage()
+        await storage.async_set(
+            "ts:conv:c1",
+            json.dumps({"ts": time.time() - 500, "data": {"a": 1}}),
+        )
+        loader = TurnStateLoader(storage, StateOptions(ttl=100))
+        container = await loader.load("c1")
+        assert container.conversation.is_empty
+
+    async def test_unexpired_blob_loads_normally(self):
+        storage = LocalStorage()
+        await storage.async_set(
+            "ts:conv:c1",
+            json.dumps({"ts": time.time(), "data": {"a": 1}}),
+        )
+        loader = TurnStateLoader(storage, StateOptions(ttl=100))
+        container = await loader.load("c1")
+        assert container.conversation["a"] == 1
+
+    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

From a81dee0b7c76b897ceaae3d9e6c5a53b773312f1 Mon Sep 17 00:00:00 2001
From: Lily Du <lilyyduu@gmail.com>
Date: Tue, 11 Aug 2026 10:58:59 -0700
Subject: [PATCH 2/6] apply suggestions - fix docstring, hardened checks for
 TTL and conversation_id/user_id

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
---
 packages/apps/src/microsoft_teams/apps/state/container.py | 2 +-
 packages/apps/src/microsoft_teams/apps/state/loader.py    | 8 +++++++-
 2 files changed, 8 insertions(+), 2 deletions(-)

diff --git a/packages/apps/src/microsoft_teams/apps/state/container.py b/packages/apps/src/microsoft_teams/apps/state/container.py
index 31eccd2e..04ee0369 100644
--- a/packages/apps/src/microsoft_teams/apps/state/container.py
+++ b/packages/apps/src/microsoft_teams/apps/state/container.py
@@ -41,7 +41,7 @@ def seal(self) -> None:
     async def delete(self) -> None:
         """Clear both scopes and remove them from the backing store.
 
-        Clearing marks the scopes empty so a later save is a no-op, and the
+        Clearing marks the scopes dirty+empty (so a later save deletes the keys), and the
         injected deleter removes the keys immediately.
         """
         self.conversation.clear()
diff --git a/packages/apps/src/microsoft_teams/apps/state/loader.py b/packages/apps/src/microsoft_teams/apps/state/loader.py
index d93b6fcf..061a48f0 100644
--- a/packages/apps/src/microsoft_teams/apps/state/loader.py
+++ b/packages/apps/src/microsoft_teams/apps/state/loader.py
@@ -73,6 +73,10 @@ async def save(self, container: TurnStateContainer) -> None:
         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.")
         await self._save_scope(self.conversation_key(container.conversation_id), container.conversation)
         if container.user is not None and container.user_id is not None:
             await self._save_scope(self.user_key(container.conversation_id, container.user_id), container.user)
@@ -116,7 +120,9 @@ def _deserialize(self, raw: Optional[Any]) -> Dict[str, Any]:
 
         if self._options.ttl is not None:
             saved_at = blob.get("ts")
-            if isinstance(saved_at, (int, float)) and (time.time() - saved_at) > self._options.ttl:
+            if not isinstance(saved_at, (int, float)):
+                return {}
+            if (time.time() - saved_at) > self._options.ttl:
                 return {}
 
         data = blob.get("data")

From 45a9af7fb33c14831904caadd2161bd7443ab9b9 Mon Sep 17 00:00:00 2001
From: lilyydu <54044854+lilyydu@users.noreply.github.com>
Date: Tue, 11 Aug 2026 11:22:06 -0700
Subject: [PATCH 3/6] refactor(state): make TurnStateContainer kw-only with
 required conversation_id
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

TurnStateContainer is on the public API surface, so its constructor
signature is a compatibility contract. Make all fields keyword-only via
@dataclass(kw_only=True) so fields can be reordered or added later
without breaking callers, and drop the empty-string default on
conversation_id (it is the storage key identity — a silent "" default
would persist under a garbage key instead of failing loudly).

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
---
 packages/apps/src/microsoft_teams/apps/state/container.py | 7 +++++--
 packages/apps/tests/test_state.py                         | 5 +++--
 2 files changed, 8 insertions(+), 4 deletions(-)

diff --git a/packages/apps/src/microsoft_teams/apps/state/container.py b/packages/apps/src/microsoft_teams/apps/state/container.py
index 04ee0369..fccd01d1 100644
--- a/packages/apps/src/microsoft_teams/apps/state/container.py
+++ b/packages/apps/src/microsoft_teams/apps/state/container.py
@@ -13,7 +13,7 @@
 _Deleter = Callable[[], Awaitable[None]]
 
 
-@dataclass
+@dataclass(kw_only=True)
 class TurnStateContainer:
     """The state scopes loaded for one turn, together with the identity they
     were loaded for.
@@ -24,11 +24,14 @@ class TurnStateContainer:
     ``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
-    conversation_id: str = ""
     user_id: Optional[str] = None
     _deleter: Optional[_Deleter] = field(default=None, repr=False, compare=False)
 
diff --git a/packages/apps/tests/test_state.py b/packages/apps/tests/test_state.py
index 2cd047e9..8e0de61e 100644
--- a/packages/apps/tests/test_state.py
+++ b/packages/apps/tests/test_state.py
@@ -102,13 +102,13 @@ def test_seal_still_allows_metadata(self):
 
 class TestTurnStateContainer:
     def test_seal_seals_both_scopes(self):
-        container = TurnStateContainer(conversation=TurnState(), user=TurnState())
+        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(), user=None)
+        container = TurnStateContainer(conversation=TurnState(), conversation_id="c1", user=None)
         container.seal()  # must not raise
         assert container.conversation.is_sealed
 
@@ -120,6 +120,7 @@ async def deleter():
 
         container = TurnStateContainer(
             conversation=TurnState({"a": 1}),
+            conversation_id="c1",
             user=TurnState({"b": 2}),
             _deleter=deleter,
         )

From 85ec3561d4072160780a60d706ca3ca7d3aa9367 Mon Sep 17 00:00:00 2001
From: lilyydu <54044854+lilyydu@users.noreply.github.com>
Date: Wed, 12 Aug 2026 16:45:32 -0700
Subject: [PATCH 4/6] fixes to comments

---
 .../microsoft_teams/apps/state/container.py   | 10 ++-
 .../src/microsoft_teams/apps/state/loader.py  | 88 +++++++++++++------
 .../src/microsoft_teams/apps/state/options.py |  8 +-
 .../microsoft_teams/apps/state/turn_state.py  | 19 ++--
 packages/apps/tests/test_state.py             | 79 +++++++++++++++++
 5 files changed, 163 insertions(+), 41 deletions(-)

diff --git a/packages/apps/src/microsoft_teams/apps/state/container.py b/packages/apps/src/microsoft_teams/apps/state/container.py
index fccd01d1..0b903a83 100644
--- a/packages/apps/src/microsoft_teams/apps/state/container.py
+++ b/packages/apps/src/microsoft_teams/apps/state/container.py
@@ -44,11 +44,13 @@ def seal(self) -> None:
     async def delete(self) -> None:
         """Clear both scopes and remove them from the backing store.
 
-        Clearing marks the scopes dirty+empty (so a later save deletes the keys), and the
-        injected deleter removes the keys immediately.
+        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()
         if self.user is not None:
             self.user.clear()
-        if self._deleter is not None:
-            await self._deleter()
diff --git a/packages/apps/src/microsoft_teams/apps/state/loader.py b/packages/apps/src/microsoft_teams/apps/state/loader.py
index 061a48f0..54fab4bf 100644
--- a/packages/apps/src/microsoft_teams/apps/state/loader.py
+++ b/packages/apps/src/microsoft_teams/apps/state/loader.py
@@ -9,6 +9,7 @@
 import logging
 import time
 from typing import Any, Dict, Optional, cast
+from urllib.parse import quote
 
 from microsoft_teams.common import Storage
 
@@ -40,12 +41,12 @@ def options(self) -> StateOptions:
         return self._options
 
     def conversation_key(self, conversation_id: str) -> str:
-        """Key for the conversation-scoped blob (mirrors C#'s ``ts:conv:{id}``)."""
-        return f"{self._options.key_prefix}:conv:{conversation_id}"
+        """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 (mirrors C#'s ``ts:user:{convId}:{userId}``)."""
-        return f"{self._options.key_prefix}:user:{conversation_id}:{user_id}"
+        """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."""
@@ -77,9 +78,27 @@ async def save(self, container: TurnStateContainer) -> None:
             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.")
-        await self._save_scope(self.conversation_key(container.conversation_id), container.conversation)
+
+        pending_deletes: list[str] = []
+        pending_sets: list[tuple[str, str]] = []
+        self._prepare_scope_save(
+            self.conversation_key(container.conversation_id),
+            container.conversation,
+            pending_deletes,
+            pending_sets,
+        )
         if container.user is not None and container.user_id is not None:
-            await self._save_scope(self.user_key(container.conversation_id, container.user_id), container.user)
+            self._prepare_scope_save(
+                self.user_key(container.conversation_id, container.user_id),
+                container.user,
+                pending_deletes,
+                pending_sets,
+            )
+
+        for key in pending_deletes:
+            await self._storage.async_delete(key)
+        for key, value in pending_sets:
+            await self._storage.async_set(key, value)
 
     async def delete(self, conversation_id: str, user_id: Optional[str] = None) -> None:
         """Delete both scope blobs for the turn's identity."""
@@ -89,43 +108,54 @@ async def delete(self, conversation_id: str, user_id: Optional[str] = None) -> N
 
     async def _load_scope(self, key: str) -> TurnState:
         raw = await self._storage.async_get(key)
-        return TurnState(self._deserialize(raw))
+        if raw is None:
+            return TurnState()
 
-    async def _save_scope(self, key: str, scope: TurnState) -> None:
+        blob = self._deserialize(raw)
+        if blob is None or self._is_expired(blob):
+            await self._storage.async_delete(key)
+            return TurnState()
+
+        data = blob.get("data")
+        if not isinstance(data, dict):
+            await self._storage.async_delete(key)
+            return TurnState()
+
+        return TurnState(cast(Dict[str, Any], data))
+
+    def _prepare_scope_save(
+        self,
+        key: str,
+        scope: TurnState,
+        pending_deletes: list[str],
+        pending_sets: list[tuple[str, str]],
+    ) -> None:
         if not scope.is_dirty:
             return
         if scope.is_empty:
-            await self._storage.async_delete(key)
+            pending_deletes.append(key)
             return
         blob = {"ts": time.time(), "data": scope.to_dict()}
-        await self._storage.async_set(key, json.dumps(blob))
+        pending_sets.append((key, json.dumps(blob)))
 
-    def _deserialize(self, raw: Optional[Any]) -> Dict[str, Any]:
-        """Turn a stored blob back into a scope dict.
+    def _deserialize(self, raw: Any) -> Optional[Dict[str, Any]]:
+        """Parse a stored blob.
 
-        Never raises: an absent, unreadable, or expired blob is treated as an
-        empty scope.
+        Never raises: unreadable or malformed blobs are treated as missing.
         """
-        if not raw:
-            return {}
         try:
             parsed: Any = json.loads(raw)
         except (ValueError, TypeError):
             logger.debug("Discarding unreadable state blob at load")
-            return {}
+            return None
 
         if not isinstance(parsed, dict):
-            return {}
-        blob = cast(Dict[str, Any], parsed)
+            return None
+        return cast(Dict[str, Any], parsed)
 
-        if self._options.ttl is not None:
-            saved_at = blob.get("ts")
-            if not isinstance(saved_at, (int, float)):
-                return {}
-            if (time.time() - saved_at) > self._options.ttl:
-                return {}
+    def _is_expired(self, blob: Dict[str, Any]) -> bool:
+        if self._options.ttl is None:
+            return False
 
-        data = blob.get("data")
-        if isinstance(data, dict):
-            return cast(Dict[str, Any], data)
-        return {}
+        saved_at = blob.get("ts")
+        return not isinstance(saved_at, (int, float)) or (time.time() - saved_at) > self._options.ttl
diff --git a/packages/apps/src/microsoft_teams/apps/state/options.py b/packages/apps/src/microsoft_teams/apps/state/options.py
index 400e61d4..20178de5 100644
--- a/packages/apps/src/microsoft_teams/apps/state/options.py
+++ b/packages/apps/src/microsoft_teams/apps/state/options.py
@@ -27,4 +27,10 @@ class StateOptions:
     """Namespace prefix for scope keys (``{prefix}:conv:...`` / ``{prefix}:user:...``)."""
 
     ttl: Optional[int] = None
-    """Optional time-to-live, in **seconds**, applied by the loader on load."""
+    """Optional lazy, sliding time-to-live in **seconds**.
+
+    The loader stamps each successful state write and treats the scope as
+    expired when that saved timestamp is older than ``ttl`` during a later load.
+    Expiry is sliding from the last write (not absolute from creation) and is
+    enforced lazily on load because ``Storage`` has no native TTL concept.
+    """
diff --git a/packages/apps/src/microsoft_teams/apps/state/turn_state.py b/packages/apps/src/microsoft_teams/apps/state/turn_state.py
index 911001a0..96bd8281 100644
--- a/packages/apps/src/microsoft_teams/apps/state/turn_state.py
+++ b/packages/apps/src/microsoft_teams/apps/state/turn_state.py
@@ -5,6 +5,8 @@
 
 from __future__ import annotations
 
+import hashlib
+import json
 from collections.abc import Iterator, Mapping, MutableMapping
 from typing import Any, Dict, Optional
 
@@ -18,8 +20,8 @@ class TurnState(MutableMapping[str, Any]):
 
     Behaves like a ``dict`` but adds two things the loader relies on:
 
-    * **Dirty tracking** — the loader only writes a scope back when it was
-      mutated, so an untouched scope costs nothing to "save".
+    * **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`.
 
@@ -32,13 +34,13 @@ class TurnState(MutableMapping[str, Any]):
 
     def __init__(self, data: Optional[Mapping[str, Any]] = None) -> None:
         self._data: Dict[str, Any] = dict(data) if data else {}
-        self._dirty = False
+        self._baseline = self._fingerprint(self._data)
         self._sealed = False
 
     @property
     def is_dirty(self) -> bool:
         """Whether the scope has been mutated since it was loaded."""
-        return self._dirty
+        return self._fingerprint(self._data) != self._baseline
 
     @property
     def is_empty(self) -> bool:
@@ -66,6 +68,11 @@ def _ensure_active(self) -> None:
         if self._sealed:
             raise TurnStateSealedError("TurnState has been sealed and can no longer be accessed.")
 
+    @staticmethod
+    def _fingerprint(data: Mapping[str, Any]) -> str:
+        canonical = json.dumps(data, sort_keys=True, default=repr, separators=(",", ":"))
+        return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
+
     def __getitem__(self, key: str) -> Any:
         self._ensure_active()
         return self._data[key]
@@ -73,12 +80,10 @@ def __getitem__(self, key: str) -> Any:
     def __setitem__(self, key: str, value: Any) -> None:
         self._ensure_active()
         self._data[key] = value
-        self._dirty = True
 
     def __delitem__(self, key: str) -> None:
         self._ensure_active()
         del self._data[key]
-        self._dirty = True
 
     def __iter__(self) -> Iterator[str]:
         self._ensure_active()
@@ -93,5 +98,5 @@ def __contains__(self, key: object) -> bool:
         return key in self._data
 
     def __repr__(self) -> str:
-        status = "sealed" if self._sealed else ("dirty" if self._dirty else "clean")
+        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
index 8e0de61e..bc62fb44 100644
--- a/packages/apps/tests/test_state.py
+++ b/packages/apps/tests/test_state.py
@@ -69,6 +69,23 @@ def test_to_dict_returns_copy(self):
         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_seal_blocks_access(self):
         state = TurnState({"a": 1})
         state.seal()
@@ -129,6 +146,19 @@ async def deleter():
         assert container.user is not None and container.user.is_empty
         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
@@ -145,6 +175,11 @@ def test_key_layout_matches_csharp(self):
         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"
@@ -173,6 +208,20 @@ async def test_round_trip(self):
         assert reloaded.conversation["greeted"] is True
         assert reloaded.user is not None and reloaded.user["step"] == 3
 
+    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)
@@ -194,6 +243,27 @@ async def test_clean_scope_is_not_written(self):
         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)
@@ -247,6 +317,15 @@ async def test_expired_blob_loads_as_empty(self):
         loader = TurnStateLoader(storage, StateOptions(ttl=100))
         container = await loader.load("c1")
         assert container.conversation.is_empty
+        assert storage.get("ts:conv:c1") is None
+
+    async def test_malformed_ttl_blob_is_deleted(self):
+        storage = LocalStorage()
+        await storage.async_set("ts:conv:c1", json.dumps({"data": {"a": 1}}))
+        loader = TurnStateLoader(storage, StateOptions(ttl=100))
+        container = await loader.load("c1")
+        assert container.conversation.is_empty
+        assert storage.get("ts:conv:c1") is None
 
     async def test_unexpired_blob_loads_normally(self):
         storage = LocalStorage()

From 0597bb631928344a7db527380283c15298d2cbcd Mon Sep 17 00:00:00 2001
From: lilyydu <54044854+lilyydu@users.noreply.github.com>
Date: Thu, 13 Aug 2026 15:34:29 -0700
Subject: [PATCH 5/6] resolve Corina's comments

---
 .../src/microsoft_teams/apps/state/loader.py  |  8 +++++
 .../src/microsoft_teams/apps/state/options.py |  3 +-
 .../microsoft_teams/apps/state/turn_state.py  | 20 ++++++++---
 packages/apps/tests/test_state.py             | 36 +++++++++++++++++++
 4 files changed, 62 insertions(+), 5 deletions(-)

diff --git a/packages/apps/src/microsoft_teams/apps/state/loader.py b/packages/apps/src/microsoft_teams/apps/state/loader.py
index 54fab4bf..2576cf78 100644
--- a/packages/apps/src/microsoft_teams/apps/state/loader.py
+++ b/packages/apps/src/microsoft_teams/apps/state/loader.py
@@ -81,11 +81,13 @@ async def save(self, container: TurnStateContainer) -> None:
 
         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(
@@ -93,12 +95,15 @@ async def save(self, container: TurnStateContainer) -> None:
                 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."""
@@ -129,14 +134,17 @@ def _prepare_scope_save(
         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
         blob = {"ts": time.time(), "data": scope.to_dict()}
         pending_sets.append((key, json.dumps(blob)))
+        pending_clean.append(scope)
 
     def _deserialize(self, raw: Any) -> Optional[Dict[str, Any]]:
         """Parse a stored blob.
diff --git a/packages/apps/src/microsoft_teams/apps/state/options.py b/packages/apps/src/microsoft_teams/apps/state/options.py
index 20178de5..ad34807b 100644
--- a/packages/apps/src/microsoft_teams/apps/state/options.py
+++ b/packages/apps/src/microsoft_teams/apps/state/options.py
@@ -32,5 +32,6 @@ class StateOptions:
     The loader stamps each successful state write and treats the scope as
     expired when that saved timestamp is older than ``ttl`` during a later load.
     Expiry is sliding from the last write (not absolute from creation) and is
-    enforced lazily on load because ``Storage`` has no native TTL concept.
+    enforced lazily on load because ``Storage`` has no native TTL concept. Load
+    hits do not refresh the timestamp; only a later successful save does.
     """
diff --git a/packages/apps/src/microsoft_teams/apps/state/turn_state.py b/packages/apps/src/microsoft_teams/apps/state/turn_state.py
index 96bd8281..ab9fe7dc 100644
--- a/packages/apps/src/microsoft_teams/apps/state/turn_state.py
+++ b/packages/apps/src/microsoft_teams/apps/state/turn_state.py
@@ -34,13 +34,16 @@ class TurnState(MutableMapping[str, Any]):
 
     def __init__(self, data: Optional[Mapping[str, Any]] = None) -> None:
         self._data: Dict[str, Any] = dict(data) if data else {}
-        self._baseline = self._fingerprint(self._data)
+        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."""
-        return self._fingerprint(self._data) != self._baseline
+        fingerprint = self._try_fingerprint(self._data)
+        if fingerprint is None:
+            return True
+        return fingerprint != self._baseline
 
     @property
     def is_empty(self) -> bool:
@@ -56,6 +59,12 @@ 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).
 
@@ -69,8 +78,11 @@ def _ensure_active(self) -> None:
             raise TurnStateSealedError("TurnState has been sealed and can no longer be accessed.")
 
     @staticmethod
-    def _fingerprint(data: Mapping[str, Any]) -> str:
-        canonical = json.dumps(data, sort_keys=True, default=repr, separators=(",", ":"))
+    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:
diff --git a/packages/apps/tests/test_state.py b/packages/apps/tests/test_state.py
index bc62fb44..7970f5f9 100644
--- a/packages/apps/tests/test_state.py
+++ b/packages/apps/tests/test_state.py
@@ -86,6 +86,15 @@ def test_mutate_then_revert_is_clean(self):
         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()
@@ -208,6 +217,33 @@ async def test_round_trip(self):
         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)

From fed21b395a02e6d86e3c2a24f26e76c8a0aa4880 Mon Sep 17 00:00:00 2001
From: lilydu <lilydu@Lilys-MacBook-Pro.local>
Date: Fri, 14 Aug 2026 14:48:03 -0700
Subject: [PATCH 6/6] refactor(state): delegate TTL to storage

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
---
 .../src/microsoft_teams/apps/state/loader.py  | 54 +++++++-------
 .../src/microsoft_teams/apps/state/options.py | 23 +++---
 packages/apps/tests/test_state.py             | 65 ++++++++++-------
 .../common/storage/__init__.py                |  4 +-
 .../common/storage/local_storage.py           | 49 ++++++++++---
 .../microsoft_teams/common/storage/storage.py | 31 ++++++++
 packages/common/tests/test_local_storage.py   | 72 ++++++++++++++++++-
 7 files changed, 218 insertions(+), 80 deletions(-)

diff --git a/packages/apps/src/microsoft_teams/apps/state/loader.py b/packages/apps/src/microsoft_teams/apps/state/loader.py
index 2576cf78..a8171d54 100644
--- a/packages/apps/src/microsoft_teams/apps/state/loader.py
+++ b/packages/apps/src/microsoft_teams/apps/state/loader.py
@@ -7,7 +7,6 @@
 
 import json
 import logging
-import time
 from typing import Any, Dict, Optional, cast
 from urllib.parse import quote
 
@@ -24,17 +23,16 @@ 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. Each blob carries a save timestamp
-    that powers the loader-applied TTL, since ``Storage`` has no native expiry.
-
+    regardless of how it serializes values. Expiry and other write behavior are
+    delegated to the storage implementation through ``StorageOptions``.
     """
 
-    def __init__(self, storage: Optional[Storage[str, str]] = None, options: Optional[StateOptions] = None) -> None:
+    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, str] = resolved
+        self._storage: Storage[str, Any] = resolved
 
     @property
     def options(self) -> StateOptions:
@@ -101,7 +99,10 @@ async def save(self, container: TurnStateContainer) -> None:
         for key in pending_deletes:
             await self._storage.async_delete(key)
         for key, value in pending_sets:
-            await self._storage.async_set(key, value)
+            if self._options.storage_options is None:
+                await self._storage.async_set(key, value)
+            else:
+                await self._storage.async_set_with_options(key, value, self._options.storage_options)
         for scope in pending_clean:
             scope.mark_clean()
 
@@ -116,17 +117,12 @@ async def _load_scope(self, key: str) -> TurnState:
         if raw is None:
             return TurnState()
 
-        blob = self._deserialize(raw)
-        if blob is None or self._is_expired(blob):
-            await self._storage.async_delete(key)
-            return TurnState()
-
-        data = blob.get("data")
-        if not isinstance(data, dict):
+        data = self._deserialize(raw)
+        if data is None:
             await self._storage.async_delete(key)
             return TurnState()
 
-        return TurnState(cast(Dict[str, Any], data))
+        return TurnState(data)
 
     def _prepare_scope_save(
         self,
@@ -142,8 +138,7 @@ def _prepare_scope_save(
             pending_deletes.append(key)
             pending_clean.append(scope)
             return
-        blob = {"ts": time.time(), "data": scope.to_dict()}
-        pending_sets.append((key, json.dumps(blob)))
+        pending_sets.append((key, json.dumps(scope.to_dict())))
         pending_clean.append(scope)
 
     def _deserialize(self, raw: Any) -> Optional[Dict[str, Any]]:
@@ -151,19 +146,20 @@ def _deserialize(self, raw: Any) -> Optional[Dict[str, Any]]:
 
         Never raises: unreadable or malformed blobs are treated as missing.
         """
-        try:
-            parsed: Any = json.loads(raw)
-        except (ValueError, TypeError):
-            logger.debug("Discarding unreadable state blob at load")
+        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
-        return cast(Dict[str, Any], parsed)
-
-    def _is_expired(self, blob: Dict[str, Any]) -> bool:
-        if self._options.ttl is None:
-            return False
-
-        saved_at = blob.get("ts")
-        return not isinstance(saved_at, (int, float)) or (time.time() - saved_at) > self._options.ttl
+        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
index ad34807b..609f5292 100644
--- a/packages/apps/src/microsoft_teams/apps/state/options.py
+++ b/packages/apps/src/microsoft_teams/apps/state/options.py
@@ -6,32 +6,25 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Optional
+from typing import Any, Optional
 
-from microsoft_teams.common import Storage
+from microsoft_teams.common import Storage, StorageOptions
 
 
 @dataclass(frozen=True)
 class StateOptions:
     """Configuration for the per-turn state layer.
 
-    Scope keys are namespaced
-    under ``key_prefix`` and expiry is applied by the loader, since ``Storage``
-    has no native TTL concept.
+    Scope keys are namespaced under ``key_prefix``. Storage-specific behavior,
+    including expiry, is configured through ``storage_options`` and enforced by
+    the selected storage implementation.
     """
 
-    storage: Optional[Storage[str, str]] = None
+    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:...``)."""
 
-    ttl: Optional[int] = None
-    """Optional lazy, sliding time-to-live in **seconds**.
-
-    The loader stamps each successful state write and treats the scope as
-    expired when that saved timestamp is older than ``ttl`` during a later load.
-    Expiry is sliding from the last write (not absolute from creation) and is
-    enforced lazily on load because ``Storage`` has no native TTL concept. Load
-    hits do not refresh the timestamp; only a later successful save does.
-    """
+    storage_options: Optional[StorageOptions] = None
+    """Optional settings passed to storage whenever a state scope is written."""
diff --git a/packages/apps/tests/test_state.py b/packages/apps/tests/test_state.py
index 7970f5f9..167dc104 100644
--- a/packages/apps/tests/test_state.py
+++ b/packages/apps/tests/test_state.py
@@ -4,7 +4,7 @@
 """
 
 import json
-import time
+from typing import Any
 
 import pytest
 from microsoft_teams.apps.state import (
@@ -14,7 +14,7 @@
     TurnStateLoader,
     TurnStateSealedError,
 )
-from microsoft_teams.common import LocalStorage
+from microsoft_teams.common import LocalStorage, StorageOptions
 
 # ---------------------------------------------------------------------------
 # TurnState
@@ -268,8 +268,7 @@ async def test_save_persists_json_string(self):
         stored = storage.get("ts:conv:c1")
         assert isinstance(stored, str)  # design §13.1: always a str
         parsed = json.loads(stored)
-        assert parsed["data"] == {"k": "v"}
-        assert "ts" in parsed
+        assert parsed == {"k": "v"}
 
     async def test_clean_scope_is_not_written(self):
         storage = LocalStorage()
@@ -336,42 +335,58 @@ async def test_corrupt_blob_loads_as_empty(self):
         loader = TurnStateLoader(storage)
         container = await loader.load("c1")
         assert container.conversation.is_empty
+        assert storage.get("ts:conv:c1") is None
 
-    async def test_blob_without_data_loads_as_empty(self):
+    async def test_non_mapping_blob_loads_as_empty_and_is_deleted(self):
         storage = LocalStorage()
-        await storage.async_set("ts:conv:c1", json.dumps({"ts": time.time()}))
+        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_expired_blob_loads_as_empty(self):
-        storage = LocalStorage()
-        await storage.async_set(
-            "ts:conv:c1",
-            json.dumps({"ts": time.time() - 500, "data": {"a": 1}}),
-        )
-        loader = TurnStateLoader(storage, StateOptions(ttl=100))
+    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_malformed_ttl_blob_is_deleted(self):
-        storage = LocalStorage()
-        await storage.async_set("ts:conv:c1", json.dumps({"data": {"a": 1}}))
-        loader = TurnStateLoader(storage, StateOptions(ttl=100))
+    async def test_storage_managed_ttl_expires_state(self, monkeypatch):
+        now = [100.0]
+        monkeypatch.setattr("microsoft_teams.common.storage.local_storage.monotonic", lambda: now[0])
+        storage: LocalStorage[str] = LocalStorage()
+        loader = TurnStateLoader(
+            storage,
+            StateOptions(storage_options=StorageOptions(ttl=10)),
+        )
         container = await loader.load("c1")
-        assert container.conversation.is_empty
+        container.conversation["a"] = 1
+        await loader.save(container)
+
+        reloaded = await loader.load("c1")
+        assert reloaded.conversation["a"] == 1
+
+        now[0] = 111.0
+        expired = await loader.load("c1")
+        assert expired.conversation.is_empty
         assert storage.get("ts:conv:c1") is None
 
-    async def test_unexpired_blob_loads_normally(self):
+    async def test_storage_options_are_optional_for_legacy_storage(self):
         storage = LocalStorage()
-        await storage.async_set(
-            "ts:conv:c1",
-            json.dumps({"ts": time.time(), "data": {"a": 1}}),
-        )
-        loader = TurnStateLoader(storage, StateOptions(ttl=100))
+        loader = TurnStateLoader(storage, StateOptions(storage_options=StorageOptions()))
         container = await loader.load("c1")
-        assert container.conversation["a"] == 1
+        container.conversation["a"] = 1
+        await loader.save(container)
+        assert storage.get("ts:conv:c1") is not None
 
     async def test_storage_from_options_is_used(self):
         storage = LocalStorage()
diff --git a/packages/common/src/microsoft_teams/common/storage/__init__.py b/packages/common/src/microsoft_teams/common/storage/__init__.py
index 990ea03f..14666fc0 100644
--- a/packages/common/src/microsoft_teams/common/storage/__init__.py
+++ b/packages/common/src/microsoft_teams/common/storage/__init__.py
@@ -5,6 +5,6 @@
 
 from .list_local_storage import ListLocalStorage
 from .local_storage import LocalStorage, LocalStorageOptions
-from .storage import ListStorage, Storage
+from .storage import ListStorage, Storage, StorageOptions
 
-__all__ = ["Storage", "ListStorage", "LocalStorage", "ListLocalStorage", "LocalStorageOptions"]
+__all__ = ["Storage", "StorageOptions", "ListStorage", "LocalStorage", "ListLocalStorage", "LocalStorageOptions"]
diff --git a/packages/common/src/microsoft_teams/common/storage/local_storage.py b/packages/common/src/microsoft_teams/common/storage/local_storage.py
index 3d6716f2..558898b3 100644
--- a/packages/common/src/microsoft_teams/common/storage/local_storage.py
+++ b/packages/common/src/microsoft_teams/common/storage/local_storage.py
@@ -5,9 +5,10 @@
 
 from collections import OrderedDict
 from dataclasses import dataclass
+from time import monotonic
 from typing import Dict, List, Optional, TypeVar
 
-from .storage import Storage
+from .storage import Storage, StorageOptions
 
 V = TypeVar("V")
 
@@ -25,6 +26,7 @@ class LocalStorage(Storage[str, V]):
 
     @property
     def store(self) -> OrderedDict[str, V]:
+        self._purge_expired()
         return self._store
 
     @property
@@ -33,10 +35,12 @@ def options(self) -> LocalStorageOptions:
 
     @property
     def keys(self) -> List[str]:
+        self._purge_expired()
         return list(self._store.keys())
 
     @property
     def size(self) -> int:
+        self._purge_expired()
         return len(self._store)
 
     def __init__(
@@ -45,10 +49,11 @@ def __init__(
         options: Optional[LocalStorageOptions] = None,
     ):
         self._store = OrderedDict(data or {})
+        self._expires_at: Dict[str, float] = {}
         self._options = options or LocalStorageOptions()
 
     def get(self, key: str) -> Optional[V]:
-        if key not in self._store:
+        if self._delete_if_expired(key) or key not in self._store:
             return None
 
         value = self._store.pop(key)
@@ -59,19 +64,47 @@ async def async_get(self, key: str) -> Optional[V]:
         return self.get(key)
 
     def set(self, key: str, value: V) -> None:
+        self._set(key, value)
+
+    async def async_set(self, key: str, value: V) -> None:
+        return self.set(key, value)
+
+    def set_with_options(self, key: str, value: V, options: StorageOptions) -> None:
+        self._set(key, value, options.ttl)
+
+    async def async_set_with_options(self, key: str, value: V, options: StorageOptions) -> None:
+        return self.set_with_options(key, value, options)
+
+    def _set(self, key: str, value: V, ttl: Optional[int] = None) -> None:
+        self._purge_expired()
+        self._expires_at.pop(key, None)
+
         if key in self._store:
             del self._store[key]
         elif self._options.max and len(self._store) >= self._options.max:
-            self._store.popitem(last=False)
+            evicted_key, _ = self._store.popitem(last=False)
+            self._expires_at.pop(evicted_key, None)
 
         self._store[key] = value
-
-    async def async_set(self, key: str, value: V) -> None:
-        return self.set(key, value)
+        if ttl is not None:
+            self._expires_at[key] = monotonic() + ttl
 
     def delete(self, key: str) -> None:
-        if key in self._store:
-            del self._store[key]
+        self._store.pop(key, None)
+        self._expires_at.pop(key, None)
 
     async def async_delete(self, key: str) -> None:
         return self.delete(key)
+
+    def _delete_if_expired(self, key: str) -> bool:
+        expires_at = self._expires_at.get(key)
+        if expires_at is None or monotonic() < expires_at:
+            return False
+        self.delete(key)
+        return True
+
+    def _purge_expired(self) -> None:
+        now = monotonic()
+        for key, expires_at in list(self._expires_at.items()):
+            if key not in self._store or now >= expires_at:
+                self.delete(key)
diff --git a/packages/common/src/microsoft_teams/common/storage/storage.py b/packages/common/src/microsoft_teams/common/storage/storage.py
index a21975e1..e6701f01 100644
--- a/packages/common/src/microsoft_teams/common/storage/storage.py
+++ b/packages/common/src/microsoft_teams/common/storage/storage.py
@@ -4,12 +4,21 @@
 """
 
 from abc import ABC, abstractmethod
+from dataclasses import dataclass
 from typing import Callable, Generic, List, Optional, TypeVar
 
 K = TypeVar("K")
 V = TypeVar("V")
 
 
+@dataclass(frozen=True)
+class StorageOptions:
+    """Options applied when writing a value to storage."""
+
+    ttl: Optional[int] = None
+    """Optional time-to-live in seconds after the value is written."""
+
+
 class Storage(Generic[K, V], ABC):
     """A storage container that can get/set/delete items by a unique key."""
 
@@ -33,6 +42,28 @@ async def async_set(self, key: K, value: V) -> None:
         """Asynchronously set a value by key."""
         pass
 
+    def set_with_options(self, key: K, value: V, options: StorageOptions) -> None:
+        """Synchronously set a value with storage-specific write options.
+
+        Existing storage implementations inherit this method. Writes without
+        effective options delegate to :meth:`set`; implementations that support
+        TTL should override it.
+        """
+        if options.ttl is not None:
+            raise NotImplementedError(f"{type(self).__name__} does not support TTL")
+        self.set(key, value)
+
+    async def async_set_with_options(self, key: K, value: V, options: StorageOptions) -> None:
+        """Asynchronously set a value with storage-specific write options.
+
+        Existing storage implementations inherit this method. Writes without
+        effective options delegate to :meth:`async_set`; implementations that
+        support TTL should override it.
+        """
+        if options.ttl is not None:
+            raise NotImplementedError(f"{type(self).__name__} does not support TTL")
+        await self.async_set(key, value)
+
     @abstractmethod
     def delete(self, key: K) -> None:
         """Synchronously delete a value by key."""
diff --git a/packages/common/tests/test_local_storage.py b/packages/common/tests/test_local_storage.py
index 82286c6f..123b71a0 100644
--- a/packages/common/tests/test_local_storage.py
+++ b/packages/common/tests/test_local_storage.py
@@ -3,7 +3,8 @@
 Licensed under the MIT License.
 """
 
-from microsoft_teams.common.storage import LocalStorage, LocalStorageOptions
+import pytest
+from microsoft_teams.common.storage import ListLocalStorage, LocalStorage, LocalStorageOptions, StorageOptions
 
 
 def test_get_undefined() -> None:
@@ -40,3 +41,72 @@ def test_max_size() -> None:
     assert storage.get("d") == 4
     assert storage.keys == ["b", "c", "d"]
     assert storage.size == 3
+
+
+async def test_inherited_options_write_preserves_existing_storage_implementations() -> None:
+    storage = ListLocalStorage[int]([1])
+
+    storage.set_with_options(0, 2, StorageOptions())
+    await storage.async_set_with_options(0, 3, StorageOptions())
+
+    assert storage.get(0) == 3
+    with pytest.raises(NotImplementedError, match="does not support TTL"):
+        await storage.async_set_with_options(0, 4, StorageOptions(ttl=10))
+
+
+def test_ttl_expiry_removes_value_from_all_surfaces(monkeypatch) -> None:
+    now = [100.0]
+    monkeypatch.setattr("microsoft_teams.common.storage.local_storage.monotonic", lambda: now[0])
+    storage = LocalStorage[str]()
+
+    storage.set_with_options("key", "value", StorageOptions(ttl=10))
+    assert storage.get("key") == "value"
+
+    now[0] = 110.0
+
+    assert storage.keys == []
+    assert storage.size == 0
+    assert dict(storage.store) == {}
+    assert storage.get("key") is None
+
+
+def test_regular_set_replaces_value_and_clears_ttl(monkeypatch) -> None:
+    now = [100.0]
+    monkeypatch.setattr("microsoft_teams.common.storage.local_storage.monotonic", lambda: now[0])
+    storage = LocalStorage[str]()
+
+    storage.set_with_options("key", "expiring", StorageOptions(ttl=10))
+    storage.set("key", "persistent")
+    now[0] = 111.0
+
+    assert storage.get("key") == "persistent"
+
+
+def test_delete_clears_ttl_metadata(monkeypatch) -> None:
+    now = [100.0]
+    monkeypatch.setattr("microsoft_teams.common.storage.local_storage.monotonic", lambda: now[0])
+    storage = LocalStorage[str]()
+
+    storage.set_with_options("key", "expiring", StorageOptions(ttl=10))
+    storage.delete("key")
+    storage.set("key", "replacement")
+    now[0] = 111.0
+
+    assert storage.get("key") == "replacement"
+
+
+def test_expired_values_do_not_evict_live_values_at_capacity(monkeypatch) -> None:
+    now = [100.0]
+    monkeypatch.setattr("microsoft_teams.common.storage.local_storage.monotonic", lambda: now[0])
+    storage = LocalStorage[int](options=LocalStorageOptions(max=2))
+
+    storage.set_with_options("expired", 1, StorageOptions(ttl=10))
+    storage.set("live", 2)
+    assert storage.get("expired") == 1
+
+    now[0] = 111.0
+    storage.set("new", 3)
+
+    assert storage.get("expired") is None
+    assert storage.get("live") == 2
+    assert storage.get("new") == 3