feat: add per-turn state foundation (state scopes, container, loader)

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",
+]

packages/apps/src/microsoft_teams/apps/state/container.py

@@ -0,0 +1,56 @@
+"""
+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()
+        if self.user is not None:
+            self.user.clear()

packages/apps/src/microsoft_teams/apps/state/loader.py

@@ -0,0 +1,165 @@
+"""
+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
+    delegated to the storage implementation through ``StorageOptions``.
+    """
+
+    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:
+            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()
+
+    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)

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 Any, Optional
+
+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``. Storage-specific behavior,
+    including expiry, is configured through ``storage_options`` and enforced by
+    the selected storage implementation.
+    """
+
+    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:...``)."""
+
+    storage_options: Optional[StorageOptions] = None
+    """Optional settings passed to storage whenever a state scope is written."""

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})"