diff --git a/AGENTS.md b/AGENTS.md index 48621a5c5..ce4eeec6b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -90,7 +90,7 @@ - 后端 JSON 不得因为“方便”改用裸 `fetch`;若确需例外,必须在所属 service 写清响应合同、状态检查和错误转换。 - 资源 URL 使用 `resource:` 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//text-events` 配置 SSE 的 flush、长超时、禁缓冲和禁缓存,不把这套配置复制给所有接口。 +- Nginx/Caddy 只对文本任务和明确的系统模型事件流路径配置 flush、长超时、禁缓冲和禁缓存;自定义渠道依赖响应头动态关闭缓冲,不把这套配置复制给所有接口。 ## 9. 验证纪律 diff --git a/CHANGELOG.md b/CHANGELOG.md index 3948aab36..13a0fc29b 100644 --- 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 读取保留同源代理兼容。 diff --git a/README.md b/README.md index 7bbeb744d..cbddea4c1 100644 --- 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//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 } ``` diff --git a/backend/internal/handler/auth.go b/backend/internal/handler/auth.go index 18c73f804..bbc2ffb79 100644 --- 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) } diff --git a/backend/internal/handler/system_proxy_stream.go b/backend/internal/handler/system_proxy_stream.go new file mode 100644 index 000000000..e29169358 --- /dev/null +++ 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 + } + } +} diff --git a/backend/internal/handler/system_proxy_stream_test.go b/backend/internal/handler/system_proxy_stream_test.go new file mode 100644 index 000000000..e9c783875 --- /dev/null +++ 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()) + } +} diff --git a/nginx.conf b/nginx.conf index 0841e6152..653f2ab80 100644 --- 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; }