diff --git a/examples/graph/src/main.py b/examples/graph/src/main.py
index dad84d1c8..f36df8603 100644
--- a/examples/graph/src/main.py
+++ b/examples/graph/src/main.py
@@ -9,7 +9,7 @@
 
 from azure.core.exceptions import ClientAuthenticationError
 from microsoft_teams.api import MessageActivity
-from microsoft_teams.apps import ActivityContext, App, AppOptions, ErrorEvent, SignInEvent
+from microsoft_teams.apps import ActivityContext, App, ErrorEvent, SignInEvent
 from microsoft_teams.common import ConsoleFormatter
 from microsoft_teams.graph import get_graph_client
 from msgraph.generated.users.item.messages.messages_request_builder import (  # type: ignore
@@ -23,8 +23,8 @@
 logging.getLogger().addHandler(stream_handler)
 logger = logging.getLogger(__name__)
 
-app_options = AppOptions(default_connection_name=os.getenv("CONNECTION_NAME", "graph"))
-app = App(**app_options)
+app = App()
+graph_flow = app.add_oauth_flow(os.getenv("CONNECTION_NAME", "graph"))
 
 
 async def get_authenticated_graph_client(ctx: ActivityContext[MessageActivity]):
@@ -34,40 +34,40 @@ async def get_authenticated_graph_client(ctx: ActivityContext[MessageActivity]):
     Returns:
         Graph client if successful, None if authentication failed.
     """
-    # Check if user is signed in
-    if not ctx.is_signed_in:
+    token = await graph_flow.get_token(ctx)
+    if token is None:
         await ctx.send("🔐 Please sign in first to access Microsoft Graph.")
-        await ctx.sign_in()
+        await graph_flow.sign_in(ctx)
         return None
 
     try:
         # Create Graph client using the user token
-        return get_graph_client(ctx.user_token)
+        return get_graph_client(token)
 
     except Exception as e:
         logger.error(f"Failed to create Graph client: {e}")
         await ctx.send("🔐 Failed to create authenticated client. Please try signing in again.")
-        await ctx.sign_in()
+        await graph_flow.sign_in(ctx)
         return None
 
 
 @app.on_message_pattern("signin")
 async def handle_signin_command(ctx: ActivityContext[MessageActivity]):
     """Handle sign-in command."""
-    if ctx.is_signed_in:
+    if await graph_flow.is_signed_in(ctx):
         await ctx.send("✅ You are already signed in!")
     else:
         await ctx.send("🔐 Please sign in to access Microsoft Graph...")
-        await ctx.sign_in()
+        await graph_flow.sign_in(ctx)
 
 
 @app.on_message_pattern("signout")
 async def handle_signout_command(ctx: ActivityContext[MessageActivity]):
     """Handle sign-out command."""
-    if not ctx.is_signed_in:
+    if not await graph_flow.is_signed_in(ctx):
         await ctx.send("ℹ️ You are not currently signed in.")
     else:
-        await ctx.sign_out()
+        await graph_flow.sign_out(ctx)
         await ctx.send("👋 You have been signed out successfully!")
 
 
@@ -98,7 +98,7 @@ async def handle_profile_command(ctx: ActivityContext[MessageActivity]):
     except ClientAuthenticationError as e:
         logger.error(f"Authentication error: {e}")
         await ctx.send("🔐 Authentication failed. Please try signing in again.")
-        await ctx.sign_in()
+        await graph_flow.sign_in(ctx)
 
     except Exception as e:
         logger.error(f"Error getting profile: {e}")
diff --git a/examples/oauth/pyproject.toml b/examples/oauth/pyproject.toml
index 87fce76bc..590b95582 100644
--- a/examples/oauth/pyproject.toml
+++ b/examples/oauth/pyproject.toml
@@ -7,7 +7,9 @@ requires-python = ">=3.11,<4.0"
 dependencies = [
     "dotenv>=0.9.9",
     "microsoft-teams-apps",
+    "microsoft-teams-graph",
 ]
 
 [tool.uv.sources]
 microsoft-teams-apps = { workspace = true }
+microsoft-teams-graph = { workspace = true }
diff --git a/examples/oauth/src/main.py b/examples/oauth/src/main.py
index 9215f91d6..7dfd459db 100644
--- a/examples/oauth/src/main.py
+++ b/examples/oauth/src/main.py
@@ -7,55 +7,106 @@
 import logging
 
 from microsoft_teams.api import MessageActivity
-from microsoft_teams.api.activities.invoke.sign_in import SignInFailureInvokeActivity
 from microsoft_teams.apps import ActivityContext, App, SignInEvent
-from microsoft_teams.apps.events.types import ErrorEvent
+from microsoft_teams.apps.events.types import ErrorEvent, SignInFailureEvent
 from microsoft_teams.common import ConsoleFormatter
+from microsoft_teams.graph import get_graph_client
 
 # Setup logging
-logging.getLogger().setLevel(logging.DEBUG)
+logging.getLogger().setLevel(logging.INFO)
 stream_handler = logging.StreamHandler()
 stream_handler.setFormatter(ConsoleFormatter())
 logging.getLogger().addHandler(stream_handler)
 logger = logging.getLogger(__name__)
 
-app = App()
+# Pending sign-in hints are stored in per-turn state, which lets connection-less
+# callbacks be routed back to the flow that started them.
+app = App(state=True)
 
+# Multi-connection OAuth: two named connections registered with add_oauth_flow, each
+# driven through the OAuthFlow it returns.
+#
+# Bot needs two OAuth connections configured in Azure, named "profile" and
+# "mail", matching the names passed below. Grant "User.Read" to the first and
+# "Mail.Read" to the second.
+profile = app.add_oauth_flow("profile", oauth_card_text="Sign in to read your profile")
+mail = app.add_oauth_flow("mail", oauth_card_text="Sign in to read your mail")
+
+
+@profile.on_signin
+async def on_profile_signin(event: SignInEvent) -> None:
+    """Only fires for the `profile` connection, using that connection's token."""
+    client = get_graph_client(event.token_response.token)
+    me = await client.me.get()
+    name = me.display_name if me else "unknown"
+    await event.activity_ctx.send(f"Signed in as **{name}**.")
+
+
+@mail.on_signin
+async def on_mail_signin(event: SignInEvent) -> None:
+    """Only fires for the `mail` connection — a token the profile flow does not have."""
+    client = get_graph_client(event.token_response.token)
+    page = await client.me.messages.get()
+    subjects = [m.subject or "(no subject)" for m in (page.value or [])[:3]] if page else []
+    body = "\n".join(f"- {s}" for s in subjects) if subjects else "_no messages_"
+    await event.activity_ctx.send(f"Latest mail:\n{body}")
+
+
+@profile.on_signin_failure
+async def on_profile_failure(event: SignInFailureEvent) -> None:
+    await event.activity_ctx.send(f"Profile sign-in failed: {event.code} - {event.message}")
 
-@app.on_message
-async def handle_message(ctx: ActivityContext[MessageActivity]):
-    """Handle message activities using the new generated handler system."""
-    print(f"[GENERATED onMessage] Message received: {ctx.activity.text}")
-    print(f"[GENERATED onMessage] From: {ctx.activity.from_}")
 
-    logger.info("User requested sign-in.")
-    if ctx.is_signed_in:
-        await ctx.send("You are already signed in. Logging you out.")
-        await ctx.sign_out()
-    else:
-        await ctx.sign_in()
+@mail.on_signin_failure
+async def on_mail_failure(event: SignInFailureEvent) -> None:
+    await event.activity_ctx.send(f"Mail sign-in failed: {event.code} - {event.message}")
 
 
 @app.event("sign_in")
-async def handle_sign_in(event: SignInEvent):
-    """Handle sign-in events."""
-    await event.activity_ctx.send("You are now signed in!")
+async def on_any_signin(event: SignInEvent) -> None:
+    """Fires for every connection. `connection_name` says which one completed."""
+    logger.info("sign-in completed on connection %r", event.connection_name)
+
+
+@app.on_message
+async def handle_message(ctx: ActivityContext[MessageActivity]) -> None:
+    # The bot is @mentioned in group chats and channels, so drop the mention
+    # before dispatching to keep the same commands working in every scope.
+    text = (ctx.activity.strip_mentions_text().text or "").strip().lower()
+
+    if text == "sign in profile":
+        # Returns the token directly when one is already cached, otherwise sends a
+        # card and returns None; the flow's on_signin handler fires once it completes.
+        if await profile.sign_in(ctx):
+            await ctx.send("Already signed in for profile access.")
+        return
+
+    if text == "sign in mail":
+        if await mail.sign_in(ctx):
+            await ctx.send("Already signed in for mail access.")
+        return
+
+    if text == "sign out":
+        await profile.sign_out(ctx)
+        await mail.sign_out(ctx)
+        await ctx.send("Signed out of both connections.")
+        return
 
+    if text == "status":
+        # Tokens are tracked per connection, so these can differ.
+        lines: list[str] = []
+        for flow in (profile, mail):
+            state = "signed in" if await flow.is_signed_in(ctx) else "signed out"
+            lines.append(f"- `{flow.connection_name}`: {state}")
+        await ctx.send("\n".join(lines))
+        return
 
-@app.on_signin_failure()
-async def handle_signin_failure(ctx: ActivityContext[SignInFailureInvokeActivity]):
-    """Handle sign-in failure events."""
-    failure = ctx.activity.value
-    print(f"Sign-in failed: {failure.code} - {failure.message}")
-    await ctx.send("Sign-in failed.")
+    await ctx.send("Try `sign in profile`, `sign in mail`, `status`, or `sign out`.")
 
 
 @app.event("error")
-async def handle_error(event: ErrorEvent):
-    """Handle error events."""
-    print(f"Error occurred: {event.error}")
-    if event.context:
-        print(f"Context: {event.context}")
+async def handle_error(event: ErrorEvent) -> None:
+    logger.error("error: %s (context=%s)", event.error, event.context)
 
 
 if __name__ == "__main__":
diff --git a/packages/apps/src/microsoft_teams/apps/app_oauth.py b/packages/apps/src/microsoft_teams/apps/app_oauth.py
index a579549c2..7be5c9f50 100644
--- a/packages/apps/src/microsoft_teams/apps/app_oauth.py
+++ b/packages/apps/src/microsoft_teams/apps/app_oauth.py
@@ -3,9 +3,12 @@
 Licensed under the MIT License.
 """
 
+import asyncio
 import logging
-from time import perf_counter
-from typing import Optional, Union
+from asyncio import Future
+from dataclasses import dataclass
+from time import perf_counter, time
+from typing import Dict, Optional, Union
 
 from httpx import HTTPStatusError
 from microsoft_teams.api import (
@@ -32,10 +35,30 @@
 from .events import ErrorEvent, EventType, SignInEvent, SignInFailureEvent
 from .oauth_connection import connection_lookup_key
 from .oauth_flow import OAuthFlow, OAuthFlowRegistry
+from .oauth_state import (
+    TOKEN_EXCHANGE_DEDUP_TTL_SECONDS,
+    has_completed_token_exchange,
+    record_completed_token_exchange,
+)
 from .routing import ActivityContext
 
 logger = logging.getLogger(__name__)
 
+TokenExchangeResult = Union[TokenExchangeInvokeResponseType, InvokeResponse[TokenExchangeInvokeResponseType]]
+
+_TOKEN_EXCHANGE_DEDUP_MAX_ENTRIES = 1000
+"""Cap on the in-memory completed-marker set, so a long-lived process cannot grow it
+without bound even if entries are added faster than the TTL retires them."""
+
+
+@dataclass(frozen=True)
+class _TokenExchangeOutcome:
+    """How an owned token exchange settled, replayed to concurrent duplicates."""
+
+    response: TokenExchangeResult = None
+    error: Optional[BaseException] = None
+    token_redeemed: bool = False
+
 
 class OauthHandlers:
     def __init__(
@@ -47,17 +70,90 @@ def __init__(
         self.default_connection_name = default_connection_name
         self.event_emitter = event_emitter
         self.oauth_registry = oauth_registry
+        # Dedup bookkeeping is per-``App`` instance state rather than module-level
+        # globals, so two apps in one process (and two tests in one session) never
+        # short-circuit each other's exchanges.
+        self._token_exchange_in_flight: Dict[str, Future[_TokenExchangeOutcome]] = {}
+        self._token_exchange_completed: Dict[str, float] = {}
 
     async def sign_in_token_exchange(
         self, ctx: ActivityContext[SignInTokenExchangeInvokeActivity]
-    ) -> Union[TokenExchangeInvokeResponseType, InvokeResponse[TokenExchangeInvokeResponseType]]:
-        """
-        Decorator to register a function that handles the sign-in token exchange.
+    ) -> TokenExchangeResult:
+        """Handle ``signin/tokenExchange``, deduplicating repeats of the same exchange.
+
+        Teams fans the same exchange out to every signed-in client endpoint, so the
+        bot can see it several times. Duplicates short-circuit to a ``200`` no-op and
+        deliberately skip ``ctx.next()``: the whole point of dedup is that the sign-in
+        side effects — the ``sign_in`` event, the flow callbacks, and the rest of the
+        middleware chain — run exactly once per exchange.
         """
+        exchange_id = ctx.activity.value.id
+        if not exchange_id:
+            # Teams normally stamps every exchange with an id. Without one there is
+            # nothing safe to key on, and collapsing every id-less exchange onto a
+            # shared empty key would drop unrelated sign-ins, so run undeduplicated.
+            return await self._run_token_exchange(ctx)
+
+        # Claim the exchange. The in-flight lookup, the in-flight insert and the
+        # completed-marker check below contain no ``await``, so the event loop cannot
+        # switch coroutines part way through: on a single loop this claim is atomic,
+        # which is what stops two concurrent duplicates from both starting an exchange.
+        #
+        # In-flight is checked first so a running exchange always wins over its own
+        # completed marker. The marker is stamped the moment the token is redeemed,
+        # while the owner still has sign-in callbacks to run, and a duplicate that
+        # landed in that window must mirror the owner rather than answer ahead of it.
+        in_flight = self._token_exchange_in_flight.get(exchange_id)
+        if in_flight is not None:
+            return await self._await_token_exchange(ctx, exchange_id, in_flight)
+        if self._is_completed_token_exchange(ctx, exchange_id):
+            return self._replay_completed_token_exchange(ctx, exchange_id)
+        owned: Future[_TokenExchangeOutcome] = asyncio.get_running_loop().create_future()
+        self._token_exchange_in_flight[exchange_id] = owned
+
+        try:
+            response = await self._run_token_exchange(ctx)
+        except BaseException as error:
+            # ``BaseException`` so cancellation also releases the entry and wakes
+            # waiters, instead of leaking the id until the process restarts.
+            self._settle_token_exchange(
+                exchange_id,
+                owned,
+                # ``token_redeemed`` matters on this path too: the exchange can fail
+                # *after* the token was already spent, and a waiter that does not know
+                # the marker was written would let its own stale snapshot erase it.
+                _TokenExchangeOutcome(
+                    error=error,
+                    token_redeemed=exchange_id in self._token_exchange_completed,
+                ),
+            )
+            raise
+        self._settle_token_exchange(
+            exchange_id,
+            owned,
+            _TokenExchangeOutcome(
+                response=response,
+                token_redeemed=exchange_id in self._token_exchange_completed,
+            ),
+        )
+        return response
+
+    async def _run_token_exchange(self, ctx: ActivityContext[SignInTokenExchangeInvokeActivity]) -> TokenExchangeResult:
+        """Perform the exchange itself for the request that owns this exchange id."""
         activity = ctx.activity
         api = ctx.api
         next_handler = ctx.next
         connection_name = activity.value.connection_name
+        # Resolved before the ``try`` so the ``finally`` below can always report it. The
+        # lookup is a plain dict access; computing it inside the ``try`` left it unbound
+        # whenever anything above it raised, which would turn the metric write in
+        # ``finally`` into an ``UnboundLocalError`` masking the original exception.
+        flow = self.oauth_registry.get(connection_name)
+        # Teams echoes back whatever casing the sign-in card carried, so telemetry keyed
+        # on the raw name splits one connection into several series. ``connection_name``
+        # stays raw: it is what goes on the wire to the Token Service, and rewriting that
+        # would be a behavior change rather than a telemetry fix.
+        event_connection_name = flow.connection_name if flow is not None else connection_name
         result = APP_OAUTH_RESULTS.failure
         started_at = perf_counter()
         try:
@@ -66,12 +162,9 @@ async def sign_in_token_exchange(
                 record_exception=False,
                 set_status_on_exception=False,
             ) as span:
-                span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_connection, connection_name)
+                span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_connection, event_connection_name)
                 span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_operation, APP_OAUTH_OPERATIONS.token_exchange)
 
-                flow = self.oauth_registry.get(connection_name)
-                event_connection_name = flow.connection_name if flow is not None else connection_name
-
                 if (
                     connection_lookup_key(connection_name) != connection_lookup_key(self.default_connection_name)
                     and flow is None
@@ -97,6 +190,9 @@ async def sign_in_token_exchange(
                 except HTTPStatusError as e:
                     status = e.response.status_code
                     if status not in (404, 400, 412):
+                        self.oauth_registry._clear_pending(  # pyright: ignore[reportPrivateUsage]
+                            ctx, event_connection_name
+                        )
                         logger.error(
                             f"Error exchanging token for user {activity.from_.id} in "
                             f"conversation {activity.conversation.id}: {e}"
@@ -108,7 +204,7 @@ async def sign_in_token_exchange(
                         error_type = APP_OAUTH_ERROR_TYPES.http_error
                         span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_error_type, error_type)
                         record_exception(span, e)
-                        record_oauth_error(connection_name, APP_OAUTH_OPERATIONS.token_exchange, error_type)
+                        record_oauth_error(event_connection_name, APP_OAUTH_OPERATIONS.token_exchange, error_type)
                         status = status or 500
                         result = APP_OAUTH_RESULTS.failure
                         span.set_attribute(APP_ATTRIBUTE_NAMES.invoke_response_status, status)
@@ -144,11 +240,16 @@ async def sign_in_token_exchange(
                     error_type = APP_OAUTH_ERROR_TYPES.exception
                     span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_error_type, error_type)
                     record_exception(span, e)
-                    record_oauth_error(connection_name, APP_OAUTH_OPERATIONS.token_exchange, error_type)
+                    record_oauth_error(event_connection_name, APP_OAUTH_OPERATIONS.token_exchange, error_type)
                     result = APP_OAUTH_RESULTS.failure
                     span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_result, result)
                     raise
 
+                # Recorded as soon as the exchange succeeds, before any sign-in side
+                # effects: the exchange token is spent at this point, so a retry could
+                # never succeed anyway. A failed exchange is never marked, leaving the
+                # id free for a genuine retry.
+                await self._record_completed_token_exchange(ctx, activity.value.id)
                 ctx.is_signed_in = True
                 ctx.user_token = token.token
                 self.oauth_registry._clear_pending(  # pyright: ignore[reportPrivateUsage]
@@ -168,13 +269,211 @@ async def sign_in_token_exchange(
                 return None
         finally:
             record_oauth_operation(
-                connection_name,
+                event_connection_name,
                 APP_OAUTH_OPERATIONS.token_exchange,
                 result,
                 (perf_counter() - started_at) * 1000,
             )
             await next_handler()
 
+    def _is_completed_token_exchange(
+        self, ctx: ActivityContext[SignInTokenExchangeInvokeActivity], exchange_id: str
+    ) -> bool:
+        self._prune_completed_token_exchanges()
+        if exchange_id in self._token_exchange_completed:
+            return True
+        return self._read_persisted_marker(ctx, exchange_id)
+
+    def _resolved_connection_name(self, connection_name: str) -> str:
+        """The registered casing for ``connection_name``.
+
+        Teams echoes back whatever casing the sign-in card carried, so the same
+        connection can arrive as ``Graph`` on one request and ``graph`` on the next.
+        The duplicate paths report this name to telemetry, and an unresolved name
+        would split one connection into several series.
+
+        Resolution is a registry dict lookup, exactly what ``_run_token_exchange``
+        does for ``event_connection_name``. Note that ``_run_token_exchange`` still
+        reports the *raw* name to its own telemetry; making that consistent is a
+        wider change than this dedup path and is deliberately left alone here.
+        """
+        flow = self.oauth_registry.get(connection_name)
+        return flow.connection_name if flow is not None else connection_name
+
+    def _read_persisted_marker(self, ctx: ActivityContext[SignInTokenExchangeInvokeActivity], exchange_id: str) -> bool:
+        """Read the cross-instance marker without ever failing the turn.
+
+        This runs before the owning request reaches ``_run_token_exchange``'s
+        ``try``/``finally``, so an escaping state error would skip ``ctx.next()`` and
+        stall the middleware chain. The persisted layer is best-effort by design -- the
+        in-memory guard is the authoritative same-instance one -- so an unreadable
+        store degrades to in-memory-only dedup rather than taking the turn down.
+
+        ``Exception``, not ``BaseException``: a cancellation still belongs to the task.
+        """
+        try:
+            return has_completed_token_exchange(ctx.state, exchange_id)
+        except Exception:
+            logger.exception("Unable to read persisted OAuth token exchange state; deduplicating in memory only.")
+            return False
+
+    async def _record_completed_token_exchange(
+        self, ctx: ActivityContext[SignInTokenExchangeInvokeActivity], exchange_id: str
+    ) -> None:
+        if not exchange_id:
+            return
+
+        self._prune_completed_token_exchanges()
+        self._token_exchange_completed[exchange_id] = time()
+        while len(self._token_exchange_completed) > _TOKEN_EXCHANGE_DEDUP_MAX_ENTRIES:
+            del self._token_exchange_completed[next(iter(self._token_exchange_completed))]
+        # Persisted too, so a duplicate handled by another process instance still sees
+        # it. Best-effort only: state has no compare-and-set, so the in-memory layer
+        # above remains the authoritative same-instance guard.
+        #
+        # Flushed mid-turn rather than at end of turn: the owner still has sign-in
+        # callbacks to run, and a duplicate racing on another instance would otherwise
+        # load a snapshot with no marker and redeem the exchange a second time. This
+        # mirrors the mid-turn save ``ctx.sign_in()`` performs for its pending hint.
+        try:
+            record_completed_token_exchange(ctx.state, exchange_id)
+            if ctx.state is not None:
+                await ctx.state._save()  # pyright: ignore[reportPrivateUsage]
+        except Exception:
+            logger.exception("Unable to persist completed OAuth token exchange; deduplicating in memory only.")
+
+    def _prune_completed_token_exchanges(self) -> None:
+        cutoff = time() - TOKEN_EXCHANGE_DEDUP_TTL_SECONDS
+        expired = [
+            exchange_id
+            for exchange_id, completed_at in self._token_exchange_completed.items()
+            if completed_at <= cutoff
+        ]
+        for exchange_id in expired:
+            del self._token_exchange_completed[exchange_id]
+
+    def _settle_token_exchange(
+        self, exchange_id: str, owned: Future[_TokenExchangeOutcome], outcome: _TokenExchangeOutcome
+    ) -> None:
+        self._token_exchange_in_flight.pop(exchange_id, None)
+        if not owned.done():
+            owned.set_result(outcome)
+
+    def _stamp_completed_token_exchange(
+        self, ctx: ActivityContext[SignInTokenExchangeInvokeActivity], exchange_id: str
+    ) -> None:
+        """Copy the completed marker into a deduplicated request's own turn state.
+
+        Each turn loads its own state snapshot, and saves are last-write-wins with no
+        compare-and-set. A duplicate whose snapshot predates the owner's write would
+        otherwise erase that freshly persisted marker when its own save lands last.
+        Only the in-memory marker is left alone here, so the TTL stays anchored to the
+        moment the token was actually redeemed rather than being extended by every
+        duplicate that arrives.
+
+        Best-effort like every other persisted-marker touch: a state failure must not
+        turn a successful duplicate into a failed invoke response.
+        """
+        try:
+            if not has_completed_token_exchange(ctx.state, exchange_id):
+                record_completed_token_exchange(ctx.state, exchange_id)
+        except Exception:
+            logger.exception("Unable to stamp completed OAuth token exchange into turn state.")
+
+    def _replay_completed_token_exchange(
+        self, ctx: ActivityContext[SignInTokenExchangeInvokeActivity], exchange_id: str
+    ) -> TokenExchangeResult:
+        """Answer a duplicate that arrived after its exchange already completed."""
+        logger.debug("Duplicate signin/tokenExchange with id '%s' - returning 200 no-op.", exchange_id)
+        self._stamp_completed_token_exchange(ctx, exchange_id)
+        connection_name = self._resolved_connection_name(ctx.activity.value.connection_name)
+        started_at = perf_counter()
+        try:
+            with get_tracer().start_as_current_span(
+                APP_SPAN_NAMES.oauth_token_exchange,
+                record_exception=False,
+                set_status_on_exception=False,
+            ) as span:
+                span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_connection, connection_name)
+                span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_operation, APP_OAUTH_OPERATIONS.token_exchange)
+                span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_result, APP_OAUTH_RESULTS.duplicate)
+                span.set_attribute(APP_ATTRIBUTE_NAMES.invoke_response_status, 200)
+                return InvokeResponse(status=200)
+        finally:
+            record_oauth_operation(
+                connection_name,
+                APP_OAUTH_OPERATIONS.token_exchange,
+                APP_OAUTH_RESULTS.duplicate,
+                (perf_counter() - started_at) * 1000,
+            )
+
+    async def _await_token_exchange(
+        self,
+        ctx: ActivityContext[SignInTokenExchangeInvokeActivity],
+        exchange_id: str,
+        in_flight: Future[_TokenExchangeOutcome],
+    ) -> TokenExchangeResult:
+        """Answer a duplicate that arrived while its exchange is still running.
+
+        The waiter mirrors whatever the owning request produced, so a caller that lost
+        the race still learns that the exchange failed (``412``) instead of being told
+        the sign-in succeeded. It mirrors the owner's *result*, never its exception
+        object -- see the failure branch below.
+        """
+        connection_name = self._resolved_connection_name(ctx.activity.value.connection_name)
+        result = APP_OAUTH_RESULTS.duplicate
+        started_at = perf_counter()
+        try:
+            with get_tracer().start_as_current_span(
+                APP_SPAN_NAMES.oauth_token_exchange,
+                record_exception=False,
+                set_status_on_exception=False,
+            ) as span:
+                span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_connection, connection_name)
+                span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_operation, APP_OAUTH_OPERATIONS.token_exchange)
+                # Shielded: cancelling this waiter must not cancel the future the
+                # owning request still has to resolve.
+                outcome = await asyncio.shield(in_flight)
+                # Stamped before the failure check: an exchange can fail after the
+                # token was already spent, and the marker still has to survive this
+                # request's own last-write-wins save.
+                if outcome.token_redeemed:
+                    self._stamp_completed_token_exchange(ctx, exchange_id)
+                if outcome.error is not None:
+                    result = APP_OAUTH_RESULTS.failure
+                    span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_result, result)
+                    span.set_attribute(APP_ATTRIBUTE_NAMES.invoke_response_status, 412)
+                    # Reported as a 412 rather than re-raised. Re-raising would hand
+                    # this task an exception it never incurred: a ``CancelledError``
+                    # from the owner would make an uncancelled waiter report itself as
+                    # cancelled and trip ``except CancelledError`` cleanup, and one
+                    # exception object shared between several waiters would have them
+                    # all append frames to the same ``__traceback__``. Mirroring the
+                    # result is what the TypeScript SDK does and what Teams needs.
+                    return InvokeResponse(
+                        status=412,
+                        body=TokenExchangeInvokeResponse(
+                            id=exchange_id,
+                            connection_name=connection_name,
+                            failure_detail=str(outcome.error) or "unable to exchange token...",
+                        ),
+                    )
+                response = outcome.response
+                # The owning request signals success by returning ``None``, which the
+                # activity processor materializes as a 200. Duplicates say so
+                # explicitly instead, matching the C# SDK's 200 no-op.
+                status = response.status if isinstance(response, InvokeResponse) else 200
+                span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_result, result)
+                span.set_attribute(APP_ATTRIBUTE_NAMES.invoke_response_status, status)
+                return response if isinstance(response, InvokeResponse) else InvokeResponse(status=200)
+        finally:
+            record_oauth_operation(
+                connection_name,
+                APP_OAUTH_OPERATIONS.token_exchange,
+                result,
+                (perf_counter() - started_at) * 1000,
+            )
+
     async def sign_in_failure(
         self, ctx: ActivityContext[SignInFailureInvokeActivity]
     ) -> Optional[InvokeResponse[None]]:
@@ -262,9 +561,11 @@ async def sign_in_failure(
                     f"registration has 'Expose an API' configured with the correct "
                     f"Application ID URI matching your OAuth connection's Token Exchange URL."
                 )
-                if target_flow is not None:
-                    self.oauth_registry._mark_sso_consumed(  # pyright: ignore[reportPrivateUsage]
-                        ctx, target_flow.connection_name
+                callback_flows = [target_flow] if target_flow is not None else registered_flows
+
+                for flow in callback_flows:
+                    self.oauth_registry._clear_pending(  # pyright: ignore[reportPrivateUsage]
+                        ctx, flow.connection_name
                     )
                 await self.event_emitter.emit_async(
                     "error",
@@ -282,7 +583,6 @@ async def sign_in_failure(
                 await self.event_emitter.emit_async("sign_in_failure", event)
                 span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_callback_invoked, True)
                 span.set_attribute(APP_ATTRIBUTE_NAMES.oauth_result, result)
-                callback_flows = [target_flow] if target_flow is not None else registered_flows
                 for flow in callback_flows:
                     flow_event = (
                         event
@@ -400,6 +700,9 @@ async def sign_in_verify_state(
                                 f"{activity.conversation.id} (HTTP {status})."
                             )
                             continue
+                        self.oauth_registry._clear_pending(  # pyright: ignore[reportPrivateUsage]
+                            ctx, connection_name
+                        )
                         logger.error(
                             f"Error verifying sign-in state for user {activity.from_.id} in conversation"
                             f"{activity.conversation.id}: {e}"
diff --git a/packages/apps/src/microsoft_teams/apps/diagnostics/_constants.py b/packages/apps/src/microsoft_teams/apps/diagnostics/_constants.py
index 8fa0cab61..3827c8fb9 100644
--- a/packages/apps/src/microsoft_teams/apps/diagnostics/_constants.py
+++ b/packages/apps/src/microsoft_teams/apps/diagnostics/_constants.py
@@ -81,6 +81,7 @@ class _AppOAuthOperations:
 
 @dataclass(frozen=True)
 class _AppOAuthResults:
+    duplicate: str = "duplicate"
     failure: str = "failure"
     no_token: str = "no_token"
     notified: str = "notified"
diff --git a/packages/apps/src/microsoft_teams/apps/oauth_flow.py b/packages/apps/src/microsoft_teams/apps/oauth_flow.py
index aa198fd39..9689ca278 100644
--- a/packages/apps/src/microsoft_teams/apps/oauth_flow.py
+++ b/packages/apps/src/microsoft_teams/apps/oauth_flow.py
@@ -9,12 +9,14 @@
 from dataclasses import replace
 from typing import Any, Awaitable, Callable, Iterator, List, Mapping, Optional, Sequence, Tuple, TypeVar, Union
 
+from httpx import HTTPStatusError
+from microsoft_teams.api import GetUserTokenParams, SignOutUserParams
+
 from .events import SignInEvent, SignInFailureEvent
 from .oauth_connection import connection_lookup_key, normalize_connection_name
 from .oauth_state import (
     clear_pending_oauth_sign_in,
     get_pending_oauth_sign_ins,
-    mark_pending_oauth_sso_consumed,
 )
 from .routing import ActivityContext, SignInOptions
 
@@ -150,15 +152,33 @@ async def sign_in(self, ctx: ActivityContext[Any], options: Optional[SignInOptio
 
     async def sign_out(self, ctx: ActivityContext[Any]) -> None:
         """Sign the user out of this connection."""
-        await ctx.sign_out(connection_name=self.connection_name)
+        await ctx.api.users.sign_out(
+            SignOutUserParams(
+                channel_id=ctx.activity.channel_id,
+                user_id=ctx.activity.from_.id,
+                connection_name=self.connection_name,
+            )
+        )
 
     async def get_token(self, ctx: ActivityContext[Any]) -> Optional[str]:
         """The user's token for this connection, or ``None`` if not signed in."""
-        return await ctx.get_user_token(connection_name=self.connection_name)
+        try:
+            res = await ctx.api.users.get_token(
+                GetUserTokenParams(
+                    channel_id=ctx.activity.channel_id,
+                    user_id=ctx.activity.from_.id,
+                    connection_name=self.connection_name,
+                )
+            )
+            return res.token
+        except HTTPStatusError as e:
+            if e.response.status_code == 404:
+                return None
+            raise
 
     async def is_signed_in(self, ctx: ActivityContext[Any]) -> bool:
         """Whether the user currently has a token for this connection."""
-        return await ctx.get_user_token(connection_name=self.connection_name) is not None
+        return await self.get_token(ctx) is not None
 
 
 class OAuthFlowRegistry(Mapping[str, OAuthFlow]):
@@ -219,10 +239,6 @@ def _clear_pending(self, ctx: ActivityContext[Any], connection_name: Optional[st
         conversation_id, user_id = _pending_scope(ctx)
         clear_pending_oauth_sign_in(ctx.state, connection_name, conversation_id, user_id)
 
-    def _mark_sso_consumed(self, ctx: ActivityContext[Any], connection_name: str) -> None:
-        conversation_id, user_id = _pending_scope(ctx)
-        mark_pending_oauth_sso_consumed(ctx.state, connection_name, conversation_id, user_id)
-
 
 def _pending_scope(ctx: ActivityContext[Any]) -> Tuple[Optional[str], Optional[str]]:
     """Conversation and user identifiers used to scope process-local pending hints."""
diff --git a/packages/apps/src/microsoft_teams/apps/oauth_pending_local.py b/packages/apps/src/microsoft_teams/apps/oauth_pending_local.py
index c521b059a..32c28518b 100644
--- a/packages/apps/src/microsoft_teams/apps/oauth_pending_local.py
+++ b/packages/apps/src/microsoft_teams/apps/oauth_pending_local.py
@@ -103,18 +103,6 @@ def replace(conversation_id: str, user_id: str, restored: List[_Entry]) -> None:
             _entries[(conversation_id, user_id, connection_key)] = (name.strip(), created_at, sso_offered)
 
 
-def mark_sso_consumed(conversation_id: str, user_id: str, connection_name: str) -> None:
-    """Retire a hint's silent-SSO marker, keeping it available for routing."""
-    key = _key(conversation_id, user_id, connection_name)
-    if key is None:
-        return
-
-    with _lock:
-        existing = _entries.get(key)
-        if existing is not None:
-            _entries[key] = (existing[0], existing[1], False)
-
-
 def _key(conversation_id: str, user_id: str, connection_name: str) -> Optional[_Key]:
     if not conversation_id or not user_id:
         return None
diff --git a/packages/apps/src/microsoft_teams/apps/oauth_state.py b/packages/apps/src/microsoft_teams/apps/oauth_state.py
index 1190c3b88..ddba809b2 100644
--- a/packages/apps/src/microsoft_teams/apps/oauth_state.py
+++ b/packages/apps/src/microsoft_teams/apps/oauth_state.py
@@ -32,6 +32,28 @@
 _PENDING_OAUTH_MAX_AGE_SECONDS = 5 * 60
 _PENDING_OAUTH_MAX_CLOCK_SKEW_SECONDS = 60
 
+# Reserved conversation-state key prefix holding the completed marker for a single
+# ``signin/tokenExchange``. The full key is ``__oauth:exchange:{id}`` and the value is an
+# ISO 8601 UTC timestamp, mirroring the C# SDK (``OAuthFlow.cs``) so all three SDKs
+# describe this state identically. That is design parity, not wire compatibility: the
+# SDKs encode the enclosing scope key differently, so they never resolve to the same
+# stored document. Conversation scope rather than user scope, because duplicates arrive
+# from several of the user's clients but always on the same conversation. Treat the key
+# as private: app code should neither read nor write it.
+_COMPLETED_EXCHANGE_STATE_KEY_PREFIX = "__oauth:exchange:"
+
+# Completed markers age out on the same schedule as pending sign-ins.
+TOKEN_EXCHANGE_DEDUP_TTL_SECONDS = _PENDING_OAUTH_MAX_AGE_SECONDS
+
+# Hard ceiling on how many completed markers one conversation document may carry.
+# The TTL above is the primary bound and the only one that should ever bind in
+# practice: these markers are scoped to a single conversation, and every duplicate of
+# an exchange reuses its id, so a conversation would have to begin a thousand
+# *distinct* sign-ins inside five minutes to reach this. The cap is a backstop that
+# keeps a pathological burst from growing the stored document without limit, matching
+# the 1000-entry bound the TypeScript SDK places on its completed list.
+_COMPLETED_EXCHANGE_MAX_ENTRIES = 1000
+
 
 @dataclass(frozen=True)
 class PendingOAuthSignIn:
@@ -106,31 +128,6 @@ def clear_pending_oauth_sign_in(
     _remove_connection(state, connection_name)
 
 
-def mark_pending_oauth_sso_consumed(
-    state: Optional[TurnStateContainer],
-    connection_name: str,
-    conversation_id: Optional[str] = None,
-    user_id: Optional[str] = None,
-) -> None:
-    """Retire a hint's silent-SSO marker while keeping it for callback routing.
-
-    Teams renders the sign-in button on the same OAuth card after a silent-SSO
-    failure, so the sign-in is still pending even though its SSO attempt is
-    spent. Dropping ``sso_offered`` stops the hint from re-attributing later
-    ``signin/failure`` callbacks while a follow-up ``signin/verifyState`` can
-    still be routed to the right connection. ``created_at`` is preserved so the
-    hint expires on its original schedule.
-    """
-    if state is None or state.user is None:
-        oauth_pending_local.mark_sso_consumed(conversation_id or "", user_id or "", connection_name)
-        return
-
-    target = connection_lookup_key(connection_name)
-    for key, name, is_sso_marker in _iter_markers(state.user):
-        if is_sso_marker and connection_lookup_key(name) == target:
-            state.user.pop(key, None)
-
-
 def replace_pending_oauth_sign_ins(
     state: Optional[TurnStateContainer],
     pending: List[PendingOAuthSignIn],
@@ -280,3 +277,125 @@ def _parse_timestamp(raw: Any) -> Optional[float]:
     if parsed.tzinfo is None:
         parsed = parsed.replace(tzinfo=timezone.utc)
     return parsed.timestamp()
+
+
+def completed_token_exchange_state_key(exchange_id: str) -> str:
+    """Reserved conversation-state key holding the completed marker for ``exchange_id``."""
+    return f"{_COMPLETED_EXCHANGE_STATE_KEY_PREFIX}{exchange_id}"
+
+
+def has_completed_token_exchange(state: Optional[TurnStateContainer], exchange_id: str) -> bool:
+    """Whether ``exchange_id`` has already been redeemed on this conversation.
+
+    Read from conversation state so a duplicate routed to a different process instance
+    still short-circuits. Turn state is a last-write-wins document with no ETag/CAS in
+    ``Storage``, so this cross-instance layer is best-effort; the caller's in-process
+    guard is what makes same-instance dedup atomic. The C# SDK carries the same caveat.
+    """
+    if state is None or not exchange_id:
+        return False
+
+    key = completed_token_exchange_state_key(exchange_id)
+    raw = state.conversation.get(key)
+    if raw is None:
+        return False
+
+    completed_at = _parse_completed_at(raw)
+    if completed_at is None:
+        logger.warning("Discarding malformed completed OAuth token exchange state.")
+        state.conversation.pop(key, None)
+        return False
+    if time() - completed_at > TOKEN_EXCHANGE_DEDUP_TTL_SECONDS:
+        state.conversation.pop(key, None)
+        return False
+    return True
+
+
+def record_completed_token_exchange(state: Optional[TurnStateContainer], exchange_id: str) -> None:
+    """Persist the completed marker for ``exchange_id``.
+
+    The marker is deliberately never cleared when the exchange finishes: a late
+    duplicate from a second Teams endpoint can arrive after the original settles, and
+    an already-removed marker would let it run as a brand new exchange. Markers are
+    dropped only once they age past :data:`TOKEN_EXCHANGE_DEDUP_TTL_SECONDS`, or -- far
+    more rarely -- when a conversation exceeds
+    :data:`_COMPLETED_EXCHANGE_MAX_ENTRIES` and the oldest are trimmed to fit.
+    """
+    if state is None or not exchange_id:
+        return
+    _write_completed_token_exchange(state, exchange_id)
+
+
+def _write_completed_token_exchange(state: TurnStateContainer, exchange_id: str) -> None:
+    """Single write chokepoint for completed markers.
+
+    Pruning here keeps the invariant that the conversation document never carries an
+    expired or unparsable marker, no matter which caller wrote it. The ceiling is
+    enforced afterwards, with the new marker already in place, so the exchange being
+    recorded can never be the one evicted to make room. Every marker enters through
+    this function, so bounding on write bounds the stored set for good.
+    """
+    _prune_completed_token_exchanges(state)
+    key = completed_token_exchange_state_key(exchange_id)
+    state.conversation[key] = _format_timestamp(time())
+    _enforce_completed_token_exchange_cap(state, key)
+
+
+def _enforce_completed_token_exchange_cap(state: TurnStateContainer, keep: str) -> None:
+    """Drop the oldest markers once a conversation exceeds the cap.
+
+    Expired markers are pruned before this runs, so everything still stored is live and
+    evicting any of it costs real dedup coverage. Oldest-first is the least damaging
+    order available: those markers are the closest to ageing out on their own, so they
+    have the least protection left to give.
+
+    ``keep`` is the marker just written and is never evicted -- it is the one the
+    current exchange depends on, and it would otherwise become a candidate whenever its
+    timestamp ties with another. Remaining ties are broken on the key so that eviction
+    is deterministic across instances instead of following dict insertion order.
+    """
+    keys = [key for key in state.conversation if key.startswith(_COMPLETED_EXCHANGE_STATE_KEY_PREFIX)]
+    overflow = len(keys) - _COMPLETED_EXCHANGE_MAX_ENTRIES
+    if overflow <= 0:
+        return
+
+    def _age_order(key: str) -> tuple[float, str]:
+        completed_at = _parse_completed_at(state.conversation.get(key))
+        # Unparsable markers sort first. Pruning should have removed them already, and
+        # anything that slipped through carries no usable expiry, so it is the safest
+        # thing to give up.
+        return (completed_at if completed_at is not None else float("-inf"), key)
+
+    for key in sorted(keys, key=_age_order):
+        if overflow <= 0:
+            break
+        if key == keep:
+            continue
+        state.conversation.pop(key, None)
+        overflow -= 1
+
+
+def _prune_completed_token_exchanges(state: TurnStateContainer) -> None:
+    """Drop expired or malformed markers so the conversation document stays bounded."""
+    now = time()
+    for key in list(state.conversation):
+        if not key.startswith(_COMPLETED_EXCHANGE_STATE_KEY_PREFIX):
+            continue
+        completed_at = _parse_completed_at(state.conversation.get(key))
+        if completed_at is None or now - completed_at > TOKEN_EXCHANGE_DEDUP_TTL_SECONDS:
+            state.conversation.pop(key, None)
+
+
+def _parse_completed_at(raw: Any) -> Optional[float]:
+    """Parse a stored marker, rejecting timestamps too far in the future.
+
+    ``_parse_timestamp`` already rejects anything that is not a parseable ISO 8601
+    string. The extra guard is for clock skew between instances: a marker stamped well
+    ahead of this instance's clock would otherwise read as fresh long past its TTL.
+    """
+    completed_at = _parse_timestamp(raw)
+    if completed_at is None:
+        return None
+    if completed_at > time() + _PENDING_OAUTH_MAX_CLOCK_SKEW_SECONDS:
+        return None
+    return completed_at
diff --git a/packages/apps/src/microsoft_teams/apps/options.py b/packages/apps/src/microsoft_teams/apps/options.py
index 56fa2f69e..8f4eb227c 100644
--- a/packages/apps/src/microsoft_teams/apps/options.py
+++ b/packages/apps/src/microsoft_teams/apps/options.py
@@ -120,7 +120,14 @@ class AppOptions(TypedDict, total=False):
 
     # OAuth
     default_connection_name: Optional[str]
-    """The OAuth connection name to use for authentication. Defaults to 'graph'."""
+    """The OAuth connection name to use for authentication. Defaults to 'graph'.
+
+    .. deprecated::
+        Names a single connection for the whole app, which does not generalize
+        past one. Register each connection with ``app.add_oauth_flow(name)`` and
+        hold on to the returned ``OAuthFlow`` instead. Still honoured as the
+        default for the deprecated ``ctx.sign_in`` / ``ctx.sign_out`` /
+        ``ctx.get_user_token`` surface."""
 
     fetch_user_token: Optional[bool]
     """Whether to eagerly look up the user's OAuth token on every inbound activity.
@@ -163,7 +170,10 @@ class InternalAppOptions:
     dangerously_allow_unauthenticated_requests: bool = False
     """Whether to accept incoming requests without JWT validation."""
     default_connection_name: str = "graph"
-    """The OAuth connection name to use for authentication."""
+    """The OAuth connection name to use for authentication.
+
+    .. deprecated::
+        Use ``app.add_oauth_flow(name)`` and the returned ``OAuthFlow``."""
     fetch_user_token: bool = False
     """When True, eagerly looks up the user's OAuth token on every inbound activity.
     The token is used to compute ``ctx.is_signed_in`` and ``ctx.user_token``, and to authenticate
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 57863f2cd..b553b5345 100644
--- a/packages/apps/src/microsoft_teams/apps/routing/activity_context.py
+++ b/packages/apps/src/microsoft_teams/apps/routing/activity_context.py
@@ -97,7 +97,18 @@ class SignInOptions:
 
 
 class ActivityContext(Generic[T]):
-    """Context object passed to activity handlers with middleware support."""
+    """Context object passed to activity handlers with middleware support.
+
+    .. note::
+        The single-connection OAuth surface here - :attr:`is_signed_in`,
+        :attr:`user_token`, :attr:`user_graph`, :meth:`sign_in`,
+        :meth:`sign_out` and :meth:`get_user_token` - predates
+        multi-connection support and is deprecated.
+
+        Register connections with
+        ``app.add_oauth_flow(name)`` and use the returned ``OAuthFlow``, which
+        pins every operation to the connection that owns it.
+    """
 
     def __init__(
         self,
@@ -162,6 +173,14 @@ def user_graph(self) -> "GraphServiceClient":
         """
         Get a Microsoft Graph client configured with the user's token.
 
+        .. deprecated::
+            Built from :attr:`user_token`, which holds the token for whichever
+            connection signed in most recently rather than the one that talks to
+            Microsoft Graph.
+
+            Read the token from the owning flow instead::
+                token = await app.get_oauth_flow("graph").get_token(ctx)
+
         Raises:
             ValueError: If the user is not signed in or doesn't have a valid token.
             RuntimeError: If the graph client cannot be created.
@@ -350,6 +369,12 @@ async def sign_in(self, options: Optional[SignInOptions] = None) -> Optional[str
         """
         Initiate a sign-in flow for the user.
 
+        .. deprecated::
+            Targets the app's single ``connection_name``, which does not
+            generalize past one connection. Use
+            ``app.get_oauth_flow(name).sign_in(ctx, options)``, which pins the
+            connection to the flow that owns it.
+
         Args:
             options: Optional signin options to customize the flow
 
@@ -480,6 +505,10 @@ async def sign_out(self, connection_name: Optional[str] = None) -> None:
 
         This method will remove the user's token from the storage.
 
+        .. deprecated::
+            Use ``app.get_oauth_flow(name).sign_out(ctx)``, which cannot sign
+            out of the wrong connection.
+
         Args:
             connection_name: The connection to sign out of. Defaults to the
                 app's default connection.
@@ -502,6 +531,10 @@ async def get_user_token(self, connection_name: Optional[str] = None) -> Optiona
         """
         Get the user's token for a connection.
 
+        .. deprecated::
+            Use ``app.get_oauth_flow(name).get_token(ctx)``, which reads the
+            token from the flow that owns the connection.
+
         Args:
             connection_name: The connection to read. Defaults to the app's
                 default connection.
@@ -530,7 +563,7 @@ async def get_user_token(self, connection_name: Optional[str] = None) -> Optiona
                 return None
             raise
 
-    async def get_token_status(self) -> List[TokenStatus]:
+    async def get_connection_status(self) -> List[TokenStatus]:
         """
         Get the token status for every OAuth connection registered on the bot.
 
diff --git a/packages/apps/tests/test_activity_context.py b/packages/apps/tests/test_activity_context.py
index bc9567a03..bf2198ae1 100644
--- a/packages/apps/tests/test_activity_context.py
+++ b/packages/apps/tests/test_activity_context.py
@@ -1079,7 +1079,7 @@ async def test_sign_out_propagates_token_service_failure(self) -> None:
 
 
 class TestActivityContextTokenHelpers:
-    """Tests for sign_out(connection_name=), get_user_token, and get_token_status."""
+    """Tests for sign_out(connection_name=), get_user_token, and get_connection_status."""
 
     @pytest.mark.asyncio
     async def test_sign_out_uses_override_connection_name(self) -> None:
@@ -1190,8 +1190,8 @@ async def test_get_user_token_propagates_non_http_errors(self) -> None:
             await ctx.get_user_token()
 
     @pytest.mark.asyncio
-    async def test_get_token_status_returns_all_connections(self) -> None:
-        """get_token_status makes a single call and returns the status list unfiltered."""
+    async def test_get_connection_status_returns_all_connections(self) -> None:
+        """get_connection_status makes a single call and returns the status list unfiltered."""
         mock_activity = MagicMock()
         mock_activity.channel_id = "msteams"
         mock_activity.from_.id = "user-1"
@@ -1200,7 +1200,7 @@ async def test_get_token_status_returns_all_connections(self) -> None:
         statuses = [MagicMock(), MagicMock()]
         ctx.api.users.get_token_status = AsyncMock(return_value=statuses)
 
-        result = await ctx.get_token_status()
+        result = await ctx.get_connection_status()
 
         assert result is statuses
         ctx.api.users.get_token_status.assert_awaited_once()
@@ -1210,8 +1210,8 @@ async def test_get_token_status_returns_all_connections(self) -> None:
         assert params.channel_id == "msteams"
 
     @pytest.mark.asyncio
-    async def test_get_token_status_propagates_errors(self) -> None:
-        """Unlike get_user_token, get_token_status lets service failures surface."""
+    async def test_get_connection_status_propagates_errors(self) -> None:
+        """Unlike get_user_token, get_connection_status lets service failures surface."""
         mock_activity = MagicMock()
         mock_activity.channel_id = "msteams"
         mock_activity.from_.id = "user-1"
@@ -1220,7 +1220,7 @@ async def test_get_token_status_propagates_errors(self) -> None:
         ctx.api.users.get_token_status = AsyncMock(side_effect=RuntimeError("service down"))
 
         with pytest.raises(RuntimeError):
-            await ctx.get_token_status()
+            await ctx.get_connection_status()
 
 
 class TestActivityContextPromptPreview:
@@ -1520,7 +1520,7 @@ async def test_override_send_failure_rolls_back_pending(self) -> None:
 
 
 class TestGetTokenStatusRegistryAware:
-    """get_token_status corrects the bulk call for registered flows."""
+    """get_connection_status corrects the bulk call for registered flows."""
 
     @staticmethod
     def _context(connection_names: list[str] | None = None):
@@ -1546,7 +1546,7 @@ async def test_without_registered_flows_the_bulk_result_is_returned_verbatim(sel
         ctx.api.users.get_token_status = AsyncMock(return_value=statuses)
         ctx.api.users.get_token = AsyncMock(side_effect=AssertionError("must not probe"))
 
-        assert await ctx.get_token_status() is statuses
+        assert await ctx.get_connection_status() is statuses
 
     @pytest.mark.asyncio
     async def test_false_status_for_a_registered_flow_is_corrected(self) -> None:
@@ -1559,7 +1559,7 @@ async def test_false_status_for_a_registered_flow_is_corrected(self) -> None:
         ctx.api.users.get_token_status = AsyncMock(return_value=[self._status("graph", False)])
         ctx.api.users.get_token = AsyncMock(return_value=MagicMock(token="a-token"))
 
-        result = await ctx.get_token_status()
+        result = await ctx.get_connection_status()
 
         assert [(s.connection_name, s.has_token) for s in result] == [("graph", True)]
 
@@ -1570,7 +1570,7 @@ async def test_a_genuine_miss_stays_false(self) -> None:
         ctx.api.users.get_token_status = AsyncMock(return_value=[self._status("graph", False)])
         ctx.api.users.get_token = AsyncMock(side_effect=_http_status_error(404))
 
-        result = await ctx.get_token_status()
+        result = await ctx.get_connection_status()
 
         assert [(s.connection_name, s.has_token) for s in result] == [("graph", False)]
 
@@ -1581,7 +1581,7 @@ async def test_a_registered_flow_missing_from_the_bulk_result_is_added(self) ->
         ctx.api.users.get_token_status = AsyncMock(return_value=[self._status("legacy", True)])
         ctx.api.users.get_token = AsyncMock(return_value=MagicMock(token="a-token"))
 
-        result = await ctx.get_token_status()
+        result = await ctx.get_connection_status()
 
         assert [(s.connection_name, s.has_token) for s in result] == [("legacy", True), ("graph", True)]
 
@@ -1599,7 +1599,7 @@ async def probe(params):
 
         ctx.api.users.get_token = AsyncMock(side_effect=probe)
 
-        result = await ctx.get_token_status()
+        result = await ctx.get_connection_status()
 
         assert result[0] is legacy
         assert probes == []
@@ -1611,7 +1611,7 @@ async def test_registry_casing_is_matched_case_insensitively(self) -> None:
         ctx.api.users.get_token_status = AsyncMock(return_value=[self._status("GRAPH", False)])
         ctx.api.users.get_token = AsyncMock(return_value=MagicMock(token="a-token"))
 
-        result = await ctx.get_token_status()
+        result = await ctx.get_connection_status()
 
         # Corrected in place: one row, keeping the service's own casing.
         assert [(s.connection_name, s.has_token) for s in result] == [("GRAPH", True)]
@@ -1625,7 +1625,7 @@ async def test_order_and_shape_are_preserved(self) -> None:
         )
         ctx.api.users.get_token = AsyncMock(side_effect=_http_status_error(404))
 
-        result = await ctx.get_token_status()
+        result = await ctx.get_connection_status()
 
         assert [s.connection_name for s in result] == ["a", "b", "c", "missing"]
 
@@ -1637,4 +1637,27 @@ async def test_non_404_lookup_failure_propagates(self) -> None:
         ctx.api.users.get_token = AsyncMock(side_effect=_http_status_error(503))
 
         with pytest.raises(HTTPStatusError):
-            await ctx.get_token_status()
+            await ctx.get_connection_status()
+
+
+class TestConnectionStatusRename:
+    """``ctx.get_token_status`` was renamed to ``ctx.get_connection_status``."""
+
+    def test_old_context_name_is_gone(self) -> None:
+        """Renamed outright rather than aliased - the method was never released."""
+        ctx, _ = _create_activity_context()
+
+        assert not hasattr(ctx, "get_token_status")
+        assert hasattr(ctx, "get_connection_status")
+
+    def test_api_client_keeps_the_endpoint_name(self) -> None:
+        """The API layer must keep its name; renaming both would undo the fix.
+
+        ``api.users.get_token_status`` is the Token Service endpoint and returns
+        the service answer verbatim, while ``ctx.get_connection_status`` adds the
+        silent-SSO correction pass. One name for two behaviors was the problem.
+        """
+        from microsoft_teams.api.clients.user.client import UserClient
+
+        assert hasattr(UserClient, "get_token_status")
+        assert not hasattr(UserClient, "get_connection_status")
diff --git a/packages/apps/tests/test_app_oauth.py b/packages/apps/tests/test_app_oauth.py
index a32c4606b..a2527e4de 100644
--- a/packages/apps/tests/test_app_oauth.py
+++ b/packages/apps/tests/test_app_oauth.py
@@ -1005,9 +1005,14 @@ async def still_runs(_):
         mock_context.next.assert_awaited_once()
 
     @pytest.mark.asyncio
-    async def test_duplicate_token_exchange_is_not_deduplicated_in_pr4(
+    async def test_duplicate_token_exchange_runs_sign_in_side_effects_once(
         self, oauth_handlers, mock_context, token_exchange_activity, mock_token_response
     ):
+        """A repeat of the same exchange id is a 200 no-op.
+
+        Broader dedup coverage lives in ``test_app_oauth_dedup.py``; this guards the
+        interaction with the PR4 routing path that resolves the flow.
+        """
         flow = oauth_handlers.oauth_registry.add(OAuthFlow("test-connection"))
         callback_count = 0
 
@@ -1019,11 +1024,14 @@ async def on_signin(_):
         mock_context.activity = token_exchange_activity
         mock_context.api.users.exchange_token.return_value = mock_token_response
 
-        await oauth_handlers.sign_in_token_exchange(mock_context)
-        await oauth_handlers.sign_in_token_exchange(mock_context)
+        assert await oauth_handlers.sign_in_token_exchange(mock_context) is None
+        duplicate = await oauth_handlers.sign_in_token_exchange(mock_context)
 
-        assert mock_context.api.users.exchange_token.await_count == 2
-        assert callback_count == 2
+        assert isinstance(duplicate, InvokeResponse)
+        assert duplicate.status == 200
+        assert mock_context.api.users.exchange_token.await_count == 1
+        assert callback_count == 1
+        assert mock_context.next.await_count == 1
 
     @pytest.mark.asyncio
     async def test_verify_state_routes_to_pending_non_default_flow(
@@ -1095,13 +1103,14 @@ async def on_github(event):
         assert pending_marker_keys(state) == {"__oauth:pending:graph", "__oauth:pending:sso:graph"}
 
     @pytest.mark.asyncio
-    async def test_sso_failure_keeps_hint_so_button_click_still_routes_to_that_flow(
+    async def test_sso_failure_clears_hint_and_verify_state_still_resolves_by_probing(
         self, oauth_handlers, mock_context, failure_activity, verify_state_activity, mock_token_response
     ):
-        """After silent SSO fails, Teams shows the sign-in button on the same card.
+        """The retired hint costs extra probes, not correctness.
 
-        The follow-up verify-state carries no connection name, so the retired hint is what
-        keeps it on GitHub instead of probing (and possibly mis-attributing to) Graph.
+        Clearing on failure matches C#/TS. The follow-up verify-state carries no
+        connection name, so it probes each flow; the code is single-use and
+        connection-scoped, so only the issuing connection accepts it.
         """
         oauth_handlers.oauth_registry.add(OAuthFlow("Graph"))
         github = oauth_handlers.oauth_registry.add(OAuthFlow("GitHub"))
@@ -1117,19 +1126,28 @@ async def on_github(event):
         mock_context.activity = failure_activity
         await oauth_handlers.sign_in_failure(mock_context)
 
+        # Both keys retired, so nothing routes the follow-up.
+        assert state.user is not None
+        assert pending_marker_keys(state) == set()
+
+        # Only GitHub issued this code; Graph rejects it the way the Token Service would.
+        def get_token(params):
+            if params.connection_name == "GitHub":
+                return mock_token_response
+            raise oauth_http_error(400, "wrong connection")
+
         mock_context.activity = verify_state_activity
-        mock_context.api.users.get_token.return_value = mock_token_response
+        mock_context.api.users.get_token.side_effect = get_token
+
         result = await oauth_handlers.sign_in_verify_state(mock_context)
 
         assert result is None
         attempted = [call.args[0].connection_name for call in mock_context.api.users.get_token.await_args_list]
-        assert attempted == ["GitHub"]
+        assert attempted == ["Graph", "GitHub"]
         assert calls == ["github:GitHub"]
-        assert state.user is not None
-        assert pending_marker_keys(state) == set()
 
     @pytest.mark.asyncio
-    async def test_retired_sso_hint_does_not_attribute_a_second_failure(
+    async def test_cleared_hint_does_not_attribute_a_second_failure(
         self, oauth_handlers, mock_context, failure_activity
     ):
         github = oauth_handlers.oauth_registry.add(OAuthFlow("GitHub"))
@@ -1145,13 +1163,12 @@ async def on_github(event):
         await oauth_handlers.sign_in_failure(mock_context)
         assert calls == ["github:GitHub"]
 
-        # Second failure: the hint's SSO marker is spent, so this falls back to the
+        # Second failure: the hint is gone, so this falls back to the
         # notify-all-registered-flows path rather than re-attributing to GitHub.
         await oauth_handlers.sign_in_failure(mock_context)
         assert calls == ["github:GitHub", "github:GitHub"]
         assert mock_context.state.user is not None
-        # The SSO marker is retired; the sign-in itself is still pending.
-        assert pending_marker_keys(mock_context.state) == {"__oauth:pending:GitHub"}
+        assert pending_marker_keys(mock_context.state) == set()
 
     @pytest.mark.asyncio
     async def test_legacy_default_connection_hint_is_cleared_without_warning(
@@ -1208,12 +1225,11 @@ async def on_github(event):
             "github:GitHub",
         ]
         assert state.user is not None
-        # The failed connection keeps its hint (the card's sign-in button is still live) but
-        # loses its SSO marker so it cannot re-attribute a second failure.
+        # Only the flow that was notified loses its hint; the unrelated pending
+        # sign-in on test-connection is left intact.
         assert pending_marker_keys(state) == {
             "__oauth:pending:test-connection",
             "__oauth:pending:sso:test-connection",
-            "__oauth:pending:GitHub",
         }
 
     @pytest.mark.asyncio
@@ -1227,7 +1243,9 @@ async def test_sign_in_failure_preserves_replacement_hint_created_by_global_hand
         async def replace_hint(event):
             assert event.connection_name == "GitHub"
             assert state.user is not None
-            assert pending_marker_keys(state) == {"__oauth:pending:GitHub"}
+            # Already retired by the time handlers run, so what this records is a
+            # genuinely new sign-in rather than a survivor of the failed one.
+            assert pending_marker_keys(state) == set()
             state.user["__oauth:pending:GitHub"] = datetime.now(timezone.utc).isoformat()
             state.user["__oauth:pending:sso:GitHub"] = state.user["__oauth:pending:GitHub"]
 
@@ -1243,6 +1261,91 @@ async def replace_hint(event):
             f"__oauth:pending:sso:{github.connection_name}",
         }
 
+    @pytest.mark.asyncio
+    async def test_sign_in_failure_clears_both_keys_for_the_resolved_flow(
+        self, oauth_handlers, mock_context, failure_activity
+    ):
+        """Parity with C#/TS: the base hint retires alongside its SSO marker."""
+        oauth_handlers.oauth_registry.add(OAuthFlow("GitHub"))
+        mock_context.activity = failure_activity
+        state = create_pending_state(("GitHub", time.time(), True))
+        mock_context.state = state
+        assert pending_marker_keys(state) == {"__oauth:pending:GitHub", "__oauth:pending:sso:GitHub"}
+
+        await oauth_handlers.sign_in_failure(mock_context)
+
+        assert pending_marker_keys(state) == set()
+
+    @pytest.mark.asyncio
+    async def test_sign_in_failure_without_a_hint_clears_every_notified_flow(
+        self, oauth_handlers, mock_context, failure_activity
+    ):
+        """With nothing to attribute the callback, every notified flow retires its hint."""
+        oauth_handlers.oauth_registry.add(OAuthFlow("Graph"))
+        oauth_handlers.oauth_registry.add(OAuthFlow("GitHub"))
+        mock_context.activity = failure_activity
+        # Non-SSO hints, so none of them can resolve a target for this callback.
+        state = create_pending_state(
+            ("Graph", time.time(), False),
+            ("GitHub", time.time(), False),
+        )
+        mock_context.state = state
+
+        await oauth_handlers.sign_in_failure(mock_context)
+
+        assert pending_marker_keys(state) == set()
+
+    @pytest.mark.asyncio
+    async def test_unexpected_token_exchange_error_clears_pending(
+        self, oauth_handlers, mock_context, token_exchange_activity
+    ):
+        """An operational fault ends the sign-in, so its hint must not outlive it."""
+        mock_context.activity = token_exchange_activity
+        state = create_pending_state(("test-connection", time.time(), True))
+        mock_context.state = state
+        mock_context.api.users.exchange_token.side_effect = oauth_http_error(500, "boom")
+
+        result = await oauth_handlers.sign_in_token_exchange(mock_context)
+
+        assert result is not None and result.status == 500
+        assert pending_marker_keys(state) == set()
+
+    @pytest.mark.asyncio
+    async def test_expected_token_exchange_failure_keeps_pending(
+        self, oauth_handlers, mock_context, token_exchange_activity
+    ):
+        """Regression guard: the 412 fallback needs its hint to route the button click."""
+        mock_context.activity = token_exchange_activity
+        state = create_pending_state(("test-connection", time.time(), True))
+        mock_context.state = state
+        mock_context.api.users.exchange_token.side_effect = oauth_http_error(404, "no token")
+
+        result = await oauth_handlers.sign_in_token_exchange(mock_context)
+
+        assert result is not None and result.status == 412
+        assert pending_marker_keys(state) == {
+            "__oauth:pending:test-connection",
+            "__oauth:pending:sso:test-connection",
+        }
+
+    @pytest.mark.asyncio
+    async def test_unexpected_verify_state_error_clears_pending(
+        self, oauth_handlers, mock_context, verify_state_activity
+    ):
+        """An operational fault ends the sign-in, unlike an expected candidate miss."""
+        # Registered so the hint survives candidate resolution and reaches the probe;
+        # an unregistered connection's hint is discarded before this branch is hit.
+        oauth_handlers.oauth_registry.add(OAuthFlow("test-connection"))
+        mock_context.activity = verify_state_activity
+        state = create_pending_state(("test-connection", time.time(), True))
+        mock_context.state = state
+        mock_context.api.users.get_token.side_effect = oauth_http_error(500, "boom")
+
+        result = await oauth_handlers.sign_in_verify_state(mock_context)
+
+        assert result is not None and result.status == 500
+        assert pending_marker_keys(state) == set()
+
     @pytest.mark.asyncio
     async def test_sign_in_failure_ignores_more_recent_non_sso_hint(
         self, oauth_handlers, mock_context, failure_activity
diff --git a/packages/apps/tests/test_app_oauth_dedup.py b/packages/apps/tests/test_app_oauth_dedup.py
new file mode 100644
index 000000000..2f2f9fcae
--- /dev/null
+++ b/packages/apps/tests/test_app_oauth_dedup.py
@@ -0,0 +1,1065 @@
+"""
+Copyright (c) Microsoft Corporation. All rights reserved.
+Licensed under the MIT License.
+
+Teams fans ``signin/tokenExchange`` out to every signed-in client endpoint, so the
+same exchange reaches the bot several times. These tests pin down the resulting
+deduplication contract end to end through the public handler entry points.
+"""
+
+import asyncio
+import logging
+from datetime import datetime, timezone
+from typing import Any, List, Optional
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from httpx import HTTPStatusError, Request, Response
+from microsoft_teams.api import (
+    InvokeResponse,
+    SignInFailureInvokeActivity,
+    SignInTokenExchangeInvokeActivity,
+    SignInVerifyStateInvokeActivity,
+)
+from microsoft_teams.api.models import (
+    Account,
+    ConversationAccount,
+    SignInExchangeToken,
+    SignInFailure,
+    SignInStateVerifyQuery,
+    TokenResponse,
+)
+from microsoft_teams.apps.app_oauth import OauthHandlers
+from microsoft_teams.apps.oauth_flow import OAuthFlow, OAuthFlowRegistry
+from microsoft_teams.apps.routing import ActivityContext
+from microsoft_teams.apps.state import TurnState, TurnStateContainer
+from microsoft_teams.common import EventEmitter
+
+# pyright: basic
+
+DEDUP_TTL_SECONDS = 5 * 60
+"""Mirror of the production TTL. Duplicated rather than imported so a silent change
+to the production value shows up here as a failure instead of passing vacuously."""
+
+DEDUP_MAX_ENTRIES = 1000
+"""Mirror of the production cap on the in-memory completed-marker set."""
+
+CONNECTION_NAME = "test-connection"
+EXCHANGE_STATE_KEY_PREFIX = "__oauth:exchange:"
+
+
+def iso(epoch_seconds: float) -> str:
+    """Render an epoch as the ISO 8601 UTC string a marker is stored as.
+
+    Mirrors the production storage boundary rather than importing it, so a change to
+    the persisted format surfaces here as a failure instead of passing vacuously.
+    """
+    return datetime.fromtimestamp(epoch_seconds, tz=timezone.utc).isoformat()
+
+
+def token_response() -> TokenResponse:
+    return TokenResponse(connection_name=CONNECTION_NAME, token="access-token", expiration="2024-12-31T23:59:59Z")
+
+
+def oauth_http_error(status: int, message: str = "boom") -> HTTPStatusError:
+    request = Request("GET", "https://token.example")
+    response = Response(status, request=request)
+    return HTTPStatusError(message, request=request, response=response)
+
+
+def exchange_activity(exchange_id: str = "exchange-1", token: str = "sso-token") -> SignInTokenExchangeInvokeActivity:
+    return SignInTokenExchangeInvokeActivity(
+        type="invoke",
+        id="activity-789",
+        from_=Account(id="user-123", name="Test User", role="user"),
+        recipient=Account(id="bot-456", name="Test Bot", role="bot"),
+        conversation=ConversationAccount(id="conv-456", conversation_type="personal"),
+        channel_id="msteams",
+        name="signin/tokenExchange",
+        value=SignInExchangeToken(id=exchange_id, connection_name=CONNECTION_NAME, token=token),
+    )
+
+
+def verify_state_activity() -> SignInVerifyStateInvokeActivity:
+    return SignInVerifyStateInvokeActivity(
+        type="invoke",
+        id="activity-789",
+        from_=Account(id="user-123", name="Test User", role="user"),
+        recipient=Account(id="bot-456", name="Test Bot", role="bot"),
+        conversation=ConversationAccount(id="conv-456", conversation_type="personal"),
+        channel_id="msteams",
+        name="signin/verifyState",
+        value=SignInStateVerifyQuery(state="verify-code"),
+    )
+
+
+def failure_activity() -> SignInFailureInvokeActivity:
+    return SignInFailureInvokeActivity(
+        type="invoke",
+        id="activity-789",
+        from_=Account(id="user-123", name="Test User", role="user"),
+        recipient=Account(id="bot-456", name="Test Bot", role="bot"),
+        conversation=ConversationAccount(id="conv-456", conversation_type="personal"),
+        channel_id="msteams",
+        name="signin/failure",
+        value=SignInFailure(code="invokeerror", message="nope"),
+    )
+
+
+def make_api(*, exchange: Any = None, get_token: Any = None) -> MagicMock:
+    """A stand-in ``ctx.api``. Shared between contexts so call counts span requests."""
+    api = MagicMock()
+    api.users.exchange_token = exchange if exchange is not None else AsyncMock(return_value=token_response())
+    api.users.get_token = get_token if get_token is not None else AsyncMock(return_value=token_response())
+    return api
+
+
+def slow_exchange(*outcomes: Any, delay: float = 0.01) -> AsyncMock:
+    """An ``exchange_token`` mock that yields to the event loop before answering.
+
+    The suspension is what lets a second concurrent request reach the dedup gate while
+    the first exchange is still in flight.
+    """
+    queued: List[Any] = list(outcomes) or [token_response()]
+
+    async def _exchange(_params: Any) -> Any:
+        await asyncio.sleep(delay)
+        outcome = queued.pop(0) if len(queued) > 1 else queued[0]
+        if isinstance(outcome, BaseException):
+            raise outcome
+        return outcome
+
+    return AsyncMock(side_effect=_exchange)
+
+
+def make_context(activity: Any, api: MagicMock, state: Optional[TurnStateContainer] = None) -> MagicMock:
+    ctx = MagicMock(spec=ActivityContext)
+    ctx.activity = activity
+    ctx.api = api
+    ctx.logger = MagicMock()
+    ctx.next = AsyncMock()
+    ctx.state = state
+    return ctx
+
+
+def make_state() -> TurnStateContainer:
+    return TurnStateContainer(
+        conversation=TurnState(),
+        conversation_id="conv-456",
+        user=TurnState(),
+        user_id="user-123",
+    )
+
+
+def make_handlers() -> tuple[OauthHandlers, MagicMock, OAuthFlow]:
+    emitter = MagicMock(spec=EventEmitter)
+    registry = OAuthFlowRegistry()
+    flow = registry.add(OAuthFlow(CONNECTION_NAME))
+    return OauthHandlers(CONNECTION_NAME, emitter, registry), emitter, flow
+
+
+def status_of(result: Any) -> int:
+    """The status Teams ends up seeing.
+
+    The owning request signals success by returning ``None``, which the activity
+    processor materializes as a 200; duplicates return that 200 explicitly.
+    """
+    return result.status if isinstance(result, InvokeResponse) else 200
+
+
+def emitted(emitter: MagicMock, name: str) -> List[Any]:
+    return [call.args[1] for call in emitter.emit_async.await_args_list if call.args[0] == name]
+
+
+class TestConcurrentTokenExchangeDedup:
+    @pytest.mark.asyncio
+    async def test_concurrent_duplicates_run_sign_in_side_effects_exactly_once(self):
+        handlers, emitter, flow = make_handlers()
+        signin_calls: List[str] = []
+
+        @flow.on_signin
+        async def on_signin(event):
+            signin_calls.append(event.connection_name)
+
+        api = make_api(exchange=slow_exchange())
+        first = make_context(exchange_activity(), api)
+        second = make_context(exchange_activity(), api)
+
+        results = await asyncio.gather(
+            handlers.sign_in_token_exchange(first),
+            handlers.sign_in_token_exchange(second),
+        )
+
+        assert [status_of(result) for result in results] == [200, 200]
+        assert api.users.exchange_token.await_count == 1
+        assert len(emitted(emitter, "sign_in")) == 1
+        assert signin_calls == [CONNECTION_NAME]
+        # The winner runs the middleware chain; the duplicate is a no-op, so `next`
+        # fires once across the pair rather than once per request.
+        assert sorted([first.next.await_count, second.next.await_count]) == [0, 1]
+
+    @pytest.mark.asyncio
+    async def test_concurrent_duplicates_mirror_the_original_failure(self):
+        handlers, emitter, flow = make_handlers()
+
+        @flow.on_signin
+        async def on_signin(_event):
+            pytest.fail("sign-in handlers must not run when the exchange fails")
+
+        api = make_api(exchange=slow_exchange(oauth_http_error(400, "bad exchange")))
+        first = make_context(exchange_activity("exchange-fail"), api)
+        second = make_context(exchange_activity("exchange-fail"), api)
+
+        results = await asyncio.gather(
+            handlers.sign_in_token_exchange(first),
+            handlers.sign_in_token_exchange(second),
+        )
+
+        assert api.users.exchange_token.await_count == 1
+        assert [status_of(result) for result in results] == [412, 412]
+        for result in results:
+            assert isinstance(result, InvokeResponse)
+            assert result.body is not None
+            assert result.body.id == "exchange-fail"
+            assert result.body.connection_name == CONNECTION_NAME
+        assert emitted(emitter, "sign_in") == []
+
+    @pytest.mark.asyncio
+    async def test_broken_sign_in_handler_does_not_break_dedup(self):
+        """A raising ``on_signin`` listener is contained by the flow.
+
+        Handler isolation means one broken listener must not turn a successful
+        callback into a failed invoke response -- so neither the owner nor its
+        duplicate fails, and the token is still exchanged exactly once.
+        """
+        handlers, emitter, flow = make_handlers()
+
+        @flow.on_signin
+        async def on_signin(_event):
+            raise RuntimeError("handler failed")
+
+        api = make_api(exchange=slow_exchange())
+        first = make_context(exchange_activity("exchange-raise"), api)
+        second = make_context(exchange_activity("exchange-raise"), api)
+
+        results = await asyncio.gather(
+            handlers.sign_in_token_exchange(first),
+            handlers.sign_in_token_exchange(second),
+            return_exceptions=True,
+        )
+
+        assert api.users.exchange_token.await_count == 1
+        assert not any(isinstance(result, BaseException) for result in results)
+        assert [status_of(result) for result in results] == [200, 200]
+        assert len(emitted(emitter, "sign_in")) == 1
+
+    @pytest.mark.asyncio
+    async def test_duplicate_arriving_during_sign_in_callbacks_awaits_the_owner(self):
+        """The completed marker is stamped the moment the token is redeemed.
+
+        A duplicate that lands in the window between redemption and the sign-in
+        callbacks finishing must still mirror the owner, not answer ahead of it.
+        """
+        handlers, _emitter, flow = make_handlers()
+        handler_started = asyncio.Event()
+        release_handler = asyncio.Event()
+        finished: List[str] = []
+
+        @flow.on_signin
+        async def on_signin(_event):
+            handler_started.set()
+            await release_handler.wait()
+            finished.append("handler")
+
+        api = make_api()
+        first = make_context(exchange_activity(), api)
+        second = make_context(exchange_activity(), api)
+
+        owner = asyncio.create_task(handlers.sign_in_token_exchange(first))
+        await handler_started.wait()
+        duplicate = asyncio.create_task(handlers.sign_in_token_exchange(second))
+        await asyncio.sleep(0)
+
+        # Still parked on the owner's future: the marker alone must not let it answer.
+        assert not duplicate.done()
+        assert finished == []
+        release_handler.set()
+
+        results = await asyncio.gather(owner, duplicate)
+
+        assert api.users.exchange_token.await_count == 1
+        # The owner's callbacks ran to completion before the duplicate was answered.
+        assert finished == ["handler"]
+        assert [status_of(result) for result in results] == [200, 200]
+
+    @pytest.mark.asyncio
+    async def test_concurrent_exchanges_with_distinct_ids_both_run(self):
+        handlers, emitter, _flow = make_handlers()
+        api = make_api(exchange=slow_exchange())
+        first = make_context(exchange_activity("exchange-a"), api)
+        second = make_context(exchange_activity("exchange-b"), api)
+
+        results = await asyncio.gather(
+            handlers.sign_in_token_exchange(first),
+            handlers.sign_in_token_exchange(second),
+        )
+
+        assert results == [None, None]
+        assert api.users.exchange_token.await_count == 2
+        assert len(emitted(emitter, "sign_in")) == 2
+        assert first.next.await_count == 1
+        assert second.next.await_count == 1
+
+    @pytest.mark.asyncio
+    async def test_concurrent_exchanges_without_an_id_are_never_collapsed(self):
+        """An empty id is not a shared key: two unrelated sign-ins must both proceed."""
+        handlers, emitter, _flow = make_handlers()
+        api = make_api(exchange=slow_exchange())
+        first = make_context(exchange_activity("", token="sso-token-a"), api)
+        second = make_context(exchange_activity("", token="sso-token-b"), api)
+
+        results = await asyncio.gather(
+            handlers.sign_in_token_exchange(first),
+            handlers.sign_in_token_exchange(second),
+        )
+
+        assert results == [None, None]
+        assert api.users.exchange_token.await_count == 2
+        exchanged = [call.args[0].exchange_request.token for call in api.users.exchange_token.await_args_list]
+        assert sorted(exchanged) == ["sso-token-a", "sso-token-b"]
+        assert len(emitted(emitter, "sign_in")) == 2
+        assert first.next.await_count == 1
+        assert second.next.await_count == 1
+
+
+class TestLateTokenExchangeDedup:
+    @pytest.mark.asyncio
+    async def test_late_duplicate_is_a_200_no_op(self):
+        handlers, emitter, flow = make_handlers()
+        signin_calls: List[str] = []
+
+        @flow.on_signin
+        async def on_signin(event):
+            signin_calls.append(event.connection_name)
+
+        api = make_api()
+        first = make_context(exchange_activity(), api)
+        second = make_context(exchange_activity(), api)
+
+        assert await handlers.sign_in_token_exchange(first) is None
+        late = await handlers.sign_in_token_exchange(second)
+
+        assert isinstance(late, InvokeResponse)
+        assert late.status == 200
+        assert late.body is None
+        assert api.users.exchange_token.await_count == 1
+        assert signin_calls == [CONNECTION_NAME]
+        assert len(emitted(emitter, "sign_in")) == 1
+        assert first.next.await_count == 1
+        assert second.next.await_count == 0
+
+    @pytest.mark.asyncio
+    async def test_sequential_exchanges_with_distinct_ids_both_run(self):
+        handlers, emitter, _flow = make_handlers()
+        api = make_api()
+
+        first = make_context(exchange_activity("exchange-a"), api)
+        second = make_context(exchange_activity("exchange-b"), api)
+        assert await handlers.sign_in_token_exchange(first) is None
+        assert await handlers.sign_in_token_exchange(second) is None
+
+        assert api.users.exchange_token.await_count == 2
+        assert len(emitted(emitter, "sign_in")) == 2
+
+    @pytest.mark.asyncio
+    async def test_sequential_exchanges_without_an_id_are_never_collapsed(self):
+        handlers, emitter, _flow = make_handlers()
+        api = make_api()
+
+        first = make_context(exchange_activity(""), api)
+        second = make_context(exchange_activity(""), api)
+        assert await handlers.sign_in_token_exchange(first) is None
+        assert await handlers.sign_in_token_exchange(second) is None
+
+        assert api.users.exchange_token.await_count == 2
+        assert len(emitted(emitter, "sign_in")) == 2
+        assert first.next.await_count == 1
+        assert second.next.await_count == 1
+
+    @pytest.mark.asyncio
+    async def test_dedup_works_without_state_configured(self):
+        handlers, _emitter, _flow = make_handlers()
+        api = make_api()
+        first = make_context(exchange_activity(), api, state=None)
+        second = make_context(exchange_activity(), api, state=None)
+
+        await handlers.sign_in_token_exchange(first)
+        late = await handlers.sign_in_token_exchange(second)
+
+        assert status_of(late) == 200
+        assert api.users.exchange_token.await_count == 1
+
+    @pytest.mark.asyncio
+    async def test_a_second_app_instance_does_not_share_in_memory_dedup(self):
+        """Guards against the marker set living in module scope instead of per app."""
+        first_handlers, _first_emitter, _first_flow = make_handlers()
+        second_handlers, _second_emitter, _second_flow = make_handlers()
+        api = make_api()
+
+        await first_handlers.sign_in_token_exchange(make_context(exchange_activity(), api))
+        assert await second_handlers.sign_in_token_exchange(make_context(exchange_activity(), api)) is None
+
+        assert api.users.exchange_token.await_count == 2
+
+    @pytest.mark.asyncio
+    async def test_sign_in_handler_failure_still_marks_the_exchange_as_spent(self):
+        """The exchange token is single-use, so a duplicate could never redeem it again."""
+        handlers, _emitter, flow = make_handlers()
+
+        @flow.on_signin
+        async def on_signin(_event):
+            raise RuntimeError("handler failed")
+
+        api = make_api()
+        first = make_context(exchange_activity(), api)
+        second = make_context(exchange_activity(), api)
+
+        # Handler isolation contains the failure, so the exchange still succeeds --
+        # but the marker must be stamped either way, which is what the duplicate proves.
+        assert await handlers.sign_in_token_exchange(first) is None
+        # PR4 guarantee: the owning request still advances the middleware chain.
+        assert first.next.await_count == 1
+
+        assert status_of(await handlers.sign_in_token_exchange(second)) == 200
+        assert api.users.exchange_token.await_count == 1
+
+
+class TestTokenExchangeDedupFailures:
+    @pytest.mark.asyncio
+    async def test_failed_exchange_can_be_retried_with_the_same_id(self):
+        """The in-flight entry is released on settle and a failure is never marked."""
+        handlers, emitter, _flow = make_handlers()
+        api = make_api(exchange=AsyncMock(side_effect=[oauth_http_error(400, "bad"), token_response()]))
+
+        first = make_context(exchange_activity("exchange-retry"), api)
+        second = make_context(exchange_activity("exchange-retry"), api)
+
+        failed = await handlers.sign_in_token_exchange(first)
+        assert status_of(failed) == 412
+
+        assert await handlers.sign_in_token_exchange(second) is None
+        assert api.users.exchange_token.await_count == 2
+        assert len(emitted(emitter, "sign_in")) == 1
+        assert first.next.await_count == 1
+        assert second.next.await_count == 1
+
+    @pytest.mark.asyncio
+    async def test_unexpected_service_error_keeps_its_status_and_is_not_marked_complete(self):
+        handlers, emitter, _flow = make_handlers()
+        api = make_api(exchange=AsyncMock(side_effect=[oauth_http_error(503, "unavailable"), token_response()]))
+
+        first = make_context(exchange_activity("exchange-503"), api)
+        second = make_context(exchange_activity("exchange-503"), api)
+
+        result = await handlers.sign_in_token_exchange(first)
+
+        assert isinstance(result, InvokeResponse)
+        assert result.status == 503
+        assert result.body is None
+        assert len(emitted(emitter, "error")) == 1
+        assert emitted(emitter, "sign_in") == []
+
+        assert await handlers.sign_in_token_exchange(second) is None
+        assert api.users.exchange_token.await_count == 2
+        assert len(emitted(emitter, "sign_in")) == 1
+
+    @pytest.mark.asyncio
+    async def test_concurrent_duplicate_mirrors_an_unexpected_service_status(self):
+        handlers, emitter, _flow = make_handlers()
+        api = make_api(exchange=slow_exchange(oauth_http_error(503, "unavailable")))
+        first = make_context(exchange_activity("exchange-503"), api)
+        second = make_context(exchange_activity("exchange-503"), api)
+
+        results = await asyncio.gather(
+            handlers.sign_in_token_exchange(first),
+            handlers.sign_in_token_exchange(second),
+        )
+
+        assert [status_of(result) for result in results] == [503, 503]
+        assert api.users.exchange_token.await_count == 1
+        # Only the request that actually talked to the service reports the error.
+        assert len(emitted(emitter, "error")) == 1
+
+
+class TestTokenExchangeDedupState:
+    @pytest.mark.asyncio
+    async def test_completion_is_persisted_under_the_reserved_conversation_key(self):
+        handlers, _emitter, _flow = make_handlers()
+        api = make_api()
+        state = make_state()
+
+        await handlers.sign_in_token_exchange(make_context(exchange_activity(), api, state))
+
+        marker = state.conversation[f"{EXCHANGE_STATE_KEY_PREFIX}exchange-1"]
+        # A bare ISO 8601 UTC string, matching the value shape the C# SDK stores.
+        assert isinstance(marker, str)
+        assert datetime.fromisoformat(marker).tzinfo is not None
+        # Conversation scope, not user scope — a duplicate can arrive from any of the
+        # user's clients but always on the same conversation.
+        assert state.user is not None
+        assert f"{EXCHANGE_STATE_KEY_PREFIX}exchange-1" not in state.user
+
+    @pytest.mark.asyncio
+    async def test_persisted_marker_dedups_across_app_instances(self):
+        first_handlers, _first_emitter, _first_flow = make_handlers()
+        second_handlers, second_emitter, _second_flow = make_handlers()
+        api = make_api()
+        state = make_state()
+
+        await first_handlers.sign_in_token_exchange(make_context(exchange_activity(), api, state))
+        late = make_context(exchange_activity(), api, state)
+        result = await second_handlers.sign_in_token_exchange(late)
+
+        assert status_of(result) == 200
+        assert api.users.exchange_token.await_count == 1
+        assert emitted(second_emitter, "sign_in") == []
+        assert late.next.await_count == 0
+
+    @pytest.mark.asyncio
+    async def test_failure_is_not_persisted_as_a_completion(self):
+        handlers, _emitter, _flow = make_handlers()
+        api = make_api(exchange=AsyncMock(side_effect=oauth_http_error(400, "bad")))
+        state = make_state()
+
+        await handlers.sign_in_token_exchange(make_context(exchange_activity(), api, state))
+
+        assert [key for key in state.conversation if key.startswith(EXCHANGE_STATE_KEY_PREFIX)] == []
+
+    @pytest.mark.asyncio
+    async def test_concurrent_duplicate_stamps_the_marker_into_its_own_snapshot(self):
+        """Each turn saves its own snapshot, last-write-wins.
+
+        An unstamped duplicate snapshot would erase the owner's freshly written marker
+        if its save happened to land last.
+        """
+        handlers, _emitter, _flow = make_handlers()
+        api = make_api(exchange=slow_exchange())
+        owner_state = make_state()
+        waiter_state = make_state()
+
+        await asyncio.gather(
+            handlers.sign_in_token_exchange(make_context(exchange_activity(), api, owner_state)),
+            handlers.sign_in_token_exchange(make_context(exchange_activity(), api, waiter_state)),
+        )
+
+        key = f"{EXCHANGE_STATE_KEY_PREFIX}exchange-1"
+        assert isinstance(owner_state.conversation[key], str)
+        assert isinstance(waiter_state.conversation[key], str)
+        assert api.users.exchange_token.await_count == 1
+
+    @pytest.mark.asyncio
+    async def test_late_duplicate_stamps_the_marker_into_its_own_snapshot(self):
+        handlers, _emitter, _flow = make_handlers()
+        api = make_api()
+        owner_state = make_state()
+        late_state = make_state()
+
+        await handlers.sign_in_token_exchange(make_context(exchange_activity(), api, owner_state))
+        await handlers.sign_in_token_exchange(make_context(exchange_activity(), api, late_state))
+
+        key = f"{EXCHANGE_STATE_KEY_PREFIX}exchange-1"
+        assert isinstance(late_state.conversation[key], str)
+        assert api.users.exchange_token.await_count == 1
+
+    @pytest.mark.asyncio
+    async def test_duplicate_does_not_extend_an_existing_marker(self, monkeypatch):
+        """Re-stamping a snapshot that already has the marker would push out its TTL."""
+        handlers, _emitter, _flow = make_handlers()
+        api = make_api()
+        state = make_state()
+        clock = {"now": 1_000.0}
+        monkeypatch.setattr("microsoft_teams.apps.app_oauth.time", lambda: clock["now"])
+        monkeypatch.setattr("microsoft_teams.apps.oauth_state.time", lambda: clock["now"])
+
+        await handlers.sign_in_token_exchange(make_context(exchange_activity(), api, state))
+        clock["now"] += 60
+        await handlers.sign_in_token_exchange(make_context(exchange_activity(), api, state))
+
+        assert state.conversation[f"{EXCHANGE_STATE_KEY_PREFIX}exchange-1"] == iso(1_000.0)
+
+    @pytest.mark.parametrize(
+        "corrupt",
+        [
+            "not-an-iso-timestamp",
+            "",
+            123,
+            None,
+            True,
+            [],
+            {"version": 1, "completed_at": 1.0},
+        ],
+        ids=[
+            "unparseable-string",
+            "empty-string",
+            "number",
+            "null",
+            "bool",
+            "list",
+            "legacy-dict-format",
+        ],
+    )
+    @pytest.mark.asyncio
+    async def test_corrupt_persisted_marker_does_not_crash_the_turn(self, corrupt):
+        handlers, emitter, _flow = make_handlers()
+        api = make_api()
+        state = make_state()
+        state.conversation[f"{EXCHANGE_STATE_KEY_PREFIX}exchange-1"] = corrupt
+
+        result = await handlers.sign_in_token_exchange(make_context(exchange_activity(), api, state))
+
+        assert result is None
+        assert api.users.exchange_token.await_count == 1
+        assert len(emitted(emitter, "sign_in")) == 1
+        # The unusable value is replaced by a well-formed marker.
+        assert isinstance(state.conversation[f"{EXCHANGE_STATE_KEY_PREFIX}exchange-1"], str)
+
+    @pytest.mark.asyncio
+    async def test_corrupt_persisted_marker_is_logged(self, caplog):
+        handlers, _emitter, _flow = make_handlers()
+        api = make_api()
+        state = make_state()
+        state.conversation[f"{EXCHANGE_STATE_KEY_PREFIX}exchange-1"] = "not-an-iso-timestamp"
+
+        with caplog.at_level(logging.WARNING, logger="microsoft_teams.apps.oauth_state"):
+            await handlers.sign_in_token_exchange(make_context(exchange_activity(), api, state))
+
+        assert "malformed completed OAuth token exchange state" in caplog.text
+
+    @pytest.mark.asyncio
+    async def test_marker_leaves_unrelated_state_untouched(self):
+        handlers, _emitter, _flow = make_handlers()
+        api = make_api()
+        state = make_state()
+        state.conversation["app-data"] = {"counter": 1}
+
+        await handlers.sign_in_token_exchange(make_context(exchange_activity(), api, state))
+
+        assert state.conversation["app-data"] == {"counter": 1}
+
+
+class TestTokenExchangeDedupExpiry:
+    @pytest.mark.asyncio
+    async def test_marker_expires_after_the_ttl(self, monkeypatch):
+        handlers, emitter, _flow = make_handlers()
+        api = make_api()
+        state = make_state()
+        clock = {"now": 1_000.0}
+        monkeypatch.setattr("microsoft_teams.apps.app_oauth.time", lambda: clock["now"])
+        monkeypatch.setattr("microsoft_teams.apps.oauth_state.time", lambda: clock["now"])
+
+        await handlers.sign_in_token_exchange(make_context(exchange_activity(), api, state))
+
+        clock["now"] += DEDUP_TTL_SECONDS - 1
+        assert status_of(await handlers.sign_in_token_exchange(make_context(exchange_activity(), api, state))) == 200
+        assert api.users.exchange_token.await_count == 1
+
+        clock["now"] += 2
+        assert await handlers.sign_in_token_exchange(make_context(exchange_activity(), api, state)) is None
+        assert api.users.exchange_token.await_count == 2
+        assert len(emitted(emitter, "sign_in")) == 2
+
+    @pytest.mark.asyncio
+    async def test_expired_persisted_markers_are_pruned_from_conversation_state(self, monkeypatch):
+        handlers, _emitter, _flow = make_handlers()
+        api = make_api()
+        state = make_state()
+        clock = {"now": 1_000.0}
+        monkeypatch.setattr("microsoft_teams.apps.app_oauth.time", lambda: clock["now"])
+        monkeypatch.setattr("microsoft_teams.apps.oauth_state.time", lambda: clock["now"])
+
+        await handlers.sign_in_token_exchange(make_context(exchange_activity("old"), api, state))
+        clock["now"] += DEDUP_TTL_SECONDS + 1
+        await handlers.sign_in_token_exchange(make_context(exchange_activity("new"), api, state))
+
+        markers = [key for key in state.conversation if key.startswith(EXCHANGE_STATE_KEY_PREFIX)]
+        assert markers == [f"{EXCHANGE_STATE_KEY_PREFIX}new"]
+
+    @pytest.mark.asyncio
+    async def test_marker_from_the_future_is_treated_as_corrupt(self, monkeypatch):
+        """A clock jump backwards must not pin a marker in place forever."""
+        handlers, _emitter, _flow = make_handlers()
+        api = make_api()
+        state = make_state()
+        monkeypatch.setattr("microsoft_teams.apps.oauth_state.time", lambda: 1_000.0)
+        # A well-formed ISO value, so this exercises the clock-skew guard rather than
+        # merely failing to parse.
+        state.conversation[f"{EXCHANGE_STATE_KEY_PREFIX}exchange-1"] = iso(1_000.0 + (60 * 60))
+
+        assert await handlers.sign_in_token_exchange(make_context(exchange_activity(), api, state)) is None
+        assert api.users.exchange_token.await_count == 1
+
+    @pytest.mark.asyncio
+    async def test_in_memory_markers_are_bounded(self):
+        """Past the cap the oldest ids are evicted, so the set cannot grow forever."""
+        handlers, _emitter, _flow = make_handlers()
+        api = make_api()
+
+        for index in range(DEDUP_MAX_ENTRIES + 1):
+            await handlers.sign_in_token_exchange(make_context(exchange_activity(f"exchange-{index}"), api))
+        assert api.users.exchange_token.await_count == DEDUP_MAX_ENTRIES + 1
+
+        # The newest id is still remembered...
+        newest = f"exchange-{DEDUP_MAX_ENTRIES}"
+        assert status_of(await handlers.sign_in_token_exchange(make_context(exchange_activity(newest), api))) == 200
+        assert api.users.exchange_token.await_count == DEDUP_MAX_ENTRIES + 1
+
+        # ...while the oldest was evicted to keep the set bounded.
+        assert await handlers.sign_in_token_exchange(make_context(exchange_activity("exchange-0"), api)) is None
+        assert api.users.exchange_token.await_count == DEDUP_MAX_ENTRIES + 2
+
+    @pytest.mark.asyncio
+    async def test_expired_in_memory_markers_are_pruned_without_state(self, monkeypatch):
+        handlers, _emitter, _flow = make_handlers()
+        api = make_api()
+        clock = {"now": 1_000.0}
+        monkeypatch.setattr("microsoft_teams.apps.app_oauth.time", lambda: clock["now"])
+
+        await handlers.sign_in_token_exchange(make_context(exchange_activity(), api))
+        clock["now"] += DEDUP_TTL_SECONDS + 1
+
+        assert await handlers.sign_in_token_exchange(make_context(exchange_activity(), api)) is None
+        assert api.users.exchange_token.await_count == 2
+
+
+class TestOtherSignInCallbacksAreNotDeduplicated:
+    @pytest.mark.asyncio
+    async def test_verify_state_is_not_deduplicated(self):
+        """The verify code is single-use, so repeats are naturally idempotent."""
+        handlers, emitter, flow = make_handlers()
+        signin_calls: List[str] = []
+
+        @flow.on_signin
+        async def on_signin(event):
+            signin_calls.append(event.connection_name)
+
+        api = make_api()
+        first = make_context(verify_state_activity(), api)
+        second = make_context(verify_state_activity(), api)
+
+        assert await handlers.sign_in_verify_state(first) is None
+        assert await handlers.sign_in_verify_state(second) is None
+
+        assert api.users.get_token.await_count == 2
+        assert len(emitted(emitter, "sign_in")) == 2
+        assert signin_calls == [CONNECTION_NAME, CONNECTION_NAME]
+        assert first.next.await_count == 1
+        assert second.next.await_count == 1
+
+    @pytest.mark.asyncio
+    async def test_concurrent_verify_states_are_not_deduplicated(self):
+        handlers, emitter, _flow = make_handlers()
+        get_token = AsyncMock(side_effect=slow_exchange().side_effect)
+        api = make_api(get_token=get_token)
+        first = make_context(verify_state_activity(), api)
+        second = make_context(verify_state_activity(), api)
+
+        await asyncio.gather(
+            handlers.sign_in_verify_state(first),
+            handlers.sign_in_verify_state(second),
+        )
+
+        assert get_token.await_count == 2
+        assert len(emitted(emitter, "sign_in")) == 2
+
+    @pytest.mark.asyncio
+    async def test_sign_in_failure_is_not_deduplicated(self):
+        """Failure is a single informational notice, not a state-changing operation."""
+        handlers, emitter, flow = make_handlers()
+        failures: List[str] = []
+
+        @flow.on_signin_failure
+        async def on_failure(event):
+            failures.append(event.code or "")
+
+        api = make_api()
+        first = make_context(failure_activity(), api)
+        second = make_context(failure_activity(), api)
+
+        assert await handlers.sign_in_failure(first) is None
+        assert await handlers.sign_in_failure(second) is None
+
+        assert len(emitted(emitter, "sign_in_failure")) == 2
+        assert failures == ["invokeerror", "invokeerror"]
+        assert first.next.await_count == 1
+        assert second.next.await_count == 1
+
+
+class TestDedupFailureIsolation:
+    """Dedup must not turn a state hiccup, a cancellation, or a post-redemption
+    failure into a stalled turn or a lost completion marker."""
+
+    @pytest.mark.asyncio
+    async def test_unreadable_marker_state_still_completes_the_turn(self, monkeypatch):
+        """An unreadable store degrades to in-memory dedup instead of stalling.
+
+        The persisted read happens before ``_run_token_exchange`` opens its
+        ``try``/``finally``, so an escaping error would skip ``ctx.next()`` and leave
+        the middleware chain hanging.
+        """
+        handlers, emitter, _flow = make_handlers()
+        monkeypatch.setattr(
+            "microsoft_teams.apps.app_oauth.has_completed_token_exchange",
+            MagicMock(side_effect=RuntimeError("state store unavailable")),
+        )
+
+        api = make_api()
+        ctx = make_context(exchange_activity(), api, make_state())
+
+        assert await handlers.sign_in_token_exchange(ctx) is None
+        assert api.users.exchange_token.await_count == 1
+        assert len(emitted(emitter, "sign_in")) == 1
+        assert ctx.next.await_count == 1
+
+    @pytest.mark.asyncio
+    async def test_completed_marker_is_flushed_before_sign_in_callbacks_run(self):
+        """The marker reaches storage mid-turn, not at end of turn.
+
+        The owner still has callbacks to run after redeeming the token. A duplicate
+        racing on another process instance loads its own snapshot, so the marker has
+        to be durable before those callbacks start or the duplicate exchanges again.
+        """
+        handlers, _emitter, flow = make_handlers()
+        state = make_state()
+        marker_key = f"{EXCHANGE_STATE_KEY_PREFIX}exchange-1"
+        order: List[str] = []
+
+        async def capture_save() -> None:
+            order.append("save" if marker_key in state.conversation else "save-without-marker")
+
+        state._save = capture_save
+
+        @flow.on_signin
+        async def on_signin(_event):
+            order.append("signin")
+
+        ctx = make_context(exchange_activity(), make_api(), state)
+        await handlers.sign_in_token_exchange(ctx)
+
+        assert order == ["save", "signin"]
+
+    @pytest.mark.asyncio
+    async def test_failure_after_redemption_still_lets_duplicates_stamp_the_marker(self):
+        """A failure *after* the token is spent must not cost the completion marker.
+
+        ``_clear_pending`` runs once the marker is already recorded, which is the
+        window this covers. A raising sign-in handler cannot reach it -- the flow
+        isolates listener failures -- so a state operation is the honest trigger.
+
+        Both halves of the bug are load-bearing here: if the owner reports
+        ``token_redeemed=False``, or if the waiter checks ``outcome.error`` before
+        stamping, the waiter's last-write-wins save erases the owner's completion and
+        a later duplicate redeems the spent exchange again.
+        """
+        handlers, _emitter, _flow = make_handlers()
+        handlers.oauth_registry._clear_pending = MagicMock(side_effect=RuntimeError("state store unavailable"))
+
+        api = make_api(exchange=slow_exchange())
+        owner_state, waiter_state = make_state(), make_state()
+        owner = make_context(exchange_activity(), api, owner_state)
+        waiter = make_context(exchange_activity(), api, waiter_state)
+
+        owner_result, waiter_result = await asyncio.gather(
+            handlers.sign_in_token_exchange(owner),
+            handlers.sign_in_token_exchange(waiter),
+            return_exceptions=True,
+        )
+
+        marker_key = f"{EXCHANGE_STATE_KEY_PREFIX}exchange-1"
+        assert api.users.exchange_token.await_count == 1
+        assert isinstance(owner_result, RuntimeError)
+        assert marker_key in owner_state.conversation
+        assert isinstance(waiter_result, InvokeResponse)
+        assert waiter_result.status == 412
+        assert marker_key in waiter_state.conversation
+
+    @pytest.mark.asyncio
+    async def test_cancelling_the_owner_does_not_cancel_its_duplicate(self):
+        """A duplicate never incurred the owner's cancellation, so it must not report one.
+
+        Re-raising the owner's ``CancelledError`` would make an uncancelled task claim
+        it was cancelled and trip any ``except CancelledError`` cleanup above it.
+        """
+        handlers, _emitter, _flow = make_handlers()
+        api = make_api(exchange=slow_exchange(delay=0.05))
+        owner = make_context(exchange_activity(), api)
+        waiter = make_context(exchange_activity(), api)
+
+        owner_task = asyncio.create_task(handlers.sign_in_token_exchange(owner))
+        await asyncio.sleep(0.01)
+        waiter_task = asyncio.create_task(handlers.sign_in_token_exchange(waiter))
+        await asyncio.sleep(0.01)
+        owner_task.cancel()
+
+        result = await waiter_task
+
+        assert owner_task.cancelled()
+        assert not waiter_task.cancelled()
+        assert isinstance(result, InvokeResponse)
+        assert result.status == 412
+
+    @pytest.mark.asyncio
+    async def test_duplicates_are_not_handed_the_owners_exception_object(self):
+        """Several waiters must not share one exception instance.
+
+        A shared instance has every waiter appending frames to the same
+        ``__traceback__``, so each report contaminates the others.
+        """
+        handlers, _emitter, _flow = make_handlers()
+        api = make_api(exchange=slow_exchange(RuntimeError("token service exploded")))
+        contexts = [make_context(exchange_activity(), api) for _ in range(3)]
+
+        owner_result, *waiter_results = await asyncio.gather(
+            *(handlers.sign_in_token_exchange(ctx) for ctx in contexts),
+            return_exceptions=True,
+        )
+
+        assert api.users.exchange_token.await_count == 1
+        assert isinstance(owner_result, RuntimeError)
+        assert [status_of(result) for result in waiter_results] == [412, 412]
+        for result in waiter_results:
+            assert isinstance(result, InvokeResponse)
+        assert waiter_results[0] is not waiter_results[1]
+
+    @pytest.mark.asyncio
+    async def test_duplicate_paths_report_the_registered_connection_casing(self, monkeypatch):
+        """Teams echoes the card's casing back, which must not split one connection
+        into several telemetry series."""
+        emitter = MagicMock(spec=EventEmitter)
+        registry = OAuthFlowRegistry()
+        registry.add(OAuthFlow("Test-Connection"))
+        handlers = OauthHandlers("Test-Connection", emitter, registry)
+
+        recorded: List[str] = []
+        monkeypatch.setattr(
+            "microsoft_teams.apps.app_oauth.record_oauth_operation",
+            lambda connection_name, *_args, **_kwargs: recorded.append(connection_name),
+        )
+
+        api = make_api(exchange=slow_exchange())
+        activity = exchange_activity()
+        activity.value.connection_name = "test-connection"
+
+        # Concurrent pair exercises the in-flight waiter, then a late request
+        # exercises the completed-marker replay.
+        await asyncio.gather(
+            handlers.sign_in_token_exchange(make_context(activity, api)),
+            handlers.sign_in_token_exchange(make_context(activity, api)),
+        )
+        await handlers.sign_in_token_exchange(make_context(activity, api))
+
+        # Owner, in-flight waiter and late replay all land on one series.
+        assert recorded == ["Test-Connection"] * 3
+        assert "test-connection" not in recorded
+
+
+class TestConnectionNameTelemetryCasing:
+    """Teams echoes back whatever casing the sign-in card carried, so one connection
+    must not fan out into several telemetry series."""
+
+    @staticmethod
+    def _capture(monkeypatch) -> tuple[List[str], List[str]]:
+        operations: List[str] = []
+        errors: List[str] = []
+        monkeypatch.setattr(
+            "microsoft_teams.apps.app_oauth.record_oauth_operation",
+            lambda connection_name, *_a, **_kw: operations.append(connection_name),
+        )
+        monkeypatch.setattr(
+            "microsoft_teams.apps.app_oauth.record_oauth_error",
+            lambda connection_name, *_a, **_kw: errors.append(connection_name),
+        )
+        return operations, errors
+
+    @staticmethod
+    def _handlers() -> OauthHandlers:
+        registry = OAuthFlowRegistry()
+        registry.add(OAuthFlow("Test-Connection"))
+        return OauthHandlers("Test-Connection", MagicMock(spec=EventEmitter), registry)
+
+    @staticmethod
+    def _activity() -> SignInTokenExchangeInvokeActivity:
+        activity = exchange_activity()
+        # The casing Teams echoes back, which differs from the registered name.
+        activity.value.connection_name = "test-connection"
+        return activity
+
+    @pytest.mark.asyncio
+    async def test_success_reports_the_registered_casing(self, monkeypatch):
+        operations, _errors = self._capture(monkeypatch)
+        handlers = self._handlers()
+        api = make_api()
+
+        await handlers.sign_in_token_exchange(make_context(self._activity(), api))
+
+        assert operations == ["Test-Connection"]
+        # The wire call keeps the name Teams sent: canonicalizing it would change what
+        # reaches the Token Service, which is a behavior change, not a telemetry fix.
+        assert api.users.exchange_token.await_args.args[0].connection_name == "test-connection"
+
+    @pytest.mark.asyncio
+    async def test_failure_reports_the_registered_casing(self, monkeypatch):
+        """The failure path is what exercises the ``finally`` after an exception.
+
+        A 500 rather than a 400: 404/400/412 are expected exchange misses that report
+        no error, so only an unexpected status reaches ``record_oauth_error``.
+        """
+        operations, errors = self._capture(monkeypatch)
+        handlers = self._handlers()
+        api = make_api(exchange=AsyncMock(side_effect=oauth_http_error(500)))
+
+        await handlers.sign_in_token_exchange(make_context(self._activity(), api))
+
+        assert operations == ["Test-Connection"]
+        assert errors == ["Test-Connection"]
+
+    @pytest.mark.asyncio
+    async def test_expected_miss_reports_the_registered_casing(self, monkeypatch):
+        """A 412 fall-back-to-card miss still has to land on the canonical series."""
+        operations, errors = self._capture(monkeypatch)
+        handlers = self._handlers()
+        api = make_api(exchange=AsyncMock(side_effect=oauth_http_error(404)))
+
+        result = await handlers.sign_in_token_exchange(make_context(self._activity(), api))
+
+        assert isinstance(result, InvokeResponse)
+        assert result.status == 412
+        assert operations == ["Test-Connection"]
+        assert errors == []
+
+    @pytest.mark.asyncio
+    async def test_registry_failure_does_not_mask_itself_in_the_finally(self, monkeypatch):
+        """The name must be resolved before the ``try``.
+
+        Resolving inside it left the variable unbound when anything above raised, so
+        the metric write in ``finally`` died with ``UnboundLocalError`` and buried the
+        real exception.
+        """
+        self._capture(monkeypatch)
+        handlers = self._handlers()
+        handlers.oauth_registry.get = MagicMock(side_effect=RuntimeError("registry exploded"))
+
+        with pytest.raises(RuntimeError, match="registry exploded"):
+            await handlers.sign_in_token_exchange(make_context(self._activity(), make_api()))
+
+    @pytest.mark.asyncio
+    async def test_unregistered_connection_is_reported_as_teams_sent_it(self, monkeypatch):
+        """An unknown connection has no registered casing to fall back to, and the
+        diagnostic is only useful if it shows what actually arrived."""
+        operations, _errors = self._capture(monkeypatch)
+        handlers = self._handlers()
+        activity = exchange_activity()
+        activity.value.connection_name = "not-registered"
+
+        await handlers.sign_in_token_exchange(make_context(activity, make_api()))
+
+        assert operations == ["not-registered"]
diff --git a/packages/apps/tests/test_oauth_flow.py b/packages/apps/tests/test_oauth_flow.py
index fbaf42aca..c5d7294b4 100644
--- a/packages/apps/tests/test_oauth_flow.py
+++ b/packages/apps/tests/test_oauth_flow.py
@@ -9,14 +9,34 @@
 from unittest.mock import AsyncMock, MagicMock
 
 import pytest
+from httpx import HTTPStatusError, Request, Response
 from microsoft_teams.apps import App, OAuthFlow, OAuthFlowRegistry
 from microsoft_teams.apps.routing import SignInOptions
 from microsoft_teams.apps.state import StateOptions, TurnStateLoader
 from microsoft_teams.common.storage import LocalStorage
 
 
+def _api_ctx() -> MagicMock:
+    """Context mock whose activity fields survive pydantic validation.
+
+    ``sign_out`` and ``get_token`` build Token Service params themselves, so
+    ``channel_id`` and ``from_.id`` must be real values rather than MagicMocks.
+    """
+    ctx = MagicMock()
+    ctx.activity.channel_id = "msteams"
+    ctx.activity.from_.id = "user-1"
+    return ctx
+
+
+def _http_status_error(status_code: int) -> HTTPStatusError:
+    """Build an httpx.HTTPStatusError carrying the given status code."""
+    request = Request("GET", "https://token.example/api/usertoken/GetToken")
+    response = Response(status_code, request=request)
+    return HTTPStatusError(f"HTTP {status_code}", request=request, response=response)
+
+
 class TestOAuthFlowOperations:
-    """sign_in / sign_out / get_token / is_signed_in delegate to the context."""
+    """sign_in / sign_out / get_token / is_signed_in, always on this flow's connection."""
 
     @pytest.mark.asyncio
     async def test_sign_in_forces_flow_connection_name(self) -> None:
@@ -70,37 +90,63 @@ async def test_sign_in_without_state_remains_supported(self) -> None:
     @pytest.mark.asyncio
     async def test_sign_out_targets_flow_connection(self) -> None:
         flow = OAuthFlow("graph")
-        ctx = MagicMock()
-        ctx.sign_out = AsyncMock(return_value=None)
+        ctx = _api_ctx()
+        ctx.api.users.sign_out = AsyncMock(return_value=None)
 
         await flow.sign_out(ctx)
 
-        ctx.sign_out.assert_awaited_once_with(connection_name="graph")
+        ctx.api.users.sign_out.assert_awaited_once()
+        params = ctx.api.users.sign_out.call_args[0][0]
+        assert params.connection_name == "graph"
+        assert params.channel_id == "msteams"
+        assert params.user_id == "user-1"
 
     @pytest.mark.asyncio
-    async def test_get_token_returns_ctx_token(self) -> None:
+    async def test_get_token_targets_flow_connection(self) -> None:
         flow = OAuthFlow("graph")
-        ctx = MagicMock()
-        ctx.get_user_token = AsyncMock(return_value="tok")
+        ctx = _api_ctx()
+        ctx.api.users.get_token = AsyncMock(return_value=MagicMock(token="tok"))
 
         result = await flow.get_token(ctx)
 
         assert result == "tok"
-        ctx.get_user_token.assert_awaited_once_with(connection_name="graph")
+        params = ctx.api.users.get_token.call_args[0][0]
+        assert params.connection_name == "graph"
+        assert params.channel_id == "msteams"
+        assert params.user_id == "user-1"
+
+    @pytest.mark.asyncio
+    async def test_get_token_returns_none_when_no_token_cached(self) -> None:
+        """404 is the Token Service saying "not signed in", not a failure."""
+        flow = OAuthFlow("graph")
+        ctx = _api_ctx()
+        ctx.api.users.get_token = AsyncMock(side_effect=_http_status_error(404))
+
+        assert await flow.get_token(ctx) is None
+
+    @pytest.mark.asyncio
+    async def test_get_token_reraises_non_404(self) -> None:
+        """An outage must not be reported as a logged-out user."""
+        flow = OAuthFlow("graph")
+        ctx = _api_ctx()
+        ctx.api.users.get_token = AsyncMock(side_effect=_http_status_error(503))
+
+        with pytest.raises(HTTPStatusError):
+            await flow.get_token(ctx)
 
     @pytest.mark.asyncio
     async def test_is_signed_in_true_when_token_present(self) -> None:
         flow = OAuthFlow("graph")
-        ctx = MagicMock()
-        ctx.get_user_token = AsyncMock(return_value="tok")
+        ctx = _api_ctx()
+        ctx.api.users.get_token = AsyncMock(return_value=MagicMock(token="tok"))
 
         assert await flow.is_signed_in(ctx) is True
 
     @pytest.mark.asyncio
     async def test_is_signed_in_false_when_no_token(self) -> None:
         flow = OAuthFlow("graph")
-        ctx = MagicMock()
-        ctx.get_user_token = AsyncMock(return_value=None)
+        ctx = _api_ctx()
+        ctx.api.users.get_token = AsyncMock(side_effect=_http_status_error(404))
 
         assert await flow.is_signed_in(ctx) is False
 
diff --git a/packages/apps/tests/test_oauth_pending_local.py b/packages/apps/tests/test_oauth_pending_local.py
index 68f681f86..04538b67d 100644
--- a/packages/apps/tests/test_oauth_pending_local.py
+++ b/packages/apps/tests/test_oauth_pending_local.py
@@ -214,15 +214,13 @@ def test_clearing_without_a_name_clears_the_whole_scope(self) -> None:
 
         assert get_pending_oauth_sign_ins(None, "c", "u") == []
 
-    def test_consuming_sso_keeps_the_hint_for_routing(self) -> None:
-        """The sign-in is still pending; only its silent-SSO attempt is spent."""
+    def test_clearing_retires_an_sso_hint_entirely(self) -> None:
+        """The whole hint goes, SSO marker included, so nothing survives to re-route."""
         record_pending_oauth_sign_in(None, "graph", sso_offered=True, conversation_id="c", user_id="u")
 
-        local.mark_sso_consumed("c", "u", "GRAPH")
+        clear_pending_oauth_sign_in(None, "GRAPH", "c", "u")
 
-        hints = get_pending_oauth_sign_ins(None, "c", "u")
-        assert len(hints) == 1
-        assert hints[0].sso_offered is False
+        assert get_pending_oauth_sign_ins(None, "c", "u") == []
 
     def test_replace_restores_the_original_timestamps(self) -> None:
         """Rollback puts back what was there, not a fresh set of hints."""
diff --git a/packages/apps/tests/test_oauth_state.py b/packages/apps/tests/test_oauth_state.py
index bbf23a495..a6053c775 100644
--- a/packages/apps/tests/test_oauth_state.py
+++ b/packages/apps/tests/test_oauth_state.py
@@ -10,10 +10,15 @@
 import pytest
 from microsoft_teams.apps import TurnState, TurnStateContainer
 from microsoft_teams.apps.oauth_state import (
+    _COMPLETED_EXCHANGE_MAX_ENTRIES,  # pyright: ignore[reportPrivateUsage]
+    TOKEN_EXCHANGE_DEDUP_TTL_SECONDS,
     PendingOAuthSignIn,
+    _enforce_completed_token_exchange_cap,  # pyright: ignore[reportPrivateUsage]
     clear_pending_oauth_sign_in,
+    completed_token_exchange_state_key,
     get_pending_oauth_sign_ins,
-    mark_pending_oauth_sso_consumed,
+    has_completed_token_exchange,
+    record_completed_token_exchange,
     record_pending_oauth_sign_in,
     replace_pending_oauth_sign_ins,
 )
@@ -152,18 +157,16 @@ def test_equal_timestamps_break_ties_deterministically(self) -> None:
 
 
 class TestSsoMarkerHandling:
-    def test_marking_sso_consumed_retires_only_the_sso_marker(self) -> None:
+    def test_clearing_retires_the_hint_and_its_sso_marker_together(self) -> None:
+        """Both keys retire as a unit, so no orphan can re-attribute a later callback."""
         state = make_state()
         record_pending_oauth_sign_in(state, "Graph", sso_offered=True)
-        assert state.user is not None
-        original = state.user["__oauth:pending:Graph"]
+        assert stored_keys(state) == {"__oauth:pending:Graph", "__oauth:pending:sso:Graph"}
 
-        mark_pending_oauth_sso_consumed(state, "graph")
+        clear_pending_oauth_sign_in(state, "graph")
 
-        # The sign-in is still pending on its original schedule; only SSO is spent.
-        assert stored_keys(state) == {"__oauth:pending:Graph"}
-        assert state.user["__oauth:pending:Graph"] == original
-        assert [(h.connection_name, h.sso_offered) for h in get_pending_oauth_sign_ins(state)] == [("Graph", False)]
+        assert stored_keys(state) == set()
+        assert get_pending_oauth_sign_ins(state) == []
 
     def test_connection_named_sso_is_not_mistaken_for_an_sso_marker(self) -> None:
         # ``sso:`` is a legal start to a connection name. Without its own base marker,
@@ -286,7 +289,6 @@ def test_helpers_are_no_ops_without_state(self) -> None:
         assert get_pending_oauth_sign_ins(None) == []
         record_pending_oauth_sign_in(None, "Graph", sso_offered=True)
         clear_pending_oauth_sign_in(None, "Graph")
-        mark_pending_oauth_sso_consumed(None, "Graph")
         replace_pending_oauth_sign_ins(None, [])
 
     def test_helpers_are_no_ops_without_a_user_scope(self) -> None:
@@ -300,5 +302,149 @@ def test_helpers_are_no_ops_without_a_user_scope(self) -> None:
         assert get_pending_oauth_sign_ins(state) == []
         record_pending_oauth_sign_in(state, "Graph", sso_offered=True)
         clear_pending_oauth_sign_in(state, "Graph")
-        mark_pending_oauth_sso_consumed(state, "Graph")
         replace_pending_oauth_sign_ins(state, [])
+
+
+# Declared rather than imported, so a change to the stored layout has to be a
+# deliberate edit here too. Mirrors ``test_app_oauth_dedup.py``.
+EXCHANGE_STATE_KEY_PREFIX = "__oauth:exchange:"
+
+
+def completed_keys(state: TurnStateContainer) -> set[str]:
+    return {key for key in state.conversation if key.startswith(EXCHANGE_STATE_KEY_PREFIX)}
+
+
+def seed_completed(state: TurnStateContainer, exchange_ids: list[str], *, age_seconds: float) -> None:
+    """Write markers straight into the document, bypassing the prune-and-cap write path."""
+    stamp = iso(time.time() - age_seconds)
+    for exchange_id in exchange_ids:
+        state.conversation[completed_token_exchange_state_key(exchange_id)] = stamp
+
+
+class TestCompletedExchangeCap:
+    def test_cap_is_one_thousand(self):
+        """Pinned deliberately: it is the bound the TypeScript SDK also promises."""
+        assert _COMPLETED_EXCHANGE_MAX_ENTRIES == 1000
+
+    def test_writing_up_to_the_cap_evicts_nothing(self):
+        state = make_state()
+        seed_completed(
+            state,
+            [f"old-{index:05d}" for index in range(_COMPLETED_EXCHANGE_MAX_ENTRIES - 1)],
+            age_seconds=10,
+        )
+
+        record_completed_token_exchange(state, "newest")
+
+        # Exactly at the limit, so every marker survives.
+        assert len(completed_keys(state)) == _COMPLETED_EXCHANGE_MAX_ENTRIES
+        assert has_completed_token_exchange(state, "newest")
+        assert has_completed_token_exchange(state, "old-00000")
+
+    def test_overflow_evicts_the_oldest_marker(self):
+        state = make_state()
+        now = time.time()
+        # Ages ascending, so ``old-00000`` is the oldest and first to go.
+        for index in range(_COMPLETED_EXCHANGE_MAX_ENTRIES):
+            key = completed_token_exchange_state_key(f"old-{index:05d}")
+            state.conversation[key] = iso(now - 250 + index * 0.01)
+
+        record_completed_token_exchange(state, "newest")
+
+        assert len(completed_keys(state)) == _COMPLETED_EXCHANGE_MAX_ENTRIES
+        assert not has_completed_token_exchange(state, "old-00000")
+        assert has_completed_token_exchange(state, "old-00001")
+        assert has_completed_token_exchange(state, "newest")
+
+    def test_expired_markers_are_pruned_before_the_cap_applies(self):
+        """Expiry must run first, or a live marker is evicted while dead ones stay."""
+        state = make_state()
+        seed_completed(
+            state,
+            [f"stale-{index:05d}" for index in range(_COMPLETED_EXCHANGE_MAX_ENTRIES)],
+            age_seconds=TOKEN_EXCHANGE_DEDUP_TTL_SECONDS + 60,
+        )
+        seed_completed(state, ["live"], age_seconds=1)
+
+        record_completed_token_exchange(state, "newest")
+
+        assert completed_keys(state) == {
+            completed_token_exchange_state_key("live"),
+            completed_token_exchange_state_key("newest"),
+        }
+
+    def test_ties_are_broken_deterministically_across_instances(self):
+        """Identical timestamps must not leave eviction up to dict insertion order."""
+        first = make_state()
+        second = make_state()
+        exchange_ids = [f"tie-{index:05d}" for index in range(_COMPLETED_EXCHANGE_MAX_ENTRIES)]
+        stamp = iso(time.time() - 5)
+        for exchange_id in exchange_ids:
+            first.conversation[completed_token_exchange_state_key(exchange_id)] = stamp
+        for exchange_id in reversed(exchange_ids):
+            second.conversation[completed_token_exchange_state_key(exchange_id)] = stamp
+
+        record_completed_token_exchange(first, "newest")
+        record_completed_token_exchange(second, "newest")
+
+        assert completed_keys(first) == completed_keys(second)
+        assert not has_completed_token_exchange(first, "tie-00000")
+
+    def test_the_marker_being_written_is_never_evicted(self):
+        """The current exchange still depends on the marker being recorded for it."""
+        state = make_state()
+        stamp = iso(time.time() - 5)
+        keep = completed_token_exchange_state_key("aaa-current")
+        state.conversation[keep] = stamp
+        for index in range(_COMPLETED_EXCHANGE_MAX_ENTRIES + 25):
+            state.conversation[completed_token_exchange_state_key(f"tie-{index:05d}")] = stamp
+
+        # Everything shares a timestamp and ``aaa-current`` sorts first on the key
+        # tie-break, so an unguarded sweep would drop the very marker being written.
+        _enforce_completed_token_exchange_cap(state, keep)
+
+        assert keep in completed_keys(state)
+        assert len(completed_keys(state)) == _COMPLETED_EXCHANGE_MAX_ENTRIES
+
+    def test_malformed_markers_are_dropped_rather_than_filling_the_cap(self):
+        state = make_state()
+        state.conversation[completed_token_exchange_state_key("unparsable")] = "not-a-timestamp"
+        state.conversation[completed_token_exchange_state_key("wrong-type")] = 12345
+        state.conversation[completed_token_exchange_state_key("far-future")] = iso(time.time() + 3600)
+
+        record_completed_token_exchange(state, "newest")
+
+        assert completed_keys(state) == {completed_token_exchange_state_key("newest")}
+
+    def test_capping_leaves_unrelated_conversation_state_alone(self):
+        state = make_state()
+        state.conversation["app:counter"] = 7
+        seed_completed(
+            state,
+            [f"old-{index:05d}" for index in range(_COMPLETED_EXCHANGE_MAX_ENTRIES + 40)],
+            age_seconds=10,
+        )
+
+        record_completed_token_exchange(state, "newest")
+
+        assert state.conversation["app:counter"] == 7
+        assert len(completed_keys(state)) == _COMPLETED_EXCHANGE_MAX_ENTRIES
+
+    def test_recorded_marker_round_trips_after_an_overflowing_write(self):
+        state = make_state()
+        now = time.time()
+        for index in range(_COMPLETED_EXCHANGE_MAX_ENTRIES + 10):
+            key = completed_token_exchange_state_key(f"old-{index:05d}")
+            state.conversation[key] = iso(now - 250 + index * 0.01)
+
+        record_completed_token_exchange(state, "round-trip")
+
+        assert len(completed_keys(state)) == _COMPLETED_EXCHANGE_MAX_ENTRIES
+        # Reload the way conversation state is actually rehydrated on the next turn.
+        reloaded = TurnStateContainer(
+            conversation=TurnState(dict(state.conversation)),
+            conversation_id="conv-1",
+            user=TurnState(),
+            user_id="user-1",
+        )
+        assert has_completed_token_exchange(reloaded, "round-trip")
diff --git a/uv.lock b/uv.lock
index 1fd5729dd..48b16f0c1 100644
--- a/uv.lock
+++ b/uv.lock
@@ -2354,12 +2354,14 @@ source = { virtual = "examples/oauth" }
 dependencies = [
     { name = "dotenv" },
     { name = "microsoft-teams-apps" },
+    { name = "microsoft-teams-graph" },
 ]
 
 [package.metadata]
 requires-dist = [
     { name = "dotenv", specifier = ">=0.9.9" },
     { name = "microsoft-teams-apps", editable = "packages/apps" },
+    { name = "microsoft-teams-graph", editable = "packages/graph" },
 ]
 
 [[package]]