From 36f43c270670ad5d81e9beac7e79744655176ad5 Mon Sep 17 00:00:00 2001
From: fangliquanflq <fangliquan@qq.com>
Date: Sun, 13 Sep 2026 00:48:35 +0800
Subject: [PATCH] fix(a2a): finalize tasks after stream disconnect

---
 plugins/platforms/a2a/adapter.py  |  4 ++++
 tests/plugins/test_a2a_phase23.py | 35 +++++++++++++++++++++++++++++++
 2 files changed, 39 insertions(+)

diff --git a/plugins/platforms/a2a/adapter.py b/plugins/platforms/a2a/adapter.py
index 73d1f29169..73568595ff 100644
--- a/plugins/platforms/a2a/adapter.py
+++ b/plugins/platforms/a2a/adapter.py
@@ -684,6 +684,7 @@ class A2AAdapter(BasePlatformAdapter):
         """message/stream as an SSE response of JSON-RPC-wrapped StreamResponse events (§9.4)."""
         protocol.metrics.streams_started += 1
         self._sse_headers(handler)
+        pending = None
         try:
             terminal, pending = self._prepare_task(params, peer, agent=agent)
             if terminal is not None:
@@ -694,8 +695,11 @@ class A2AAdapter(BasePlatformAdapter):
             self._sse_write(handler, protocol.sse_data(protocol.stream_task(submitted), req_id))
             self._sse_write(handler, protocol.sse_data(protocol.status_update(task_id, context_id, protocol.STATE_WORKING), req_id))
             state, reply = self._finalize_task(pending, *self._await_reply(pending, keepalive=self._keepalive(handler)))
+            pending = None
             self._emit_terminal(handler, task_id, context_id, state, reply, req_id=req_id)
         except (BrokenPipeError, ConnectionResetError):
+            if pending is not None:
+                self._finalize_task(pending, protocol.STATE_FAILED, "[client disconnected]")
             logger.debug("A2A: stream client disconnected")
 
     def _rpc_tasks_subscribe(self, handler, req_id: Any, params: dict, agent: Optional[dict] = None) -> None:
diff --git a/tests/plugins/test_a2a_phase23.py b/tests/plugins/test_a2a_phase23.py
index 101d17e090..67ac1ca66c 100644
--- a/tests/plugins/test_a2a_phase23.py
+++ b/tests/plugins/test_a2a_phase23.py
@@ -572,6 +572,41 @@ class TestTaskStore:
         assert result == [(protocol.STATE_COMPLETED, "reply")]
         assert adapter.tasks.get("t-live")["state"] == protocol.STATE_COMPLETED
 
+    def test_stream_disconnect_releases_active_request(self, monkeypatch):
+        adapter, _base = _make_live_adapter(monkeypatch)
+        rec = adapter.tasks.create("t-live", "c1", "peer")
+        adapter.tasks.set_state("t-live", protocol.STATE_WORKING)
+        pending = {
+            "task_id": "t-live", "context_id": "c1", "peer": "peer",
+            "future": adapter._add_pending("t-live", "c1"),
+            "created_iso": rec["created_iso"], "started": time.time(),
+        }
+        monkeypatch.setattr(adapter, "_prepare_task", lambda *_args, **_kwargs: (None, pending))
+
+        class BrokenWriter:
+            def write(self, _chunk):
+                raise BrokenPipeError
+
+        class Handler:
+            wfile = BrokenWriter()
+
+            def send_response(self, _status):
+                pass
+
+            def send_header(self, _name, _value):
+                pass
+
+            def end_headers(self):
+                pass
+
+        adapter._rpc_message_stream(Handler(), 1, {}, "peer")
+
+        stored = adapter.tasks.get("t-live")
+        assert stored["state"] == protocol.STATE_FAILED
+        assert stored["reply"] == "[client disconnected]"
+        assert "t-live" not in adapter._pending
+        assert "t-live" not in adapter._active_tasks
+
     def test_list_newest_first_with_filters(self):
         store = protocol.TaskStore()
         store.create("t1", "c1", "p")