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

Original-Commit: b2fef297e8a50d519a0480d120a32eb81234a48f fix(text): 文本生成 - 修复声明式协议流式请求与增量输出
2026-09-20 23:18:26 +08:00
parent 19c9a207c9
commit 15946e5a86
5 changed files with 178 additions and 2 deletions
--- a/backend/internal/app/provider_declarative_text_stream_test.go
+++ b/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())
+					}
+				}
+			})
+		}
+	}
+}
--- a/backend/internal/app/provider_protocol.go
+++ b/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
 		}
--- a/backend/internal/app/provider_text.go
+++ b/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 {
--- a/docs/content/docs/backend/task-runtime-stability.mdx
+++ b/docs/content/docs/backend/task-runtime-stability.mdx
@@ -44,6 +44,12 @@ SSE 缓存 key 包含用户、任务和 `after` 游标，回源仍检查归属

 这是合并相同游标的轮询读取，不是事件驱动推送；不同游标和不同实例仍可能分别查询数据库。连接总量、客户端重连退避及任务删除后的缓存失效仍需后续治理。

+## 上游文本流
+
+创作台 / 画布文本任务开启流式输出且模型允许流式时，声明式 `chat-completion`、`openai-response` 和 `claude-api` 协议会发送 `stream=true`，通过宿主网络层实时解析 SSE 并发布文本增量。Chat Completions 同时请求 usage 终态块；普通 JSON 响应仍由插件声明的响应映射解析。插件的请求字段、路径、鉴权以及宿主的 SSRF、取消和审计边界保持有效，未知协议不套用这些 SSE 格式。
+
+流式传输可减少等待完整正文期间的网关超时，但不能保证上游及时发出首个数据块。收到 HTTP 524 或上下文取消时仍返回真实失败，不自动重发生成请求；计费结果不明确的订单保留待核对状态，缺少上游任务 ID 的历史请求无法靠本次修复自动找回。
+
 ## worker 与渠道槽位

 - worker 全局槽位由固定 45 分钟 TTL 改为 1 分钟租约，每 15 秒与数据库任务租约一起续期；数据库任务租约仍为 45 秒。
--- a/docs/content/docs/progress/pending-test.mdx
+++ b/docs/content/docs/progress/pending-test.mdx
@@ -38,6 +38,12 @@ description: 当前项目已实现功能的手工验证清单与未确认边界
 - [x] 58 项专项及相邻测试通过，覆盖同渠道同名不同价模型的选择、实际生成配置、报价目标、能力隔离，以及真实价格组件的明暗主题静态渲染；前端类型检查和生产构建通过。构建仍提示部分 chunk 超过 500 kB。
 - [ ] 云端部署前端后，在真实 MiniMax H3 列表逐项切换高速版、多图一致性、音画融合及满血等选项，确认选中项不跳回、价目与后台一致；调整参数后检查精确报价和实际任务使用的渠道与模型。本轮未读取云端配置或日志、未调用真实付费生成、未部署，静态渲染不代表云端端到端验收完成。

+## 声明式文本任务流式输出
+
+- 已修复：创作台 / 画布文本开启流式且模型支持时，声明式 Chat Completions、Responses、Claude 协议发送 `stream=true` 并实时发布文本增量；保留插件映射、普通 JSON 响应处理和宿主网络安全校验。Chat Completions 请求 usage 终态块，524 和取消不会触发自动重发。
+- [x] 三种协议的 15 组场景通过，覆盖响应结束前发布增量、普通 JSON 回退、关闭流式、524 单次调用和取消；声明式执行、Agent 流解析及计费核对相关回归通过。正式后端编译、部署后版本与 API 就绪检查通过。
+- [ ] 真实渠道核对首个文本增量、长回复、取消与最终用量；历史 524 任务没有上游任务 ID，仍需供应商核对执行和费用，不自动重新生成或改写账务。
+
 ## 多模态列表与分镜表合并回归

 - [x] 合入 PR #549 的图片 / 视频列表分析、动态表格、视频脚本转换和批量生成设置；参考图上限前后端统一为 10。保留本地主分支表格布局、参考图提及、删除列、结果定位、手工行增量同步及单行重新生成。