Hacktoberfest 2026：维护者为十月标记出来的 issue，仍然开放、适合新手。 浏览 Hacktoberfest issue

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

未关闭
  • 哪里不对？
#1,323 0 条评论 0 个 reaction 已指派 2 人 在 GitHub 查看

维护者通常 2 天内回复

@rohityan 已经在做这个了。

开始于 2026年10月7日。

评估

这个 Issue 还没有评估数据。

描述

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
主要语言
Python
星标
2.2k
派生
509
平均合并
3 天 18 小时
30 天内合并 PR
44

环境准备

从这里开始

  1. 先读完整个 Issue，再读项目的贡献指南。
  2. 在 Issue 下留言说明你要接手 —— 这能避免两个人做同样的事。
  3. Fork 仓库，在一个分支上完成修改。
  4. 提交 Pull Request，并在描述里引用这个 Issue 编号。

a2aproject/a2a-python 的其他 Issue

查看 a2aproject/a2a-python 的全部 Issue

相似的 Issue

更多 Python Issue

把新 issue 发到你的邮箱

精选适合新手参与的 GitHub issue 摘要。