[marcinbelczewski]

Description

Problem

Spec 3.1.2: "If the agent returns a Task, the stream MUST begin with the Task object". On a follow-up to an existing task, on_message_send_stream subscribes with include_initial_task=False, so the leading Task has to come from the executor. The v1.0 migration guide shows how: task = context.current_task or new_task_from_user_message(...), then enqueue_event(task).

When the task already exists, EventConsumer._handle_initial_task logs Task ... already exists. Ignoring task replacement. at ERROR and clears message_to_save. As a result, the follow-up user message never reaches Task.history.

Fix

In _handle_initial_task, when the task already exists:

• If the follow-up message is still pending (this Task is the first task event of the request), it's added to the stored task's history right away. The leading streamed Task then already has it. As with the other path, update_with_message first moves the agent's status.message into history, so the order is user 1, agent, user 2.
• Otherwise, e.g. a second Task in the same request, the ERROR is logged as before.

Both paths now share _add_message_to_history, which is the message_id check from #1106 moved out of _handle_task_modification_event. So a follow-up resent with the same message_id is saved once, no matter whether the executor enqueues the Task or only status updates.

Thanks @ConnorMoss02 for the review that led to this version.

Behaviour changes

• A Task enqueued as the first event of a follow-up no longer logs an ERROR. Its content is still ignored, and the stored task is kept.
• An executor that only enqueues the current task on a follow-up, with no further events, now keeps the follow-up message. Before, it was lost.

Known limitation

The leading streamed Task includes the follow-up message only when the executor enqueues context.current_task itself, as the migration guide does. That object is the one TaskManager keeps, and update_with_message updates it in place. If the executor enqueues a copy, history is still saved correctly, but the first streamed Task is stale, same as on main. Subscribers get the raw executor event in _process_event_inner. This is discussed in the review thread.

Tests

In tests/integration/test_scenarios.py:

• test_scenario_streaming_resumption_enqueues_current_task (once / twice): the executor enqueues the current task once or twice on the follow-up. It checks the event order, that the first streamed Task has history msg-start, agent-ask, msg-resume and no status.message, the final history, and that the replacement ERROR is logged 0 or 1 times.
• test_scenario_resent_follow_up_message_saved_once (status_only / current_task): resending the same follow-up adds it to history once for both executor patterns.

On main, both resumption cases and resent…[current_task] fail. resent…[status_only] passes there, because it covers the existing check that moved into the shared helper. resent…[current_task] also fails without the message_id check on the new path.

---

• Follow the CONTRIBUTING Guide.
• Make your Pull Request title in the https://www.conventionalcommits.org/ specification.
• Ensure the tests and linter pass (Run bash scripts/format.sh from the repository root to format)
• Appropriate docs were updated (if necessary)

Fixes #1285 🦕

[github-actions]

🧪 Code Coverage (vs main)

⬇️ Download Full Report

	Base	PR	Delta
src/a2a/server/agent_execution/active_task.py	96.10%	96.14%	🟢 +0.04%
src/a2a/server/cluster/database_event_stream.py	97.27%	93.64%	🔴 -3.64%
Total	93.46%	93.43%	🔴 -0.04%

Generated by coverage-comment.yml

[ConnorMoss02]

Ran this locally with an agent like the one in your test. Two things.

1. The leading Task on the follow-up stream still has the old history: ['msg-start']. msg-resume only lands with the next status update. The final get_task is right, but the first streamed Task is what 3.1.2 is about.
2. On a follow-up message_to_save is still set for every Task, so enqueueing current_task twice drops the second one with no ERROR.

Saving the message here fixes both. The first Task then has both messages, the double enqueue logs once, and tests/integration plus tests/server pass:

if existing_task:
    if self.message_to_save is None:
        logger.error(
            'Task %s already exists. Ignoring task replacement.',
            self.active_task._task_id,
        )
    else:
        await self.active_task._task_manager.save_task_event(
            self.active_task._task_manager.update_with_message(
                self.message_to_save, existing_task
            )
        )
else:
    await self.active_task._task_manager.save_task_event(event)
self.message_to_save = None

[marcinbelczewski]

@ConnorMoss02 thanks a lot for taking the time to actually run it and for the patch! You are right on both points and sorry for that. My test only checked that the stream starts with a Task and what ends up in history at the end, but never what is inside that first Task, which is the whole point of 3.1.2... And the claim that a second Task still logs the ERROR I based on test_scenario_enqueue_task_twice, which only covers a brand new task, not a follow-up. My bad.

I applied your approach with one small change: the follow-up message now goes through the same message_id check that _handle_task_modification_event already had (from #1106). I moved it to _add_message_to_history so both paths use it. Without it, if the executor enqueues current_task and the client resends the same follow-up message (same message_id), the message lands in history twice, while for an executor that only sends status updates it's saved once. Let me know if you'd rather keep it exactly as you proposed.

Tests:

• test_scenario_streaming_resumption_enqueues_current_task is now parametrized with once/twice. It checks the history of the first streamed Task and that the ERROR is logged exactly once when the Task is enqueued twice.
• New test_scenario_resent_follow_up_message_saved_once checks that both executor patterns save a resent message only once.

I checked that the streaming test fails on the previous version and the resend one fails with your patch as is. I also broke each part of the change on purpose to make sure some test catches it. Squashed into a single commit and force pushed.

Thanks again for catching this!

[marcinbelczewski]

One more thing I noticed after posting the above. The first streamed Task is right only if the executor enqueues context.current_task itself (as in the migration guide), because that's the same object TaskManager keeps. If it enqueues a copy, history is fine but the first streamed Task still has only ['msg-start']. Subscribers get the raw executor event here:

a2a-python/src/a2a/server/agent_execution/active_task.py

Lines 280 to 286 in aa8e966

	handled_event = updated_task if isinstance(event, Task) else event
	if updated_task is not None and handled_event is not None:
	await self._update_task_state(updated_task, handled_event)
	self.active_task._task_created.set()
	await self._enqueue_to_subscribers(event, updated_task)

Looks like a leftover from #1053. Before it the consumer sent the stored task to subscribers, but maybe I'm missing something. Passing handled_event there fixes it and all tests pass, but it changes what's streamed for every Task event, so I'd rather not sneak it into this PR. Should I add it here or open a separate issue?

[marcinbelczewski]

Small update: the test now also has an agent message in the INPUT_REQUIRED status, to check the user 1 → agent → user 2 order.