From d41ff5bcd890b6062dc03efb8002fc803a791550 Mon Sep 17 00:00:00 2001
From: Jeremiah Lowin <153965+jlowin@users.noreply.github.com>
Date: Wed, 22 Jul 2026 12:38:46 -0400
Subject: [PATCH] Rebuild guard tasks as end-and-reenter; remove imperative
 in-task elicit

A task tool that returns InputRequiredResult now ends its leg (freeing the
worker) and stores the ask as durable state; tasks/update enqueues a fresh
Docket execution (the next leg) with accumulated request_state/input_responses
injected via ctx. No worker ever blocks on input, so a parked task no longer
holds up shutdown. Imperative ctx.elicit() inside a task is removed and raises
with guard-pattern guidance.

Co-Authored-By: Claude <noreply@anthropic.com>
---
 fastmcp_slim/fastmcp/server/context.py        | 141 ++----
 fastmcp_tasks/fastmcp_tasks/components.py     |   7 +-
 fastmcp_tasks/fastmcp_tasks/context.py        |  46 +-
 fastmcp_tasks/fastmcp_tasks/creation.py       |  52 +-
 fastmcp_tasks/fastmcp_tasks/extension.py      |  13 +-
 fastmcp_tasks/fastmcp_tasks/handlers.py       | 134 ++++--
 fastmcp_tasks/fastmcp_tasks/input_loop.py     | 137 ++++++
 fastmcp_tasks/fastmcp_tasks/input_store.py    | 446 +++++++++++++-----
 fastmcp_tasks/fastmcp_tasks/keys.py           |  41 ++
 fastmcp_tasks/fastmcp_tasks/lifespan.py       |   3 +
 tests/server/test_mrtr_guards.py              |  22 +-
 .../server/test_context_background_task.py    | 114 +----
 tests/tasks/server/test_extension.py          |   9 +-
 tests/tasks/server/test_guard_reentrant.py    | 174 +++++++
 tests/tasks/server/test_reenter_shutdown.py   |  65 +++
 .../server/test_task_elicitation_relay.py     | 219 ---------
 tests/tasks/server/test_wire_production.py    |   1 +
 17 files changed, 1054 insertions(+), 570 deletions(-)
 create mode 100644 fastmcp_tasks/fastmcp_tasks/input_loop.py
 create mode 100644 tests/tasks/server/test_guard_reentrant.py
 create mode 100644 tests/tasks/server/test_reenter_shutdown.py
 delete mode 100644 tests/tasks/server/test_task_elicitation_relay.py

diff --git a/fastmcp_slim/fastmcp/server/context.py b/fastmcp_slim/fastmcp/server/context.py
index 4eeb3bd0a..e83ac83d4 100644
--- a/fastmcp_slim/fastmcp/server/context.py
+++ b/fastmcp_slim/fastmcp/server/context.py
@@ -3,7 +3,7 @@ from __future__ import annotations
 import logging
 import warnings
 import weakref
-from collections.abc import Awaitable, Callable, Generator, Mapping, Sequence
+from collections.abc import Callable, Generator, Mapping, Sequence
 from contextlib import contextmanager
 from contextvars import ContextVar, Token
 from dataclasses import dataclass
@@ -125,29 +125,17 @@ def _warn_sampling_deprecated() -> None:
 _current_context: ContextVar[Context | None] = ContextVar("context", default=None)
 
 
-#: Hook installed by the tasks extension (``fastmcp-tasks``) so ``ctx.elicit()``
-#: works inside a background-task worker, where there is no live request to
-#: carry the elicitation. Core ships no task engine; the extension registers a
-#: handler here at construction and ``Context._elicit_for_task`` delegates to it.
-#: ``None`` (the default) means no tasks extension is active, so in-task
-#: elicitation raises a clear install hint.
-_task_elicitation_handler: (
-    Callable[[Context, str, dict[str, Any]], Awaitable[mcp_types.ElicitResult]] | None
-) = None
-
-
-def set_task_elicitation_handler(
-    handler: Callable[[Context, str, dict[str, Any]], Awaitable[mcp_types.ElicitResult]]
-    | None,
-) -> None:
-    """Install (or clear) the in-task elicitation handler.
-
-    Called by the tasks extension so a worker's ``ctx.elicit()`` parks an input
-    request the client answers via ``tasks/update`` (SEP-2663 poll-based input).
-    Passing ``None`` restores the default "requires the tasks extension" error.
-    """
-    global _task_elicitation_handler
-    _task_elicitation_handler = handler
+#: Error raised when a tool calls ``ctx.elicit()`` inside a background task.
+#: Background tasks gather input with the guard/return pattern (return an
+#: ``InputRequiredResult``), which the end-and-reenter machinery drives across
+#: worker legs. Imperative elicitation would require blocking a worker on a
+#: client round-trip, which end-and-reenter deliberately does not do.
+_TASK_ELICIT_ERROR = (
+    "Imperative ctx.elicit() is not supported inside a background task. Gather "
+    "input with the guard pattern instead: return an InputRequiredResult from "
+    "the tool (with input_requests), and read ctx.input_responses / "
+    "ctx.request_state when the task re-runs after the client answers."
+)
 
 
 TransportType = Literal["stdio", "sse", "streamable-http"]
@@ -273,6 +261,14 @@ class Context:
         self._origin_request_id: str | None = origin_request_id
         # Request-scoped state for non-serializable values (serializable=False)
         self._request_state: dict[str, Any] = {}
+        # Multi-round-trip input carried in-task (SEP-2322 guard channel). A
+        # foreground round recovers `input_responses`/`request_state` from the
+        # wire request; a worker has no wire request, so the tasks extension's
+        # in-task loop sets these between rounds and the properties below fall
+        # back to them. The guard tool reads `ctx.input_responses` identically
+        # in both modes — only the transport differs (task store vs wire params).
+        self._task_input_responses: mcp_types.InputResponses | None = None
+        self._task_request_state: str | None = None
 
     @property
     def is_background_task(self) -> bool:
@@ -441,9 +437,14 @@ class Context:
         keys match the `input_requests` map the tool minted; each value is the
         client's result for that request (an `ElicitResult`, `CreateMessageResult`,
         or `ListRootsResult`).
+
+        In a background task there is no wire request, so this falls back to the
+        responses the in-task guard loop delivered (see the tasks extension).
         """
         params = self._input_response_params()
-        return params.input_responses if params else None
+        if params is not None and params.input_responses is not None:
+            return params.input_responses
+        return self._task_input_responses
 
     @property
     def request_state(self) -> str | None:
@@ -455,9 +456,14 @@ class Context:
         before the tool runs, so tampering is rejected before this is read).
         `None` on the initial round. Use it to carry a small amount of computed
         state across rounds without re-deriving it.
+
+        In a background task there is no wire request, so this falls back to the
+        state the in-task guard loop re-injected (see the tasks extension).
         """
         params = self._input_response_params()
-        return params.request_state if params else None
+        if params is not None and params.request_state is not None:
+            return params.request_state
+        return self._task_request_state
 
     @property
     def lifespan_context(self) -> dict[str, Any]:
@@ -1349,9 +1355,10 @@ class Context:
                 ``value`` field. Same scope rules as ``response_title``.
 
         Note:
-            This method works transparently in both request and background task
-            contexts. In background task mode (SEP-1686), it will set the task
-            status to "input_required" and wait for the client to provide input.
+            Imperative elicitation is not available inside a background task
+            (calling it there raises a ``ToolError``). A task gathers input with
+            the guard pattern: return an ``InputRequiredResult`` and read
+            ``ctx.input_responses`` / ``ctx.request_state`` when the task re-runs.
         """
         if response_type is None and fastmcp.settings.deprecation_warnings:
             warnings.warn(
@@ -1371,24 +1378,22 @@ class Context:
         )
 
         if self.is_background_task:
-            # Background task mode: use task-aware elicitation
-            result = await self._elicit_for_task(
-                message=message,
-                schema=config.schema,
-            )
-        else:
-            # Foreground push path: server-initiated elicitation needs a
-            # back-channel, which the 2026-07-28 era removed (SEP-2577). Raise a
-            # clear era-aware error before hitting the wire instead of the SDK's
-            # opaque "Method not found". Handshake-era behavior is unchanged.
-            if self._is_modern_protocol():
-                raise ToolError(_ELICIT_MODERN_ERROR)
-            # Standard request mode: use session.elicit directly
-            result = await self.session.elicit(
-                message=message,
-                requested_schema=config.schema,
-                related_request_id=self.request_id,
-            )
+            # Background tasks gather input with the guard/return pattern, not
+            # imperative elicitation — the worker never blocks on a client
+            # round-trip. Fail fast with the guidance to use InputRequiredResult.
+            raise ToolError(_TASK_ELICIT_ERROR)
+        # Foreground push path: server-initiated elicitation needs a back-channel,
+        # which the 2026-07-28 era removed (SEP-2577). Raise a clear era-aware
+        # error before hitting the wire instead of the SDK's opaque "Method not
+        # found". Handshake-era behavior is unchanged.
+        if self._is_modern_protocol():
+            raise ToolError(_ELICIT_MODERN_ERROR)
+        # Standard request mode: use session.elicit directly
+        result = await self.session.elicit(
+            message=message,
+            requested_schema=config.schema,
+            related_request_id=self.request_id,
+        )
 
         if result.action == "accept":
             return handle_elicit_accept(config, result.content)
@@ -1399,48 +1404,6 @@ class Context:
         else:
             raise ValueError(f"Unexpected elicitation action: {result.action}")
 
-    async def _elicit_for_task(
-        self,
-        message: str,
-        schema: dict[str, Any],
-    ) -> mcp_types.ElicitResult:
-        """Send an elicitation request from a background task (SEP-1686).
-
-        This method handles elicitation when running in a Docket worker context,
-        where there's no active MCP request. It:
-        1. Sets the task status to "input_required"
-        2. Sends the elicitation request with task metadata
-        3. Waits for the client to provide input via tasks/sendInput
-        4. Returns the result and resumes task execution
-
-        Args:
-            message: The message to display to the user
-            schema: The JSON schema for the expected response
-
-        Returns:
-            ElicitResult with the user's response
-
-        Raises:
-            RuntimeError: If not running in a background task context
-        """
-        if not self.is_background_task:
-            raise RuntimeError(
-                "_elicit_for_task called but not in a background task context"
-            )
-
-        # In-task elicitation is provided by the tasks extension (SEP-2663)
-        # from the `fastmcp-tasks` package, which installs the handler below.
-        # Core ships no task engine, so without the extension this raises a
-        # clear install hint rather than reaching a wire the worker lacks.
-        handler = _task_elicitation_handler
-        if handler is None:
-            raise RuntimeError(
-                "In-task elicitation requires the tasks extension. Install "
-                "'fastmcp[tasks]' and register the tasks extension via "
-                "mcp.add_extension(...)."
-            )
-        return await handler(self, message, schema)
-
     def _make_state_key(self, key: str) -> str:
         """Create session-prefixed key for state storage."""
         return f"{self.session_id}:{key}"
diff --git a/fastmcp_tasks/fastmcp_tasks/components.py b/fastmcp_tasks/fastmcp_tasks/components.py
index 0f3c0370a..69b301e28 100644
--- a/fastmcp_tasks/fastmcp_tasks/components.py
+++ b/fastmcp_tasks/fastmcp_tasks/components.py
@@ -37,6 +37,7 @@ from fastmcp.tools.base import Tool
 from fastmcp.tools.function_tool import FunctionTool, _resolve_param_hints
 from fastmcp.utilities.components import FastMCPComponent
 from fastmcp.utilities.types import get_cached_typeadapter
+from fastmcp_tasks.input_loop import reentrant_task_fn
 
 if TYPE_CHECKING:
     from docket import Docket
@@ -54,7 +55,11 @@ def register_component_with_docket(component: FastMCPComponent, docket: Docket)
         return
 
     if isinstance(component, FunctionTool):
-        docket.register(component.fn, names=[component.key])
+        # Run the tool through the guard loop so a body that returns an
+        # InputRequiredResult drives the reentrant in-task input cycle. The
+        # wrapper is signature-preserving, so Docket's dependency injection is
+        # unchanged for a body that never asks for input.
+        docket.register(reentrant_task_fn(component.fn), names=[component.key])
     elif isinstance(component, Tool):
         docket.register(component.run, names=[component.key])
     elif isinstance(component, FunctionResource):
diff --git a/fastmcp_tasks/fastmcp_tasks/context.py b/fastmcp_tasks/fastmcp_tasks/context.py
index ff56097c7..48100fa40 100644
--- a/fastmcp_tasks/fastmcp_tasks/context.py
+++ b/fastmcp_tasks/fastmcp_tasks/context.py
@@ -16,7 +16,11 @@ from contextvars import ContextVar
 from dataclasses import dataclass
 from typing import TYPE_CHECKING
 
-from fastmcp_tasks.keys import parse_task_key, task_redis_prefix
+from fastmcp_tasks.keys import (
+    leg_number_from_key,
+    parse_task_key,
+    task_redis_prefix,
+)
 
 try:
     from docket import TaskKey
@@ -109,6 +113,26 @@ def get_task_context() -> TaskContextInfo | None:
         return None
 
 
+def get_task_leg_number() -> int:
+    """Return the current leg number of the running task (1 outside a re-entry).
+
+    Each re-entry after client input runs as a fresh Docket execution under a
+    per-leg key; the capture wrapper reads this to scope a leg's outstanding
+    input requests so successive legs never collide in Redis.
+    """
+    from fastmcp_tasks.dependencies import is_docket_available
+
+    if not is_docket_available():
+        return 1
+
+    from docket.dependencies import current_execution
+
+    try:
+        return leg_number_from_key(current_execution.get().key)
+    except LookupError:
+        return 1
+
+
 @dataclass(frozen=True, slots=True)
 class TaskContextSnapshot:
     """All context data snapshotted at task-submission time.
@@ -388,6 +412,11 @@ async def make_task_context() -> Context | None:
     id; the server prefers the one registered at submission time so mounted
     tasks resolve to the child server. No live session is attached — SEP-2663
     input and status are polled, so the worker needs no back-channel.
+
+    For a re-entered leg (after the client answered a guard ask), the accumulated
+    per-leg state is loaded and injected so the tool reads ``ctx.input_responses``
+    / ``ctx.request_state`` identically to the foreground guard contract. Leg 1
+    loads nothing (both ``None``).
     """
     from fastmcp.server.context import Context
     from fastmcp.server.dependencies import get_server
@@ -407,4 +436,19 @@ async def make_task_context() -> Context | None:
         origin_request_id=origin_request_id,
     )
     await ctx.__aenter__()
+
+    docket = server._docket
+    if docket is None:
+        from fastmcp_tasks.dependencies import _current_docket
+
+        docket = _current_docket.get()
+    if docket is not None:
+        from fastmcp_tasks.input_store import load_pending_input
+
+        request_state, input_responses = await load_pending_input(
+            docket, task_info.task_scope, task_info.task_id
+        )
+        ctx._task_request_state = request_state
+        ctx._task_input_responses = input_responses
+
     return ctx
diff --git a/fastmcp_tasks/fastmcp_tasks/creation.py b/fastmcp_tasks/fastmcp_tasks/creation.py
index 9409c9117..678e57881 100644
--- a/fastmcp_tasks/fastmcp_tasks/creation.py
+++ b/fastmcp_tasks/fastmcp_tasks/creation.py
@@ -31,6 +31,7 @@ from fastmcp_tasks.context import (
     register_task_server,
 )
 from fastmcp_tasks.dependencies import _current_docket
+from fastmcp_tasks.input_store import save_current_leg, save_task_args
 from fastmcp_tasks.keys import build_task_key, task_redis_prefix
 from fastmcp_tasks.models import CreateTaskResult
 
@@ -74,8 +75,9 @@ async def create_task(
     # argument-splatting match what the worker will invoke.
     component = await _registered_task_component(context, tool)
 
+    raw_arguments = dict(arguments or {})
     coerced = coerce_task_arguments(
-        component, dict(arguments or {}), strict=_strict_input_validation()
+        component, raw_arguments, strict=_strict_input_validation()
     )
 
     task_id = secrets.token_urlsafe(32)
@@ -114,6 +116,13 @@ async def create_task(
         await redis.set(created_at_key, created_at, ex=ttl_seconds)
         await redis.set(poll_interval_key, str(poll_interval_ms), ex=ttl_seconds)
 
+    # End-and-reenter state: the raw (wire) arguments feed every leg, re-coerced
+    # per leg, and the leg pointer starts at leg 1 (the base task key). A guard
+    # return re-enters by enqueuing the next leg with these same arguments (see
+    # handlers.enqueue_next_leg).
+    await save_task_args(docket, task_scope, task_id, raw_arguments, ttl_seconds)
+    await save_current_leg(docket, task_scope, task_id, task_key, 1, ttl_seconds)
+
     await snapshot.save(docket, task_scope, task_id, ttl_seconds)
 
     await add_component_to_docket(
@@ -150,6 +159,47 @@ def _owning_server(tool: Tool, fallback: FastMCP) -> FastMCP:
     return fallback
 
 
+async def registered_component_for_key(server: FastMCP, component_key: str) -> Tool:
+    """Return the Docket-registered task component matching ``component_key``.
+
+    ``get_tasks()`` yields the same components registered with Docket (the
+    underlying ``FunctionTool`` for a mounted tool, not a provider wrapper), so
+    matching by ``key`` recovers the component whose calling convention agrees
+    with the worker. Used when re-entering a task leg, where only the stored
+    compound key (not the original ``Tool`` object) is available.
+    """
+    for component in await server.get_tasks():
+        if component.key == component_key and isinstance(component, Tool):
+            return component
+    raise MCPError(
+        code=INTERNAL_ERROR,
+        message=f"No task-enabled component found for {component_key!r}.",
+    )
+
+
+async def enqueue_task_leg(
+    server: FastMCP,
+    docket: Docket,
+    component: Tool,
+    raw_arguments: dict[str, object],
+    leg_key: str,
+) -> None:
+    """Enqueue a fresh Docket execution (the next leg) for a re-entered task.
+
+    Re-coerces the stored wire arguments (each leg validates independently, as a
+    foreground retry would) and adds the component's registered callable — the
+    capture wrapper — under ``leg_key``. Waits for the execution to become
+    durable so a ``tasks/get`` immediately after ``tasks/update`` resolves.
+    """
+    coerced = coerce_task_arguments(
+        component, dict(raw_arguments), strict=_strict_input_validation()
+    )
+    await add_component_to_docket(
+        component, docket, coerced, fn_key=component.key, task_key=leg_key
+    )
+    await _await_durable_creation(docket, leg_key)
+
+
 async def _registered_task_component(context: Context, tool: Tool) -> Tool:
     """Return the component Docket registered for ``tool``'s key.
 
diff --git a/fastmcp_tasks/fastmcp_tasks/extension.py b/fastmcp_tasks/fastmcp_tasks/extension.py
index 5b3e2be1f..45ea7f6ec 100644
--- a/fastmcp_tasks/fastmcp_tasks/extension.py
+++ b/fastmcp_tasks/fastmcp_tasks/extension.py
@@ -203,10 +203,10 @@ class TasksExtension(ServerExtension):
     async def lifespan(self) -> AsyncIterator[None]:
         """Start the Docket backend/worker and install the worker-side hooks.
 
-        Installs core's background-context factory and in-task elicitation
-        handler for the duration so a worker's ``ctx`` (progress, elicitation)
-        functions, then runs the Docket lifespan. The hooks are process-global
-        and refcounted: with several servers in one process (each its own
+        Installs core's background-context factory and worker-server resolver for
+        the duration so a worker's ``ctx`` (progress, server resolution) works,
+        then runs the Docket lifespan. The hooks are process-global and
+        refcounted: with several servers in one process (each its own
         runtime-tree root), the hooks stay installed until the last tasks
         extension shuts down, so one server's exit cannot strand another
         server's in-flight workers.
@@ -231,20 +231,17 @@ _active_worker_hook_holds: int = 0
 
 
 def _install_worker_hooks() -> None:
-    from fastmcp.server.context import set_task_elicitation_handler
     from fastmcp.server.dependencies import (
         set_background_context_factory,
         set_worker_server_resolver,
     )
     from fastmcp_tasks import wire_production
     from fastmcp_tasks.context import make_task_context, resolve_worker_server
-    from fastmcp_tasks.input_store import elicit_in_task
 
     global _active_worker_hook_holds
     _active_worker_hook_holds += 1
     set_background_context_factory(make_task_context)
     set_worker_server_resolver(resolve_worker_server)
-    set_task_elicitation_handler(elicit_in_task)
     # Enable server-side production of the claimed CreateTaskResult on tools/call
     # (the SDK ships only claim consumption). Refcounted independently but
     # installed/released in lockstep with the worker hooks.
@@ -252,7 +249,6 @@ def _install_worker_hooks() -> None:
 
 
 def _release_worker_hooks() -> None:
-    from fastmcp.server.context import set_task_elicitation_handler
     from fastmcp.server.dependencies import (
         set_background_context_factory,
         set_worker_server_resolver,
@@ -263,7 +259,6 @@ def _release_worker_hooks() -> None:
     _active_worker_hook_holds -= 1
     if _active_worker_hook_holds <= 0:
         _active_worker_hook_holds = 0
-        set_task_elicitation_handler(None)
         set_worker_server_resolver(None)
         set_background_context_factory(None)
     wire_production.uninstall()
diff --git a/fastmcp_tasks/fastmcp_tasks/handlers.py b/fastmcp_tasks/fastmcp_tasks/handlers.py
index e7c906b8e..c65929013 100644
--- a/fastmcp_tasks/fastmcp_tasks/handlers.py
+++ b/fastmcp_tasks/fastmcp_tasks/handlers.py
@@ -33,8 +33,21 @@ from fastmcp.tools.base import InputRequiredToolResult, Tool
 from fastmcp.utilities.tasks import DEFAULT_POLL_INTERVAL_MS
 from fastmcp.utilities.versions import VersionSpec
 from fastmcp_tasks.context import get_task_scope
-from fastmcp_tasks.input_store import deliver_input_responses, read_outstanding_inputs
-from fastmcp_tasks.keys import parse_task_key, task_redis_prefix
+from fastmcp_tasks.creation import enqueue_task_leg, registered_component_for_key
+from fastmcp_tasks.input_store import (
+    clear_outstanding,
+    load_current_leg,
+    load_task_args,
+    read_outstanding_inputs,
+    save_current_leg,
+    store_input_responses,
+    translate_responses,
+)
+from fastmcp_tasks.keys import (
+    leg_execution_key,
+    parse_task_key,
+    task_redis_prefix,
+)
 from fastmcp_tasks.models import (
     CancelTaskResult,
     GetTaskResult,
@@ -57,10 +70,6 @@ DOCKET_TO_MCP_STATE: dict[ExecutionState, str] = {
     ExecutionState.CANCELLED: "cancelled",
 }
 
-_WORKING_STATES = frozenset(
-    {ExecutionState.SCHEDULED, ExecutionState.QUEUED, ExecutionState.RUNNING}
-)
-
 
 def _task_not_found(task_id: str) -> MCPError:
     """The single "not found" error for missing, expired, or cross-scope ids.
@@ -96,12 +105,14 @@ def _ttl_ms(docket: Docket) -> int:
 
 async def _lookup_task(
     docket: Docket, task_scope: str | None, task_id: str
-) -> tuple[Any, str, str | None, int]:
-    """Resolve a task's execution and stored metadata within the caller's scope.
+) -> tuple[Any, str, int, str | None, int]:
+    """Resolve a task's current-leg execution and metadata within the scope.
 
-    Returns ``(execution, task_key, created_at, poll_interval_ms)``. Raises the
-    shared "not found" error when the scope-prefixed metadata is absent or the
-    execution has expired.
+    Returns ``(execution, base_task_key, leg_number, created_at,
+    poll_interval_ms)``. The execution is the *current leg* (the latest Docket
+    execution), which for a re-entered task differs from the base task key.
+    Raises the shared "not found" error when the scope-prefixed metadata is
+    absent or the current leg's execution has expired.
     """
     prefix = task_redis_prefix(task_scope)
     meta_key = docket.key(f"{prefix}:{task_id}")
@@ -115,11 +126,13 @@ async def _lookup_task(
         values = await redis.mget(meta_key, created_at_key, poll_key)  # ty: ignore[too-many-positional-arguments]
     task_key_bytes, created_at_bytes, poll_bytes = values
 
-    task_key = task_key_bytes.decode("utf-8") if task_key_bytes else None
-    if not task_key:
+    base_task_key = task_key_bytes.decode("utf-8") if task_key_bytes else None
+    if not base_task_key:
         raise _task_not_found(task_id)
 
-    execution = await docket.get_execution(task_key)
+    current_leg_key, leg_number = await load_current_leg(docket, task_scope, task_id)
+    execution_key = current_leg_key or base_task_key
+    execution = await docket.get_execution(execution_key)
     if not execution:
         raise _task_not_found(task_id)
 
@@ -132,7 +145,7 @@ async def _lookup_task(
     except (ValueError, UnicodeDecodeError):
         poll_interval_ms = DEFAULT_POLL_INTERVAL_MS
 
-    return execution, task_key, created_at, poll_interval_ms
+    return execution, base_task_key, leg_number, created_at, poll_interval_ms
 
 
 async def _resolve_tool(server: FastMCP, task_key: str) -> Tool:
@@ -160,17 +173,21 @@ async def _resolve_tool(server: FastMCP, task_key: str) -> Tool:
 def _inline_result(tool: Tool, raw_value: Any) -> dict[str, Any]:
     """Convert a completed task's raw return into an inlined CallToolResult dict.
 
-    A guard tool that returned an ``InputRequiredResult`` from inside a task is
-    rejected: multi-round-trip guards need a live request to answer the prompt
-    and cannot complete as a task.
+    A completed task should never carry an ``InputRequiredResult``: a function
+    tool's guard returns are captured by the end-and-reenter wrapper (see
+    ``input_loop.py``), which records the leg's outstanding requests and ends the
+    leg (returning ``None``), so ``tasks/get`` reports ``input_required`` rather
+    than inlining. Reaching here with a guard result means a component type the
+    wrapper does not wrap (e.g. a base ``Tool``) returned one, which the task
+    path cannot drive — a safety net, not an expected path.
     """
     if isinstance(raw_value, mcp_types.InputRequiredResult | InputRequiredToolResult):
         raise MCPError(
             code=mcp_types.INTERNAL_ERROR,
             message=(
-                f"Tool {tool.name!r} requested input while running as a background "
-                "task. Input-required (multi-round-trip) tools need a live request "
-                "to answer the prompt and cannot run as tasks."
+                f"Tool {tool.name!r} returned an input-required result as a task, "
+                "but its component type is not driven by the in-task guard loop. "
+                "Guard-pattern tasks are supported for function tools."
             ),
         )
     mcp_result = tool.convert_result(raw_value).to_mcp_result()
@@ -193,9 +210,13 @@ async def tasks_get(server: FastMCP, task_id: str) -> GetTaskResult:
         raise _task_not_found(task_id)
 
     task_scope = get_task_scope()
-    execution, task_key, created_at, poll_interval_ms = await _lookup_task(
-        docket, task_scope, task_id
-    )
+    (
+        execution,
+        base_task_key,
+        leg_number,
+        created_at,
+        poll_interval_ms,
+    ) = await _lookup_task(docket, task_scope, task_id)
     await execution.sync()
 
     created_at_iso = _normalize_iso_timestamp(created_at)
@@ -218,16 +239,17 @@ async def tasks_get(server: FastMCP, task_id: str) -> GetTaskResult:
             **payload,
         )
 
-    # An outstanding input request outranks the Docket "running" state: the task
-    # is parked in the worker waiting for tasks/update, so it is input_required.
-    if execution.state in _WORKING_STATES:
-        outstanding = await read_outstanding_inputs(docket, task_scope, task_id)
+    if execution.state == ExecutionState.COMPLETED:
+        # A guard leg ends its Docket execution and records outstanding input
+        # requests to Redis: a completed leg with outstanding requests is the
+        # task waiting for tasks/update (input_required), not a finished task.
+        outstanding = await read_outstanding_inputs(
+            docket, task_scope, task_id, leg_number
+        )
         if outstanding:
             return build("input_required", input_requests=outstanding)
-
-    if execution.state == ExecutionState.COMPLETED:
         raw_value = await execution.get_result(timeout=timedelta(seconds=0))
-        tool = await _resolve_tool(server, task_key)
+        tool = await _resolve_tool(server, base_task_key)
         return build("completed", result=_inline_result(tool, raw_value))
 
     if execution.state == ExecutionState.FAILED:
@@ -257,26 +279,66 @@ async def tasks_get(server: FastMCP, task_id: str) -> GetTaskResult:
 async def tasks_update(
     server: FastMCP, task_id: str, input_responses: dict[str, Any]
 ) -> UpdateTaskResult:
-    """Handle ``tasks/update``: deliver input responses to the parked worker."""
+    """Handle ``tasks/update``: answer a guard leg and re-enter the task.
+
+    The responses are keyed by the surfaced keys ``tasks/get`` reported. Unknown
+    or already-satisfied keys are ignored (SEP-2663). When at least one answer
+    matches the current leg's outstanding requests, they are translated to the
+    tool's own keys, stored for the next leg, and a fresh Docket execution (the
+    next leg) is enqueued with the task's arguments. The worker is never blocked;
+    re-entry is the whole mechanism. A stale or empty update is an idempotent
+    no-op.
+    """
     docket = server._docket
     if docket is None:
         raise _task_not_found(task_id)
 
     task_scope = get_task_scope()
     # Resolve within scope so a cross-scope update is a "not found", not a no-op.
-    await _lookup_task(docket, task_scope, task_id)
-    await deliver_input_responses(docket, task_scope, task_id, input_responses)
+    _execution, base_task_key, leg_number, _created_at, _poll = await _lookup_task(
+        docket, task_scope, task_id
+    )
+
+    translated = await translate_responses(
+        docket, task_scope, task_id, leg_number, input_responses
+    )
+    if translated is None:
+        # Nothing matched the current leg's outstanding requests: the leg was
+        # already answered, or the keys are unknown. Idempotent no-op.
+        return UpdateTaskResult()
+
+    # Store the answers for the next leg to read, then enqueue that leg. Ordering
+    # matters: the answers must be in Redis before the next leg's worker context
+    # loads them, and current_leg must not advance to an execution that is not
+    # yet durable — so enqueue (with its durable wait) precedes the pointer swap.
+    await store_input_responses(docket, task_scope, task_id, translated)
+
+    component = await registered_component_for_key(
+        server, parse_task_key(base_task_key)["component_identifier"]
+    )
+    raw_arguments = await load_task_args(docket, task_scope, task_id)
+    next_leg = leg_number + 1
+    next_leg_key = leg_execution_key(base_task_key, next_leg)
+
+    await enqueue_task_leg(server, docket, component, raw_arguments, next_leg_key)
+    ttl_seconds = int(docket.execution_ttl.total_seconds())
+    await save_current_leg(
+        docket, task_scope, task_id, next_leg_key, next_leg, ttl_seconds
+    )
+    # The answered leg's surfaced keys are now superseded; drop them so they are
+    # never reused (SEP-2663 L350).
+    await clear_outstanding(docket, task_scope, task_id, leg_number)
     return UpdateTaskResult()
 
 
 async def tasks_cancel(server: FastMCP, task_id: str) -> CancelTaskResult:
-    """Handle ``tasks/cancel``: cooperatively cancel the task, empty ack."""
+    """Handle ``tasks/cancel``: cooperatively cancel the current leg, empty ack."""
     docket = server._docket
     if docket is None:
         raise _task_not_found(task_id)
 
     task_scope = get_task_scope()
-    execution, _task_key, _created_at, _poll = await _lookup_task(
+    execution, _base_task_key, _leg, _created_at, _poll = await _lookup_task(
         docket, task_scope, task_id
     )
     await docket.cancel(execution.key)
diff --git a/fastmcp_tasks/fastmcp_tasks/input_loop.py b/fastmcp_tasks/fastmcp_tasks/input_loop.py
new file mode 100644
index 000000000..ed36def55
--- /dev/null
+++ b/fastmcp_tasks/fastmcp_tasks/input_loop.py
@@ -0,0 +1,137 @@
+"""The end-and-reenter capture wrapper for guard-pattern task tools.
+
+A guard tool asks for input by *returning* an `InputRequiredResult` rather than
+awaiting `ctx.elicit()`. Foreground, each such return is one leg of a
+multi-round-trip: the tool returns, the client answers, the framework re-invokes
+the tool with the answers on `ctx.input_responses`. The tool body is written
+once and is oblivious to how many legs it takes.
+
+As a background task the leg boundary is a *worker* boundary. This wrapper runs
+the tool body exactly once. If the body returns a real value, it is the leg's
+result. If the body returns an `InputRequiredResult`, the wrapper records the
+leg's outstanding requests (and any carried `request_state`) to Redis and
+returns — the Docket execution then completes and the worker is freed. The task
+sits in `input_required` as durable state until the client answers via
+`tasks/update`, which enqueues a fresh Docket execution (the next leg) that
+re-runs this wrapper with the accumulated state injected onto `ctx`. No worker
+is ever blocked awaiting input.
+
+The wrapper preserves the wrapped callable's signature so Docket's dependency
+injection still resolves the tool's parameters (its own args, `ctx`, and any
+Docket-native dependencies) exactly as it would for the raw callable. The
+per-leg state (`ctx.input_responses` / `ctx.request_state`) is injected by the
+worker `Context` factory (`make_task_context`) before the body runs.
+"""
+
+from __future__ import annotations
+
+import functools
+import inspect
+import logging
+from typing import TYPE_CHECKING, Any
+
+import mcp_types
+
+from fastmcp.tools.base import InputRequiredToolResult
+from fastmcp_tasks.context import get_task_context, get_task_leg_number
+from fastmcp_tasks.input_store import store_outstanding
+
+if TYPE_CHECKING:
+    from collections.abc import Awaitable, Callable
+
+    from docket import Docket
+
+logger = logging.getLogger(__name__)
+
+
+def _as_input_required(result: Any) -> mcp_types.InputRequiredResult | None:
+    """Return the `InputRequiredResult` a guard leg produced, or None.
+
+    A tool body may return the bare `InputRequiredResult` or the
+    `InputRequiredToolResult` wrapper FastMCP uses foreground; both mean the same
+    ask.
+    """
+    if isinstance(result, InputRequiredToolResult):
+        return result.input_required
+    if isinstance(result, mcp_types.InputRequiredResult):
+        return result
+    return None
+
+
+def _serialize_requests(
+    input_requests: mcp_types.InputRequests,
+) -> dict[str, dict[str, Any]]:
+    """Dump each request to the wire payload surfaced for the client to answer."""
+    return {
+        key: request.model_dump(by_alias=True, mode="json", exclude_none=True)
+        for key, request in input_requests.items()
+    }
+
+
+def _resolve_docket() -> Docket | None:
+    """Resolve the active Docket from the current context or worker default."""
+    from fastmcp.server.dependencies import get_context
+    from fastmcp_tasks.dependencies import _current_docket
+
+    try:
+        docket = get_context().fastmcp._docket
+    except RuntimeError:
+        docket = None
+    if docket is None:
+        docket = _current_docket.get()
+    return docket
+
+
+def reentrant_task_fn(
+    fn: Callable[..., Awaitable[Any]],
+) -> Callable[..., Awaitable[Any]]:
+    """Wrap a task tool's callable to capture a guard leg's ask (end-and-reenter).
+
+    Signature-preserving, so Docket injects the wrapped callable's parameters
+    unchanged. The body runs exactly once: a real return is the leg's result; an
+    `InputRequiredResult` is captured to Redis (outstanding requests + carried
+    state) and the wrapper returns, ending the leg without blocking. The next
+    leg is enqueued by ``tasks/update`` when the client answers.
+    """
+
+    @functools.wraps(fn)
+    async def wrapper(*args: Any, **kwargs: Any) -> Any:
+        result = await fn(*args, **kwargs)
+        input_required = _as_input_required(result)
+        if input_required is None:
+            return result
+
+        requests = input_required.input_requests or {}
+        request_state = input_required.request_state
+        if not requests and request_state is None:
+            # A leg that asks nothing and carries nothing can never be answered;
+            # treat it as the terminal result rather than an unanswerable park.
+            return result
+
+        task_context = get_task_context()
+        docket = _resolve_docket()
+        if task_context is None or docket is None:
+            logger.warning(
+                "guard leg produced an ask outside a task worker; returning it"
+            )
+            return result
+
+        await store_outstanding(
+            docket,
+            task_context.task_scope,
+            task_context.task_id,
+            get_task_leg_number(),
+            _serialize_requests(requests),
+            request_state,
+        )
+        # The leg ends here: the Docket execution completes and the worker is
+        # freed. The task is now input_required until tasks/update enqueues the
+        # next leg. Return None so the completed leg carries no stray result.
+        return None
+
+    # `functools.wraps` copies `__wrapped__`, so `inspect.signature` already
+    # unwraps to `fn`; set it explicitly too, so a dependency injector reading
+    # `__signature__` directly (rather than following `__wrapped__`) still sees
+    # the tool's real parameters.
+    wrapper.__signature__ = inspect.signature(fn)  # ty: ignore[unresolved-attribute]
+    return wrapper
diff --git a/fastmcp_tasks/fastmcp_tasks/input_store.py b/fastmcp_tasks/fastmcp_tasks/input_store.py
index 5c032be06..48af941ea 100644
--- a/fastmcp_tasks/fastmcp_tasks/input_store.py
+++ b/fastmcp_tasks/fastmcp_tasks/input_store.py
@@ -1,142 +1,260 @@
-"""In-task input store for SEP-2663 poll-based elicitation.
+"""Per-task Redis state for SEP-2663 end-and-reenter input gathering.
 
-When a background task calls ``ctx.elicit()`` it has no live request to carry the
-prompt. SEP-2663 handles this by polling: the worker parks an *input request*
-here, the task's ``tasks/get`` status flips to ``input_required`` with the
-outstanding requests, the client answers via ``tasks/update``, and the parked
-worker resumes.
+A background task gathers client input by *ending a leg* and re-entering, never
+by blocking a worker. When a `task=True` tool returns an `InputRequiredResult`,
+the leg's Docket execution completes and the worker is freed; the task's state
+lives here in Redis as `input_required`. When the client answers via
+`tasks/update`, a fresh Docket execution (the next leg) re-runs the tool with the
+accumulated state injected onto its `Context`. No worker ever waits for input.
 
-This is the reworked SEP-1686 elicitation module. The Redis request/response
-mechanics — a per-request hash the poll surface reads and a per-key list the
-worker blocks on with ``BLPOP`` — are preserved. What's gone is the *push
-envelope*: the old code sent a ``notifications/tasks/status`` through the
-distributed notification queue to wake the client. Under SEP-2663 the client
-discovers the outstanding request by polling ``tasks/get``, so no push is needed.
+This module owns the durable state each task carries between legs:
+
+- **args** — the original tool arguments, re-supplied to every leg.
+- **current_leg / leg** — the latest leg's Docket execution key and its number.
+- **request_state** — the opaque string a leg carried forward (SEP-2322).
+- **input_responses** — the typed answers the last `tasks/update` delivered,
+  translated to the tool's own request keys.
+- **input:requests / input:map** — the current leg's outstanding requests, keyed
+  by a server-minted surfaced key, plus the surfaced-key → tool-key mapping.
+
+Each surfaced request key is minted fresh with high-entropy suffix and never
+reused after its response is delivered (SEP-2663 L350): a task that asks twice,
+or a leg that requests several inputs at once, surfaces distinct, independently
+answerable keys, and the tool reads its *own* keys on the next leg via the
+translated `input_responses`.
 """
 
 from __future__ import annotations
 
 import json
 import logging
-from typing import TYPE_CHECKING, Any
+import secrets
+from typing import TYPE_CHECKING, Any, cast
 
 import mcp_types
-from redis.exceptions import RedisError
 
-from fastmcp_tasks.context import get_task_context
 from fastmcp_tasks.keys import task_redis_prefix
 
 if TYPE_CHECKING:
     from docket import Docket
 
-    from fastmcp.server.context import Context
-
 logger = logging.getLogger(__name__)
 
-# How long a parked input request (and any delivered response) lives before
-# expiring. A task blocked on input holds a worker slot, so this doubles as the
-# maximum time a worker waits for the client to answer.
+# How long a task's input state (outstanding requests and delivered responses)
+# lives before expiring. With end-and-reenter no worker is held while a task is
+# input_required, so this bounds only how long durable input state survives, not
+# any worker slot.
 INPUT_TTL_SECONDS = 3600
 
+# Reconstruct a typed response from its stored `{"type": name, "data": dump}`
+# form so a re-entered leg reads a real `ElicitResult` (etc.) on
+# `ctx.input_responses`, matching the foreground guard contract.
+_RESULT_TYPE_BY_NAME: dict[str, type[mcp_types.Result]] = {
+    "ElicitResult": mcp_types.ElicitResult,
+    "CreateMessageResult": mcp_types.CreateMessageResult,
+    "CreateMessageResultWithTools": mcp_types.CreateMessageResultWithTools,
+    "ListRootsResult": mcp_types.ListRootsResult,
+}
 
-def _requests_key(docket: Docket, task_scope: str | None, task_id: str) -> str:
-    """Redis hash of outstanding input requests, keyed by input key."""
-    return docket.key(f"{task_redis_prefix(task_scope)}:{task_id}:input:requests")
+# Map an outstanding request's wire method to the result type its answer
+# validates into. Elicitation is the supported in-task input; the others are
+# kept complete so a client that answers one is parsed rather than dropped.
+_RESULT_TYPE_BY_METHOD: dict[str, type[mcp_types.Result]] = {
+    "elicitation/create": mcp_types.ElicitResult,
+    "sampling/createMessage": mcp_types.CreateMessageResult,
+    "roots/list": mcp_types.ListRootsResult,
+}
 
 
-def _response_key(
-    docket: Docket, task_scope: str | None, task_id: str, input_key: str
+def result_type_for_method(method: str) -> type[mcp_types.Result]:
+    """The result type an outstanding request's answer validates into."""
+    return _RESULT_TYPE_BY_METHOD.get(method, mcp_types.ElicitResult)
+
+
+def _prefix(docket: Docket, task_scope: str | None, task_id: str) -> str:
+    return f"{task_redis_prefix(task_scope)}:{task_id}"
+
+
+def _args_key(docket: Docket, task_scope: str | None, task_id: str) -> str:
+    return docket.key(f"{_prefix(docket, task_scope, task_id)}:args")
+
+
+def _current_leg_key(docket: Docket, task_scope: str | None, task_id: str) -> str:
+    return docket.key(f"{_prefix(docket, task_scope, task_id)}:current_leg")
+
+
+def _leg_number_key(docket: Docket, task_scope: str | None, task_id: str) -> str:
+    return docket.key(f"{_prefix(docket, task_scope, task_id)}:leg")
+
+
+def _request_state_key(docket: Docket, task_scope: str | None, task_id: str) -> str:
+    return docket.key(f"{_prefix(docket, task_scope, task_id)}:request_state")
+
+
+def _input_responses_key(docket: Docket, task_scope: str | None, task_id: str) -> str:
+    return docket.key(f"{_prefix(docket, task_scope, task_id)}:input_responses")
+
+
+def _requests_key(
+    docket: Docket, task_scope: str | None, task_id: str, leg: int
 ) -> str:
-    """Redis list the worker blocks on for a single input key's response."""
-    return docket.key(
-        f"{task_redis_prefix(task_scope)}:{task_id}:input:resp:{input_key}"
-    )
+    """Redis hash of a leg's outstanding input requests, keyed by surfaced key.
 
-
-def _elicitation_input_request(message: str, schema: dict[str, Any]) -> dict[str, Any]:
-    """Build the SEP-2663 ``InputRequest`` for an elicitation (an ElicitRequest)."""
-    return {
-        "method": "elicitation/create",
-        "params": {"message": message, "requestedSchema": schema},
-    }
-
-
-async def elicit_in_task(
-    context: Context, message: str, schema: dict[str, Any]
-) -> mcp_types.ElicitResult:
-    """Park an elicitation request and block until the client answers it.
-
-    Installed as core's in-task elicitation handler by ``TasksExtension``. Parks
-    an input request keyed by the task's own id (one outstanding elicitation per
-    task at a time — the polling model is inherently sequential), flips the
-    task's polled status to ``input_required``, and blocks on the response list.
-    Returns the client's ``ElicitResult``; on timeout or a missing task context,
-    returns a ``cancel`` action so the worker never hangs indefinitely.
+    Scoped by leg number so a re-entered leg's fresh requests never collide with
+    the answered leg's stale ones in the shared keyspace.
     """
-    task_context = get_task_context()
-    if task_context is None:
-        logger.warning("elicit_in_task called outside a task worker; cancelling")
-        return mcp_types.ElicitResult(action="cancel", content=None)
+    return docket.key(f"{_prefix(docket, task_scope, task_id)}:input:{leg}:requests")
 
-    docket = context.fastmcp._docket
-    if docket is None:
-        from fastmcp_tasks.dependencies import _current_docket
 
-        docket = _current_docket.get()
-    if docket is None:
-        return mcp_types.ElicitResult(action="cancel", content=None)
+def _map_key(docket: Docket, task_scope: str | None, task_id: str, leg: int) -> str:
+    """Redis hash mapping a leg's surfaced keys back to the tool's own keys."""
+    return docket.key(f"{_prefix(docket, task_scope, task_id)}:input:{leg}:map")
 
-    task_scope = task_context.task_scope
-    task_id = task_context.task_id
-    # One elicitation outstanding per task: key the request by the task id so the
-    # inputRequests map surfaced by tasks/get is stable and answerable.
-    input_key = task_id
 
-    requests_key = _requests_key(docket, task_scope, task_id)
-    response_key = _response_key(docket, task_scope, task_id, input_key)
-    request_payload = _elicitation_input_request(message, schema)
+def _mint_surfaced_key(task_id: str) -> str:
+    """Mint a unique surfaced key for one outstanding request (SEP-2663 L350).
 
+    Namespaced by the task id and suffixed with fresh entropy so no two
+    requests — across legs or within one leg — ever collide, and a key is never
+    reused after its response is delivered.
+    """
+    return f"{task_id}:{secrets.token_hex(8)}"
+
+
+def _decode(value: Any) -> str | None:
+    if value is None:
+        return None
+    if isinstance(value, bytes):
+        return value.decode("utf-8")
+    return str(value)
+
+
+# ---------------------------------------------------------------------------
+# Task arguments and leg pointer (written at create, advanced at tasks/update)
+# ---------------------------------------------------------------------------
+
+
+async def save_task_args(
+    docket: Docket,
+    task_scope: str | None,
+    task_id: str,
+    arguments: dict[str, Any],
+    ttl_seconds: int,
+) -> None:
+    """Store the original tool arguments, re-supplied to every leg."""
     async with docket.redis() as redis:
-        await redis.hset(requests_key, input_key, json.dumps(request_payload))
-        await redis.expire(requests_key, INPUT_TTL_SECONDS)
+        await redis.set(
+            _args_key(docket, task_scope, task_id),
+            json.dumps(arguments),
+            ex=ttl_seconds,
+        )
 
+
+async def load_task_args(
+    docket: Docket, task_scope: str | None, task_id: str
+) -> dict[str, Any]:
+    """Load the stored tool arguments for a task's next leg."""
+    async with docket.redis() as redis:
+        raw = await redis.get(_args_key(docket, task_scope, task_id))
+    decoded = _decode(raw)
+    if not decoded:
+        return {}
+    parsed = json.loads(decoded)
+    return parsed if isinstance(parsed, dict) else {}
+
+
+async def save_current_leg(
+    docket: Docket,
+    task_scope: str | None,
+    task_id: str,
+    leg_key: str,
+    leg_number: int,
+    ttl_seconds: int,
+) -> None:
+    """Record the latest leg's Docket execution key and its number."""
+    async with docket.redis() as redis:
+        await redis.set(
+            _current_leg_key(docket, task_scope, task_id), leg_key, ex=ttl_seconds
+        )
+        await redis.set(
+            _leg_number_key(docket, task_scope, task_id),
+            str(leg_number),
+            ex=ttl_seconds,
+        )
+
+
+async def load_current_leg(
+    docket: Docket, task_scope: str | None, task_id: str
+) -> tuple[str | None, int]:
+    """Return the current leg's execution key and number (defaults to 1)."""
+    async with docket.redis() as redis:
+        leg_key = _decode(
+            await redis.get(_current_leg_key(docket, task_scope, task_id))
+        )
+        leg_raw = _decode(await redis.get(_leg_number_key(docket, task_scope, task_id)))
     try:
-        async with docket.redis() as redis:
-            result = await redis.blpop([response_key], timeout=INPUT_TTL_SECONDS)
-    except (RedisError, OSError) as exc:
-        logger.warning("BLPOP failed for task %s input; cancelling: %s", task_id, exc)
-        result = None
+        leg_number = int(leg_raw) if leg_raw else 1
+    except ValueError:
+        leg_number = 1
+    return leg_key, leg_number
+
+
+# ---------------------------------------------------------------------------
+# Outstanding requests (written by the capture wrapper, read by tasks/get)
+# ---------------------------------------------------------------------------
+
+
+async def store_outstanding(
+    docket: Docket,
+    task_scope: str | None,
+    task_id: str,
+    leg: int,
+    serialized_requests: dict[str, dict[str, Any]],
+    request_state: str | None,
+    ttl_seconds: int = INPUT_TTL_SECONDS,
+) -> None:
+    """Persist a leg's outstanding input requests plus its carried state.
+
+    ``serialized_requests`` maps the tool's own request keys to serialized
+    ``InputRequest`` payloads. Each is stored under a freshly minted surfaced
+    key, with the surfaced-key → tool-key mapping recorded alongside so
+    ``tasks/update`` can translate answers back. ``request_state`` is written
+    when the leg carried one and cleared otherwise, so it travels to the next
+    leg verbatim.
+    """
+    requests_key = _requests_key(docket, task_scope, task_id, leg)
+    map_key = _map_key(docket, task_scope, task_id, leg)
+    state_key = _request_state_key(docket, task_scope, task_id)
 
     async with docket.redis() as redis:
-        await redis.hdel(requests_key, input_key)
-        await redis.delete(response_key)
-
-    if not result:
-        return mcp_types.ElicitResult(action="cancel", content=None)
-
-    _key, raw = result
-    response = json.loads(raw)
-    return mcp_types.ElicitResult(
-        action=response.get("action", "accept"),
-        content=response.get("content"),
-    )
+        for tool_key, payload in serialized_requests.items():
+            surfaced = _mint_surfaced_key(task_id)
+            await redis.hset(requests_key, surfaced, json.dumps(payload))
+            await redis.hset(map_key, surfaced, tool_key)
+        await redis.expire(requests_key, ttl_seconds)
+        await redis.expire(map_key, ttl_seconds)
+        if request_state is not None:
+            await redis.set(state_key, request_state, ex=ttl_seconds)
+        else:
+            await redis.delete(state_key)
 
 
 async def read_outstanding_inputs(
-    docket: Docket, task_scope: str | None, task_id: str
+    docket: Docket, task_scope: str | None, task_id: str, leg: int
 ) -> dict[str, Any]:
-    """Return the task's outstanding input requests, keyed by input key.
+    """Return a leg's outstanding input requests, keyed by surfaced key.
 
-    Empty when the task is not waiting on input. Consumed by ``tasks/get`` to
+    Empty when the leg is not waiting on input. Consumed by ``tasks/get`` to
     build the ``input_required`` status and its ``inputRequests`` snapshot.
     """
-    requests_key = _requests_key(docket, task_scope, task_id)
     async with docket.redis() as redis:
-        raw = await redis.hgetall(requests_key)
+        raw = await redis.hgetall(_requests_key(docket, task_scope, task_id, leg))
     outstanding: dict[str, Any] = {}
     for key, value in raw.items():
-        key_str = key.decode() if isinstance(key, bytes) else key
-        value_str = value.decode() if isinstance(value, bytes) else value
+        key_str = _decode(key)
+        value_str = _decode(value)
+        if key_str is None or value_str is None:
+            continue
         try:
             outstanding[key_str] = json.loads(value_str)
         except json.JSONDecodeError:
@@ -144,26 +262,136 @@ async def read_outstanding_inputs(
     return outstanding
 
 
-async def deliver_input_responses(
+async def _read_outstanding_map(
+    docket: Docket, task_scope: str | None, task_id: str, leg: int
+) -> dict[str, str]:
+    """Return the surfaced-key → tool-key mapping for a leg."""
+    async with docket.redis() as redis:
+        raw = await redis.hgetall(_map_key(docket, task_scope, task_id, leg))
+    mapping: dict[str, str] = {}
+    for key, value in raw.items():
+        key_str = _decode(key)
+        value_str = _decode(value)
+        if key_str is None or value_str is None:
+            continue
+        mapping[key_str] = value_str
+    return mapping
+
+
+# ---------------------------------------------------------------------------
+# Responses (written by tasks/update, read by the next leg's context factory)
+# ---------------------------------------------------------------------------
+
+
+async def translate_responses(
     docket: Docket,
     task_scope: str | None,
     task_id: str,
+    leg: int,
     responses: dict[str, Any],
-) -> None:
-    """Deliver ``tasks/update`` responses to the parked worker(s).
+) -> dict[str, mcp_types.Result] | None:
+    """Translate a ``tasks/update`` payload into typed, tool-keyed responses.
 
-    For each response whose key names an outstanding request, pushes the
-    response onto that key's list (waking the worker's ``BLPOP``) and removes the
-    request. Responses for unknown or already-satisfied keys are ignored, as the
-    spec requires.
+    ``responses`` is keyed by the surfaced keys the client received for ``leg``.
+    Unknown or already-satisfied keys are ignored (SEP-2663). Each recognized
+    answer is validated into the result type its request maps to and re-keyed to
+    the tool's own request key. Returns ``None`` when nothing matched, so the
+    caller can treat a stale or empty update as an idempotent no-op.
     """
-    requests_key = _requests_key(docket, task_scope, task_id)
+    outstanding = await read_outstanding_inputs(docket, task_scope, task_id, leg)
+    if not outstanding:
+        return None
+    mapping = await _read_outstanding_map(docket, task_scope, task_id, leg)
+
+    translated: dict[str, mcp_types.Result] = {}
+    for surfaced_key, raw in responses.items():
+        payload = outstanding.get(surfaced_key)
+        if payload is None:
+            continue
+        tool_key = mapping.get(surfaced_key)
+        if tool_key is None:
+            continue
+        method = payload.get("method", "elicitation/create")
+        result_type = result_type_for_method(method)
+        translated[tool_key] = result_type.model_validate(raw)
+
+    return translated or None
+
+
+async def store_input_responses(
+    docket: Docket,
+    task_scope: str | None,
+    task_id: str,
+    translated: dict[str, mcp_types.Result],
+    ttl_seconds: int = INPUT_TTL_SECONDS,
+) -> None:
+    """Store translated responses for the next leg to read via ``ctx``.
+
+    The responses are stored typed-but-serialized (``{"type", "data"}``) so the
+    next leg's context factory reconstructs real result objects keyed by the
+    tool's own request keys.
+    """
+    stored = {
+        tool_key: {
+            "type": type(result).__name__,
+            "data": result.model_dump(by_alias=True, mode="json"),
+        }
+        for tool_key, result in translated.items()
+    }
     async with docket.redis() as redis:
-        for input_key, response in responses.items():
-            outstanding = await redis.hget(requests_key, input_key)
-            if outstanding is None:
-                continue
-            response_key = _response_key(docket, task_scope, task_id, input_key)
-            await redis.rpush(response_key, json.dumps(response))
-            await redis.expire(response_key, INPUT_TTL_SECONDS)
-            await redis.hdel(requests_key, input_key)
+        await redis.set(
+            _input_responses_key(docket, task_scope, task_id),
+            json.dumps(stored),
+            ex=ttl_seconds,
+        )
+
+
+async def clear_outstanding(
+    docket: Docket, task_scope: str | None, task_id: str, leg: int
+) -> None:
+    """Drop a leg's outstanding requests and mapping once it has been answered.
+
+    The answered surfaced keys are never reused (a later leg mints its own), so
+    a duplicate ``tasks/update`` naming them finds nothing and is a no-op.
+    """
+    async with docket.redis() as redis:
+        await redis.delete(_requests_key(docket, task_scope, task_id, leg))
+        await redis.delete(_map_key(docket, task_scope, task_id, leg))
+
+
+async def load_pending_input(
+    docket: Docket, task_scope: str | None, task_id: str
+) -> tuple[str | None, mcp_types.InputResponses | None]:
+    """Load the per-leg state a re-entered leg reads via ``ctx``.
+
+    Returns ``(request_state, input_responses)``: the opaque state carried
+    forward and the typed answers keyed by the tool's own request keys. Both are
+    ``None`` on the first leg (nothing has been asked yet).
+    """
+    async with docket.redis() as redis:
+        state_raw = _decode(
+            await redis.get(_request_state_key(docket, task_scope, task_id))
+        )
+        responses_raw = _decode(
+            await redis.get(_input_responses_key(docket, task_scope, task_id))
+        )
+
+    responses: dict[str, mcp_types.Result] | None = None
+    if responses_raw:
+        parsed = json.loads(responses_raw)
+        if isinstance(parsed, dict):
+            responses = {}
+            for tool_key, entry in parsed.items():
+                if not isinstance(entry, dict):
+                    continue
+                result_type = _RESULT_TYPE_BY_NAME.get(entry.get("type", ""))
+                if result_type is None:
+                    continue
+                responses[tool_key] = result_type.model_validate(entry.get("data"))
+
+    # The reconstructed values are the concrete result types the tool asked for;
+    # `InputResponses` is that union keyed by request key. The `Result` element
+    # type erases that for the checker, so narrow at the return.
+    if responses is None:
+        return state_raw, None
+    return state_raw, cast("mcp_types.InputResponses", responses)
diff --git a/fastmcp_tasks/fastmcp_tasks/keys.py b/fastmcp_tasks/fastmcp_tasks/keys.py
index 10af6a6f9..e0bc4ae56 100644
--- a/fastmcp_tasks/fastmcp_tasks/keys.py
+++ b/fastmcp_tasks/fastmcp_tasks/keys.py
@@ -37,6 +37,44 @@ _AUTH_TAG = "auth"
 _ANON_TAG = "anon"
 _VALID_TAGS = (_AUTH_TAG, _ANON_TAG)
 
+# Delimiter separating the stable base task key from a per-leg suffix. A single
+# background task runs as a sequence of Docket executions (legs): the first leg
+# uses the base key, and each re-entry (after the client answers input) enqueues
+# a fresh execution under `{base}{_LEG_DELIMITER}{n}`. The base key encodes every
+# segment with `quote(safe="")`, which percent-encodes `#` to `%23`, so a literal
+# `#` never appears inside the base key and is an unambiguous leg boundary. All
+# task-identity parsing strips the leg suffix, so the scope/task-id/component a
+# leg resolves to are identical across every leg of the same task.
+_LEG_DELIMITER = "#"
+
+
+def leg_execution_key(base_task_key: str, leg: int) -> str:
+    """Build the Docket execution key for a given leg of a task.
+
+    Leg 1 uses the bare base key (so existing single-leg behavior is unchanged);
+    later legs append `#leg{n}` so each re-entry is a distinct Docket execution
+    while still parsing back to the same task scope, id, and component.
+    """
+    if leg <= 1:
+        return base_task_key
+    return f"{base_task_key}{_LEG_DELIMITER}leg{leg}"
+
+
+def base_task_key(execution_key: str) -> str:
+    """Strip any per-leg suffix, returning the stable base task key."""
+    return execution_key.split(_LEG_DELIMITER, 1)[0]
+
+
+def leg_number_from_key(execution_key: str) -> int:
+    """Return the leg number a Docket execution key encodes (leg 1 = base key)."""
+    _base, sep, suffix = execution_key.partition(_LEG_DELIMITER)
+    if not sep:
+        return 1
+    try:
+        return int(suffix.removeprefix("leg"))
+    except ValueError:
+        return 1
+
 
 def build_task_key(
     task_scope: str | None,
@@ -97,6 +135,9 @@ def parse_task_key(task_key: str) -> TaskKeyParts:
         >>> parse_task_key("anon:task456:tool:my_tool")
         `{'task_scope': None, 'client_task_id': 'task456', 'task_type': 'tool', 'component_identifier': 'my_tool'}`
     """
+    # A per-leg execution key (`{base}#leg{n}`) parses to the same identity as
+    # its base: every leg of a task shares one scope, id, and component.
+    task_key = base_task_key(task_key)
     tag, _, rest = task_key.partition(":")
     if tag not in _VALID_TAGS or not rest:
         raise ValueError(
diff --git a/fastmcp_tasks/fastmcp_tasks/lifespan.py b/fastmcp_tasks/fastmcp_tasks/lifespan.py
index 925bd6325..a8738bd81 100644
--- a/fastmcp_tasks/fastmcp_tasks/lifespan.py
+++ b/fastmcp_tasks/fastmcp_tasks/lifespan.py
@@ -94,6 +94,9 @@ async def docket_lifespan(
                     try:
                         yield
                     finally:
+                        # End-and-reenter never parks a worker on input, so a
+                        # task waiting for input holds no worker slot: cancelling
+                        # run_forever drains promptly regardless of task state.
                         worker_task.cancel()
                         with suppress(asyncio.CancelledError):
                             await worker_task
diff --git a/tests/server/test_mrtr_guards.py b/tests/server/test_mrtr_guards.py
index f5efa8309..8355bdebe 100644
--- a/tests/server/test_mrtr_guards.py
+++ b/tests/server/test_mrtr_guards.py
@@ -1170,7 +1170,9 @@ class TestTaskExecution:
         if hasattr(Docket, "_memory_server"):
             delattr(Docket, "_memory_server")
 
-    async def test_guard_result_from_task_is_rejected(self, reset_docket_memory_server):
+    async def test_guard_result_from_task_parks_for_input(
+        self, reset_docket_memory_server
+    ):
         mcp = FastMCP("guard-task")
         mcp.add_extension(TasksExtension())
 
@@ -1182,14 +1184,20 @@ class TestTaskExecution:
                 request_state=None,
             )
 
-        # A guard's `InputRequiredResult` only makes sense against a live
-        # request. Submitting `book_flight` as a background task and then
-        # reading it back must reject the guard result: `tasks/get` raises when
-        # it tries to inline the completed task's InputRequiredResult.
+        # A function-tool guard is driven as a task by the in-task reentrant
+        # loop: submitting `book_flight` parks its input request on the poll
+        # surface (`input_required`), where a client answers it via
+        # `tasks/update`. The full round-trip lives in
+        # tests/tasks/server/test_guard_reentrant.py.
         async with running_task_server(mcp):
             created = await submit_task(mcp, "book_flight", {})
-            with pytest.raises(MCPError, match="background task"):
-                await wait_for_task(mcp, created.task_id)
+            parked = await wait_for_task(
+                mcp,
+                created.task_id,
+                target_states=frozenset({"input_required"}),
+            )
+            assert parked.status == "input_required"
+            assert parked.input_requests
 
 
 class TestHttpTransport:
diff --git a/tests/tasks/server/test_context_background_task.py b/tests/tasks/server/test_context_background_task.py
index 0bf470f6b..a4ab4ed61 100644
--- a/tests/tasks/server/test_context_background_task.py
+++ b/tests/tasks/server/test_context_background_task.py
@@ -31,21 +31,16 @@ from mcp_types import (
     Implementation,
     InitializeRequestParams,
 )
-from pydantic import BaseModel
 
 from fastmcp import FastMCP
+from fastmcp.exceptions import ToolError
 from fastmcp.server.auth import AccessToken
 from fastmcp.server.context import Context
 from fastmcp.server.dependencies import get_access_token
-from fastmcp.server.elicitation import (
-    AcceptedElicitation,
-    DeclinedElicitation,
-)
 from fastmcp_tasks import TasksExtension
 from tests.tasks.task_helpers import (
     running_task_server,
     submit_task,
-    update_task,
     wait_for_task,
 )
 
@@ -295,11 +290,17 @@ class TestContextClientExtensionBackgroundTask:
 
 
 class TestContextElicitBackgroundTask:
-    """Tests for Context.elicit() in background task mode."""
+    """Tests for Context.elicit() in background task mode.
 
-    async def test_elicit_raises_when_no_task_engine(self):
-        """elicit() fails fast when in a background task but no tasks extension
-        is installed to answer the request."""
+    Imperative elicitation is not supported inside a background task: the worker
+    never blocks on a client round-trip. A task gathers input with the guard
+    pattern (return an ``InputRequiredResult``), so ``ctx.elicit()`` in a task
+    fails fast with guidance rather than parking a worker.
+    """
+
+    async def test_elicit_raises_with_guard_guidance(self):
+        """elicit() inside a background task raises a ToolError pointing to the
+        guard/return pattern (InputRequiredResult)."""
         mcp = FastMCP("test")
         ctx = Context(mcp, task_id="test-task-123")
 
@@ -308,7 +309,7 @@ class TestContextElicitBackgroundTask:
 
         ctx._session = cast(ServerSession, MockSession())
 
-        with pytest.raises(RuntimeError, match="tasks extension"):
+        with pytest.raises(ToolError, match="InputRequiredResult"):
             await ctx.elicit("Need input", str)
 
 
@@ -391,99 +392,24 @@ class TestBackgroundTaskIntegration:
             "session_unavailable": True,
         }
 
-    async def test_elicit_accept_flow(self):
-        """E2E: tool elicits input, client accepts via tasks/update (poll)."""
-        mcp = FastMCP("elicit-accept-test")
+    async def test_imperative_elicit_fails_with_guard_guidance(self):
+        """A task=True tool that calls ctx.elicit() fails with the guard-pattern
+        error rather than parking a worker on a client round-trip."""
+        mcp = FastMCP("elicit-forbidden")
         mcp.add_extension(TasksExtension())
 
         @mcp.tool(task=True)
         async def ask_name(ctx: Context) -> str:
             result = await ctx.elicit("What is your name?", str)
-            if isinstance(result, AcceptedElicitation):
-                return f"Hello, {result.data}!"
-            return "No name provided"
+            return str(result)
 
         async with running_task_server(mcp):
             created = await submit_task(mcp, "ask_name", {})
-            parked = await wait_for_task(
-                mcp, created.task_id, target_states=frozenset({"input_required"})
-            )
-            assert parked.input_requests is not None
-            key = next(iter(parked.input_requests))
-            await update_task(
-                mcp,
-                created.task_id,
-                {key: {"action": "accept", "content": {"value": "Bob"}}},
-            )
             final = await wait_for_task(mcp, created.task_id)
 
-        assert final.status == "completed"
-        assert final.result is not None
-        assert final.result["structuredContent"] == {"result": "Hello, Bob!"}
-
-    async def test_elicit_decline_flow(self):
-        """E2E: tool elicits input, client declines via tasks/update (poll)."""
-        mcp = FastMCP("elicit-decline-test")
-        mcp.add_extension(TasksExtension())
-
-        @mcp.tool(task=True)
-        async def optional_input(ctx: Context) -> str:
-            result = await ctx.elicit("Want to provide a name?", str)
-            if isinstance(result, DeclinedElicitation):
-                return "User declined"
-            if isinstance(result, AcceptedElicitation):
-                return f"Got: {result.data}"
-            return "Cancelled"
-
-        async with running_task_server(mcp):
-            created = await submit_task(mcp, "optional_input", {})
-            parked = await wait_for_task(
-                mcp, created.task_id, target_states=frozenset({"input_required"})
-            )
-            assert parked.input_requests is not None
-            key = next(iter(parked.input_requests))
-            await update_task(mcp, created.task_id, {key: {"action": "decline"}})
-            final = await wait_for_task(mcp, created.task_id)
-
-        assert final.status == "completed"
-        assert final.result is not None
-        assert final.result["structuredContent"] == {"result": "User declined"}
-
-    async def test_elicit_with_pydantic_model(self):
-        """E2E: tool elicits structured Pydantic input via tasks/update (poll)."""
-
-        class UserInfo(BaseModel):
-            name: str
-            age: int
-
-        mcp = FastMCP("elicit-pydantic-test")
-        mcp.add_extension(TasksExtension())
-
-        @mcp.tool(task=True)
-        async def get_user_info(ctx: Context) -> str:
-            result = await ctx.elicit("Provide user info", UserInfo)
-            if isinstance(result, AcceptedElicitation):
-                assert isinstance(result.data, UserInfo)
-                return f"{result.data.name} is {result.data.age}"
-            return "No info"
-
-        async with running_task_server(mcp):
-            created = await submit_task(mcp, "get_user_info", {})
-            parked = await wait_for_task(
-                mcp, created.task_id, target_states=frozenset({"input_required"})
-            )
-            assert parked.input_requests is not None
-            key = next(iter(parked.input_requests))
-            await update_task(
-                mcp,
-                created.task_id,
-                {key: {"action": "accept", "content": {"name": "Alice", "age": 30}}},
-            )
-            final = await wait_for_task(mcp, created.task_id)
-
-        assert final.status == "completed"
-        assert final.result is not None
-        assert final.result["structuredContent"] == {"result": "Alice is 30"}
+        assert final.status == "failed"
+        assert final.error is not None
+        assert "InputRequiredResult" in final.error["message"]
 
 
 class TestAccessTokenInBackgroundTasks:
diff --git a/tests/tasks/server/test_extension.py b/tests/tasks/server/test_extension.py
index 21fd2b7fd..9d9d49029 100644
--- a/tests/tasks/server/test_extension.py
+++ b/tests/tasks/server/test_extension.py
@@ -26,7 +26,6 @@ from mcp.shared.exceptions import MCPError
 from fastmcp import FastMCP
 from fastmcp.client import Client
 from fastmcp.exceptions import ToolError
-from fastmcp.server import context as core_context
 from fastmcp.server.dependencies import bind_request_context
 from fastmcp.tools.base import ToolResult
 from fastmcp.utilities.tasks import TASKS_EXTENSION_ID, TaskConfig
@@ -356,6 +355,8 @@ async def test_worker_hooks_survive_sibling_server_shutdown():
     each running a TasksExtension refcount them, so the hooks clear only when
     the last extension lifespan exits.
     """
+    from fastmcp.server import dependencies as core_dependencies
+
     server_a = _tasks_server()
     server_b = _tasks_server()
 
@@ -363,8 +364,8 @@ async def test_worker_hooks_survive_sibling_server_shutdown():
         await stack_b.enter_async_context(server_b._lifespan_manager())
         async with AsyncExitStack() as stack_a:
             await stack_a.enter_async_context(server_a._lifespan_manager())
-            assert core_context._task_elicitation_handler is not None
+            assert core_dependencies._background_context_factory is not None
         # Server A has shut down; server B's workers still need the hooks.
-        assert core_context._task_elicitation_handler is not None
+        assert core_dependencies._background_context_factory is not None
     # The last extension exited; hooks are cleared.
-    assert core_context._task_elicitation_handler is None
+    assert core_dependencies._background_context_factory is None
diff --git a/tests/tasks/server/test_guard_reentrant.py b/tests/tasks/server/test_guard_reentrant.py
new file mode 100644
index 000000000..99e53c481
--- /dev/null
+++ b/tests/tasks/server/test_guard_reentrant.py
@@ -0,0 +1,174 @@
+"""The guard-pattern reentrant loop driven inside a background task.
+
+A `task=True` tool that *returns* an `InputRequiredResult` (rather than awaiting
+`ctx.elicit()`) is the same guard authoring model FastMCP uses foreground. As a
+task, the worker drives the multi-round-trip itself: it parks the request on the
+poll surface, the client answers via `tasks/update`, and the tool is re-invoked
+with the answer on `ctx.input_responses` — identical to the foreground contract,
+only the transport differs. These tests exercise that loop end-to-end through
+the real interceptor and handlers via `task_helpers`.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import mcp_types
+
+from fastmcp import Context, FastMCP
+from fastmcp_tasks import TasksExtension
+from tests.tasks.task_helpers import (
+    running_task_server,
+    submit_task,
+    update_task,
+    wait_for_task,
+)
+
+
+def _elicit_request(message: str) -> mcp_types.ElicitRequest:
+    return mcp_types.ElicitRequest(
+        params=mcp_types.ElicitRequestFormParams(
+            message=message,
+            requested_schema={
+                "type": "object",
+                "properties": {"value": {"type": "string"}},
+            },
+        )
+    )
+
+
+def _answer(responses: mcp_types.InputResponses, key: str) -> str:
+    """Read the string value a client accepted for `key` (test helper)."""
+    result = responses[key]
+    assert isinstance(result, mcp_types.ElicitResult)
+    assert result.content is not None
+    return str(result.content["value"])
+
+
+def _input_required(
+    requests: dict[str, mcp_types.ElicitRequest],
+    request_state: str | None = None,
+) -> mcp_types.InputRequiredResult:
+    return mcp_types.InputRequiredResult(
+        result_type="input_required",
+        input_requests=requests,
+        request_state=request_state,
+    )
+
+
+async def _park_key(mcp: FastMCP, task_id: str) -> str:
+    parked = await wait_for_task(
+        mcp, task_id, target_states=frozenset({"input_required"})
+    )
+    assert parked.status == "input_required"
+    assert parked.input_requests is not None
+    return next(iter(parked.input_requests))
+
+
+async def test_guard_return_single_round_completes():
+    """A tool that returns InputRequiredResult once is driven to completion."""
+    mcp = FastMCP("guard")
+    mcp.add_extension(TasksExtension())
+
+    @mcp.tool(task=True)
+    async def greet(ctx: Context) -> str | mcp_types.InputRequiredResult:
+        responses = ctx.input_responses
+        if responses is None:
+            return _input_required({"name": _elicit_request("Your name?")})
+        return f"Hello, {_answer(responses, 'name')}!"
+
+    async with running_task_server(mcp):
+        created = await submit_task(mcp, "greet", {})
+        key = await _park_key(mcp, created.task_id)
+        await update_task(
+            mcp,
+            created.task_id,
+            {key: {"action": "accept", "content": {"value": "Ada"}}},
+        )
+        final = await wait_for_task(mcp, created.task_id)
+
+    assert final.status == "completed"
+    assert final.result is not None
+    assert final.result["structuredContent"] == {"result": "Hello, Ada!"}
+
+
+async def test_guard_return_multiple_rounds_use_distinct_keys():
+    """A tool that asks twice surfaces distinct keys across rounds (SEP-2663 L350).
+
+    The second round's key must differ from the first's — a client that
+    deduplicates by key must not suppress the second ask. Cross-round state
+    travels through `request_state` (each leg's `input_responses` holds only
+    that leg's answers, matching the foreground guard contract).
+    """
+    mcp = FastMCP("guard")
+    mcp.add_extension(TasksExtension())
+
+    @mcp.tool(task=True)
+    async def full_name(ctx: Context) -> str | mcp_types.InputRequiredResult:
+        responses = ctx.input_responses
+        if responses is None:
+            # Round 1: ask for the first name.
+            return _input_required({"first": _elicit_request("First name?")})
+        if ctx.request_state is None:
+            # Round 2: carry the first name forward in request_state, ask last.
+            return _input_required(
+                {"last": _elicit_request("Last name?")},
+                request_state=_answer(responses, "first"),
+            )
+        # Round 3: request_state holds the first name; responses holds the last.
+        return f"{ctx.request_state} {_answer(responses, 'last')}"
+
+    async with running_task_server(mcp):
+        created = await submit_task(mcp, "full_name", {})
+        key1 = await _park_key(mcp, created.task_id)
+        await update_task(
+            mcp,
+            created.task_id,
+            {key1: {"action": "accept", "content": {"value": "Ada"}}},
+        )
+        key2 = await _park_key(mcp, created.task_id)
+        assert key2 != key1
+        await update_task(
+            mcp,
+            created.task_id,
+            {key2: {"action": "accept", "content": {"value": "Lovelace"}}},
+        )
+        final = await wait_for_task(mcp, created.task_id)
+
+    assert final.status == "completed"
+    assert final.result is not None
+    assert final.result["structuredContent"] == {"result": "Ada Lovelace"}
+
+
+async def test_non_guard_tool_runs_once():
+    """A tool that never asks for input completes in a single invocation."""
+    calls: list[int] = []
+    mcp = FastMCP("guard")
+    mcp.add_extension(TasksExtension())
+
+    @mcp.tool(task=True)
+    async def square(n: int) -> int:
+        calls.append(n)
+        return n * n
+
+    async with running_task_server(mcp):
+        created = await submit_task(mcp, "square", {"n": 6})
+        final = await wait_for_task(mcp, created.task_id)
+
+    assert final.status == "completed"
+    assert final.result is not None
+    assert final.result["structuredContent"] == {"result": 36}
+    assert calls == [6]
+
+
+def test_reentrant_wrapper_preserves_signature():
+    """The wrapper keeps the tool's parameters so Docket DI is unchanged."""
+    import inspect
+
+    from fastmcp_tasks.input_loop import reentrant_task_fn
+
+    async def fn(n: int, ctx: Any) -> int:
+        return n
+
+    wrapped = reentrant_task_fn(fn)
+    assert list(inspect.signature(wrapped).parameters) == ["n", "ctx"]
diff --git a/tests/tasks/server/test_reenter_shutdown.py b/tests/tasks/server/test_reenter_shutdown.py
new file mode 100644
index 000000000..54544839e
--- /dev/null
+++ b/tests/tasks/server/test_reenter_shutdown.py
@@ -0,0 +1,65 @@
+"""Shutdown regression for end-and-reenter task input.
+
+The whole point of end-and-reenter is that a task waiting on client input holds
+no worker: the guard leg's Docket execution completed and the worker is free.
+This test proves it — a task parked in ``input_required`` that is never answered
+must not delay server shutdown. Under the old block-and-resume model the worker
+sat on a Redis wait for the input TTL and wedged teardown; here the lifespan
+exits promptly.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+import mcp_types
+
+from fastmcp import Context, FastMCP
+from fastmcp_tasks import TasksExtension
+from tests.tasks.task_helpers import running_task_server, submit_task, wait_for_task
+
+
+def _elicit_request(message: str) -> mcp_types.ElicitRequest:
+    return mcp_types.ElicitRequest(
+        params=mcp_types.ElicitRequestFormParams(
+            message=message,
+            requested_schema={
+                "type": "object",
+                "properties": {"value": {"type": "string"}},
+            },
+        )
+    )
+
+
+async def test_parked_task_does_not_delay_shutdown():
+    """Exiting the lifespan with a task in input_required (never answered) must
+    return promptly — no worker is parked awaiting input."""
+    mcp = FastMCP("parked-shutdown")
+    mcp.add_extension(TasksExtension())
+
+    @mcp.tool(task=True)
+    async def greet(ctx: Context) -> str | mcp_types.InputRequiredResult:
+        if ctx.input_responses is None:
+            return mcp_types.InputRequiredResult(
+                result_type="input_required",
+                input_requests={"name": _elicit_request("Your name?")},
+                request_state=None,
+            )
+        return "done"
+
+    loop = asyncio.get_event_loop()
+    manager = running_task_server(mcp)
+    await manager.__aenter__()
+    try:
+        created = await submit_task(mcp, "greet", {})
+        parked = await wait_for_task(
+            mcp, created.task_id, target_states=frozenset({"input_required"})
+        )
+        assert parked.status == "input_required"
+    finally:
+        # Never answer; time how long teardown takes.
+        started = loop.time()
+        await manager.__aexit__(None, None, None)
+        elapsed = loop.time() - started
+
+    assert elapsed < 3.0, f"lifespan took {elapsed:.2f}s to exit with a parked task"
diff --git a/tests/tasks/server/test_task_elicitation_relay.py b/tests/tasks/server/test_task_elicitation_relay.py
deleted file mode 100644
index ab1096bbd..000000000
--- a/tests/tasks/server/test_task_elicitation_relay.py
+++ /dev/null
@@ -1,219 +0,0 @@
-"""In-task elicitation under SEP-2663 (poll-based input).
-
-A background worker that calls ``ctx.elicit()`` has no live request, so SEP-2663
-parks the request and the task's ``tasks/get`` status flips to ``input_required``
-with the outstanding ``inputRequests``. The caller answers with ``tasks/update``
-and the parked worker resumes. This replaces the SEP-1686 push relay (which sent
-``elicitation/create`` over a back-channel); the accept/decline/cancel semantics,
-structured round-trips, and sequential elicitations are preserved, driven here
-in-process because there is no client task API until Phase 4.
-"""
-
-from __future__ import annotations
-
-import asyncio
-from dataclasses import dataclass
-from typing import Any
-
-import fastmcp_tasks.input_store as input_store
-from pydantic import BaseModel
-
-from fastmcp import FastMCP
-from fastmcp.server.context import Context
-from fastmcp.server.elicitation import (
-    AcceptedElicitation,
-    CancelledElicitation,
-    DeclinedElicitation,
-)
-from fastmcp_tasks import TasksExtension
-from tests.tasks.task_helpers import (
-    get_task,
-    running_task_server,
-    submit_task,
-    update_task,
-    wait_for_task,
-)
-
-
-async def _wait_for_input_required(server: FastMCP, task_id: str, timeout: float = 5.0):
-    """Poll until the task is waiting on input, returning the GetTaskResult."""
-    deadline = asyncio.get_event_loop().time() + timeout
-    while True:
-        got = await get_task(server, task_id)
-        if got.status == "input_required":
-            return got
-        if got.status in ("completed", "failed", "cancelled"):
-            raise AssertionError(
-                f"Task {task_id} reached {got.status!r} before requesting input"
-            )
-        if asyncio.get_event_loop().time() >= deadline:
-            raise TimeoutError(f"Task {task_id} never requested input")
-        await asyncio.sleep(0.02)
-
-
-async def _drive(server: FastMCP, name: str, answers: list[dict[str, Any]]) -> str:
-    """Submit a task, answer each elicitation in turn, return its result text."""
-    created = await submit_task(server, name, {})
-    for answer in answers:
-        got = await _wait_for_input_required(server, created.task_id)
-        key = next(iter(got.input_requests))
-        request = got.input_requests[key]
-        assert request["method"] == "elicitation/create"
-        await update_task(server, created.task_id, {key: answer})
-    final = await wait_for_task(server, created.task_id)
-    assert final.status == "completed", final.error
-    assert final.result is not None
-    return final.result["content"][0]["text"]
-
-
-async def test_accept_answers_the_elicitation():
-    mcp = FastMCP("relay-accept")
-    mcp.add_extension(TasksExtension())
-
-    @mcp.tool(task=True)
-    async def ask_name(ctx: Context) -> str:
-        result = await ctx.elicit("What is your name?", str)
-        if isinstance(result, AcceptedElicitation):
-            return f"Hello, {result.data}!"
-        return "No name"
-
-    async with running_task_server(mcp):
-        text = await _drive(
-            mcp, "ask_name", [{"action": "accept", "content": {"value": "Alice"}}]
-        )
-        assert text == "Hello, Alice!"
-
-
-async def test_decline_yields_declined_elicitation():
-    mcp = FastMCP("relay-decline")
-    mcp.add_extension(TasksExtension())
-
-    @mcp.tool(task=True)
-    async def optional_input(ctx: Context) -> str:
-        result = await ctx.elicit("Provide a name?", str)
-        if isinstance(result, DeclinedElicitation):
-            return "User declined"
-        return "Other"
-
-    async with running_task_server(mcp):
-        text = await _drive(mcp, "optional_input", [{"action": "decline"}])
-        assert text == "User declined"
-
-
-async def test_cancel_yields_cancelled_elicitation():
-    mcp = FastMCP("relay-cancel")
-    mcp.add_extension(TasksExtension())
-
-    @mcp.tool(task=True)
-    async def cancellable(ctx: Context) -> str:
-        result = await ctx.elicit("Input?", str)
-        if isinstance(result, CancelledElicitation):
-            return "Cancelled"
-        return "Not cancelled"
-
-    async with running_task_server(mcp):
-        text = await _drive(mcp, "cancellable", [{"action": "cancel"}])
-        assert text == "Cancelled"
-
-
-async def test_dataclass_round_trips():
-    mcp = FastMCP("relay-dataclass")
-    mcp.add_extension(TasksExtension())
-
-    @dataclass
-    class UserInfo:
-        name: str
-        age: int
-
-    @mcp.tool(task=True)
-    async def get_user(ctx: Context) -> str:
-        result = await ctx.elicit("Provide user info", UserInfo)
-        if isinstance(result, AcceptedElicitation):
-            assert isinstance(result.data, UserInfo)
-            return f"{result.data.name} is {result.data.age}"
-        return "No info"
-
-    async with running_task_server(mcp):
-        text = await _drive(
-            mcp,
-            "get_user",
-            [{"action": "accept", "content": {"name": "Bob", "age": 30}}],
-        )
-        assert text == "Bob is 30"
-
-
-async def test_pydantic_model_round_trips():
-    mcp = FastMCP("relay-pydantic")
-    mcp.add_extension(TasksExtension())
-
-    class Config(BaseModel):
-        host: str
-        port: int
-
-    @mcp.tool(task=True)
-    async def get_config(ctx: Context) -> str:
-        result = await ctx.elicit("Server config?", Config)
-        if isinstance(result, AcceptedElicitation):
-            assert isinstance(result.data, Config)
-            return f"{result.data.host}:{result.data.port}"
-        return "No config"
-
-    async with running_task_server(mcp):
-        text = await _drive(
-            mcp,
-            "get_config",
-            [{"action": "accept", "content": {"host": "localhost", "port": 8080}}],
-        )
-        assert text == "localhost:8080"
-
-
-async def test_multiple_sequential_elicitations():
-    mcp = FastMCP("relay-multi")
-    mcp.add_extension(TasksExtension())
-
-    @mcp.tool(task=True)
-    async def two_questions(ctx: Context) -> str:
-        r1 = await ctx.elicit("First name?", str)
-        r2 = await ctx.elicit("Last name?", str)
-        if isinstance(r1, AcceptedElicitation) and isinstance(r2, AcceptedElicitation):
-            return f"{r1.data} {r2.data}"
-        return "Incomplete"
-
-    async with running_task_server(mcp):
-        text = await _drive(
-            mcp,
-            "two_questions",
-            [
-                {"action": "accept", "content": {"value": "Jane"}},
-                {"action": "accept", "content": {"value": "Doe"}},
-            ],
-        )
-        assert text == "Jane Doe"
-
-
-async def test_unanswered_input_times_out_to_cancel(monkeypatch):
-    """A worker that is never answered eventually resumes with a cancel.
-
-    The poll model has no "no handler" fast path; instead the parked worker's
-    blocking wait is bounded by ``INPUT_TTL_SECONDS``. Patched short here so the
-    timeout-to-cancel behaviour is testable.
-    """
-    monkeypatch.setattr(input_store, "INPUT_TTL_SECONDS", 1)
-
-    mcp = FastMCP("relay-timeout")
-    mcp.add_extension(TasksExtension())
-
-    @mcp.tool(task=True)
-    async def needs_input(ctx: Context) -> str:
-        result = await ctx.elicit("Input?", str)
-        if isinstance(result, CancelledElicitation):
-            return "Cancelled as expected"
-        return "Other"
-
-    async with running_task_server(mcp):
-        created = await submit_task(mcp, "needs_input", {})
-        # Never answer; the worker's bounded wait resolves to cancel.
-        final = await wait_for_task(mcp, created.task_id, timeout=10.0)
-        assert final.status == "completed"
-        assert final.result is not None
-        assert final.result["content"][0]["text"] == "Cancelled as expected"
diff --git a/tests/tasks/server/test_wire_production.py b/tests/tasks/server/test_wire_production.py
index ec7ef7418..14c6381c8 100644
--- a/tests/tasks/server/test_wire_production.py
+++ b/tests/tasks/server/test_wire_production.py
@@ -12,6 +12,7 @@ from __future__ import annotations
 
 import mcp_types.methods as methods
 import pytest
+
 from fastmcp_tasks import wire_production
 
 _MODERN = "2026-07-28"