From 30b40caff8036d2aad254570db44ac947b3e62ae Mon Sep 17 00:00:00 2001 From: euvre <93761161+euvre@users.noreply.github.com> Date: Mon, 27 Jul 2026 17:39:44 +0800 Subject: [PATCH] Fix agentbot embedded chat streaming envelope (#17420) --- internal/handler/bot.go | 68 +++++++++--------------------- internal/handler/bot_test.go | 40 ++++++++++++++---- internal/service/bot_completion.go | 12 ------ 3 files changed, 52 insertions(+), 68 deletions(-) diff --git a/internal/handler/bot.go b/internal/handler/bot.go index 5eeb07c969..cd61f07605 100644 --- a/internal/handler/bot.go +++ b/internal/handler/bot.go @@ -120,13 +120,18 @@ func (h *BotHandler) AgentbotInputs(c *gin.Context) { // AgentbotCompletion POST /api/v1/agentbots//completions // // Mirrors python bot_api.py:157 (canvas_service.completion wrapper). -// Streams SSE frames in the Python envelope shape. The URL-bound -// agent_id is authoritative — the body must NOT override it. +// The URL-bound agent_id is authoritative — the body must NOT +// override it. // -// Each canvas.RunEvent is re-formatted into the Python -// {code, message, data} envelope: a "message" event's Data string is -// treated as the assistant text, "message_end" terminates the -// stream with the python completion marker. +// The shared/embedded agent chat page (web/src/pages/agent/share) +// parses the stream with the same use-send-message.ts parser as the +// in-app agent chat, so frames must carry the agent canvas envelope +// — {event, message_id, session_id, task_id, created_at, data} — +// terminated by `data:[DONE]`. Forwarding raw canvas.RunEvent via +// WriteChatbotRunEvent keeps this endpoint byte-compatible with the +// python canvas_service.completion output (which yields every +// canvas event, including node_* telemetry used by the "Thinking" +// log panel). func (h *BotHandler) AgentbotCompletion(c *gin.Context) { user, code, msg := GetUser(c) if code != common.CodeSuccess { @@ -159,50 +164,17 @@ func (h *BotHandler) AgentbotCompletion(c *gin.Context) { c.Writer.Header().Set("Content-Type", "text/event-stream") c.Writer.Header().Set("Cache-Control", "no-cache") c.Writer.Header().Set("Connection", "keep-alive") + doneSent := false for ev := range events { - switch ev.Type { - case "message": - // The python iframe_completion wrapper flattens each - // message chunk into a {code:0, data:{answer:...}} - // frame. We forward the message Data as the assistant - // text payload so the iframe SDK's `data.answer` - // parser keeps working. The agentbot path uses - // WriteAgentbotFrame (a thin alias for - // WriteChatbotFrame) to keep the two paths visually - // distinct in the handler. - frame := service.ChatbotSSEFrame{ - Data: ev.Data, - Reference: map[string]any{}, - SessionID: ev.SessionID, - } - if err := service.WriteAgentbotFrame(c.Writer, frame); err != nil { - return - } - case "message_end", "done": - // Terminator events. message_end occasionally carries - // a final payload (e.g. structured output); forward - // it as a final answer frame when present, then close - // the stream with the standard python completion - // marker. A bare `done` event closes the stream - // directly. - if ev.Data != "" { - frame := service.ChatbotSSEFrame{ - Data: ev.Data, - Reference: map[string]any{}, - SessionID: ev.SessionID, - } - _ = service.WriteAgentbotFrame(c.Writer, frame) - } - _ = service.WriteDoneFrame(c.Writer) - return - default: - // Non-message events (node_started, node_finished, …) - // are silently dropped on the agentbot path. The - // python canvas_service.completion wrapper only - // forwards the assistant text frames, not the run - // telemetry; we mirror that behaviour so external - // widgets see the same wire shape. + if ev.Type == "done" { + doneSent = true } + if err := service.WriteChatbotRunEvent(c.Writer, ev); err != nil { + return + } + } + if !doneSent { + _ = service.WriteChatbotRunEvent(c.Writer, canvas.RunEvent{Type: "done"}) } } diff --git a/internal/handler/bot_test.go b/internal/handler/bot_test.go index 5fa5aefdd4..b472a583ca 100644 --- a/internal/handler/bot_test.go +++ b/internal/handler/bot_test.go @@ -397,6 +397,12 @@ func TestChatbotCompletion_SessionTenantMismatch(t *testing.T) { // ----- AgentbotCompletion tests (criteria 17, 18, 19, 20) ----- // TestAgentbotCompletion_StreamsSSE covers criterion 17. +// +// The shared/embedded agent chat page parses the stream with the +// same use-send-message.ts parser as the in-app agent chat, so the +// handler must forward the agent canvas envelope ({event, +// message_id, session_id, data}) — NOT the chatbot {code, +// data:{answer}} envelope — terminated by [DONE]. func TestAgentbotCompletion_StreamsSSE(t *testing.T) { stub := &stubBotService{ agentbotCompleteFn: func(ctx context.Context, tenantID, agentID string, req service.AgentbotCompletionRequest) (<-chan canvas.RunEvent, common.ErrorCode, error) { @@ -415,20 +421,34 @@ func TestAgentbotCompletion_StreamsSSE(t *testing.T) { t.Fatalf("status = %d, want 200", w.Code) } frames := parseBotSSEFrames(t, w.Body.Bytes()) - if len(frames) < 2 { - t.Fatalf("expected >= 2 frames, got %d", len(frames)) + if len(frames) < 3 { + t.Fatalf("expected >= 3 frames, got %d", len(frames)) } // The last frame must be [DONE]. if frames[len(frames)-1] != "[DONE]" { t.Errorf("last frame = %q, want [DONE]", frames[len(frames)-1]) } - // First frame is the data envelope. + // First frame is the agent canvas message envelope. var env map[string]any if err := json.Unmarshal([]byte(frames[0]), &env); err != nil { t.Fatalf("bad JSON: %v", err) } - if env["code"].(float64) != 0 { - t.Errorf("frame code = %v, want 0", env["code"]) + if env["event"] != "message" { + t.Errorf("frame event = %v, want message", env["event"]) + } + if env["session_id"] != "s1" { + t.Errorf("frame session_id = %v, want s1", env["session_id"]) + } + if env["data"] != "hello" { + t.Errorf("frame data = %v, want hello", env["data"]) + } + // Second frame forwards the message_end terminator event. + var endEnv map[string]any + if err := json.Unmarshal([]byte(frames[1]), &endEnv); err != nil { + t.Fatalf("bad JSON: %v", err) + } + if endEnv["event"] != "message_end" { + t.Errorf("frame event = %v, want message_end", endEnv["event"]) } } @@ -1075,12 +1095,16 @@ func parseBotSSEFrames(t *testing.T, body []byte) []string { if line == "" { continue } - if line == "data: [DONE]" { + // Both "data: ..." (chatbot frames) and "data:..." (agent + // canvas frames via writeSSEJSON) are valid SSE; the wire + // allows an optional single space after the colon. + payload := strings.TrimSpace(strings.TrimPrefix(line, "data:")) + if payload == "[DONE]" { frames = append(frames, "[DONE]") continue } - if strings.HasPrefix(line, "data: ") { - frames = append(frames, strings.TrimPrefix(line, "data: ")) + if strings.HasPrefix(line, "data:") { + frames = append(frames, payload) } else { t.Logf("ignoring unparseable SSE line: %q", line) } diff --git a/internal/service/bot_completion.go b/internal/service/bot_completion.go index 93db35c333..48d7713eb1 100644 --- a/internal/service/bot_completion.go +++ b/internal/service/bot_completion.go @@ -270,18 +270,6 @@ func writeSSEJSON(w http.ResponseWriter, payload map[string]any) error { return nil } -// AgentbotSSEFrame mirrors ChatbotSSEFrame for the agentbot -// completion path. The envelope shape is the same; the only -// difference is that the LLM call goes through the canvas runner -// (AgentService.RunAgent) instead of the legacy dialog async_chat. -type AgentbotSSEFrame = ChatbotSSEFrame - -// WriteAgentbotFrame is an alias for WriteChatbotFrame — both bot -// completion paths emit the same python wire shape. -func WriteAgentbotFrame(w http.ResponseWriter, f ChatbotSSEFrame) error { - return WriteChatbotFrame(w, f) -} - // ChatbotCompletion streams an SSE response for // /api/v1/chatbots//completions. //