fix(agent): drop placeholder assistant bubble in chat

Author: jordan-umusu

frontend/src/components/chat/chat-session-pane.tsx

@@ -288,7 +288,7 @@ export function ChatSessionPane({
     currRunId,
     prompt: activePrompt,
   } = useSessionStatus({
-    chatId: chat?.id,
+    chatId: isReadonly ? undefined : chat?.id,
     workspaceId,
   })
   const { sendMessage, messages, status, regenerate, lastError, clearError } =

frontend/tests/chat-session-pane.test.tsx

@@ -7,7 +7,11 @@ import {
   MessagePart,
 } from "@/components/chat/chat-session-pane"
 import { TooltipProvider } from "@/components/ui/tooltip"
-import { useUpdateChat, useVercelChat } from "@/hooks/use-chat"
+import {
+  useSessionStatus,
+  useUpdateChat,
+  useVercelChat,
+} from "@/hooks/use-chat"
 import { useBuilderRegistryActions } from "@/lib/hooks"
 
 jest.mock("@/hooks/use-chat", () => ({
@@ -85,6 +89,9 @@ const mockUseVercelChat = useVercelChat as jest.MockedFunction<
 const mockUseUpdateChat = useUpdateChat as jest.MockedFunction<
   typeof useUpdateChat
 >
+const mockUseSessionStatus = useSessionStatus as jest.MockedFunction<
+  typeof useSessionStatus
+>
 const mockUseBuilderRegistryActions =
   useBuilderRegistryActions as jest.MockedFunction<
     typeof useBuilderRegistryActions
@@ -241,6 +248,30 @@ describe("ChatSessionPane", () => {
     })
   })
 
+  it("does not poll session status for read-only legacy chats", () => {
+    mockUseVercelChatStatus("ready")
+    const readonlyChat = { ...createChatFixture(), is_readonly: true }
+
+    render(
+      <QueryClientProvider client={queryClient}>
+        <TooltipProvider>
+          <ChatSessionPane
+            chat={readonlyChat}
+            workspaceId="workspace-1"
+            entityType="case"
+            entityId="case-1"
+            modelInfo={{ name: "gpt-4o-mini", provider: "openai" }}
+          />
+        </TooltipProvider>
+      </QueryClientProvider>
+    )
+
+    expect(mockUseSessionStatus).toHaveBeenCalledWith({
+      chatId: undefined,
+      workspaceId: "workspace-1",
+    })
+  })
+
   it("renders dots indicator while status is submitted", () => {
     mockUseVercelChat.mockReturnValue({
       sendMessage: jest.fn(),

tests/unit/test_agent_session_router.py

@@ -579,16 +579,20 @@ async def test_stream_session_events_returns_204_when_no_turn_started() -> None:
 async def test_stream_session_events_attaches_when_turn_in_progress_no_events_yet() -> (
     None
 ):
+    session_id = uuid.uuid4()
+    curr_run_id = uuid.uuid4()
     stream = _fake_stream()
     response = await _run_stream_endpoint(
-        session=_fake_stream_session(status_value="running"),
+        session=_fake_stream_session(status_value="running", curr_run_id=curr_run_id),
         stream=stream,
         headers={},
+        session_id=session_id,
     )
 
     assert isinstance(response, StreamingResponse)
     stream.sse.assert_called_once()
     assert stream.sse.call_args.kwargs["last_id"] == "0-0"
+    assert stream.sse.call_args.kwargs["message_id"] == f"{session_id}:{curr_run_id}"
 
 
 @pytest.mark.anyio
@@ -609,9 +613,10 @@ async def test_stream_session_events_waiting_without_cursor_uses_db_history() ->
 @pytest.mark.anyio
 async def test_stream_session_events_resumes_after_last_event_id() -> None:
     """A live cursor newer than the buffer min resumes after it (composite id)."""
+    curr_run_id = uuid.uuid4()
     stream = _fake_stream(min_id="1000-0")
     response = await _run_stream_endpoint(
-        session=_fake_stream_session(status_value="running"),
+        session=_fake_stream_session(status_value="running", curr_run_id=curr_run_id),
         stream=stream,
         headers={"Last-Event-ID": "1234-0:2"},
     )
@@ -647,9 +652,10 @@ async def test_stream_session_events_terminal_reconnect_uses_latest_run_id() ->
 @pytest.mark.anyio
 async def test_stream_session_events_stale_cursor_running_replays_from_start() -> None:
     """Cursor older than the buffer min, still running -> replay from 0-0."""
+    curr_run_id = uuid.uuid4()
     stream = _fake_stream(min_id="2000-0")
     response = await _run_stream_endpoint(
-        session=_fake_stream_session(status_value="running"),
+        session=_fake_stream_session(status_value="running", curr_run_id=curr_run_id),
         stream=stream,
         headers={"Last-Event-ID": "1000-0:0"},
     )
@@ -672,4 +678,22 @@ async def test_stream_session_events_stale_cursor_terminal_finishes() -> None:
 
     assert isinstance(response, StreamingResponse)
     stream.finished_sse.assert_called_once()
+    assert stream.finished_sse.call_args.kwargs["message_id"] is None
+    stream.sse.assert_not_called()
+
+
+@pytest.mark.anyio
+async def test_stream_session_events_terminal_without_run_id_finishes() -> None:
+    """Terminal reconnects without a run id do not synthesize a new bubble id."""
+    stream = _fake_stream(min_id="1000-0")
+    response = await _run_stream_endpoint(
+        session=_fake_stream_session(status_value="stopped"),
+        stream=stream,
+        headers={"Last-Event-ID": "1234-0:2"},
+        latest_run_id=None,
+    )
+
+    assert isinstance(response, StreamingResponse)
+    stream.finished_sse.assert_called_once()
+    assert stream.finished_sse.call_args.kwargs["message_id"] is None
     stream.sse.assert_not_called()

tests/unit/test_vercel_stream_context.py

@@ -1018,6 +1018,22 @@ async def events():
     assert id_frames[1].startswith("id: 1000-0:1\n")
 
 
+@pytest.mark.anyio
+async def test_sse_vercel_omits_start_frame_when_message_id_unknown():
+    """Terminal reconnect finish streams should not create a synthetic bubble."""
+
+    async def events():
+        return
+        yield  # pragma: no cover - establishes async generator
+
+    frames: list[str] = []
+    async for frame in sse_vercel(events(), message_id=None):
+        frames.append(frame)
+
+    assert not any('"type":"start"' in frame for frame in frames)
+    assert any('"type":"finish"' in frame for frame in frames)
+
+
 @pytest.mark.anyio
 async def test_sse_vercel_skips_frames_at_composite_resume_cursor():
     """Reconnect inside a Redis entry replays that entry but drops seen frames."""

tracecat/agent/adapter/vercel.py

@@ -766,7 +766,7 @@ class VercelStreamContext:
     consistent start/delta/end sequences required by the Vercel protocol.
     """
 
-    message_id: str
+    message_id: str | None
     # Active parts keyed by event index -> maintains per-part lifecycle state.
     part_states: dict[int, _PartState] = dataclasses.field(default_factory=dict)
     tool_finished: dict[str, bool] = dataclasses.field(default_factory=dict)
@@ -1805,14 +1805,15 @@ def convert_chat_messages_to_ui(
 async def sse_vercel(
     events: AsyncIterable[StreamEvent],
     *,
-    message_id: str,
+    message_id: str | None,
     resume_from: str | None = None,
 ) -> AsyncIterable[str]:
     """Stream Redis events as Vercel AI SDK frames without persisting adapter output.
 
     ``message_id`` is the stable assistant-bubble id (``session_id:curr_run_id``)
     so reconnects within a turn resume the same bubble instead of spawning a new
-    one. Each Redis entry can fan out to many Vercel frames, so frames carry a
+    one. It is omitted for terminal reconnects where no run id can be resolved.
+    Each Redis entry can fan out to many Vercel frames, so frames carry a
     composite ``id: {redis_id}:{frame_index}`` that the browser replays from.
     """
 
@@ -1839,7 +1840,8 @@ def emit(payload: VercelSSEPayload) -> str | None:
 
     try:
         # 1. Start of the message stream
-        yield format_sse(StartEventPayload(messageId=message_id))
+        if message_id is not None:
+            yield format_sse(StartEventPayload(messageId=message_id))
 
         # 2. Process events from Redis stream
         async for stream_event in events:

tracecat/agent/session/router.py

@@ -42,9 +42,9 @@
 router = APIRouter(prefix="/agent/sessions", tags=["agent-sessions"])
 
 
-def _bubble_id(session_id: uuid.UUID, curr_run_id: uuid.UUID | None) -> str:
-    """Stable assistant-bubble id for a turn (falls back to session if no run)."""
-    return f"{session_id}:{curr_run_id}" if curr_run_id else str(session_id)
+def _bubble_id(session_id: uuid.UUID, curr_run_id: uuid.UUID | None) -> str | None:
+    """Stable assistant-bubble id for a turn, if the turn is known."""
+    return f"{session_id}:{curr_run_id}" if curr_run_id else None
 
 
 def _redis_id_lt(a: str, b: str) -> bool:
@@ -567,6 +567,17 @@ async def stream_session_events(
             start_id = requested
             resume_from = last_event_id if cursor else None
 
+    # Terminal reconnects can outlive curr_run_id and, in edge cases, fail to
+    # resolve a persisted run id. Do not invent a session-only assistant id:
+    # finish cleanly so the client refetches DB history without rendering an
+    # empty synthetic bubble.
+    if not is_stream_attachable and message_id is None:
+        return StreamingResponse(
+            stream.finished_sse(format=format, message_id=None),
+            media_type="text/event-stream",
+            headers=headers,
+        )
+
     logger.info(
         "Starting session stream",
         last_id=start_id,

tracecat/agent/stream/connector.py

@@ -254,7 +254,7 @@ def sse(
         last_id: str,
         format: StreamFormat,
         *,
-        message_id: str,
+        message_id: str | None,
         resume_from: str | None = None,
     ) -> AsyncIterable[str]:
         cursor = parse_vercel_frame_cursor(resume_from)
@@ -277,7 +277,7 @@ def sse(
                 raise ValueError(f"Invalid format: {format}")
 
     def finished_sse(
-        self, format: StreamFormat, *, message_id: str
+        self, format: StreamFormat, *, message_id: str | None
     ) -> AsyncIterable[str]:
         """Emit an immediately-finishing stream (no live content).