Fix agentbot embedded chat streaming envelope (#17420)

internal/handler/bot.go

@@ -120,13 +120,18 @@ func (h *BotHandler) AgentbotInputs(c *gin.Context) {
 // AgentbotCompletion POST /api/v1/agentbots/<agent_id>/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"})
 	}
 }
 

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)
 		}

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/<dialog_id>/completions.
 //