Background task notification leaks into next turn after receive_response()

[garlanddiego]

Description

When a background task (spawned via run_in_background=true Agent tool) completes between turns, its TaskNotificationMessage leaks into the next receive_response() call. The model then responds to the stale task notification instead of the new user prompt.

Reproduction

import asyncio
import anyio
from claude_agent_sdk import (
    ClaudeSDKClient, ClaudeAgentOptions,
    AssistantMessage, TextBlock, ResultMessage,
    TaskStartedMessage, TaskNotificationMessage,
)

async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Glob", "Grep", "Bash"],
        permission_mode="bypassPermissions",
        cwd="/tmp",
    )

    async with ClaudeSDKClient(options=options) as client:
        # Turn 1: spawn a background task, end turn immediately
        await client.query(
            "Spawn a background agent (run_in_background=true) to run: sleep 60 && echo done\n"
            "Then immediately say 'Spawned.' and end your turn. Do NOT wait."
        )
        pending = set()
        async for msg in client.receive_response():
            if isinstance(msg, TaskStartedMessage):
                pending.add(msg.task_id)

        print(f"Turn 1 done. Pending tasks: {pending}")
        if not pending:
            print("No pending tasks — model waited for task. Re-run to reproduce.")
            return

        # Wait for the background task to complete
        print("Waiting 65s for background task to complete...")
        await asyncio.sleep(65)

        # Turn 2: simple question
        await client.query("What is 2+2? Just answer with the number.")
        async for msg in client.receive_response():
            if isinstance(msg, TaskNotificationMessage):
                print(f"*** BUG: Stale TaskNotification leaked! task_id={msg.task_id} ***")
            elif isinstance(msg, AssistantMessage):
                text = "".join(b.text for b in msg.content if isinstance(b, TextBlock) and b.text)
                if text:
                    print(f"Model response: {text[:100]}")
            elif isinstance(msg, ResultMessage):
                result = getattr(msg, "result", "") or ""
                print(f"Result: {result[:100]}")

anyio.run(main)

Note: The model may sometimes choose to wait for the background task before ending Turn 1 (no pending tasks). Re-run a few times to reproduce — it depends on model behavior.

Expected behavior

receive_response() for Turn 2 should only yield messages related to Turn 2's query. Background task notifications from Turn 1 should either:

1. Be consumed by Turn 1's receive_response() (even after ResultMessage), or
2. Be filtered out / delivered through a separate channel

Actual behavior

Turn 2's receive_response() yields a stale TaskNotificationMessage from Turn 1. The model sees this notification in its conversation context, responds to it ("The background task completed successfully...") instead of answering the new query ("2+2").

Observed message flow

Turn 2 receive_response():
  msg#1 TaskNotificationMessage task_id=btjnl316x status=completed  ← STALE from Turn 1!
  msg#2 SystemMessage subtype=init
  msg#3 AssistantMessage (tool use)
  msg#4 UserMessage (tool result)
  msg#5 AssistantMessage "The background task completed successfully..."  ← Wrong answer!
  msg#6 ResultMessage

Impact

In multi-turn applications (chat bots, agents), this causes:

• Model answering questions about stale background tasks instead of the user's new message
• "Answer wrong question" behavior that confuses users
• Workarounds like client.stop_task() are unreliable ("No task found" errors)

Environment

• claude-agent-sdk 0.1.50
• Python 3.14.2
• macOS

Related issues

• Second query hangs after first query invokes background Task #558 (Second query hangs after background Task)
• Task notifications arrive after TaskOutput already returned results claude-code#21343 (Task notifications arrive after TaskOutput)

[garlanddiego]

Workaround

We found a reliable workaround: detect stale notifications → drop the contaminated turn → re-query.

Strategy

1. Track task IDs from TaskStartedMessage during each turn
2. On ResultMessage, if tasks are still pending, mark them as "stale" and try client.stop_task() (best-effort — sometimes returns "No task found")
3. On the next turn, if receive_response() yields a TaskNotificationMessage whose task_id is in the stale set, set a found_stale flag
4. While found_stale, drop all subsequent messages until ResultMessage ends the turn
5. After the contaminated turn ends, re-send the same query() with the original prompt
6. Repeat up to N times until a clean turn (no stale notifications) is received

Reproduction + fix demo

import asyncio
import anyio
from claude_agent_sdk import (
    ClaudeSDKClient, ClaudeAgentOptions,
    AssistantMessage, TextBlock, ResultMessage,
    TaskStartedMessage, TaskNotificationMessage,
)

def msg_text(msg):
    if isinstance(msg, AssistantMessage):
        return "".join(b.text for b in msg.content if isinstance(b, TextBlock) and b.text)
    return ""

async def smart_turn(client, prompt, stale_ids, max_retries=5):
    """Turn that detects stale notifications and re-queries."""
    for attempt in range(max_retries):
        await client.query(prompt)
        found_stale = False
        result = None

        async for msg in client.receive_response():
            if isinstance(msg, TaskNotificationMessage) and msg.task_id in stale_ids:
                stale_ids.discard(msg.task_id)
                found_stale = True
                continue
            if found_stale:
                continue  # Drop contaminated messages
            if isinstance(msg, AssistantMessage):
                text = msg_text(msg)
                if text:
                    result = text

        if not found_stale:
            return result  # Clean turn
        print(f"  Stale turn dropped (attempt {attempt + 1}), retrying...")
    return result

async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Glob", "Grep", "Bash"],
        permission_mode="bypassPermissions",
        cwd="/tmp",
    )

    async with ClaudeSDKClient(options=options) as client:
        # Turn 1: spawn background task
        await client.query(
            "Spawn a background agent (run_in_background=true) to run: sleep 30 && echo done\n"
            "After spawning, say 'Spawned.' and END YOUR TURN. Do NOT wait."
        )
        pending = set()
        async for msg in client.receive_response():
            if isinstance(msg, TaskStartedMessage):
                pending.add(msg.task_id)
        print(f"Pending: {pending}")

        if not pending:
            print("No pending tasks. Re-run.")
            return

        # Wait for task to complete
        await asyncio.sleep(35)

        # Turn 2: smart turn with stale detection
        result = await smart_turn(client, "What is 7*8? Answer with just the number.", pending)
        print(f"Result: {result}")
        assert result and "56" in result, f"Expected '56', got: {result}"
        print("✓ Workaround works!")

anyio.run(main)

Notes

• stop_task() is best-effort — sometimes the task has already finished internally but the notification hasn't been emitted yet, causing "No task found" errors
• Each stale notification consumes one turn (the model sees it and responds to it), so N pending tasks may require N re-queries
• The re-query approach preserves conversation context (no need to restart the client)
• This is a workaround — ideally receive_response() should not yield notifications from tasks started in previous turns