diff --git a/examples/state/README.md b/examples/state/README.md new file mode 100644 index 00000000..c905b10a --- /dev/null +++ b/examples/state/README.md @@ -0,0 +1,15 @@ +# State + +Demonstrates the per-turn state layer: enabling it with `App(state=True)` and +reading/writing the `conversation` and `user` scopes through `ctx.state`. + +State is loaded before each turn, saved automatically after it, and then sealed +(post-turn access raises `TurnStateSealedError`). With no storage configured the +app uses in-memory `LocalStorage`; pass `App(state=StateOptions(storage=...))` +for a durable backing store. + +## Run + +```bash +uv run --directory examples/state src/main.py +``` diff --git a/examples/state/pyproject.toml b/examples/state/pyproject.toml new file mode 100644 index 00000000..adfcfe09 --- /dev/null +++ b/examples/state/pyproject.toml @@ -0,0 +1,14 @@ +[project] +name = "state" +version = "0.1.0" +description = "Per-turn state app" +readme = "README.md" +requires-python = ">=3.11,<4.0" +dependencies = [ + "dotenv>=0.9.9", + "microsoft-teams-apps", + "microsoft-teams-api", +] + +[tool.uv.sources] +microsoft-teams-apps = { workspace = true } diff --git a/examples/state/src/main.py b/examples/state/src/main.py new file mode 100644 index 00000000..72dc76c5 --- /dev/null +++ b/examples/state/src/main.py @@ -0,0 +1,50 @@ +""" +Copyright (c) Microsoft Corporation. All rights reserved. +Licensed under the MIT License. +""" + +import asyncio +import logging + +from microsoft_teams.api import MessageActivity +from microsoft_teams.apps import ActivityContext, App + +logging.basicConfig(level=logging.INFO) +logger = logging.getLogger(__name__) + +# `state=True` enables the per-turn state layer using the app's storage. +# +# With no other storage configured the app falls back to in-memory +# `LocalStorage`, so state is scoped to a single process and lost on restart — +# the SDK logs a warning to that effect. For production, pass a durable store +# via `App(state=StateOptions(storage=...))` +# +# State is loaded before each turn and saved automatically after it, then +# sealed — reading or writing `ctx.state` after the handler returns raises +# `TurnStateSealedError`. +app = App(state=True) + + +@app.on_message +async def handle_message(ctx: ActivityContext[MessageActivity]) -> None: + """Track a per-conversation message count and a per-user first-seen name.""" + assert ctx.state is not None + + # Conversation scope: shared by everyone in the chat/channel. + count = ctx.state.conversation.get("message_count", 0) + 1 + ctx.state.conversation["message_count"] = count + + # User scope: per-sender. `user` is None only when the activity has no sender. + greeting = "" + if ctx.state.user is not None: + if "name" not in ctx.state.user: + ctx.state.user["name"] = ctx.activity.from_.name + greeting = f"Nice to meet you, {ctx.activity.from_.name}! " + else: + greeting = f"Welcome back, {ctx.state.user['name']}! " + + await ctx.send(f"{greeting}This conversation has seen {count} message(s).") + + +if __name__ == "__main__": + asyncio.run(app.start()) diff --git a/packages/apps/src/microsoft_teams/apps/__init__.py b/packages/apps/src/microsoft_teams/apps/__init__.py index 230b9e50..19a191ce 100644 --- a/packages/apps/src/microsoft_teams/apps/__init__.py +++ b/packages/apps/src/microsoft_teams/apps/__init__.py @@ -17,6 +17,7 @@ from .options import AppOptions, AppTelemetryOptions from .plugins import * # noqa: F401, F403 from .routing import ActivityContext +from .state import StateOptions, TurnState, TurnStateContainer, TurnStateSealedError, create_state_loader from .token_provider import AppTokenProvider from .utils.html_widget import ( DisplayMode, @@ -45,6 +46,11 @@ "HttpStream", "ActivityContext", "AppTokenProvider", + "StateOptions", + "TurnState", + "TurnStateContainer", + "TurnStateSealedError", + "create_state_loader", "to_threaded_conversation_id", "build_html_widget_markdown", "build_html_widget_message", diff --git a/packages/apps/src/microsoft_teams/apps/app.py b/packages/apps/src/microsoft_teams/apps/app.py index b136a709..20bd830b 100644 --- a/packages/apps/src/microsoft_teams/apps/app.py +++ b/packages/apps/src/microsoft_teams/apps/app.py @@ -60,6 +60,7 @@ from .plugins import PluginBase, PluginStartEvent from .routing import ActivityHandlerMixin, ActivityRouter from .routing.activity_context import ActivityContext +from .state import create_state_loader from .token_manager import DEFAULT_TENANT_FOR_GRAPH_TOKEN, TokenManager from .token_provider import AppTokenProvider from .utils import create_graph_client @@ -92,6 +93,8 @@ def __init__(self, **options: Unpack[AppOptions]): self.storage = self.options.storage or LocalStorage() + self._state_loader = create_state_loader(self.options.state, self.storage) + self.http_client = self._init_http_client() self._events = EventEmitter[EventType]() @@ -143,6 +146,7 @@ def __init__(self, **options: Unpack[AppOptions]): self.cloud, fetch_user_token=self.options.fetch_user_token, agent365_baggage_options=self.options.telemetry.get("agent365") if self.options.telemetry else None, + state_loader=self._state_loader, ) self.event_manager = EventManager(self._events) self.activity_processor.event_manager = self.event_manager diff --git a/packages/apps/src/microsoft_teams/apps/app_process.py b/packages/apps/src/microsoft_teams/apps/app_process.py index bd6ba85e..3a5dac8d 100644 --- a/packages/apps/src/microsoft_teams/apps/app_process.py +++ b/packages/apps/src/microsoft_teams/apps/app_process.py @@ -46,6 +46,7 @@ from .plugins import PluginActivityEvent, PluginBase, StreamCancelledError from .routing.activity_context import ActivityContext from .routing.router import ActivityHandler, ActivityRouter +from .state import TurnStateLoader from .token_provider import AppTokenProvider from .utils import extract_tenant_id @@ -74,6 +75,7 @@ def __init__( cloud: CloudEnvironment = PUBLIC, fetch_user_token: bool = True, agent365_baggage_options: Agent365BaggageOptions | bool | None = None, + state_loader: Optional[TurnStateLoader] = None, ) -> None: self.router = router self.id = id @@ -86,6 +88,7 @@ def __init__( self.cloud = cloud self.fetch_user_token = fetch_user_token self.agent365_baggage_options = agent365_baggage_options + self.state_loader = state_loader # This will be set after the EventManager is initialized due to # a circular dependency @@ -288,6 +291,8 @@ async def route(ctx: ActivityContext[ActivityBase]) -> Optional[Any]: raise ValueError("EventManager was not initialized properly") try: + await self._load_turn_state(activityCtx, activity) + # If no registered handlers, middleware_result is set to None middleware_result = await self.execute_middleware_chain(activityCtx, handlers) @@ -312,12 +317,40 @@ async def route(ctx: ActivityContext[ActivityBase]) -> Optional[Any]: response = InvokeResponse[Any](status=200) except Exception as error: await self.event_manager.on_error(ErrorEvent(error=error, activity=activity), plugins) - raise error + raise + finally: + await self._persist_turn_state(activityCtx) logger.debug("Completed processing activity") return response + async def _load_turn_state(self, ctx: ActivityContext[ActivityBase], activity: ValidatedActivity) -> None: + """Load per-turn state onto ``ctx.state`` when state is enabled. + + Loads both the conversation scope and the user scope (keyed by the + activity's ``from`` identity). A no-op when state is disabled, leaving + ``ctx.state`` as ``None``. + """ + if self.state_loader is None: + return + ctx.state = await self.state_loader.load(activity.conversation.id, activity.from_.id) + + async def _persist_turn_state(self, ctx: ActivityContext[ActivityBase]) -> None: + """Save dirty scopes and seal state at the end of the turn. + + Runs in a ``finally`` so dirty state is persisted even when the handler + raised. Sealing makes any post-turn access raise, guarding against use of + per-turn state in background work. + """ + container = ctx.state + if self.state_loader is None or container is None: + return + try: + await self.state_loader.save(container) + finally: + container.seal() + def _activity_attributes(self, activity: ActivityBase) -> dict[str, str]: attributes = { APP_ATTRIBUTE_NAMES.activity_type: activity.type, diff --git a/packages/apps/src/microsoft_teams/apps/options.py b/packages/apps/src/microsoft_teams/apps/options.py index a1b2f35e..1f41cb38 100644 --- a/packages/apps/src/microsoft_teams/apps/options.py +++ b/packages/apps/src/microsoft_teams/apps/options.py @@ -19,6 +19,7 @@ from .diagnostics import Agent365BaggageOptions from .http.adapter import HttpServerAdapter from .plugins import PluginBase +from .state import StateOptions DANGEROUSLY_ALLOW_UNAUTHENTICATED_REQUESTS_ENV_VAR = "DANGEROUSLY_ALLOW_UNAUTHENTICATED_REQUESTS" _TRUE_ENV_VALUES = {"1", "true", "yes", "on"} @@ -88,6 +89,14 @@ class AppOptions(TypedDict, total=False): # Infrastructure storage: Optional[Storage[str, Any]] plugins: Optional[List[PluginBase]] + state: Optional[Union[bool, StateOptions]] + """Per-turn state opt-in. Off by default (``None``/``False``). + + ``state=True`` enables state on the app's shared ``storage`` (in-memory by + default). Pass a ``StateOptions`` to configure the + key prefix, TTL, or a dedicated ``Storage`` backend. When enabled, handlers + read/write ``ctx.state.conversation`` and ``ctx.state.user``; when off, + ``ctx.state`` is ``None``.""" dangerously_allow_unauthenticated_requests: Optional[bool] """ Whether to accept incoming requests without JWT validation. @@ -183,6 +192,9 @@ class InternalAppOptions: If not set or equals client_id, uses direct managed identity (no federation). """ storage: Optional[Storage[str, Any]] = None + state: Optional[Union[bool, StateOptions]] = None + """Per-turn state opt-in. ``None``/``False`` disables it; ``True`` enables it on + the app's shared storage; a ``StateOptions`` configures prefix/TTL/backend.""" service_url: Optional[str] = None """ Base Service URL for BotBackend. diff --git a/packages/apps/src/microsoft_teams/apps/routing/activity_context.py b/packages/apps/src/microsoft_teams/apps/routing/activity_context.py index 99b449f7..b11bb111 100644 --- a/packages/apps/src/microsoft_teams/apps/routing/activity_context.py +++ b/packages/apps/src/microsoft_teams/apps/routing/activity_context.py @@ -44,6 +44,7 @@ from ..files import FilesAccessor from ..http_stream import HttpStream from ..plugins.streamer import StreamerProtocol +from ..state import TurnStateContainer from ..utils import create_graph_client if TYPE_CHECKING: @@ -101,6 +102,7 @@ def __init__( self.connection_name = connection_name self.is_signed_in = is_signed_in self.cloud = cloud + self.state: Optional[TurnStateContainer] = None self._app_token = app_token self._stream: Optional[StreamerProtocol] = None self._files: Optional[FilesAccessor] = None diff --git a/packages/apps/src/microsoft_teams/apps/state/__init__.py b/packages/apps/src/microsoft_teams/apps/state/__init__.py index 11b1305b..24c0d4bb 100644 --- a/packages/apps/src/microsoft_teams/apps/state/__init__.py +++ b/packages/apps/src/microsoft_teams/apps/state/__init__.py @@ -4,7 +4,7 @@ """ from .container import TurnStateContainer -from .loader import TurnStateLoader +from .loader import TurnStateLoader, create_state_loader from .options import StateOptions from .turn_state import TurnState, TurnStateSealedError @@ -14,4 +14,5 @@ "TurnStateContainer", "TurnStateLoader", "StateOptions", + "create_state_loader", ] diff --git a/packages/apps/src/microsoft_teams/apps/state/loader.py b/packages/apps/src/microsoft_teams/apps/state/loader.py index 54fab4bf..4e479615 100644 --- a/packages/apps/src/microsoft_teams/apps/state/loader.py +++ b/packages/apps/src/microsoft_teams/apps/state/loader.py @@ -8,10 +8,10 @@ import json import logging import time -from typing import Any, Dict, Optional, cast +from typing import Any, Dict, Optional, Union, cast from urllib.parse import quote -from microsoft_teams.common import Storage +from microsoft_teams.common import LocalStorage, Storage from .container import TurnStateContainer from .options import StateOptions @@ -159,3 +159,28 @@ def _is_expired(self, blob: Dict[str, Any]) -> bool: saved_at = blob.get("ts") return not isinstance(saved_at, (int, float)) or (time.time() - saved_at) > self._options.ttl + + +def create_state_loader( + state: Optional[Union[bool, "StateOptions"]], + fallback_storage: Storage[str, Any], +) -> Optional[TurnStateLoader]: + """Resolve the ``App(state=...)`` option into a loader (or ``None`` when off). + + ``state`` is the opt-in value: falsy disables state; ``True`` enables it with + defaults; a ``StateOptions`` configures it. The loader's storage is the one on + ``StateOptions`` when provided, otherwise the app's shared ``fallback_storage``. + A warning is logged when that resolves to in-memory ``LocalStorage``. + """ + if not state: + return None + + options = StateOptions() if state is True else state + storage: Storage[str, Any] = options.storage if options.storage is not None else fallback_storage + + if isinstance(storage, LocalStorage): + logger.warning( + "State is enabled with in-memory storage (LocalStorage): per-turn state is lost on " + + "restart and is not shared across instances." + ) + return TurnStateLoader(storage=cast(Storage[str, str], storage), options=options) diff --git a/packages/apps/tests/test_app_process.py b/packages/apps/tests/test_app_process.py index 711b1779..2bd0e030 100644 --- a/packages/apps/tests/test_app_process.py +++ b/packages/apps/tests/test_app_process.py @@ -22,6 +22,7 @@ from microsoft_teams.apps.app_process import ActivityProcessor from microsoft_teams.apps.events import CoreActivity from microsoft_teams.apps.routing.router import ActivityHandler, ActivityRouter +from microsoft_teams.apps.state import TurnState, TurnStateContainer, TurnStateLoader, TurnStateSealedError from microsoft_teams.apps.token_provider import AppTokenProvider from microsoft_teams.common import Client, LocalStorage from opentelemetry import baggage @@ -846,3 +847,156 @@ async def test_process_activity_raises_exception(self, activity_processor): # Assert error event was called assert activity_processor.event_manager.on_error.called + + @pytest.mark.asyncio + async def test_state_load_failure_emits_error_event(self, activity_processor): + """State load failures are handled by the same on_error path as handler failures.""" + mock_plugins = [] + mock_activity_event = self._message_event("activity-load-failure") + load_error = RuntimeError("storage unavailable") + + activity_processor.state_loader = MagicMock() + activity_processor.state_loader.load = AsyncMock(side_effect=load_error) + activity_processor.state_loader.save = AsyncMock() + activity_processor.router.select_handlers = MagicMock(return_value=[]) + activity_processor.execute_middleware_chain = AsyncMock() + activity_processor.event_manager = MagicMock() + activity_processor.event_manager.on_error = AsyncMock() + + with pytest.raises(RuntimeError, match="storage unavailable"): + await activity_processor.process_activity(mock_plugins, mock_activity_event) + + activity_processor.event_manager.on_error.assert_called_once() + activity_processor.execute_middleware_chain.assert_not_called() + + # --- Per-turn state (App(state=...)) integration ----------------------- + + @staticmethod + def _message_event(activity_id: str = "activity-state") -> ActivityEvent: + core_activity = CoreActivity( + type="message", + id=activity_id, + service_url="https://service.url", + **{ + "from": {"id": "user-123", "name": "Test User"}, + "conversation": {"id": "conv-789"}, + "recipient": {"id": "bot-456", "name": "Test Bot"}, + "channelId": "msteams", + }, + ) + mock_token = MagicMock(spec=TokenProtocol) + mock_token.service_url = "https://service.url" + return ActivityEvent(body=core_activity, token=mock_token) + + def _wire_event_manager(self, activity_processor): + activity_processor.event_manager = MagicMock() + activity_processor.event_manager.on_activity_response = AsyncMock() + activity_processor.event_manager.on_error = AsyncMock() + + @pytest.mark.asyncio + async def test_state_disabled_leaves_ctx_state_none(self, activity_processor): + """With no state loader, ctx.state is None for handlers.""" + captured: dict[str, Any] = {} + + async def handler(ctx: ActivityContext[Activity]) -> None: + captured["state"] = ctx.state + + activity_processor.state_loader = None + activity_processor.router.select_handlers = MagicMock(return_value=[handler]) + self._wire_event_manager(activity_processor) + + await activity_processor.process_activity([], self._message_event()) + + assert captured["state"] is None + + @pytest.mark.asyncio + async def test_state_enabled_exposes_scopes_and_persists_across_turns(self, activity_processor): + """Handlers see loaded scopes, and writes persist to the next turn.""" + + storage: LocalStorage[str] = LocalStorage() + activity_processor.state_loader = TurnStateLoader(storage) + self._wire_event_manager(activity_processor) + + async def writer(ctx: ActivityContext[Activity]) -> None: + assert ctx.state is not None + ctx.state.conversation["greeted"] = True + assert ctx.state.user is not None + ctx.state.user["count"] = 1 + + activity_processor.router.select_handlers = MagicMock(return_value=[writer]) + await activity_processor.process_activity([], self._message_event("turn-1")) + + seen: dict[str, Any] = {} + + async def reader(ctx: ActivityContext[Activity]) -> None: + assert ctx.state is not None and ctx.state.user is not None + seen["greeted"] = ctx.state.conversation.get("greeted") + seen["count"] = ctx.state.user.get("count") + + activity_processor.router.select_handlers = MagicMock(return_value=[reader]) + await activity_processor.process_activity([], self._message_event("turn-2")) + + assert seen == {"greeted": True, "count": 1} + + @pytest.mark.asyncio + async def test_state_is_sealed_after_turn(self, activity_processor): + """State is sealed once the turn ends; later access raises.""" + + storage: LocalStorage[str] = LocalStorage() + activity_processor.state_loader = TurnStateLoader(storage) + self._wire_event_manager(activity_processor) + + holder: dict[str, Any] = {} + + async def handler(ctx: ActivityContext[Activity]) -> None: + holder["state"] = ctx.state + + activity_processor.router.select_handlers = MagicMock(return_value=[handler]) + await activity_processor.process_activity([], self._message_event()) + + with pytest.raises(TurnStateSealedError): + _ = holder["state"].conversation["greeted"] + + @pytest.mark.asyncio + async def test_state_saved_even_when_handler_raises(self, activity_processor): + """Dirty state is persisted in the finally even if the handler throws.""" + + storage: LocalStorage[str] = LocalStorage() + activity_processor.state_loader = TurnStateLoader(storage) + self._wire_event_manager(activity_processor) + + async def boom(ctx: ActivityContext[Activity]) -> None: + assert ctx.state is not None + ctx.state.conversation["partial"] = True + raise RuntimeError("boom") + + activity_processor.router.select_handlers = MagicMock(return_value=[boom]) + + with pytest.raises(RuntimeError, match="boom"): + await activity_processor.process_activity([], self._message_event()) + + reloaded = await TurnStateLoader(storage).load("conv-789", "user-123") + assert reloaded.conversation["partial"] is True + + @pytest.mark.asyncio + async def test_state_is_sealed_when_save_raises(self, activity_processor): + """State is sealed even when persistence fails.""" + container = TurnStateContainer( + conversation=TurnState({"value": "dirty"}), + conversation_id="conv-789", + user=TurnState({"value": "dirty"}), + user_id="user-123", + ) + container.conversation["value"] = "changed" + + activity_processor.state_loader = MagicMock() + activity_processor.state_loader.load = AsyncMock(return_value=container) + activity_processor.state_loader.save = AsyncMock(side_effect=RuntimeError("save failed")) + self._wire_event_manager(activity_processor) + activity_processor.router.select_handlers = MagicMock(return_value=[]) + + with pytest.raises(RuntimeError, match="save failed"): + await activity_processor.process_activity([], self._message_event()) + + assert container.conversation.is_sealed is True + assert container.user is not None and container.user.is_sealed is True diff --git a/packages/apps/tests/test_state.py b/packages/apps/tests/test_state.py index bc62fb44..79c413e0 100644 --- a/packages/apps/tests/test_state.py +++ b/packages/apps/tests/test_state.py @@ -4,7 +4,9 @@ """ import json +import logging import time +from unittest.mock import MagicMock import pytest from microsoft_teams.apps.state import ( @@ -13,8 +15,9 @@ TurnStateContainer, TurnStateLoader, TurnStateSealedError, + create_state_loader, ) -from microsoft_teams.common import LocalStorage +from microsoft_teams.common import LocalStorage, Storage # --------------------------------------------------------------------------- # TurnState @@ -344,3 +347,47 @@ async def test_storage_from_options_is_used(self): container.conversation["k"] = "v" await loader.save(container) assert storage.get("ts:conv:c1") is not None + + +# --------------------------------------------------------------------------- +# create_state_loader (App(state=...) resolution) +# --------------------------------------------------------------------------- + + +class TestCreateStateLoader: + def test_create_state_loader_is_exported_from_apps_surface(self): + import microsoft_teams.apps as apps + + assert apps.create_state_loader is create_state_loader + assert "create_state_loader" in apps.__all__ + + def test_returns_none_when_disabled(self): + assert create_state_loader(None, LocalStorage()) is None + assert create_state_loader(False, LocalStorage()) is None + + def test_true_enables_on_fallback_storage_and_warns(self, caplog): + fallback = LocalStorage() + with caplog.at_level(logging.WARNING): + loader = create_state_loader(True, fallback) + assert isinstance(loader, TurnStateLoader) + assert any("localstorage" in r.message.lower() for r in caplog.records) + + async def test_options_storage_is_used_over_fallback(self): + dedicated: LocalStorage[str] = LocalStorage() + fallback: LocalStorage[str] = LocalStorage() + loader = create_state_loader(StateOptions(storage=dedicated), fallback) + assert loader is not None + + container = await loader.load("c1") + container.conversation["k"] = "v" + await loader.save(container) + + assert dedicated.get("ts:conv:c1") is not None + assert fallback.get("ts:conv:c1") is None + + def test_non_local_storage_does_not_warn(self, caplog): + fake_storage = MagicMock(spec=Storage) + with caplog.at_level(logging.WARNING): + loader = create_state_loader(StateOptions(storage=fake_storage), LocalStorage()) + assert isinstance(loader, TurnStateLoader) + assert not any("localstorage" in r.message.lower() for r in caplog.records) diff --git a/uv.lock b/uv.lock index a5af9d7c..7f66b7bf 100644 --- a/uv.lock +++ b/uv.lock @@ -35,6 +35,7 @@ members = [ "proactive-messaging", "quoting", "reactions", + "state", "stream", "suggested-actions", "tab", @@ -3113,6 +3114,23 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/ec/bb/2799cc2ede3ed41131f8975621e7213dfc7ef4acbbaadfa440f32500c370/starlette-1.3.1-py3-none-any.whl", hash = "sha256:c7372aae11c3c3f26a42df7bd626cec2f47d03483d261d369516a615a53714c6", size = 73632, upload-time = "2026-06-12T09:23:10.017Z" }, ] +[[package]] +name = "state" +version = "0.1.0" +source = { virtual = "examples/state" } +dependencies = [ + { name = "dotenv" }, + { name = "microsoft-teams-api" }, + { name = "microsoft-teams-apps" }, +] + +[package.metadata] +requires-dist = [ + { name = "dotenv", specifier = ">=0.9.9" }, + { name = "microsoft-teams-api", editable = "packages/api" }, + { name = "microsoft-teams-apps", editable = "packages/apps" }, +] + [[package]] name = "std-uritemplate" version = "2.0.8"