From c72b5dd7f0cdeef9cdf3f4bb1c3eca64deb0e973 Mon Sep 17 00:00:00 2001
From: Chris Guidry <chris.g@prefect.io>
Date: Fri, 9 Jan 2026 13:37:09 -0500
Subject: [PATCH] Add detailed logging around docket.add() and subscription
 tasks

Tracing where the Redis ACL error occurs - the initial Redis writes
succeed but error happens somewhere after.

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
---
 src/fastmcp/server/tasks/handlers.py      | 30 ++++++++++++++++++++---
 src/fastmcp/server/tasks/subscriptions.py | 17 ++++++++++++-
 2 files changed, 42 insertions(+), 5 deletions(-)

diff --git a/src/fastmcp/server/tasks/handlers.py b/src/fastmcp/server/tasks/handlers.py
index 00a65df27..4f850b921 100644
--- a/src/fastmcp/server/tasks/handlers.py
+++ b/src/fastmcp/server/tasks/handlers.py
@@ -139,18 +139,38 @@ async def handle_tool_as_task(
 
     # Queue function to Docket by name (result storage via execution_ttl)
     # Use tool.key which matches what was registered - prefixed for mounted tools
-    await docket.add(
-        tool.key,
-        key=task_key,
-    )(**arguments)
+    _logger.info(
+        f"[{instance_id}] About to call docket.add: tool.key={tool.key}, "
+        f"task_key={task_key}, arguments={arguments}"
+    )
+    try:
+        await docket.add(
+            tool.key,
+            key=task_key,
+        )(**arguments)
+        _logger.info(f"[{instance_id}] docket.add completed successfully")
+    except Exception as e:
+        import traceback
+
+        _logger.error(
+            f"[{instance_id}] docket.add FAILED: {type(e).__name__}: {e}\n"
+            f"Traceback:\n{traceback.format_exc()}"
+        )
+        raise
 
     # Spawn subscription task to send status notifications (SEP-1686 optional feature)
     from fastmcp.server.tasks.subscriptions import subscribe_to_task_updates
 
     # Start subscription in session's task group (persists for connection lifetime)
+    _logger.info(
+        f"[{instance_id}] Checking for subscription task group: "
+        f"hasattr={hasattr(ctx.session, '_subscription_task_group')}"
+    )
     if hasattr(ctx.session, "_subscription_task_group"):
         tg = ctx.session._subscription_task_group  # type: ignore[attr-defined]
+        _logger.info(f"[{instance_id}] Task group: {tg}")
         if tg:
+            _logger.info(f"[{instance_id}] Starting subscription task")
             tg.start_soon(  # type: ignore[union-attr]
                 subscribe_to_task_updates,
                 server_task_id,
@@ -158,7 +178,9 @@ async def handle_tool_as_task(
                 ctx.session,
                 docket,
             )
+            _logger.info(f"[{instance_id}] Subscription task started")
 
+    _logger.info(f"[{instance_id}] About to return task stub")
     # Return task stub
     # Tasks MUST begin in "working" status per SEP-1686 final spec (line 381)
     return mcp.types.CallToolResult(
diff --git a/src/fastmcp/server/tasks/subscriptions.py b/src/fastmcp/server/tasks/subscriptions.py
index 9bbf0fad1..5b4269471 100644
--- a/src/fastmcp/server/tasks/subscriptions.py
+++ b/src/fastmcp/server/tasks/subscriptions.py
@@ -42,13 +42,23 @@ async def subscribe_to_task_updates(
         session: MCP ServerSession for sending notifications
         docket: Docket instance for subscribing to execution events
     """
+    logger.info(
+        f"subscribe_to_task_updates STARTING: task_id={task_id}, task_key={task_key}"
+    )
     try:
+        logger.info(
+            f"subscribe_to_task_updates: About to call docket.get_execution({task_key})"
+        )
         execution = await docket.get_execution(task_key)
+        logger.info(
+            f"subscribe_to_task_updates: docket.get_execution returned {execution}"
+        )
         if execution is None:
             logger.warning(f"No execution found for task {task_id}")
             return
 
         # Subscribe to state and progress events from Docket
+        logger.info("subscribe_to_task_updates: About to subscribe to execution events")
         async for event in execution.subscribe():
             if event["type"] == "state":
                 # Send notifications/tasks/status when state changes
@@ -70,7 +80,12 @@ async def subscribe_to_task_updates(
                 )
 
     except Exception as e:
-        logger.warning(f"Subscription task failed for {task_id}: {e}", exc_info=True)
+        import traceback
+
+        logger.error(
+            f"subscribe_to_task_updates FAILED for {task_id}: {type(e).__name__}: {e}\n"
+            f"Traceback:\n{traceback.format_exc()}"
+        )
 
 
 async def _send_status_notification(