From 72252e48c9df0b6f9cfbe26e36370ac40f82c2be Mon Sep 17 00:00:00 2001
From: Johannes du Plessis <johannes@langchain.dev>
Date: Fri, 22 May 2026 16:02:43 -0700
Subject: [PATCH] fix: stream agent tokens to the Agents UI (#1329)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

* fix(dashboard): stream agent tokens to the Agents UI

The dashboard SSE endpoint was serializing the SDK's `StreamPart`
namedtuple via `str(part)`, so every event arrived at the browser
with `event = "StreamPart(event='values', data=..., id=None)"`. The
frontend's `event.startsWith("messages")` branch never fired and
every event fell through to a query refetch — which is why tool
calls appeared to stream (state refetch) but assistant text did not.

Also opt the run into `messages-tuple` stream mode so the server
actually buffers per-token `AIMessageChunk` events, and forward
`Last-Event-ID` so reconnects resume the in-flight run instead of
restarting from `seq=0`.

* fix(dashboard): drop unsupported stream_mode arg from join_stream

threads.join_stream's stream_mode is a ThreadStreamMode literal
(run_modes/lifecycle/state_update); passing run-level modes is
invalid. The default 'run_modes' already replays the modes the
run was created with, which we set correctly on runs.create.

---------

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>
---
 agent/dashboard/routes.py     |  7 ++++++-
 agent/dashboard/thread_api.py | 28 ++++++++++++++++++++++++----
 2 files changed, 30 insertions(+), 5 deletions(-)

diff --git a/agent/dashboard/routes.py b/agent/dashboard/routes.py
index 1982e82b..5fff2e01 100644
--- a/agent/dashboard/routes.py
+++ b/agent/dashboard/routes.py
@@ -627,10 +627,15 @@ async def api_delete_thread(
 @router.get("/threads/{thread_id}/stream")
 async def api_stream_thread(
     thread_id: str,
+    request: Request,
     session: dict[str, Any] = _SESSION_DEP,
 ) -> StreamingResponse:
+    last_event_id = request.headers.get("last-event-id")
+
     async def event_generator():
-        async for chunk in stream_dashboard_thread(thread_id, session["sub"]):
+        async for chunk in stream_dashboard_thread(
+            thread_id, session["sub"], last_event_id=last_event_id
+        ):
             yield chunk
 
     return StreamingResponse(
diff --git a/agent/dashboard/thread_api.py b/agent/dashboard/thread_api.py
index 0646f008..f272469e 100644
--- a/agent/dashboard/thread_api.py
+++ b/agent/dashboard/thread_api.py
@@ -26,6 +26,7 @@ logger = logging.getLogger(__name__)
 
 _ASSISTANT_ID = "agent"
 _DASHBOARD_SOURCE = "dashboard"
+_DASHBOARD_STREAM_MODES: tuple[str, ...] = ("values", "updates", "messages-tuple")
 
 
 def _agent_version_metadata() -> dict[str, str]:
@@ -292,6 +293,8 @@ async def _start_agent_run(
         input={"messages": [{"role": "user", "content": prompt}]},
         config={"configurable": configurable, "metadata": _agent_version_metadata()},
         if_not_exists="create",
+        stream_mode=list(_DASHBOARD_STREAM_MODES),
+        stream_resumable=True,
     )
     run_id = run.get("run_id") if isinstance(run, dict) else getattr(run, "run_id", None)
     await client.threads.update(
@@ -367,6 +370,8 @@ async def send_dashboard_message(
         _ASSISTANT_ID,
         input={"messages": [{"role": "user", "content": prompt}]},
         config={"configurable": configurable, "metadata": _agent_version_metadata()},
+        stream_mode=list(_DASHBOARD_STREAM_MODES),
+        stream_resumable=True,
     )
     run_id = run.get("run_id") if isinstance(run, dict) else getattr(run, "run_id", None)
     await client.threads.update(
@@ -426,7 +431,9 @@ async def delete_dashboard_thread(thread_id: str, login: str) -> None:
     await client.threads.delete(thread_id)
 
 
-async def stream_dashboard_thread(thread_id: str, login: str) -> AsyncIterator[str]:
+async def stream_dashboard_thread(
+    thread_id: str, login: str, *, last_event_id: str | None = None
+) -> AsyncIterator[str]:
     try:
         thread = await langgraph_client().threads.get(thread_id)
     except Exception as exc:  # noqa: BLE001
@@ -435,7 +442,20 @@ async def stream_dashboard_thread(thread_id: str, login: str) -> AsyncIterator[s
     metadata = thread.get("metadata") if isinstance(thread.get("metadata"), dict) else {}
     _assert_thread_owner(metadata, login)
 
-    stream = await langgraph_client().threads.join_stream(thread_id)
+    stream = await langgraph_client().threads.join_stream(
+        thread_id,
+        last_event_id=last_event_id,
+    )
     async for part in stream:
-        payload = part if isinstance(part, dict) else {"event": str(part)}
-        yield f"data: {json.dumps(payload, default=str)}\n\n"
+        event = getattr(part, "event", None) or (
+            part.get("event") if isinstance(part, dict) else None
+        )
+        data = getattr(part, "data", None) if not isinstance(part, dict) else part.get("data")
+        event_id = getattr(part, "id", None) if not isinstance(part, dict) else part.get("id")
+        payload: dict[str, Any] = {"event": event, "data": data}
+        if event_id is not None:
+            payload["id"] = event_id
+        chunk = f"data: {json.dumps(payload, default=str)}\n\n"
+        if event_id is not None:
+            chunk = f"id: {event_id}\n{chunk}"
+        yield chunk