Hacktoberfest 2026: le issue che i maintainer hanno segnato per ottobre, aperte e adatte ai principianti. Sfoglia le issue Hacktoberfest

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

Aperta
  • Qualcosa non va?
#1,323 0 commenti 0 reazioni 1 assegnatario Vedi su GitHub

I maintainer di solito rispondono entro 2 giorni

@rohityan ci sta già lavorando.

Dal 7/10/2026.

  • #1327 di @SashaMIT — aperta

Valutazione

Questa issue non è ancora stata valutata.

Descrizione

component: server
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
Lingua principale
Python
Stelle
2.2k
Fork
509
Merge medio
3g 18h
PR unite (30g)
44

Preparare l'ambiente

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di a2aproject/a2a-python

Tutte le issue di a2aproject/a2a-python

Issue simili

Altre issue su Python

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.