Consider adding support to streaming to OnMessageReceived in TaskManager

[sucaritas]

The issue comes from the fact that the SendMessageStreamingAsync method in TaskManager checks for OnMessageReceived, and if it is found, there is no way to support streaming.

a2a-dotnet/src/A2A/Server/TaskManager.cs

Line 244 in 5bee984

if (OnMessageReceived != null)

Since receiving messages is one of the most common actions to override, this prevents people from either streaming or being able to receive messages, just so streaming can work.

Suggestions:

1.- Make the streaming be bases on the UpdateStatusAsync method. streaming should subscribe to events from the UpdateStatusAsync
2.- Add an OnMessageReceivedStreaming method that returns an IAsyncEnumerable.
3.- Add the TaskUpdateEventEnumerator object to the signature of OnMessageReceived, so that events can be pumped out as they are processed.

Pros & Cons

For #1
Pro: No major signature changes, just after creating the AgentTask imedeatly subscribe to it and pass along the enumerator.

For #2
pro: The implementation is semantically good
Con: The implementore will have to implement two methods to receive messages

For #3
Pro: The implementation is straight forward. Semantically it might be ok, not sure. i think i should as the non-streaminmg processing should be able to push notifications.

#1 Is the one i would prefer as it would allow me to use the UpdateStatusAsync naturally.

[darrelmiller]

If an Agent implements OnMessageReceived, it is making the decision that it wants to respond to an incoming message with a single response message. Keeping the stream open is unnecessary.

From the A2A documentation

When a message is sent to an agent, it can choose to reply with either:

A stateless Message.
A stateful Task followed by zero or more TaskStatusUpdateEvent or TaskArtifactUpdateEvent.
If the response is a Message, the interaction is completed.
https://a2a-protocol.org/latest/topics/life-of-a-task/

If the Agent implements OnTaskCreated, then it returns a Task and a the stream is kept open because TaskStatusEvents can signal the final event in the stream.

There is a hybrid scenario where an agent may want to initially return messages to the client and at some point decide that a Task should be created. In that case, the client would have to make multiple calls to message/stream before the stream would be kept opened once the agent creates the task.

I don't think there is any point in keeping the stream open after an agent returns a message because the agent isn't going to return anything else on that stream until a client sends a new message. If the client sends a new message and the agent does decide to create a task, the stream will be opened on the second request and the first stream would just be unused.

[sucaritas]

This makes sense however, the issue is that when an agent implemtns the OnMessageReceived it will never be able to stream.
this limitation is explicitly setup in the 'SendMessageStreamingAsync' method of the TaskManger.

which means that if a caller asks for a stream the Agent will not be able to comply.
imho this is wrong.

if the caller calls throguh 'SendMessageAsync' then yes that makes sense that there would be no streaming and thats the end on that. however if the caller is explicitly asking for a stream through 'SendMessageStreamingAsync' then it is a bug in my opinion to simply return 1 message and its done.

Which i also notice that if a task is started with a non-streaming call through SendMessageAsync and it is a long running task.
the SendMessageAsync will return the task to the user, but the user will not be able to subscribe to update for the task.
it is left with only polling which i also think is a bug. this is because the SubscribeToTaskAsync method tries to find an existing enumerator, but cannot find it because the initial call was made with SendMessageAsync

[darrelmiller]

@sucaritas I need to go reconfirm this but there should be no problem returning a stream once a Task has been created.

a2a-dotnet/src/A2A/Server/TaskManager.cs

Lines 241 to 245 in 5bee984

	if (agentTask == null)
	{
	// If the task is configured to process simple messages without tasks, pass the message directly to the agent
	if (OnMessageReceived != null)
	{

The method checks to see if there is a Task before checking to see if OnMessageReceived is present. If there is a Task then it will stream events back to the client.

Which i also notice that if a task is started with a non-streaming call through SendMessageAsync and it is a long running task.
the SendMessageAsync will return the task to the user, but the user will not be able to subscribe to update for the task. it is left with only polling which i also think is a bug. this is because the SubscribeToTaskAsync method tries to find an existing enumerator, but cannot find it because the initial call was made with SendMessageAsync

That's a great callout. Interestingly, in the JSON-RPC transport the method is called "resubscribe" https://a2a-protocol.org/latest/specification/#79-tasksresubscribe and the specification explicitly limits it to only being able to reconnect to previously created streams. When the gRPC and HTTP transport was introduced it was renamed to "subscribe" which infers that you should be able to subscribe to a task that was started synchronously. We should definitely get that clarified in the specification.

[sucaritas]

Thank you @darrelmiller I have an initial implementation by not touching the OnMessageReceived.
I know this is in prerelease trying as best i can not to fork and just use what is avilable. not implementing OnMessageReceived has solve this issue for me thanks for taking a look.