Add `streaming` support by eleonorecharles · Pull Request #43 · datalayer/fasta2a

eleonorecharles · 2026-01-08T13:20:20Z

Fixes #27

Superseds #26

@echarles @physicsrob @samuelcolvin @Kludex

echarles · 2026-01-09T15:48:22Z

CI is green - we are using this branch for various projects.

echarles · 2026-01-13T20:03:30Z

datalayer/agent-runtimes#12

Needed for:

amitgelber · 2026-02-26T14:22:13Z

@echarles can this be merged? how can we push it? thank you

echarles · 2026-02-26T17:46:41Z

@echarles can this be merged? how can we push it? thank you

I hope this PR can be reviewed soon.

Kludex · 2026-03-07T15:27:53Z

+                    # Serialize event to ensure proper camelCase conversion
+                    event_dict = stream_event_ta.dump_python(event, mode='json', by_alias=True)
+
+                    # Wrap in JSON-RPC response
+                    jsonrpc_response = {'jsonrpc': '2.0', 'id': request_id, 'result': event_dict}
+
+                    # Convert to JSON string
+                    yield json.dumps(jsonrpc_response)


Do we need to do this serialization and then deserialization? Seems a bit weird. I'll check.

But for sure we are not going to use json.dumps. There's a function in pydantic_core for it.

In d16abca

The 3-step serialize pattern in the SSE generator:

stream_event_ta.dump_python→ intermediate Python dict

Wrap in a plain dict {'jsonrpc': '2.0', ...}

json.dumps(...) → JSON string

is replaced by a single call . This skips the intermediate dict materialization and uses pydantic_core' JSON serializer instead of json.dumps. The StreamMessageResponse TypeAdapter handles camelCase aliasing for the entire envelope + nested event in one pass.

Kludex · 2026-03-07T15:33:12Z

        provider: AgentProvider | None = None,
        skills: list[Skill] | None = None,
        docs_url: str | None = '/docs',
+        streaming: bool = False,


Do you want it to be False by default?

set to True in 23907cf

Kludex · 2026-03-07T15:33:46Z

+            # Parse the streaming request
+            stream_request = stream_message_request_ta.validate_json(data)
+
+            # Create an async generator wrapper that formats events as JSON-RPC responses


There's no need for a comment explaining each line.

Suggested change

# Create an async generator wrapper that formats events as JSON-RPC responses

fixed in 33ee9c5

Kludex · 2026-03-07T15:33:53Z

        if a2a_request['method'] == 'message/send':
            jsonrpc_response = await self.task_manager.send_message(a2a_request)
+        elif a2a_request['method'] == 'message/stream':
+            # Parse the streaming request


Suggested change

# Parse the streaming request

fixed in 33ee9c5

Kludex · 2026-03-07T15:34:09Z

-        raise NotImplementedError('send_run_task is not implemented yet.')
+        ...


Why did you replace this?

reverted in 1709673

Kludex · 2026-03-07T15:34:17Z

-        raise NotImplementedError('send_cancel_task is not implemented yet.')
+        ...


Same question.

reverted in 1709673

Kludex · 2026-03-07T15:37:23Z

        return self.contexts.get(context_id)
+
+
+class StreamingStorageWrapper(Storage[ContextT]):


Why do we need this instead of using Storage?

Storage is used in 2db81b8

Kludex · 2026-03-07T15:38:07Z

+class TestFastA2AStreaming:
+    """Tests for the FastA2A message/stream SSE endpoint."""


No test classes please.

fixed in ab97e62

devin-ai-integration

Devin Review found 4 potential issues.

View 3 additional findings in Devin Review.

devin-ai-integration · 2026-03-07T17:11:49Z

+        task = await self._storage.update_task(task_id, state, new_artifacts, new_messages)
+
+        # Determine if this is a final state
+        final = state in ('completed', 'failed', 'canceled')


🚩 TaskState completeness: 'rejected', 'auth-required', 'unknown' and 'input-required' not treated as final

In fasta2a/worker.py:74, final is determined by state in ('completed', 'failed', 'canceled'). Looking at the TaskState type alias at fasta2a/schema.py:436-438, there are additional terminal-like states: 'rejected', 'auth-required', and 'unknown'. If a worker implementation ever sets the state to 'rejected', no final event would be sent to stream subscribers, causing them to hang. The current EchoWorker in tests only uses 'working', 'completed', and 'canceled', so this isn't triggered now, but could be a problem for future worker implementations.

Was this helpful? React with 👍 or 👎 to provide feedback.

Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com>

devin-ai-integration

Devin Review found 3 new potential issues.

View 6 additional findings in Devin Review.

devin-ai-integration · 2026-03-07T17:21:46Z

+        task = await self._storage.update_task(task_id, state, new_artifacts, new_messages)
+
+        # Determine if this is a final state
+        final = state in ('completed', 'failed', 'canceled')


🔴 Worker.update_task missing 'rejected' from final states causes stream to never close

The Worker.update_task method at fasta2a/worker.py:88 defines final states as ('completed', 'failed', 'canceled'), but the TaskState type at fasta2a/schema.py:502 includes 'rejected' as a valid terminal state. The TaskManager.resubscribe_task at fasta2a/task_manager.py:218 correctly treats 'rejected' as terminal: terminal_states = {'completed', 'canceled', 'failed', 'rejected'}. If a worker calls self.update_task(task_id, 'rejected'), the final flag will be False, so the method will emit a non-final status update (lines 91-101) but will skip the final status emit and event_bus.close() call (lines 122-134). This means SSE subscribers will hang indefinitely waiting for more events that will never arrive.

Was this helpful? React with 👍 or 👎 to provide feedback.

devin-ai-integration

Devin Review found 3 new potential issues.

View 5 additional findings in Devin Review.

devin-ai-integration · 2026-03-07T17:27:59Z

🚩 InMemoryStorage.load_task mutates the stored task's history in place

Pre-existing issue: InMemoryStorage.load_task at fasta2a/storage.py:85-86 does task['history'] = task['history'][-history_length:], which mutates the stored task dict in place, permanently truncating the history. This means subsequent calls to load_task without history_length will still see the truncated history. This is a pre-existing bug, not introduced by this PR, but relevant context since streaming relies on storage state.

Was this helpful? React with 👍 or 👎 to provide feedback.

echarles · 2026-03-07T17:48:15Z

@Kludex Thx for the review, I have addressed them and CI is green.

echarles · 2026-03-12T05:52:05Z

@Kludex I just see a few PRs merged on the main branch related to streaming with an event_bus. Is the streaming feature now complete in the main branch, and should this PR be closed?

devin-ai-integration

Devin Review found 2 new potential issues.

View 1 additional finding in Devin Review.

devin-ai-integration · 2026-03-12T05:59:56Z

+    async def update_task(
+        self,
+        task_id: str,
+        state: TaskState,
+        new_artifacts: list[Artifact] | None = None,
+        new_messages: list[Message] | None = None,
+    ) -> None:
+        """Update a task's state in storage and publish streaming events to the broker.
+
+        This is the primary method workers should use to update task state. It handles
+        both persisting the update and notifying any stream subscribers.
+        """
+        task = await self.storage.update_task(task_id, state, new_artifacts, new_messages)
+
+        final = state in ('completed', 'failed', 'canceled')
+
+        # For non-final updates, publish status first
+        if not final:
+            await self.broker.event_bus.emit(
+                task_id,
+                StreamResponse(
+                    status_update=TaskStatusUpdateEvent(
+                        task_id=task_id,
+                        context_id=task['context_id'],
+                        status=task['status'],
+                    ),
+                ),
+            )
+
+        # Publish message events before final status so subscribers receive them
+        if new_messages:
+            for message in new_messages:
+                await self.broker.event_bus.emit(task_id, StreamResponse(message=message))
+
+        # Publish artifact events
+        if new_artifacts:
+            for artifact in new_artifacts:
+                await self.broker.event_bus.emit(
+                    task_id,
+                    StreamResponse(
+                        artifact_update=TaskArtifactUpdateEvent(
+                            task_id=task_id,
+                            context_id=task['context_id'],
+                            artifact=artifact,
+                        ),
+                    ),
+                )
+
+        # For final updates, publish status last (after messages and artifacts)
+        if final:
+            await self.broker.event_bus.emit(
+                task_id,
+                StreamResponse(
+                    status_update=TaskStatusUpdateEvent(
+                        task_id=task_id,
+                        context_id=task['context_id'],
+                        status=task['status'],
+                    ),
+                ),
+            )
+            await self.broker.event_bus.close(task_id)


🚩 Test still uses direct storage/event_bus calls instead of new Worker.update_task helper

The test in tests/test_streaming.py:30-68 (EchoWorker) manually calls self.storage.update_task() and self.broker.event_bus.emit()/close() directly, rather than using the new Worker.update_task() helper method introduced in this PR. While not a bug, this means the new Worker.update_task method has no test coverage. If the intent is for workers to use the helper, the test should be updated to validate the helper's behavior (including the ordering of status/message/artifact events).

Was this helpful? React with 👍 or 👎 to provide feedback.

devin-ai-integration · 2026-03-12T05:59:57Z

+StreamEvent = Union[Task, Message, TaskStatusUpdateEvent, TaskArtifactUpdateEvent]
+"""A streaming event that can be sent during message/stream requests."""
+
+stream_event_ta: TypeAdapter[StreamEvent] = TypeAdapter(StreamEvent)


🚩 StreamEvent type alias may not be useful without discriminator

The new StreamEvent = Union[Task, Message, TaskStatusUpdateEvent, TaskArtifactUpdateEvent] at fasta2a/schema.py:997 and its TypeAdapter are exported publicly but never used internally. These TypedDicts share overlapping field names (e.g., task_id, context_id) which could make Pydantic's union discrimination unreliable without an explicit Discriminator. This may cause unexpected validation behavior when deserializing ambiguous payloads. Worth verifying the intended use case for this type.

Was this helpful? React with 👍 or 👎 to provide feedback.

echarles · 2026-03-12T06:05:54Z

@Kludex I have merged and resolved conflicts. Where do we go from here?

echarles and others added 3 commits January 7, 2026 09:29

fix: streaming

3e60fa1

test: streaming

8a5dbfe

fix CI

5903cff

amitgelber mentioned this pull request Feb 26, 2026

Slack invite link is expired pydantic/pydantic-ai#4440

Closed

Kludex reviewed Mar 7, 2026

View reviewed changes

simplify serialization

d16abca

devin-ai-integration Bot reviewed Mar 7, 2026

View reviewed changes

echarles and others added 4 commits March 7, 2026 18:12

set streaming true by default

23907cf

Update fasta2a/applications.py

dec5581

Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com>

remove comments

33ee9c5

revert NotImplementedError

1709673

devin-ai-integration Bot reviewed Mar 7, 2026

View reviewed changes

use storage

2db81b8

devin-ai-integration Bot reviewed Mar 7, 2026

View reviewed changes

echarles added 3 commits March 7, 2026 18:39

no test class

ab97e62

fix: ci

e5e2be8

fix: ci

447153f

merge with main

6c0f423

devin-ai-integration Bot reviewed Mar 12, 2026

View reviewed changes

		raise NotImplementedError('send_run_task is not implemented yet.')
		...

		raise NotImplementedError('send_cancel_task is not implemented yet.')
		...

		return self.contexts.get(context_id)


		class StreamingStorageWrapper(Storage[ContextT]):

		class TestFastA2AStreaming:
		"""Tests for the FastA2A message/stream SSE endpoint."""

Conversation

eleonorecharles commented Jan 8, 2026

Uh oh!

echarles commented Jan 9, 2026

Uh oh!

echarles commented Jan 13, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

amitgelber commented Feb 26, 2026

Uh oh!

echarles commented Feb 26, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

echarles Mar 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

devin-ai-integration Bot left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

devin-ai-integration Bot Mar 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

devin-ai-integration Bot left a comment

Choose a reason for hiding this comment

Uh oh!

devin-ai-integration Bot Mar 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

devin-ai-integration Bot left a comment

Choose a reason for hiding this comment

Uh oh!

devin-ai-integration Bot Mar 7, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

echarles commented Mar 7, 2026

Uh oh!

echarles commented Mar 12, 2026

Uh oh!

devin-ai-integration Bot left a comment

Choose a reason for hiding this comment

echarles commented Jan 13, 2026 •

edited

Loading

echarles Mar 7, 2026 •

edited

Loading

devin-ai-integration Bot Mar 7, 2026 •

edited

Loading

devin-ai-integration Bot Mar 7, 2026 •

edited

Loading