fix(sse): 系统模型 - 恢复 SSE 增量传输

集成 PR #257，保留原 PR 开发者为提交作者。
2026-08-19 20:32:46 +08:00
parent d2a35e10d8
commit cd42afdfdd
7 changed files with 233 additions and 14 deletions
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -90,7 +90,7 @@
 - 后端 JSON 不得因为“方便”改用裸 `fetch`；若确需例外，必须在所属 service 写清响应合同、状态检查和错误转换。
 - 资源 URL 使用 `resource:<id>` storage key；资源文件下载和 OSS URL 解析统一通过 `services/api/resources.ts`，不要在组件里直接拼接 `/resources/...`。
 - 文本 SSE 合同是 `GET /api/tasks/:id/text-events`，事件包含 `delta` 和终态 `terminal`；游标是单调递增事件 `id`，断线恢复使用 `Last-Event-ID` 或 `?after=`，不能把任务 ID 当游标。
- SSE 只对该路径关闭代理缓冲、缓存和 gzip；不得把长超时和 `proxy_buffering off` 复制到所有 `/api/` 请求。
+- SSE 只对文本任务事件路径和系统模型的 `responses`、`chat/completions`、Gemini `:streamGenerateContent` 路径关闭代理缓冲、缓存和 gzip；自定义渠道仅透传后端按响应类型设置的 `X-Accel-Buffering`。不得把长超时和 `proxy_buffering off` 复制到所有 `/api/` 请求。
 - blob 下载、图片/视频生成结果和 OSS 私有资源不得把敏感 URL、Cookie、API Key 写入日志、localStorage 或错误上报。

 ### 4.4 后端响应、权限和安全合同
@@ -169,7 +169,7 @@
 - 默认不启动 dev server；只有用户明确要求浏览器预览或联调时才启动，并先确认端口和数据目录。
 - Docker 部署只对外暴露网页容器 `3000`；后端 `8080` 留在 Compose 网络内。健康检查只能证明入口可用，不能替代 SSE/登录/生成路径验证。
 - 生产必须设置明确的 `CANVAS_CORS_ORIGINS`，保持公开注册关闭，HTTPS 终止后保留 Host、X-Forwarded-*，并限制数据库、备份、数据目录和 `.settings-key` 权限。
- Nginx/Caddy 只对 `/api/tasks/<id>/text-events` 配置 SSE 的 flush、长超时、禁缓冲和禁缓存，不把这套配置复制给所有接口。
+- Nginx/Caddy 只对文本任务和明确的系统模型事件流路径配置 flush、长超时、禁缓冲和禁缓存；自定义渠道依赖响应头动态关闭缓冲，不把这套配置复制给所有接口。

 ## 9. 验证纪律

--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,7 @@

 ## Unreleased

+- 修复系统文本模型中转先读取完整响应才返回的问题，恢复 OpenAI Responses、Chat Completions 与 Gemini 的 SSE 增量输出，并补齐多层 Nginx 的禁缓冲信号传递。
 - 火山方舟 Seedream 图片生成显式关闭“AI 生成”水印，并要求生成响应直接返回 Base64 后写入服务器或对象存储，避免为持久化结果再次请求方舟 CDN；中转服务忽略返回格式时仍保留 URL 下载兼容。
 - 项目开源协议由 AGPL-3.0 调整为 MIT，保留上游及项目贡献者版权声明。
 - 修复任务中心等资源入口仍由应用服务器转发对象存储媒体的问题：配置 CDN 加速域名后，登录资源、公开画布分享和管理员媒体预览统一重定向到 CDN；内部 Blob 读取保留同源代理兼容。
--- a/README.md
+++ b/README.md
@@ -124,9 +124,9 @@ curl -fsSL https://raw.githubusercontent.com/ddcat-ai/open-ai-canvas/main/script

 部署配置和 PostgreSQL 密码保存在 `/opt/open-ai-canvas/.env`，不要发送给他人，也不要删除 `backend-data`、`postgres-data` 和 `redis-data` 数据卷。数据卷持久化不等于备份，请定期备份 PostgreSQL 和上传文件。直接使用 IP 访问仅适合首次配置；公网长期使用必须绑定域名并配置 HTTPS。

-## 生产环境文本 SSE
+## 生产环境 SSE

-文本任务事件流是登录态接口 `GET /api/tasks/:id/text-events`。它只发送当前用户有权限访问的文本任务增量，响应类型为 `text/event-stream`；事件 `delta` 的 `id` 是单调递增的文本序号，`terminal` 表示任务已经成功、失败或取消。生产反向代理必须对这一条路径关闭响应缓冲和缓存，并允许长时间读取；不要把这些设置复制到所有 `/api/` 请求上。
+文本任务事件流是登录态接口 `GET /api/tasks/:id/text-events`。它只发送当前用户有权限访问的文本任务增量，响应类型为 `text/event-stream`；事件 `delta` 的 `id` 是单调递增的文本序号，`terminal` 表示任务已经成功、失败或取消。画布创作对话还会通过 `/api/ai/system/:channelId/responses`、`/chat/completions` 或 Gemini `:streamGenerateContent` 接口直接接收模型事件流。生产反向代理必须只对这些流式路径关闭响应缓冲和缓存，并允许长时间读取；不要把这些设置复制到所有 `/api/` 请求上。

 ### Nginx

@@ -141,15 +141,34 @@ location ~ ^/api/tasks/[^/]+/text-events$ {
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    proxy_set_header X-Forwarded-Proto $scheme;
    proxy_set_header X-Forwarded-Host $host;
+    proxy_set_header Connection "";
    proxy_buffering off;
    proxy_cache off;
+    proxy_pass_header X-Accel-Buffering;
+    gzip off;
+    proxy_read_timeout 3600s;
+    proxy_send_timeout 3600s;
+}
+
+location ~ ^/api/ai/system/[^/]+/(?:responses|chat/completions|models/[^/]+:streamGenerateContent)$ {
+    proxy_pass http://127.0.0.1:3000;
+    proxy_http_version 1.1;
+    proxy_set_header Host $host;
+    proxy_set_header X-Real-IP $remote_addr;
+    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+    proxy_set_header X-Forwarded-Proto $scheme;
+    proxy_set_header X-Forwarded-Host $host;
+    proxy_set_header Connection "";
+    proxy_buffering off;
+    proxy_cache off;
+    proxy_pass_header X-Accel-Buffering;
    gzip off;
    proxy_read_timeout 3600s;
    proxy_send_timeout 3600s;
 }
 ```

-项目镜像内的 `nginx.conf` 已包含同等规则，并只对 `/api/tasks/<id>/text-events` 关闭缓冲。外层 Nginx 和镜像内 Nginx 都存在时，两层都要保留该路径的流式设置；任一层重新缓冲都会让浏览器看起来直到任务结束才收到增量。
+项目镜像内的 `nginx.conf` 已包含同等规则，并只对文本任务和系统模型事件流关闭缓冲。自定义渠道共用 `/api/ai/custom`，由后端仅在上游实际返回 `text/event-stream` 时发送 `X-Accel-Buffering: no`；镜像内 Nginx 会继续向外层代理传递这个信号。外层 Nginx 和镜像内 Nginx 都存在时，任一层重新缓冲都会让浏览器看起来直到模型响应结束才收到增量。

 ### Caddy

@@ -167,6 +186,16 @@ canvas.example.com {
        header_down Cache-Control "no-cache, no-transform"
    }

+    @systemModelEvents path_regexp systemModelEvents ^/api/ai/system/[^/]+/(responses|chat/completions|models/[^/]+:streamGenerateContent)$
+    reverse_proxy @systemModelEvents 127.0.0.1:3000 {
+        flush_interval -1
+        transport http {
+            read_timeout 1h
+        }
+        header_up X-Forwarded-Proto {scheme}
+        header_down Cache-Control "no-cache, no-transform"
+    }
+
    reverse_proxy 127.0.0.1:3000
 }
 ```
--- a/backend/internal/handler/auth.go
+++ b/backend/internal/handler/auth.go
@@ -829,13 +829,27 @@ func proxySystemRequest(c *gin.Context, svc *service.Service, user *model.User,
 		status = model.ApiCallStatusFailed
 	}
 	responseLimit := policy.Request.SystemRelayResponseMB << 20
-	responseBody, readErr := io.ReadAll(io.LimitReader(resp.Body, responseLimit+1))
+	streamed := isSystemProxyEventStream(resp)
+	var responseBody []byte
+	var readErr error
+	if streamed {
+		responseBody, readErr = streamSystemProxyResponse(c, resp, responseLimit)
+	} else {
+		responseBody, readErr = io.ReadAll(io.LimitReader(resp.Body, responseLimit+1))
+	}
 	if readErr != nil {
 		status = model.ApiCallStatusFailed
 		errorText = readErr.Error()
-		_ = svc.MarkBillingUncertain(billingOrderID, "系统渠道响应读取失败，费用状态待核对")
-		logSystemProxyCall(svc, apiCallLog(user, channel, billingOrderID, capability, protocol, c.Request.Method, path, target, body, c.GetHeader("Content-Type"), status, statusCode, time.Since(startedAt), errorText, concurrencyLimit), nil)
-		fail(c, http.StatusBadGateway, errors.New("系统渠道响应读取失败"))
+		billingNote := "系统渠道响应读取失败，费用状态待核对"
+		if errors.Is(readErr, errSystemProxyResponseTooLarge) {
+			billingNote = "上游已响应但流式响应体超过限制，费用状态待核对"
+		}
+		_ = svc.MarkBillingUncertain(billingOrderID, billingNote)
+		logSystemProxyCall(svc, apiCallLog(user, channel, billingOrderID, capability, protocol, c.Request.Method, path, target, body, c.GetHeader("Content-Type"), status, statusCode, time.Since(startedAt), errorText, concurrencyLimit), responseBody)
+		// SSE 响应头已经发送，流中断后只能关闭连接；非流式响应仍返回结构化错误。
+		if !streamed {
+			fail(c, http.StatusBadGateway, errors.New("系统渠道响应读取失败"))
+		}
 		return
 	}
 	if int64(len(responseBody)) > responseLimit {
@@ -855,12 +869,10 @@ func proxySystemRequest(c *gin.Context, svc *service.Service, user *model.User,
 	} else {
 		_ = svc.RefundBilling(billingOrderID, "上游明确返回失败")
 	}
-	for _, key := range []string{"Content-Type", "Cache-Control", "Content-Disposition"} {
-		if value := resp.Header.Get(key); value != "" {
-			c.Header(key, value)
-		}
+	if streamed {
+		return
 	}
-	c.Header("X-Content-Type-Options", "nosniff")
+	copySystemProxyResponseHeaders(c, resp)
 	c.Data(resp.StatusCode, resp.Header.Get("Content-Type"), responseBody)
 }

--- a/backend/internal/handler/system_proxy_stream.go
+++ b/backend/internal/handler/system_proxy_stream.go
@@ -0,0 +1,70 @@
+package handler
+
+import (
+	"bytes"
+	"errors"
+	"io"
+	"mime"
+	"net/http"
+	"strings"
+
+	"github.com/gin-gonic/gin"
+)
+
+var errSystemProxyResponseTooLarge = errors.New("系统渠道流式响应超过配置上限")
+
+func isSystemProxyEventStream(resp *http.Response) bool {
+	mediaType, _, err := mime.ParseMediaType(resp.Header.Get("Content-Type"))
+	return err == nil && strings.EqualFold(mediaType, "text/event-stream")
+}
+
+func copySystemProxyResponseHeaders(c *gin.Context, resp *http.Response) {
+	for _, key := range []string{"Content-Type", "Cache-Control", "Content-Disposition"} {
+		if value := resp.Header.Get(key); value != "" {
+			c.Header(key, value)
+		}
+	}
+	c.Header("X-Content-Type-Options", "nosniff")
+}
+
+func streamSystemProxyResponse(c *gin.Context, resp *http.Response, responseLimit int64) ([]byte, error) {
+	copySystemProxyResponseHeaders(c, resp)
+	// 同时通知镜像内和可能存在的外层 Nginx，不得重新缓冲模型事件流。
+	c.Header("Cache-Control", "no-cache, no-store, no-transform")
+	c.Header("X-Accel-Buffering", "no")
+	c.Status(resp.StatusCode)
+	c.Writer.WriteHeaderNow()
+	c.Writer.Flush()
+
+	var captured bytes.Buffer
+	buffer := make([]byte, 32<<10)
+	remaining := responseLimit
+	for {
+		readLimit := int64(len(buffer))
+		if remaining < readLimit {
+			readLimit = remaining + 1
+		}
+		if readLimit < 1 {
+			readLimit = 1
+		}
+		read, readErr := resp.Body.Read(buffer[:int(readLimit)])
+		if read > 0 {
+			if int64(read) > remaining {
+				return captured.Bytes(), errSystemProxyResponseTooLarge
+			}
+			chunk := buffer[:read]
+			_, _ = captured.Write(chunk)
+			if _, writeErr := c.Writer.Write(chunk); writeErr != nil {
+				return captured.Bytes(), writeErr
+			}
+			c.Writer.Flush()
+			remaining -= int64(read)
+		}
+		if readErr != nil {
+			if errors.Is(readErr, io.EOF) {
+				return captured.Bytes(), nil
+			}
+			return captured.Bytes(), readErr
+		}
+	}
+}
--- a/backend/internal/handler/system_proxy_stream_test.go
+++ b/backend/internal/handler/system_proxy_stream_test.go
@@ -0,0 +1,85 @@
+package handler
+
+import (
+	"bufio"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/gin-gonic/gin"
+)
+
+func TestSystemProxyEventStreamFlushesBeforeUpstreamCompletes(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	release := make(chan struct{})
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream; charset=utf-8")
+		_, _ = io.WriteString(w, "data: first\n\n")
+		w.(http.Flusher).Flush()
+		<-release
+		_, _ = io.WriteString(w, "data: second\n\n")
+	}))
+	defer upstream.Close()
+
+	proxy := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		resp, err := http.Get(upstream.URL)
+		if err != nil {
+			http.Error(w, err.Error(), http.StatusBadGateway)
+			return
+		}
+		defer resp.Body.Close()
+		context, _ := gin.CreateTestContext(w)
+		context.Request = r
+		_, _ = streamSystemProxyResponse(context, resp, 1<<20)
+	}))
+	defer proxy.Close()
+
+	client := &http.Client{Timeout: 3 * time.Second}
+	response, err := client.Get(proxy.URL)
+	if err != nil {
+		close(release)
+		t.Fatal(err)
+	}
+	if response.Header.Get("X-Accel-Buffering") != "no" {
+		close(release)
+		_ = response.Body.Close()
+		t.Fatalf("X-Accel-Buffering = %q", response.Header.Get("X-Accel-Buffering"))
+	}
+	line, err := bufio.NewReader(response.Body).ReadString('\n')
+	if err != nil {
+		close(release)
+		_ = response.Body.Close()
+		t.Fatal(err)
+	}
+	if line != "data: first\n" {
+		close(release)
+		_ = response.Body.Close()
+		t.Fatalf("first streamed line = %q", line)
+	}
+	close(release)
+	_ = response.Body.Close()
+}
+
+func TestSystemProxyEventStreamEnforcesResponseLimit(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	response := httptest.NewRecorder()
+	context, _ := gin.CreateTestContext(response)
+	context.Request = httptest.NewRequest(http.MethodPost, "/api/ai/system/channel/responses", nil)
+	upstream := &http.Response{
+		StatusCode: http.StatusOK,
+		Header:     http.Header{"Content-Type": []string{"text/event-stream"}},
+		Body:       io.NopCloser(strings.NewReader("data: too large\n\n")),
+	}
+
+	captured, err := streamSystemProxyResponse(context, upstream, 4)
+	if !errors.Is(err, errSystemProxyResponseTooLarge) {
+		t.Fatalf("error = %v", err)
+	}
+	if len(captured) != 0 || response.Body.Len() != 0 {
+		t.Fatalf("captured = %q, response = %q", captured, response.Body.String())
+	}
+}
--- a/nginx.conf
+++ b/nginx.conf
@@ -44,8 +44,28 @@ server {
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $canvas_forwarded_proto;
        proxy_set_header X-Forwarded-Host $host;
+        proxy_set_header Connection "";
        proxy_buffering off;
        proxy_cache off;
+        proxy_pass_header X-Accel-Buffering;
+        gzip off;
+        proxy_read_timeout 3600s;
+        proxy_send_timeout 3600s;
+    }
+
+    # 系统渠道的 OpenAI / Gemini 文本接口直接向浏览器返回模型事件流。
+    location ~ ^/api/ai/system/[^/]+/(?:responses|chat/completions|models/[^/]+:streamGenerateContent)$ {
+        proxy_pass http://backend:8080;
+        proxy_http_version 1.1;
+        proxy_set_header Host $host;
+        proxy_set_header X-Real-IP $remote_addr;
+        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+        proxy_set_header X-Forwarded-Proto $canvas_forwarded_proto;
+        proxy_set_header X-Forwarded-Host $host;
+        proxy_set_header Connection "";
+        proxy_buffering off;
+        proxy_cache off;
+        proxy_pass_header X-Accel-Buffering;
        gzip off;
        proxy_read_timeout 3600s;
        proxy_send_timeout 3600s;
@@ -59,6 +79,8 @@ server {
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $canvas_forwarded_proto;
        proxy_set_header X-Forwarded-Host $host;
+        # 自定义渠道共用一个入口；只把后端明确标记的 SSE 禁缓冲信号传给外层代理。
+        proxy_pass_header X-Accel-Buffering;
        proxy_read_timeout 3600s;
    }