feat: end a task's stream when its worker is done, and stream the bridge's answer

[echarles]

Problem

The SSE plumbing from #51 is on main (event bus, message/stream, tasks/resubscribe), but a stream only ends when the worker publishes a final status and closes the bus itself. A worker that just writes storage never does — and the pydantic-ai bridge's AgentWorker is such a worker. Verified against main with agent_to_a2a(Agent(TestModel(...))): the task reaches completed in storage within a second, while the message/stream response has sent nothing and never ends.

Fix

• Worker — once run_task / cancel_task returns, the task's state is read back from storage and, if it is final (completed, canceled, failed, rejected) or waiting on the client (input-required, auth-required), published and the stream closed. The failure path does what it did before. A worker that already published and closed sees no duplicate: nothing is subscribed by then. New publish_status / publish_artifact helpers give workers a plain way to report progress.
• InMemoryEventBus.emit — drops a subscriber whose connection is gone (BrokenResourceError / ClosedResourceError) instead of letting a disconnected SSE client break the worker.
• pydantic-ai bridge — publishes working when it starts, the model's text as chunks of the answer's artifact while it is written (append=True, via Agent.iter, available on every pydantic-ai release the bridge supports), and the whole artifact as the last chunk under the same artifact_id. With TestModel(custom_output_text='hello streaming world') the stream is now: task → working → 'hello ' → 'streaming ' → 'world' → whole artifact (last chunk) → completed, and ends.

Tests and docs

• tests/test_streaming.py: a worker that only writes storage ends its stream with completed; one that raises ends it with failed. The existing echo-worker test is unchanged (still exactly three events).
• tests/test_pydantic_ai_streaming.py: the bridge streams the answer and ends.
• README: a Streaming subsection under Design.

Succeeds the closed #43, on top of the event_bus design rather than beside it.

🤖 Generated with Claude Code

[Copilot]

🟡 Changes recommended

resubscribe_task currently doesn’t treat input-required/auth-required as stream-ending states, so resubscribe can hang indefinitely now that workers may close streams in those states.

Once you've addressed the issues Copilot identified, you can request another Copilot review.

Pull request overview

This PR improves SSE streaming reliability in FastA2A by ensuring task streams end even when workers only write to storage (not the event bus), and by making the pydantic-ai bridge stream model output incrementally as artifact chunks.

Changes:

• Update Worker to publish a final status (based on storage state) and close the event bus stream after task operations return or fail.
• Harden InMemoryEventBus.emit() to drop disconnected subscribers instead of letting broken SSE connections break publishers.
• Add/extend tests and documentation for streaming behavior, including pydantic-ai streaming of artifact chunks.

File summaries

File	Description
tests/test_streaming.py	Adds coverage for streams ending when workers only update storage, and for failure ending behavior.
tests/test_pydantic_ai_streaming.py	New test ensuring the pydantic-ai bridge streams answer text as artifact chunks and ends the stream.
README.md	Documents expected message/stream behavior and the bridge’s streaming semantics.
fasta2a/worker.py	Adds stream-ending logic based on reading final task state from storage; introduces publish helpers for status/artifacts.
fasta2a/pydantic_ai/_bridge.py	Emits working, streams model text deltas as artifact chunks, then publishes the final artifact.
fasta2a/event_bus.py	Drops dead subscribers on send errors to avoid breaking workers publishing to SSE clients.

Review details

• Files reviewed: 6/6 changed files
• Comments generated: 3
• Review effort level: Lite

---

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

[echarles]

The 3.10+ test jobs were hanging in tests/test_pydantic_ai.py (3.9 skips the bridge tests). Cause: the bridge streamed through node.stream(), and FunctionModel without a stream_function — what those tests use — refuses streamed requests; pydantic-ai does not let a run continue once a node has tried to stream, so the task failed and the tests' while task := await get_task(...) loops, which only exit on completed, waited forever.

Fixed in ec07a38: the bridge attempts the streamed run and, if the model refuses on entering the first stream (before any request is made), redoes the run without streaming and remembers it for the next tasks. A refusal after streaming has begun is raised as the real error it is. New test test_a_model_that_cannot_stream_still_answers_and_the_stream_ends covers it under a 10 s deadline.

Reproduced and verified locally on Python 3.12 with the locked dependencies (uv sync --frozen --extra pydantic-ai --python 3.12): before the fix tests/test_pydantic_ai.py hangs; after it the whole suite passes (34 tests), ruff and pyright clean.

Worth a follow-up: those poll loops in test_pydantic_ai.py hang the suite on a failed task instead of failing it.