diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/__init__.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/__init__.py index a143c6a3..053b72dc 100644 --- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/__init__.py +++ b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/__init__.py @@ -11,7 +11,7 @@ from .input_file import InputFile, InputFileDownloader from .query import Query from ._routes import _RouteList, _Route, RouteRank -from .typing_indicator import TypingIndicator +from .typing_indicator import TypingChannelStrategy, TypingIndicator, TypingOptions from ._type_defs import RouteHandler, RouteSelector, StateT # Auth @@ -46,7 +46,9 @@ "Query", "Route", "RouteHandler", + "TypingChannelStrategy", "TypingIndicator", + "TypingOptions", "StatePropertyAccessor", "ConversationState", "state", diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/agent_application.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/agent_application.py index fbfdf8a2..b8b19331 100644 --- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/agent_application.py +++ b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/agent_application.py @@ -5,6 +5,7 @@ from __future__ import annotations import logging +from contextlib import nullcontext from copy import copy from functools import partial @@ -697,62 +698,69 @@ async def on_turn(self, context: TurnContext): await self._start_long_running_call(context, self._on_turn) async def _on_turn(self, context: TurnContext): - typing = None try: with spans.AppOnTurn(context) as on_turn_span: - if context.activity.type != ActivityTypes.typing: - if self._options.start_typing_timer: - typing = TypingIndicator(context) - typing.start() - - self._remove_mentions(context) - - logger.debug("Initializing turn state") - turn_state = await self._initialize_state(context) - if ( - context.activity.type == ActivityTypes.message - or context.activity.type == ActivityTypes.invoke - ): + use_typing = ( + self._options.start_typing_timer + and context.activity.type != ActivityTypes.typing + ) + typing_context = ( + TypingIndicator( + context, + typing_options=self._options.typing, + ) + if use_typing + else nullcontext() + ) - ( - auth_intercepts, - continuation_activity, - ) = await self._auth._on_turn_auth_intercept(context, turn_state) - if auth_intercepts: - if continuation_activity: - new_context = copy(context) - new_context.activity = continuation_activity - logger.info( - "Resending continuation activity %s", - continuation_activity.text, - ) - await self.on_turn(new_context) - await turn_state.save(context) + async with typing_context: + self._remove_mentions(context) + + logger.debug("Initializing turn state") + turn_state = await self._initialize_state(context) + if context.activity.type in [ + ActivityTypes.message, + ActivityTypes.invoke, + ]: + + ( + auth_intercepts, + continuation_activity, + ) = await self._auth._on_turn_auth_intercept( + context, turn_state + ) + if auth_intercepts: + if continuation_activity: + new_context = copy(context) + new_context.activity = continuation_activity + logger.info( + "Resending continuation activity %s", + continuation_activity.text, + ) + await self.on_turn(new_context) + await turn_state.save(context) + return + + logger.debug("Running before turn middleware") + if not await self._run_before_turn_middleware(context, turn_state): return - logger.debug("Running before turn middleware") - if not await self._run_before_turn_middleware(context, turn_state): - return + logger.debug("Running file downloads") + await self._handle_file_downloads(context, turn_state) - logger.debug("Running file downloads") - await self._handle_file_downloads(context, turn_state) + logger.debug("Running activity handlers") + await self._on_activity(context, turn_state, on_turn_span) - logger.debug("Running activity handlers") - await self._on_activity(context, turn_state, on_turn_span) - - logger.debug("Running after turn middleware") - if await self._run_after_turn_middleware(context, turn_state): - await turn_state.save(context) - return + logger.debug("Running after turn middleware") + if await self._run_after_turn_middleware(context, turn_state): + await turn_state.save(context) + return except ApplicationError as err: logger.error( f"An application error occurred in the AgentApplication: {err}", exc_info=True, ) await self._on_error(context, err) - finally: - if typing: - typing.stop() def _remove_mentions(self, context: TurnContext): if ( diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/app_options.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/app_options.py index a66ca494..506462cf 100644 --- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/app_options.py +++ b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/app_options.py @@ -13,6 +13,7 @@ from microsoft_agents.hosting.core.storage import Storage # from .auth import AuthOptions +from .typing_indicator import TypingOptions from .input_file import InputFileDownloader from ..channel_service_adapter import ChannelServiceAdapter @@ -62,6 +63,13 @@ class ApplicationOptions: the request. Defaults to true. """ + typing: Optional[TypingOptions] = None + """ + Optional. Typing indicator timing options. Controls initial delay, interval, + and per-channel strategies. If not provided, defaults are used when + ``start_typing_timer`` is true. + """ + long_running_messages: bool = False """ Optional. If true, the bot supports long running messages that can take longer then the 10 - 15 diff --git a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/typing_indicator.py b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/typing_indicator.py index e3841e85..cedab434 100644 --- a/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/typing_indicator.py +++ b/libraries/microsoft-agents-hosting-core/microsoft_agents/hosting/core/app/typing_indicator.py @@ -7,47 +7,181 @@ import asyncio import logging -from typing import Optional +from dataclasses import dataclass, field +from typing import Dict, Optional from microsoft_agents.hosting.core import TurnContext -from microsoft_agents.activity import Activity, ActivityTypes +from microsoft_agents.activity import Activity, ActivityTypes, Channels, EntityTypes logger = logging.getLogger(__name__) +DEFAULT_INITIAL_DELAY_MS = 500 +DEFAULT_INTERVAL_MS = 10000 + + +@dataclass +class TypingChannelStrategy: + """Per-channel timing parameters for the typing indicator. + + :param initial_delay_ms: Delay in milliseconds before the first typing activity. + :param interval_ms: Interval in milliseconds between subsequent typing activities. + """ + + initial_delay_ms: int = DEFAULT_INITIAL_DELAY_MS + interval_ms: int = DEFAULT_INTERVAL_MS + + +@dataclass +class TypingOptions: + """Configuration options for the automatic typing indicator. + + :param initial_delay_ms: Default delay in milliseconds before the first typing activity. + :param interval_ms: Default interval in milliseconds between subsequent typing activities. + :param channel_strategies: Optional per-channel timing overrides keyed by channel ID. + """ + + initial_delay_ms: int = DEFAULT_INITIAL_DELAY_MS + interval_ms: int = DEFAULT_INTERVAL_MS + channel_strategies: Dict[str, TypingChannelStrategy] = field(default_factory=dict) + + def __post_init__(self): + # Apply default channel overrides (matching .NET's M365Copilot default) + if Channels.copilot_studio.value not in self.channel_strategies: + # make a copy to avoid mutating input + self.channel_strategies = dict(self.channel_strategies) + self.channel_strategies[Channels.copilot_studio.value] = ( + TypingChannelStrategy(initial_delay_ms=250, interval_ms=1000) + ) + + def get_strategy_for_channel(self, channel: str) -> TypingChannelStrategy: + """Returns the timing strategy for the given channel, falling back to defaults.""" + if channel in self.channel_strategies: + return self.channel_strategies[channel] + return TypingChannelStrategy( + initial_delay_ms=self.initial_delay_ms, + interval_ms=self.interval_ms, + ) + class TypingIndicator: """ Encapsulates the logic for sending "typing" activity to the user. Scoped to a single turn of conversation with the user. + + Automatically stops when a message or streaming activity is about to be + sent on the same turn, preventing bare typing activities from overlapping + with real responses. + + Can be used as an async context manager:: + + async with TypingIndicator(context, typing_options=opts) as typing: + # typing indicators are sent automatically + ... """ - def __init__(self, context: TurnContext, interval_seconds: float = 10.0) -> None: + def __init__( + self, + context: TurnContext, + interval_seconds: Optional[float] = None, + typing_options: Optional[TypingOptions] = None, + ) -> None: """Initializes a new instance of the TypingIndicator class. :param context: The turn context. - :param interval_seconds: The interval in seconds between typing indicators. + :param interval_seconds: Interval in seconds between typing indicators. + When set, overrides the interval from ``typing_options``. + :param typing_options: Typing timing options including per-channel + strategies. If not provided, defaults are used. An explicit + ``interval_seconds`` value takes precedence over the interval + in ``typing_options`` for backward compatibility. """ - if interval_seconds <= 0: - raise ValueError("interval_seconds must be greater than 0") + options = typing_options or TypingOptions() + + channel = ( + context.activity.channel_id.channel if context.activity.channel_id else "" + ) or "" + strategy = options.get_strategy_for_channel(channel) + + interval = ( + interval_seconds + if interval_seconds is not None + else strategy.interval_ms / 1000.0 + ) + initial_delay = strategy.initial_delay_ms / 1000.0 + + if interval <= 0: + raise ValueError("interval must be greater than 0") + + if initial_delay < 0: + raise ValueError("initial_delay must be greater than or equal to 0") + self._context: TurnContext = context - self._interval: float = interval_seconds + self._interval: float = interval + self._initial_delay: float = initial_delay self._task: Optional[asyncio.Task[None]] = None + self._last_send: Optional[asyncio.Task[None]] = None + self._stopped: bool = False + self._hook_registered: bool = False + + async def __aenter__(self) -> "TypingIndicator": + self.start() + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: + await self._stop_async() + + async def _send_typing(self) -> None: + """Sends a single typing activity via the adapter, bypassing middleware.""" + ref = self._context.activity.get_conversation_reference() + typing_activity = TurnContext.apply_conversation_reference( + Activity(type=ActivityTypes.typing), ref + ) + await self._context.adapter.send_activities(self._context, [typing_activity]) async def _run(self) -> None: - """Sends typing indicators at regular intervals.""" - - running_task = self._task + """Sends typing indicators at regular intervals after an initial delay.""" try: - while running_task is self._task: - await self._context.send_activity(Activity(type=ActivityTypes.typing)) + if self._initial_delay > 0: + await asyncio.sleep(self._initial_delay) + + while not self._stopped: + send_task = asyncio.ensure_future(self._send_typing()) + self._last_send = send_task + # Shield the send so task cancellation doesn't interrupt + # an in-flight adapter call. + try: + await asyncio.shield(send_task) + except asyncio.CancelledError: + # Task was cancelled while sending — wait for the send + # to finish naturally, then exit the loop. + break await asyncio.sleep(self._interval) except asyncio.CancelledError: - # Task was cancelled, exit gracefully pass + except Exception: + logger.debug( + "Typing indicator send failed for conversation %s", + self._context.activity.conversation.id, + exc_info=True, + ) + + @staticmethod + def _has_streaminfo(activity: Activity) -> bool: + """Check if an activity contains a streaminfo entity.""" + if not activity.entities: + return False + for entity in activity.entities: + entity_type = getattr(entity, "type", None) + if entity_type is None and isinstance(entity, dict): + entity_type = entity.get("type") + if entity_type == EntityTypes.STREAM_INFO.value: + return True + return False def start(self) -> None: - """Starts sending typing indicators.""" + """Starts sending typing indicators and registers a send-activity hook + that auto-stops the indicator when a real response is about to be sent.""" if self._task is not None: logger.warning( @@ -57,25 +191,53 @@ def start(self) -> None: return logger.debug( - "Starting typing indicator with interval: %s seconds in conversation %s", + "Starting typing indicator (initial_delay=%.1fs, interval=%.1fs) " + "for conversation %s", + self._initial_delay, self._interval, self._context.activity.conversation.id, ) + + self._stopped = False self._task = asyncio.create_task(self._run()) - def stop(self) -> None: - """Stops sending typing indicators.""" + if not self._hook_registered: - if self._task is None: - logger.warning( - "Typing indicator is not running for conversation %s", - self._context.activity.conversation.id, - ) + async def _on_send_activities_handler(ctx, activities, next_handler): + should_stop = any( + a.type == ActivityTypes.message or self._has_streaminfo(a) + for a in activities + ) + if should_stop: + self._stop_loop() + return await next_handler() + + self._context.on_send_activities(_on_send_activities_handler) + self._hook_registered = True + + def _stop_loop(self) -> None: + """Cancels the background loop task. Does not await in-flight sends.""" + if self._stopped: return + self._stopped = True - logger.debug( - "Stopping typing indicator for conversation %s", - self._context.activity.conversation.id, - ) - self._task.cancel() - self._task = None + if self._task is not None: + self._task.cancel() + self._task = None + + def stop(self) -> None: + """Stops sending typing indicators synchronously.""" + self._stop_loop() + + async def _stop_async(self) -> None: + """Stops typing and waits for any in-flight send to finish.""" + self._stop_loop() + + # Wait for any in-flight typing send to finish so we don't + # interrupt an adapter call mid-flight. + if self._last_send and not self._last_send.done(): + try: + await asyncio.shield(self._last_send) + except (asyncio.CancelledError, Exception): + pass + self._last_send = None diff --git a/tests/hosting_core/app/test_agent_application.py b/tests/hosting_core/app/test_agent_application.py index b5bad921..40e184d5 100644 --- a/tests/hosting_core/app/test_agent_application.py +++ b/tests/hosting_core/app/test_agent_application.py @@ -1,33 +1,96 @@ -# from microsoft_agents.authentication.msal.msal_connection_manager import MsalConnectionManager -# from microsoft_agents.hosting.core.turn_context import TurnContext -# import pytest +""" +Copyright (c) Microsoft Corporation. All rights reserved. +Licensed under the MIT License. +""" -# from microsoft_agents.authentication.msal import MsalAuthentication -# from microsoft_agents.hosting.core import ( -# MemoryStorage, -# AgentApplication, -# ApplicationOptions, -# Connections -# ) +import asyncio +from unittest.mock import AsyncMock, patch -# # def mock_send_activity(mocker): -# # mocker.patch.object(TurnContext, 'send_activity', new=) +import pytest -# class TestUtils: +from microsoft_agents.activity import Activity, ActivityTypes +from microsoft_agents.hosting.core import MemoryStorage +from microsoft_agents.hosting.core.app import ( + AgentApplication, + ApplicationOptions, + TurnState, +) -# @pytest.fixture -# def options(self): -# return ApplicationOptions() -# @pytest.fixture -# def storage(self): -# return MemoryStorage() +class StubAdapter: + """Minimal adapter for testing AgentApplication._on_turn.""" -# @pytest.fixture -# def connection_manager(self): -# return MsalConnectionManager() + def __init__(self): + self.sent_activities: list[Activity] = [] + async def send_activities(self, context, activities): + self.sent_activities.extend(activities) + return [None] * len(activities) -# class TestAgentApplication: -# pass +class StubTurnContext: + """Minimal TurnContext double for AgentApplication tests.""" + + def __init__(self, activity: Activity, adapter=None): + self.activity = activity + self.adapter = adapter or StubAdapter() + self._on_send_handlers = [] + self.turn_state = {} + self.responded = False + + def on_send_activities(self, handler): + self._on_send_handlers.append(handler) + + async def send_activity(self, activity): + self.responded = True + + +@pytest.mark.asyncio +async def test_on_turn_no_typing_when_start_typing_timer_false(): + """When start_typing_timer=False, no typing indicators should be sent.""" + adapter = StubAdapter() + app = AgentApplication[TurnState]( + options=ApplicationOptions( + storage=MemoryStorage(), + start_typing_timer=False, + ) + ) + + context = StubTurnContext( + Activity( + type=ActivityTypes.message, + text="hello", + channel_id="test", + conversation={"id": "conv1"}, + from_property={"id": "user1"}, + recipient={"id": "bot1"}, + service_url="https://test", + ), + adapter=adapter, + ) + + # Patch internals to avoid needing full infrastructure + with patch.object(app, "_remove_mentions"), patch.object( + app, "_initialize_state", new_callable=AsyncMock, return_value=TurnState() + ), patch.object( + app, "_run_before_turn_middleware", new_callable=AsyncMock, return_value=True + ), patch.object( + app, "_handle_file_downloads", new_callable=AsyncMock + ), patch.object( + app, "_on_activity", new_callable=AsyncMock + ), patch.object( + app, "_run_after_turn_middleware", new_callable=AsyncMock, return_value=True + ): + await app._on_turn(context) + + # Give any background tasks a chance to fire (they shouldn't) + await asyncio.sleep(0.1) + + # No typing activities should have been sent via adapter + typing_activities = [ + a for a in adapter.sent_activities if a.type == ActivityTypes.typing + ] + assert len(typing_activities) == 0 + + # No on_send_activities hook should have been registered + assert len(context._on_send_handlers) == 0 diff --git a/tests/hosting_core/app/test_typing_indicator.py b/tests/hosting_core/app/test_typing_indicator.py index 70b9d75b..3fc2d71c 100644 --- a/tests/hosting_core/app/test_typing_indicator.py +++ b/tests/hosting_core/app/test_typing_indicator.py @@ -7,37 +7,152 @@ import pytest -from microsoft_agents.activity import Activity, ActivityTypes -from microsoft_agents.hosting.core.app.typing_indicator import TypingIndicator +from microsoft_agents.activity import Activity, ActivityTypes, Channels +from microsoft_agents.hosting.core.app.typing_indicator import ( + TypingChannelStrategy, + TypingIndicator, + TypingOptions, +) -class StubTurnContext: - """Test double that tracks sent activities.""" +class StubAdapter: + """Fake adapter that records activities sent via send_activities.""" def __init__(self, should_raise: bool = False) -> None: - self.sent_activities = [] + self.sent_activities: list[Activity] = [] self.should_raise = should_raise - self.activity = Activity(type="text", conversation={"id": "test_convo"}) - async def send_activity(self, activity: Activity): + async def send_activities(self, context, activities): if self.should_raise: raise RuntimeError("send_activity failure") - self.sent_activities.append(activity) - return None + self.sent_activities.extend(activities) + return [None] * len(activities) + + +class StubTurnContext: + """Test double that tracks sent activities.""" + + def __init__(self, should_raise: bool = False, channel_id: str = "test") -> None: + self.adapter = StubAdapter(should_raise) + self.activity = Activity( + type="message", + conversation={"id": "test_convo"}, + channel_id=channel_id, + service_url="https://test", + from_property={"id": "bot"}, + recipient={"id": "user"}, + ) + self._on_send_handlers = [] + + def on_send_activities(self, handler): + self._on_send_handlers.append(handler) + + @property + def sent_activities(self): + return self.adapter.sent_activities + + +# --------------------------------------------------------------------------- +# Helper to create fast options for timing-sensitive tests +# --------------------------------------------------------------------------- +def _fast_options(initial_delay_ms: int = 5, interval_ms: int = 10) -> TypingOptions: + """Create TypingOptions with fast timing for tests. + + Uses a sentinel channel so the built-in copilot_studio default doesn't + interfere. + """ + return TypingOptions( + initial_delay_ms=initial_delay_ms, + interval_ms=interval_ms, + channel_strategies={}, + ) + + +# =========================================================================== +# TypingOptions / TypingChannelStrategy unit tests +# =========================================================================== + + +class TestTypingOptions: + def test_defaults(self): + opts = TypingOptions() + assert opts.initial_delay_ms == 500 + assert opts.interval_ms == 10000 + + def test_copilot_studio_default_strategy(self): + opts = TypingOptions() + strategy = opts.get_strategy_for_channel(Channels.copilot_studio.value) + assert strategy.initial_delay_ms == 250 + assert strategy.interval_ms == 1000 + + def test_custom_channel_strategy(self): + opts = TypingOptions( + channel_strategies={ + "msteams": TypingChannelStrategy(initial_delay_ms=100, interval_ms=500) + } + ) + strategy = opts.get_strategy_for_channel("msteams") + assert strategy.initial_delay_ms == 100 + assert strategy.interval_ms == 500 + + def test_unknown_channel_falls_back_to_defaults(self): + opts = TypingOptions(initial_delay_ms=300, interval_ms=1500) + strategy = opts.get_strategy_for_channel("some_unknown_channel") + assert strategy.initial_delay_ms == 300 + assert strategy.interval_ms == 1500 + + def test_empty_channel_id_falls_back_to_defaults(self): + opts = TypingOptions() + strategy = opts.get_strategy_for_channel("") + assert strategy.initial_delay_ms == 500 + assert strategy.interval_ms == 10000 + + def test_override_copilot_studio_default(self): + """User can override the built-in copilot_studio default.""" + custom = TypingChannelStrategy(initial_delay_ms=999, interval_ms=8888) + opts = TypingOptions(channel_strategies={Channels.copilot_studio.value: custom}) + strategy = opts.get_strategy_for_channel(Channels.copilot_studio.value) + assert strategy.initial_delay_ms == 999 + assert strategy.interval_ms == 8888 + + +# =========================================================================== +# TypingIndicator tests (existing + new per-channel tests) +# =========================================================================== @pytest.mark.asyncio -async def test_start_sends_typing_activity(): - """Test that start() begins sending typing activities at regular interval_secondss.""" +async def test_start_sends_typing_activity_after_initial_delay(): + """Test that start() sends typing activities after the initial delay.""" context = StubTurnContext() - indicator = TypingIndicator(context, interval_seconds=0.01) # 10ms interval_seconds + opts = _fast_options(initial_delay_ms=150, interval_ms=10) + indicator = TypingIndicator(context, typing_options=opts) indicator.start() - await asyncio.sleep(0.05) # Wait 50ms to allow multiple typing activities - indicator.stop() + await asyncio.sleep(0.05) + # Should NOT have sent yet (still in initial delay) + assert len(context.sent_activities) == 0 + + await asyncio.sleep(0.25) + await indicator._stop_async() - # Should have sent at least 3 typing activities (50ms / 10ms = 5, but accounting for timing) - assert len(context.sent_activities) >= 3 + assert len(context.sent_activities) >= 1 + assert all( + activity.type == ActivityTypes.typing for activity in context.sent_activities + ) + + +@pytest.mark.asyncio +async def test_start_sends_typing_at_interval(): + """Test that start() sends multiple typing activities at regular intervals.""" + context = StubTurnContext() + indicator = TypingIndicator(context, typing_options=_fast_options()) + + indicator.start() + await asyncio.sleep(0.08) + await indicator._stop_async() + + assert len(context.sent_activities) >= 2 assert all( activity.type == ActivityTypes.typing for activity in context.sent_activities ) @@ -47,69 +162,67 @@ async def test_start_sends_typing_activity(): async def test_start_creates_task(): """Test that start() creates an asyncio task.""" context = StubTurnContext() - indicator = TypingIndicator(context) + indicator = TypingIndicator(context, typing_options=_fast_options()) indicator.start() assert indicator._task is not None assert isinstance(indicator._task, asyncio.Task) - indicator.stop() + await indicator._stop_async() @pytest.mark.asyncio async def test_start_if_already_running(): """Test that start() is idempotent if already running.""" context = StubTurnContext() - indicator = TypingIndicator(context) + indicator = TypingIndicator(context, typing_options=_fast_options()) indicator.start() indicator.start() - indicator.stop() + await indicator._stop_async() @pytest.mark.asyncio async def test_stop_if_not_running(): """Test that stop() is idempotent if not running.""" context = StubTurnContext() - indicator = TypingIndicator(context) - indicator.stop() + indicator = TypingIndicator(context, typing_options=_fast_options()) + await indicator._stop_async() @pytest.mark.asyncio async def test_stop_prevents_further_typing_activities(): """Test that stop() prevents further typing activities from being sent.""" context = StubTurnContext() - indicator = TypingIndicator(context, interval_seconds=0.01) + indicator = TypingIndicator(context, typing_options=_fast_options()) indicator.start() - await asyncio.sleep(0.025) # Let it run briefly - indicator.stop() + await asyncio.sleep(0.025) + await indicator._stop_async() count_before = len(context.sent_activities) - await asyncio.sleep(0.03) # Wait more time + await asyncio.sleep(0.03) count_after = len(context.sent_activities) - assert count_before == count_after # No new activities after stop + assert count_before == count_after @pytest.mark.asyncio async def test_multiple_start_stop_cycles(): """Test that the indicator can be started and stopped multiple times.""" context = StubTurnContext() - indicator = TypingIndicator(context, interval_seconds=0.01) + indicator = TypingIndicator(context, typing_options=_fast_options()) - # First cycle indicator.start() await asyncio.sleep(0.02) - indicator.stop() + await indicator._stop_async() count_first = len(context.sent_activities) - # Second cycle indicator.start() await asyncio.sleep(0.02) - indicator.stop() + await indicator._stop_async() count_second = len(context.sent_activities) assert count_second > count_first @@ -119,13 +232,299 @@ async def test_multiple_start_stop_cycles(): async def test_typing_activity_format(): """Test that sent activities are properly formatted typing activities.""" context = StubTurnContext() - indicator = TypingIndicator(context, interval_seconds=0.01) + indicator = TypingIndicator(context, typing_options=_fast_options()) indicator.start() - await asyncio.sleep(0.015) # Wait for at least one activity - indicator.stop() + await asyncio.sleep(0.05) + await indicator._stop_async() assert len(context.sent_activities) >= 1 for activity in context.sent_activities: assert isinstance(activity, Activity) assert activity.type == ActivityTypes.typing + + +@pytest.mark.asyncio +async def test_typing_activity_has_conversation_reference(): + """Test that typing activities include conversation reference from the turn.""" + context = StubTurnContext() + indicator = TypingIndicator(context, typing_options=_fast_options()) + + indicator.start() + await asyncio.sleep(0.05) + await indicator._stop_async() + + assert len(context.sent_activities) >= 1 + activity = context.sent_activities[0] + assert activity.conversation is not None + assert activity.conversation.id == "test_convo" + + +@pytest.mark.asyncio +async def test_stop_is_idempotent(): + """Calling stop() multiple times should not error.""" + context = StubTurnContext() + indicator = TypingIndicator(context, typing_options=_fast_options()) + + indicator.start() + await asyncio.sleep(0.05) + await indicator._stop_async() + await indicator._stop_async() # should not raise + + +@pytest.mark.asyncio +async def test_send_failure_stops_gracefully(): + """If the adapter throws, the indicator should stop without raising.""" + context = StubTurnContext(should_raise=True) + indicator = TypingIndicator(context, typing_options=_fast_options()) + + indicator.start() + await asyncio.sleep(0.03) + await indicator._stop_async() + + +@pytest.mark.asyncio +async def test_send_hook_does_not_block_on_inflight_typing_send(): + """Hook should stop typing without delaying real outbound activities.""" + context = StubTurnContext() + opts = _fast_options(initial_delay_ms=5000, interval_ms=1000) + indicator = TypingIndicator(context, typing_options=opts) + indicator.start() + + assert len(context._on_send_handlers) == 1 + handler = context._on_send_handlers[0] + + release_send = asyncio.Event() + + async def _blocked_send(): + await release_send.wait() + + indicator._last_send = asyncio.create_task(_blocked_send()) + + next_handler_called = False + + async def _next_handler(): + nonlocal next_handler_called + next_handler_called = True + + # This call would block if the hook awaited stop() while _last_send is in flight. + await asyncio.wait_for( + handler(context, [Activity(type=ActivityTypes.message)], _next_handler), + timeout=0.05, + ) + + assert next_handler_called + + release_send.set() + await indicator._stop_async() + + +# --------------------------------------------------------------------------- +# Per-channel strategy tests +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_channel_strategy_controls_timing(): + """A per-channel strategy should control the initial delay and interval.""" + context = StubTurnContext(channel_id="msteams") + opts = TypingOptions( + initial_delay_ms=9999, # very large default — would time-out + interval_ms=9999, + channel_strategies={ + "msteams": TypingChannelStrategy(initial_delay_ms=5, interval_ms=10) + }, + ) + indicator = TypingIndicator(context, typing_options=opts) + + indicator.start() + await asyncio.sleep(0.06) + await indicator._stop_async() + + # Should have sent typing despite the large defaults, + # because the msteams strategy overrides. + assert len(context.sent_activities) >= 1 + + +@pytest.mark.asyncio +async def test_unknown_channel_uses_default_timing(): + """Unknown channels should fall back to the default timing.""" + context = StubTurnContext(channel_id="some_custom_channel") + opts = TypingOptions(initial_delay_ms=5, interval_ms=10) + indicator = TypingIndicator(context, typing_options=opts) + + indicator.start() + await asyncio.sleep(0.06) + await indicator._stop_async() + + assert len(context.sent_activities) >= 1 + + +@pytest.mark.asyncio +async def test_copilot_studio_uses_builtin_defaults(): + """Copilot Studio should use the built-in 250ms/1000ms defaults.""" + context = StubTurnContext(channel_id=Channels.copilot_studio.value) + opts = TypingOptions() # uses built-in defaults + indicator = TypingIndicator(context, typing_options=opts) + + # Verify the resolved timing + assert indicator._initial_delay == 0.25 + assert indicator._interval == 1.0 + + await indicator._stop_async() + + +@pytest.mark.asyncio +async def test_no_options_uses_global_defaults(): + """When no TypingOptions are provided, global defaults (500ms/10000ms) apply.""" + context = StubTurnContext(channel_id="test") + indicator = TypingIndicator(context) + + assert indicator._initial_delay == 0.5 + assert indicator._interval == 10.0 + + await indicator._stop_async() + + +@pytest.mark.asyncio +async def test_invalid_interval_raises(): + """interval_ms <= 0 should raise ValueError.""" + context = StubTurnContext() + opts = TypingOptions(interval_ms=0) + with pytest.raises(ValueError, match="interval"): + TypingIndicator(context, typing_options=opts) + + +@pytest.mark.asyncio +async def test_negative_initial_delay_strategy_raises(): + """initial_delay_ms < 0 in a channel strategy should raise ValueError + at the strategy level (TypingIndicator won't accept it).""" + context = StubTurnContext() + # A negative initial_delay_ms in the options still resolves to a negative + # float internally — but since initial_delay is not validated in __init__ + # (only interval is), this is a TypingOptions-level concern. + # We verify the indicator uses the value correctly. + channel_strategies = { + "test": TypingChannelStrategy(initial_delay_ms=-1, interval_ms=10) + } + opts = TypingOptions( + initial_delay_ms=0, interval_ms=10, channel_strategies=channel_strategies + ) + with pytest.raises(ValueError, match="initial_delay"): + TypingIndicator(context, typing_options=opts) + + +@pytest.mark.asyncio +async def test_negative_initial_delay_raises(): + """initial_delay_ms < 0 should raise ValueError + at the strategy level (TypingIndicator won't accept it).""" + context = StubTurnContext() + # A negative initial_delay_ms in the options still resolves to a negative + # float internally — but since initial_delay is not validated in __init__ + # (only interval is), this is a TypingOptions-level concern. + # We verify the indicator uses the value correctly. + opts = TypingOptions(initial_delay_ms=-1, interval_ms=10) + with pytest.raises(ValueError, match="initial_delay"): + TypingIndicator(context, typing_options=opts) + + +@pytest.mark.asyncio +async def test_negative_interval_ms_raises(): + """initial_delay_ms < 0 in a channel strategy should raise ValueError + at the strategy level (TypingIndicator won't accept it).""" + context = StubTurnContext() + # A negative initial_delay_ms in the options still resolves to a negative + # float internally — but since initial_delay is not validated in __init__ + # (only interval is), this is a TypingOptions-level concern. + # We verify the indicator uses the value correctly. + opts = TypingOptions(initial_delay_ms=1000, interval_ms=-1) + with pytest.raises(ValueError, match="interval"): + TypingIndicator(context, typing_options=opts) + + +# --------------------------------------------------------------------------- +# Backward compatibility tests (legacy constructor parameter) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_legacy_interval_seconds_parameter(): + """The old interval_seconds parameter should still work.""" + context = StubTurnContext() + indicator = TypingIndicator(context, interval_seconds=0.01) + + assert indicator._interval == 0.01 + # initial_delay comes from default strategy (500ms for "test" channel) + assert indicator._initial_delay == 0.5 + + indicator.start() + await asyncio.sleep(0.6) + await indicator._stop_async() + assert len(context.sent_activities) >= 1 + + +@pytest.mark.asyncio +async def test_legacy_params_override_typing_options(): + """Explicit interval_seconds takes precedence over typing_options.""" + context = StubTurnContext() + opts = TypingOptions(initial_delay_ms=5, interval_ms=9999) + indicator = TypingIndicator( + context, + interval_seconds=0.01, + typing_options=opts, + ) + + assert indicator._interval == 0.01 + # initial_delay still comes from typing_options + assert indicator._initial_delay == 0.005 + + +# --------------------------------------------------------------------------- +# Async context manager tests +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_async_context_manager_starts_and_stops(): + """TypingIndicator can be used as an async context manager.""" + context = StubTurnContext() + opts = _fast_options() + + async with TypingIndicator(context, typing_options=opts) as indicator: + await asyncio.sleep(0.05) + assert indicator._task is not None + + # After exiting, should be stopped + assert indicator._stopped is True + assert len(context.sent_activities) >= 1 + + +@pytest.mark.asyncio +async def test_async_context_manager_stops_on_exception(): + """Context manager stops typing even if body raises.""" + context = StubTurnContext() + opts = _fast_options() + + with pytest.raises(RuntimeError): + async with TypingIndicator(context, typing_options=opts) as indicator: + await asyncio.sleep(0.02) + raise RuntimeError("test error") + + assert indicator._stopped is True + + +# --------------------------------------------------------------------------- +# nullcontext fallback test (simulates start_typing_timer=False in AgentApplication) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_nullcontext_fallback_with_async_with(): + """When typing is None, `async with typing or nullcontext()` should work.""" + from contextlib import nullcontext + + typing = None # simulates start_typing_timer=False + + # This must not raise + async with typing or nullcontext(): + pass # turn logic would run here