feat(mrtr): linear continuation-based handler — Option H

The Option B footgun was: await elicit() looks like a suspension point but
is actually a re-entry point, so everything above it runs twice. Option H
fixes that by making it a REAL suspension point — the coroutine frame is
held in a ContinuationStore across MRTR rounds, keyed by request_state.

Handler code stays exactly as it was in the SSE era:

    async def my_tool(ctx: LinearCtx, location: str) -> str:
        audit_log(location)      # runs exactly once
        units = await ctx.elicit("Which units?", UnitsSchema)
        return f"{location}: 22°{units.u}"

The wrapper linear_mrtr(my_tool, store=...) translates this into a standard
MRTR on_call_tool handler. Round 1 starts the coroutine; elicit() sends
IncompleteResult back through the wrapper and parks on a stream. Round 2's
retry wakes it with the answer. The coroutine continues from where it
stopped — no re-entry, no double-execution.

Trade-off: server holds the frame in memory between rounds. Client sees
pure MRTR (no SSE, independent requests), but server is stateful within
a single tool call. Horizontally-scaled deployments need sticky routing on
the request_state token. Same operational shape as Option A's SSE hold,
without the long-lived connection.

SDK pieces (src/mcp/server/experimental/mrtr/linear.py):
- LinearCtx with async elicit(message, PydanticSchema) -> instance
- ContinuationStore — owns the task group, TTL-based frame expiry
- linear_mrtr(handler, store=...) — the wrapper
- ElicitDeclined raised when user declines/cancels

7 E2E tests including the key assertion: side-effects above await fire
exactly once (the test measures audit_log count).

Author: maxisbey

examples/servers/mrtr-options/README.md

@@ -45,15 +45,16 @@ infra." The rows collapse for E, which is why it's the SDK default.
 
 ## Options
 
-|                                | Author writes                   | SDK does                         | Hidden re-entry | Old client gets                   |
-| ------------------------------ | ------------------------------- | -------------------------------- | --------------- | --------------------------------- |
-| [E](mrtr_options/option_e_degrade.py)        | MRTR-native only                | Nothing                          | No              | Result w/ default, or error       |
-| [A](mrtr_options/option_a_sse_shim.py)       | MRTR-native only                | Retry-loop over SSE              | Yes, safe       | Full elicitation                  |
-| [B](mrtr_options/option_b_await_shim.py)     | `await elicit()`                | Exception → `IncompleteResult`   | **Yes, unsafe** | Full elicitation                  |
-| [C](mrtr_options/option_c_version_branch.py) | One handler, `if version` branch | Version accessor                | No              | Full elicitation                  |
-| [D](mrtr_options/option_d_dual_handler.py)   | Two handlers                    | Picks by version                 | No              | Full elicitation                  |
-| [F](mrtr_options/option_f_ctx_once.py)       | MRTR-native + `ctx.once` wraps  | `once()` guard in request_state  | No              | (same as E)                       |
-| [G](mrtr_options/option_g_tool_builder.py)   | Step functions + `.build()`     | Step-tracking in request_state   | No              | (same as E)                       |
+|                                | Author writes                   | SDK does                         | Hidden re-entry | Server state         | Old client gets                   |
+| ------------------------------ | ------------------------------- | -------------------------------- | --------------- | -------------------- | --------------------------------- |
+| [E](mrtr_options/option_e_degrade.py)        | MRTR-native only                | Nothing                          | No              | None                 | Result w/ default, or error       |
+| [A](mrtr_options/option_a_sse_shim.py)       | MRTR-native only                | Retry-loop over SSE              | Yes, safe       | SSE connection       | Full elicitation                  |
+| [B](mrtr_options/option_b_await_shim.py)     | `await elicit()`                | Exception → `IncompleteResult`   | **Yes, unsafe** | None                 | Full elicitation                  |
+| [C](mrtr_options/option_c_version_branch.py) | One handler, `if version` branch | Version accessor                | No              | SSE (old-client arm) | Full elicitation                  |
+| [D](mrtr_options/option_d_dual_handler.py)   | Two handlers                    | Picks by version                 | No              | SSE (old-client arm) | Full elicitation                  |
+| [F](mrtr_options/option_f_ctx_once.py)       | MRTR-native + `ctx.once` wraps  | `once()` guard in request_state  | No              | None                 | (same as E)                       |
+| [G](mrtr_options/option_g_tool_builder.py)   | Step functions + `.build()`     | Step-tracking in request_state   | No              | None                 | (same as E)                       |
+| [H](mrtr_options/option_h_linear.py)         | `await ctx.elicit()` (linear)   | Holds coroutine frame in memory  | No              | Coroutine frame      | (same as E)                       |
 
 "Hidden re-entry" = the handler function is invoked more than once for a
 single logical tool call, and the author can't tell from the source text.
@@ -120,6 +121,17 @@ per tool. Likely SDK answer: ship F as a primitive on the context, ship G
 as an opt-in builder, recommend G for multi-round tools and F for
 single-question tools.
 
+**H (linear continuation)** is the Option B footgun, *fixed*. Handler code
+reads exactly like the SSE era — `await ctx.elicit()` is a genuine
+suspension point, side-effects above it fire once — because the coroutine
+frame is held in memory across rounds. The trade: server is stateful
+*within* a single tool call (frame keyed by `request_state`), so
+horizontally-scaled deployments need sticky routing on the token. Same
+operational shape as A's SSE hold but without the long-lived connection.
+Use for migrating existing SSE-era tools without rewriting, or when the
+linear style is genuinely clearer than guard-first. Don't use if you need
+true statelessness — E/F/G encode everything in `request_state` itself.
+
 ## The invariant test
 
 `tests/server/experimental/test_mrtr_options.py` parametrises all seven

examples/servers/mrtr-options/mrtr_options/option_h_linear.py

@@ -0,0 +1,63 @@
+"""Option H: continuation-based linear MRTR. ``await ctx.elicit()`` is genuine.
+
+The Option B footgun was: ``await elicit()`` *looks* like a suspension point
+but is actually a re-entry point, so everything above it runs twice. This
+fixes that by making it a *real* suspension point — the coroutine frame is
+held in a ``ContinuationStore`` across MRTR rounds, keyed by
+``request_state``.
+
+Handler code stays exactly as it was in the SSE era. Side-effects above
+the await fire once because the function never restarts — it resumes.
+
+Trade-off: the server holds the frame in memory between rounds. Client
+still sees pure MRTR (no SSE), but the server is stateful *within* a
+single tool call. Horizontally-scaled deployments need sticky routing on
+the ``request_state`` token. Same operational shape as Option A's SSE
+hold, without the long-lived connection.
+
+When to use: migrating existing SSE-era tools to MRTR wire protocol
+without rewriting the handler, or when the linear style is genuinely
+clearer than guard-first (complex branching, many rounds).
+
+When not to: if you need true statelessness across server instances.
+Use E/F/G — they encode everything the server needs in ``request_state``
+itself.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from pydantic import BaseModel
+
+from mcp.server.experimental.mrtr import ContinuationStore, LinearCtx, linear_mrtr
+
+from ._shared import audit_log, build_server, lookup_weather
+
+
+class UnitsPref(BaseModel):
+    units: str
+
+
+# ───────────────────────────────────────────────────────────────────────────
+# This is what the tool author writes. Linear, front-to-back, no re-entry
+# contract to reason about. The ``audit_log`` above the await fires
+# exactly once — the await is a real suspension point.
+# ───────────────────────────────────────────────────────────────────────────
+
+
+async def weather(ctx: LinearCtx, args: dict[str, Any]) -> str:
+    location = args["location"]
+    audit_log(location)  # runs once — unlike Option B
+    prefs = await ctx.elicit("Which units?", UnitsPref)
+    return lookup_weather(location, prefs.units)
+
+
+# ───────────────────────────────────────────────────────────────────────────
+# Registration. The store must be entered as an async context manager
+# around the server's run loop — it owns the task group that keeps the
+# suspended coroutines alive.
+# ───────────────────────────────────────────────────────────────────────────
+
+store = ContinuationStore()
+server = build_server("mrtr-option-h", on_call_tool=linear_mrtr(weather, store=store))

src/mcp/server/experimental/mrtr/__init__.py

@@ -30,13 +30,18 @@
 from mcp.server.experimental.mrtr.builder import EndStep, IncompleteStep, ToolBuilder
 from mcp.server.experimental.mrtr.compat import MrtrHandler, dispatch_by_version, sse_retry_shim
 from mcp.server.experimental.mrtr.context import MrtrCtx
+from mcp.server.experimental.mrtr.linear import ContinuationStore, ElicitDeclined, LinearCtx, linear_mrtr
 
 __all__ = [
     "MrtrCtx",
     "ToolBuilder",
     "IncompleteStep",
     "EndStep",
     "MrtrHandler",
+    "LinearCtx",
+    "ContinuationStore",
+    "ElicitDeclined",
+    "linear_mrtr",
     "input_response",
     "encode_state",
     "decode_state",