Hacktoberfest 2026: die Issues, die Maintainer für den Oktober markiert haben – offen und einsteigerfreundlich. Hacktoberfest-Issues durchsuchen

[Bug]: Cluster mode: SubscribeToTask on a replica holding a paused ActiveTask returns a stale INPUT_REQUIRED snapshot and closes

Offen
  • Stimmt etwas nicht?
#1,323 0 Kommentare 0 Reaktionen 2 zugewiesene Personen Auf GitHub ansehen

Maintainer antworten meist innerhalb von 2 Tagen

@rohityan arbeitet bereits daran.

Seit 07.10.2026.

Bewertung

Dieses Issue wurde noch nicht bewertet.

Beschreibung

What happened?

Version: a2a-sdk 1.2.2 (code unchanged on main), cluster mode (VersionedDatabaseTaskStore + DatabaseTaskEventStream).

In cluster mode, SubscribeToTask on a replica that ran an earlier turn of a multi-turn task returns a stale snapshot and closes the stream right away, while the current turn is still running on another replica. A replica that never saw the task streams the same turn correctly.

Steps

  1. Replica A runs turn 1. The agent pauses with TASK_STATE_INPUT_REQUIRED. A keeps its ActiveTask in ActiveTaskRegistry, because only terminal states trigger cleanup.
  2. The client's reply lands on replica B, which runs turn 2 (WORKING … COMPLETED).
  3. While turn 2 is running, the client resubscribes. The load balancer sends it to replica A.

Expected: a snapshot of the current state (WORKING), then turn 2's events from the shared event stream, ending at COMPLETED. This is what replica C returns.

Actual: replica A returns Task(TASK_STATE_INPUT_REQUIRED), the state from turn 1, and closes the stream. The client concludes the agent is waiting for input, even though the task is running and later completes.

resubscribe on C (never saw the task): Task(WORKING) -> StatusUpdate(WORKING) x3 -> StatusUpdate(COMPLETED) -> <stream closed>
resubscribe on A (ran turn 1):         Task(INPUT_REQUIRED) -> <stream closed>
GetTask afterwards: TASK_STATE_COMPLETED

This is consistent: 5 out of 5 runs.

Cause

on_subscribe_to_task (default_request_handler_v2.py, around line 575) takes the local fast path whenever the replica's registry has an entry for the task:

# Shared-stream mode. Fast path: this replica runs the agent -> tap it.
local = await self._active_task_registry.get(task_id)
if local is not None:
    async for event in local.subscribe(include_initial_task=True):
        yield event
    return

Once a turn has ended in an interrupted state, having a registry entry no longer means "this replica runs the agent". The paused ActiveTask on A is idle:

  • local.subscribe(include_initial_task=True) returns its TaskManager's cached snapshot, which still says INPUT_REQUIRED. Unlike the producer loop, this path doesn't call invalidate() first.
  • A then follows its own in-memory queue, which never receives turn 2's events because they are produced on B.

The handler already read the current snapshot and version from _versioned_store a few lines earlier, so the remote path (_subscribe_remote) would return the right answer.

Possible fixes

  • In shared-stream mode, take the local fast path only while the local ActiveTask has a request in flight. Otherwise use _subscribe_remote(task_id, task, snapshot_version, stream).
  • Or evict an ActiveTask from the registry when its turn ends in an interrupted state, when cluster mode is on. That would also stop paused tasks from staying in memory indefinitely on every replica that ever ran a turn for them.

Related: #1188 and #1281. This is the same root cause (a paused ActiveTask outliving its turn), but on the subscribe path, which #1281's per-request invalidate() doesn't cover.

Repro (standalone; needs only a2a-sdk[sqlite] or aiosqlite):

repro_resubscribe_stale.py 
"""Resubscribe on a replica that holds a paused ActiveTask returns a stale snapshot.

a2a-sdk 1.2.2, cluster mode (VersionedDatabaseTaskStore + DatabaseTaskEventStream),
three replicas sharing one SQLite database.
"""
import asyncio
import os
import tempfile

from sqlalchemy.ext.asyncio import create_async_engine

from a2a.helpers.proto_helpers import get_message_text
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.cluster import DatabaseTaskEventStream, VersionedDatabaseTaskStore
from a2a.server.context import ServerCallContext
from a2a.server.events.event_queue import EventQueue
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks.task_updater import TaskUpdater
from a2a.types import a2a_pb2 as pb

DB = os.path.join(tempfile.mkdtemp(), "cluster.db")


class TwoTurnAgent(AgentExecutor):
    """Turn 1 asks for input; turn 2 works for ~1.6s, then completes."""

    async def execute(self, context: RequestContext, queue: EventQueue) -> None:
        if context.current_task is None:
            await queue.enqueue_event(pb.Task(
                id=context.task_id, context_id=context.context_id,
                status=pb.TaskStatus(state=pb.TASK_STATE_SUBMITTED),
                history=[context.message]))
        updater = TaskUpdater(queue, context.task_id, context.context_id)
        if get_message_text(context.message) == "hi":
            await updater.requires_input()
            return
        for _ in range(4):
            await updater.start_work()
            await asyncio.sleep(0.4)
        await updater.complete()

    async def cancel(self, context: RequestContext, queue: EventQueue) -> None:
        await TaskUpdater(queue, context.task_id, context.context_id).cancel()


CARD = pb.AgentCard(name="repro", capabilities=pb.AgentCapabilities(streaming=True))
STORES = []


def replica() -> DefaultRequestHandler:
    engine = create_async_engine(f"sqlite+aiosqlite:///{DB}")
    store, stream = VersionedDatabaseTaskStore(engine), DatabaseTaskEventStream(engine)
    STORES.extend([store, stream])
    return DefaultRequestHandler(agent_executor=TwoTurnAgent(), task_store=store,
                                 event_stream=stream, agent_card=CARD)


def send(text: str, task_id: str | None = None) -> pb.SendMessageRequest:
    msg = pb.Message(message_id=text, role=pb.ROLE_USER, context_id="ctx",
                     parts=[pb.Part(text=text)])
    if task_id:
        msg.task_id = task_id
    return pb.SendMessageRequest(
        message=msg, configuration=pb.SendMessageConfiguration(return_immediately=True))


def label(event) -> str:
    status = getattr(event, "status", None)
    return f"{type(event).__name__}({pb.TaskState.Name(status.state)})" if status else type(event).__name__


async def resubscribe(name: str, handler: DefaultRequestHandler, task_id: str) -> None:
    events = []

    async def tail() -> None:
        async for event in handler.on_subscribe_to_task(
            pb.SubscribeToTaskRequest(id=task_id), ServerCallContext()
        ):
            events.append(label(event))

    try:
        await asyncio.wait_for(tail(), timeout=5)
        events.append("<stream closed>")
    except asyncio.TimeoutError:
        events.append("<still open after 5s>")
    print(f"resubscribe on {name}: {' -> '.join(events)}")


async def main() -> None:
    a, b, c = replica(), replica(), replica()
    for s in STORES:
        await s.initialize()
    ctx = ServerCallContext()

    task = await a.on_message_send(send("hi"), ctx)          # turn 1 on A -> INPUT_REQUIRED
    await asyncio.sleep(0.3)
    await b.on_message_send(send("go", task.id), ctx)        # turn 2 runs on B
    await asyncio.sleep(0.1)

    await asyncio.gather(
        resubscribe("A (ran turn 1)", a, task.id),
        resubscribe("C (never saw the task)", c, task.id),
    )
    final = await c.on_get_task(pb.GetTaskRequest(id=task.id), ctx)
    print("GetTask afterwards:", pb.TaskState.Name(final.status.state))


asyncio.run(main())
Relevant log output
resubscribe on C (never saw the task): Task(TASK_STATE_WORKING) -> TaskStatusUpdateEvent(TASK_STATE_WORKING) -> TaskStatusUpdateEvent(TASK_STATE_WORKING) -> TaskStatusUpdateEvent(TASK_STATE_WORKING) -> TaskStatusUpdateEvent(TASK_STATE_COMPLETED) -> <stream closed>
resubscribe on A (ran turn 1): Task(TASK_STATE_INPUT_REQUIRED) -> <stream closed>
GetTask afterwards: TASK_STATE_COMPLETED
Code of Conduct
  • I agree to follow this project's Code of Conduct
Vorherrschende Sprache
Python
Sterne
2.2k
Forks
509
Ø Merge
3 T. 18 Std.
Gemergte PRs (30 T.)
44

Entwicklungsumgebung

Erste Schritte

  1. Lesen Sie das ganze Issue und danach den Beitragsleitfaden des Projekts.
  2. Schreiben Sie ins Issue, dass Sie es übernehmen — das erspart doppelte Arbeit.
  3. Forken Sie das Repository und arbeiten Sie in einem Branch.
  4. Öffnen Sie einen Pull Request, der die Issue-Nummer nennt.

Mehr aus a2aproject/a2a-python

Alle Issues in a2aproject/a2a-python

Ähnliche Issues

Weitere Issues zu Python

Neue Issues direkt in Ihr Postfach

Eine kurze Übersicht über anfängerfreundliche GitHub-Issues.