fix(text): 文本生成 - 修复声明式协议流式请求与增量输出 (#558)

Original-Commit: b2fef297e8a50d519a0480d120a32eb81234a48f
fix(text): 文本生成 - 修复声明式协议流式请求与增量输出

backend/internal/app/provider_declarative_text_stream_test.go

@@ -0,0 +1,120 @@
+package app
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestDeclarativeTextStreaming(t *testing.T) {
+	t.Setenv("CANVAS_ALLOWED_PRIVATE_UPSTREAM_HOSTS", "127.0.0.1")
+	for _, wire := range []struct {
+		id, path, event, response string
+	}{
+		{"chat-completion", "/v1/chat/completions", "data: {\"choices\":[{\"delta\":{\"content\":\"hello\"}}]}\n\n", `{"choices":[{"message":{"content":"hello"}}]}`},
+		{"openai-response", "/v1/responses", "event: response.output_text.delta\ndata: {\"delta\":\"hello\"}\n\n", `{"output_text":"hello"}`},
+		{"claude-api", "/v1/messages", "event: content_block_delta\ndata: {\"type\":\"content_block_delta\",\"delta\":{\"type\":\"text_delta\",\"text\":\"hello\"}}\n\n", `{"content":[{"type":"text","text":"hello"}]}`},
+	} {
+		for _, scenario := range []string{"stream", "json_fallback", "stream_disabled", "http_524", "cancel"} {
+			t.Run(wire.id+"/"+scenario, func(t *testing.T) {
+				ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+				defer cancel()
+				var calls atomic.Int32
+				deltaReceived := make(chan struct{}, 1)
+				server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+					calls.Add(1)
+					if r.URL.Path != wire.path {
+						t.Errorf("path = %q, want %q", r.URL.Path, wire.path)
+					}
+					var body map[string]any
+					if err := json.NewDecoder(r.Body).Decode(&body); err != nil {
+						t.Errorf("decode request: %v", err)
+						return
+					}
+					if body["model"] != "test-model" {
+						t.Errorf("plugin model mapping lost: %#v", body)
+					}
+					stream, _ := body["stream"].(bool)
+					if stream != (scenario != "stream_disabled") {
+						t.Errorf("stream = %v for %s", body["stream"], scenario)
+					}
+					if stream {
+						if r.Header.Get("Accept") != "text/event-stream" {
+							t.Errorf("Accept = %q", r.Header.Get("Accept"))
+						}
+						if wire.id == "chat-completion" {
+							options, _ := body["stream_options"].(map[string]any)
+							if options["include_usage"] != true {
+								t.Error("stream usage was not requested")
+							}
+						}
+					}
+					if scenario == "http_524" {
+						w.WriteHeader(524)
+						fmt.Fprint(w, "upstream HTTP 524")
+						return
+					}
+					if scenario == "json_fallback" || scenario == "stream_disabled" {
+						w.Header().Set("Content-Type", "application/json")
+						fmt.Fprint(w, wire.response)
+						return
+					}
+					w.Header().Set("Content-Type", "text/event-stream; charset=utf-8")
+					fmt.Fprint(w, wire.event)
+					w.(http.Flusher).Flush()
+					// A delta must reach the caller before the upstream response ends.
+					select {
+					case <-deltaReceived:
+					case <-time.After(3 * time.Second):
+						t.Error("text was buffered until the response ended")
+					}
+					if scenario == "cancel" {
+						<-r.Context().Done()
+					}
+				}))
+				defer server.Close()
+				var deltas strings.Builder
+				input := canvasGenerationInput{
+					Mode: "text", Prompt: "hello", StreamText: scenario != "stream_disabled",
+					Config: providerConfig{BaseURL: server.URL, APIKey: "test-key", Model: "test-model", InterfaceType: wire.id},
+					OnTextDelta: func(delta string) {
+						deltas.WriteString(delta)
+						deltaReceived <- struct{}{}
+						if scenario == "cancel" {
+							cancel()
+						}
+					},
+				}
+				result, err := runTextTask(withProtocolRegistry(ctx, loadOfficialFallbackRegistry()), input)
+				if calls.Load() != 1 {
+					t.Fatalf("upstream calls = %d, want exactly one", calls.Load())
+				}
+				switch scenario {
+				case "http_524":
+					var upstream providerHTTPError
+					if !errors.As(err, &upstream) || upstream.StatusCode != 524 || !billingFailureUncertain(err) {
+						t.Fatalf("524 must remain an uncertain failure: %v", err)
+					}
+				case "cancel":
+					if !errors.Is(err, context.Canceled) {
+						t.Fatalf("expected cancellation, got %v", err)
+					}
+				default:
+					if err != nil || result["text"] != "hello" {
+						t.Fatalf("result = %#v, err = %v", result, err)
+					}
+					if scenario == "stream" && deltas.String() != "hello" {
+						t.Fatalf("deltas = %q", deltas.String())
+					}
+				}
+			})
+		}
+	}
+}

backend/internal/app/provider_protocol.go

@@ -74,11 +74,15 @@ func runProtocolAdapterTaskWithPolicy(ctx context.Context, input canvasGeneratio
 		if err != nil {
 			return nil, err
 		}
-		body, err := executeProtocolRequest(withProviderRequestKind(ctx, "create"), input.Config, spec)
+		body, streamedResult, err := executeProtocolCreateRequest(withProviderRequestKind(ctx, "create"), input, spec)
 		if err != nil {
 			return nil, err
 		}
-		created, err = adapter.ParseCreate(ctx, body)
+		if streamedResult != nil {
+			created = protocol.CreateResult{Status: protocol.StatusSucceeded, Result: streamedResult}
+		} else {
+			created, err = adapter.ParseCreate(ctx, body)
+		}
 		if err != nil {
 			return nil, err
 		}

backend/internal/app/provider_text.go

@@ -728,6 +728,46 @@ func runTextTask(ctx context.Context, input canvasGenerationInput) (map[string]i
 	return runLegacyTextTask(ctx, input)
 }
 
+// Known text protocols keep the plugin's request mapping and host transport,
+// while sharing the SSE parser used by text generation and Agent requests.
+func executeProtocolCreateRequest(ctx context.Context, input canvasGenerationInput, spec protocol.RequestSpec) ([]byte, *protocol.Result, error) {
+	wire := input.Config.InterfaceType
+	if wire == string(model.ChannelInterfaceOpenAIResponse) {
+		wire = "responses"
+	}
+	if input.Mode != "text" || !input.StreamText || (wire != "chat-completion" && wire != "responses" && wire != "claude-api") {
+		data, err := executeProtocolRequest(ctx, input.Config, spec)
+		return data, nil, err
+	}
+	body := protocolBodyObject(spec.Body)
+	if body == nil {
+		return nil, nil, errors.New("声明式流式文本请求体必须是 JSON 对象")
+	}
+	body["stream"] = true
+	if wire == "chat-completion" {
+		if err := ensureChatCompletionStreamUsage(body); err != nil {
+			return nil, nil, err
+		}
+	}
+	spec.Body = body
+	parser := newStreamingAgentParser(wire, input.OnTextDelta)
+	parser.emitReasoning = input.OnReasoningDelta
+	data, mimeType, err := executeProtocolBinaryRequestWithConsumer(ctx, input.Config, spec, parser.consume)
+	if err != nil || !strings.Contains(strings.ToLower(mimeType), "event-stream") {
+		return data, nil, err
+	}
+	parser.flush()
+	parsed, err := parser.result()
+	if err != nil {
+		return nil, nil, err
+	}
+	text := stringField(parsed, "text")
+	if text == "" {
+		return nil, nil, errors.New("流式文本接口没有返回内容")
+	}
+	return data, &protocol.Result{Text: text, Reasoning: stringField(parsed, "reasoning")}, nil
+}
+
 func runLegacyTextTask(ctx context.Context, input canvasGenerationInput) (map[string]interface{}, error) {
 	responseInput, err := textResponseInput(input)
 	if err != nil {