From b13392ef50d467b414a6934a2e36cb798a4e3cf8 Mon Sep 17 00:00:00 2001 From: wangbo Date: Wed, 5 Aug 2026 18:15:06 +0800 Subject: [PATCH] =?UTF-8?q?fix(media):=20=E7=BB=9F=E4=B8=80=E5=9B=BE?= =?UTF-8?q?=E7=89=87=E7=BB=93=E6=9E=9C=20URL=20=E5=8C=96=E5=B9=B6=E9=99=90?= =?UTF-8?q?=E5=88=B6=E5=90=8C=E6=AD=A5=20Base64?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 将上游 URL 直接持久化，内联媒体经对象存储后仅保留 URL 与内部定位元数据；异步轮询、任务详情和幂等重放统一使用零对象读取的 URL 投影，并增加 64KiB 响应门禁。 OpenAI 图片接口接受 url 与 b64_json，同步 Base64 限制为 20MiB 和每 Pod 2 并发；新增历史结果迁移清零门禁、结果指标和 API GOMEMLIMIT。 验证：API go test ./...、go vet、聚焦 race、pnpm openapi、pnpm lint/test/build、迁移安全检查与 docker compose config 均通过。 --- apps/api/cmd/backfill-binary-results/main.go | 38 +- apps/api/docs/swagger.json | 14 +- apps/api/docs/swagger.yaml | 13 +- apps/api/internal/httpapi/binary_results.go | 2 +- .../internal/httpapi/compat_protocol_test.go | 50 ++ apps/api/internal/httpapi/easyai_compat.go | 98 ++-- .../internal/httpapi/easyai_compat_test.go | 139 +++-- apps/api/internal/httpapi/handlers.go | 128 ++++- apps/api/internal/httpapi/openapi_models.go | 34 +- apps/api/internal/httpapi/public_errors.go | 5 + .../internal/httpapi/public_errors_test.go | 20 +- apps/api/internal/httpapi/response.go | 31 ++ .../httpapi/volces_compat_handlers.go | 4 +- apps/api/internal/publicerror/public_error.go | 12 + apps/api/internal/runner/binary_results.go | 498 +++++++++++++++++- .../internal/runner/binary_results_test.go | 176 +++++++ apps/api/internal/runner/service.go | 39 +- apps/api/internal/runner/upload.go | 118 +++-- apps/api/internal/runner/upload_test.go | 78 ++- apps/api/internal/securityevents/metrics.go | 33 ++ .../internal/securityevents/metrics_test.go | 9 + deploy/kubernetes/production/application.yaml | 4 + 22 files changed, 1367 insertions(+), 176 deletions(-) diff --git a/apps/api/cmd/backfill-binary-results/main.go b/apps/api/cmd/backfill-binary-results/main.go index cffadec..ef42fe5 100644 --- a/apps/api/cmd/backfill-binary-results/main.go +++ b/apps/api/cmd/backfill-binary-results/main.go @@ -14,6 +14,7 @@ import ( func main() { apply := flag.Bool("apply", false, "persist compacted results; default is dry-run") + requireClean := flag.Bool("require-clean", false, "exit non-zero unless a complete dry-run finds no results requiring URL migration") batchSize := flag.Int("batch-size", 100, "rows per batch, maximum 100") maxBatches := flag.Int("max-batches", 10, "maximum batches for one invocation") afterID := flag.String("after-id", "", "resume after this task UUID") @@ -29,6 +30,10 @@ func main() { logger.Error("invalid backfill bounds", "batchSize", *batchSize, "maxBatches", *maxBatches) os.Exit(1) } + if *apply && *requireClean { + logger.Error("--apply and --require-clean cannot be combined") + os.Exit(1) + } ctx := context.Background() db, err := store.Connect(ctx, cfg.DatabaseURL) @@ -42,8 +47,11 @@ func main() { cursor := *afterID scanned := 0 matched := 0 + blockingMatched := 0 updated := 0 expired := 0 + expiredLocalPlaceholders := 0 + complete := false for batch := 0; batch < *maxBatches; batch++ { items, err := db.ListTaskBinaryResultBackfillBatch(ctx, cursor, *batchSize) if err != nil { @@ -51,26 +59,30 @@ func main() { os.Exit(1) } if len(items) == 0 { + complete = true break } for _, item := range items { cursor = item.ID scanned++ - if !runner.TaskResultHasInlineBinary(item.Result) { + if !runner.TaskResultNeedsURLMigration(item.Result) { continue } matched++ + isExpired := item.FinishedAt.Before(time.Now().Add(-time.Duration(localResultTTLHours(cfg)) * time.Hour)) + hasLocalPlaceholder := runner.TaskResultHasLocalPlaceholder(item.Result) + if isExpired && hasLocalPlaceholder { + expiredLocalPlaceholders++ + if *apply { + logger.Warn("skip expired local result placeholder without overwriting stored result", "taskId", item.ID) + } + continue + } + blockingMatched++ if !*apply { continue } - isExpired := item.FinishedAt.Before(time.Now().Add(-time.Duration(localResultTTLHours(cfg)) * time.Hour)) - var persistent map[string]any - var changed bool - if isExpired { - persistent, changed, err = service.CompactExpiredTaskResultForStorage(ctx, item.ID, item.Result) - } else { - persistent, changed, err = service.MaterializeTaskResultForStorage(ctx, item.ID, item.Result) - } + persistent, changed, err := service.MigrateTaskResultToURLs(ctx, item.ID, item.Result) if err != nil { logger.Error("materialize historical binary result failed", "taskId", item.ID, "error", err) os.Exit(1) @@ -92,6 +104,7 @@ func main() { } } if len(items) < *batchSize { + complete = true break } } @@ -99,10 +112,17 @@ func main() { "apply", *apply, "scanned", scanned, "matched", matched, + "blockingMatched", blockingMatched, "updated", updated, "expired", expired, + "expiredLocalPlaceholders", expiredLocalPlaceholders, + "complete", complete, "resumeAfterId", cursor, ) + if *requireClean && (!complete || blockingMatched > 0) { + logger.Error("binary result URL migration gate failed", "complete", complete, "blockingMatched", blockingMatched, "expiredLocalPlaceholders", expiredLocalPlaceholders, "resumeAfterId", cursor) + os.Exit(1) + } } func localResultTTLHours(cfg config.Config) int { diff --git a/apps/api/docs/swagger.json b/apps/api/docs/swagger.json index d6ed1b3..d9712ac 100644 --- a/apps/api/docs/swagger.json +++ b/apps/api/docs/swagger.json @@ -5656,6 +5656,16 @@ "description": "OK", "schema": { "$ref": "#/definitions/httpapi.EasyAIGeneratedResponse" + }, + "headers": { + "Deprecation": { + "type": "string", + "description": "output_content 与 result 兼容别名的废弃标记" + }, + "X-Gateway-Response-Format": { + "type": "string", + "description": "实际结果格式，异步轮询固定为 url" + } } }, "404": { @@ -11167,6 +11177,7 @@ } }, "output_content": { + "description": "OutputContent is a one-release URL-only compatibility alias for Data.", "type": "array", "items": { "$ref": "#/definitions/httpapi.EasyAIMediaOutput" @@ -11177,6 +11188,7 @@ "example": "/api/v1/ai/result/9f4d8f3d-5f5f-4bb7-a4be-344a9f930e25" }, "result": { + "description": "Result is a one-release URL-only compatibility alias and is deprecated.", "type": "object", "additionalProperties": true }, @@ -11218,7 +11230,7 @@ "type": "string" }, "b64_json": { - "description": "B64JSON is retained when result media transfer is disabled.", + "description": "B64JSON is available only to bounded synchronous OpenAI image responses.", "type": "string" }, "content": { diff --git a/apps/api/docs/swagger.yaml b/apps/api/docs/swagger.yaml index 51df397..93c914b 100644 --- a/apps/api/docs/swagger.yaml +++ b/apps/api/docs/swagger.yaml @@ -409,6 +409,8 @@ definitions: type: string type: array output_content: + description: OutputContent is a one-release URL-only compatibility alias for + Data. items: $ref: '#/definitions/httpapi.EasyAIMediaOutput' type: array @@ -417,6 +419,7 @@ definitions: type: string result: additionalProperties: true + description: Result is a one-release URL-only compatibility alias and is deprecated. type: object status: enum: @@ -446,7 +449,8 @@ definitions: audio_url: type: string b64_json: - description: B64JSON is retained when result media transfer is disabled. + description: B64JSON is available only to bounded synchronous OpenAI image + responses. type: string content: type: string @@ -8093,6 +8097,13 @@ paths: responses: "200": description: OK + headers: + Deprecation: + description: output_content 与 result 兼容别名的废弃标记 + type: string + X-Gateway-Response-Format: + description: 实际结果格式，异步轮询固定为 url + type: string schema: $ref: '#/definitions/httpapi.EasyAIGeneratedResponse' "404": diff --git a/apps/api/internal/httpapi/binary_results.go b/apps/api/internal/httpapi/binary_results.go index 2455c3d..d290f1f 100644 --- a/apps/api/internal/httpapi/binary_results.go +++ b/apps/api/internal/httpapi/binary_results.go @@ -12,7 +12,7 @@ func (s *Server) hydrateTaskResult(ctx context.Context, task store.GatewayTask) if task.Status != "succeeded" || len(task.Result) == 0 { return task, nil } - result, err := s.runner.HydrateTaskResult(ctx, task.ID, task.Result) + result, err := s.runner.ProjectTaskResultURLs(ctx, task.ID, task.Result) if err != nil { return store.GatewayTask{}, err } diff --git a/apps/api/internal/httpapi/compat_protocol_test.go b/apps/api/internal/httpapi/compat_protocol_test.go index 6b7d20c..6bbeebe 100644 --- a/apps/api/internal/httpapi/compat_protocol_test.go +++ b/apps/api/internal/httpapi/compat_protocol_test.go @@ -2,6 +2,7 @@ package httpapi import ( "context" + "encoding/base64" "encoding/json" "errors" "net/http" @@ -14,6 +15,55 @@ import ( "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" ) +func TestSynchronousInlineResponseLimitAndCapacity(t *testing.T) { + request := map[string]any{"response_format": "b64_json"} + if !synchronousInlineResponseRequested("images.generations", request) { + t.Fatal("explicit b64_json image response was not detected") + } + encoded := strings.Repeat("A", base64.StdEncoding.EncodedLen(int(maxSynchronousInlineResponseBytes+1))) + if got := inlineMediaDecodedSize(map[string]any{"data": []any{map[string]any{"b64_json": encoded}}}); got <= maxSynchronousInlineResponseBytes { + t.Fatalf("decoded size=%d", got) + } + if got := base64DecodedSize(base64.StdEncoding.EncodeToString([]byte{1})); got != 1 { + t.Fatalf("padded Base64 decoded size=%d", got) + } + + releaseFirst, err := acquireSynchronousInlineResponseSlot(t.Context(), "images.generations", request) + if err != nil { + t.Fatal(err) + } + defer releaseFirst() + releaseSecond, err := acquireSynchronousInlineResponseSlot(t.Context(), "images.generations", request) + if err != nil { + t.Fatal(err) + } + defer releaseSecond() + + cancelled, cancel := context.WithCancel(t.Context()) + cancel() + _, err = acquireSynchronousInlineResponseSlot(cancelled, "images.generations", request) + if clients.ErrorCode(err) != "response_format_capacity_timeout" { + t.Fatalf("unexpected capacity error: %v", err) + } +} + +func TestValidateMediaResponseFormat(t *testing.T) { + for _, value := range []string{"url", "b64_json", " B64_JSON "} { + body := map[string]any{"response_format": value} + if err := validateMediaResponseFormat("images.generations", body); err != nil { + t.Fatalf("validate %q: %v", value, err) + } + } + for _, value := range []any{"base64", 1, map[string]any{"type": "url"}} { + if err := validateMediaResponseFormat("images.generations", map[string]any{"response_format": value}); err == nil { + t.Fatalf("accepted invalid response_format: %#v", value) + } + } + if err := validateMediaResponseFormat("chat.completions", map[string]any{"response_format": map[string]any{"type": "json_object"}}); err != nil { + t.Fatalf("image validation affected chat response_format: %v", err) + } +} + func TestProtocolAPIKeyStoreFailureIs503InsteadOf401(t *testing.T) { authenticator := auth.New("test-secret", "", "") authenticator.LocalAPIKeyVerifier = func(context.Context, string) (*auth.User, error) { diff --git a/apps/api/internal/httpapi/easyai_compat.go b/apps/api/internal/httpapi/easyai_compat.go index cc8fc20..2a0d3de 100644 --- a/apps/api/internal/httpapi/easyai_compat.go +++ b/apps/api/internal/httpapi/easyai_compat.go @@ -1,7 +1,6 @@ package httpapi import ( - "encoding/json" "net/http" "strings" @@ -109,32 +108,31 @@ func easyAIFileUploadResponse(upload map[string]any) map[string]any { func easyAITaskResultResponse(task store.GatewayTask) map[string]any { sourceResult := cloneEasyAIMap(task.Result) - cleanResult := cloneEasyAIMap(sourceResult) - delete(cleanResult, "raw") - delete(cleanResult, "raw_data") - normalizeEasyAIInlineMediaFields(cleanResult) - data := easyAITaskResultData(task, sourceResult) status := easyAITaskResultStatus(task.Status) if status == "failed" { data = []any{} } - cleanResult["data"] = data - cleanResult["output_content"] = data - - response := cloneEasyAIMap(cleanResult) - response["status"] = status - response["task_id"] = task.ID - response["taskId"] = task.ID - response["query_url"] = "/api/v1/ai/result/" + task.ID - response["created"] = task.CreatedAt.UnixMilli() - response["data"] = data - response["output_content"] = data - response["output"] = easyAIOutputURLs(data) - response["result"] = cleanResult + output := easyAIOutputURLs(data) + compatResult := map[string]any{ + "data": data, + "output": output, + "output_content": data, + } + response := map[string]any{ + "status": status, + "task_id": task.ID, + "taskId": task.ID, + "query_url": "/api/v1/ai/result/" + task.ID, + "created": task.CreatedAt.UnixMilli(), + "data": data, + "output": output, + "output_content": data, + "result": compatResult, + } upstreamTaskID := firstNonEmpty( - easyAIString(cleanResult["upstream_task_id"]), + easyAIString(sourceResult["upstream_task_id"]), task.RemoteTaskID, ) if upstreamTaskID != "" { @@ -144,12 +142,14 @@ func easyAITaskResultResponse(task store.GatewayTask) map[string]any { response["usage"] = task.Usage } - cancelState := runner.DescribeTaskCancellation(task) - response["cancellable"] = cancelState.Cancellable - response["submitted"] = cancelState.Submitted + if status != "success" { + cancelState := runner.DescribeTaskCancellation(task) + response["cancellable"] = cancelState.Cancellable + response["submitted"] = cancelState.Submitted + } message := firstNonEmpty( - easyAIString(cleanResult["message"]), + easyAIString(sourceResult["message"]), task.ErrorMessage, task.Error, task.Message, @@ -164,10 +164,10 @@ func easyAITaskResultResponse(task store.GatewayTask) map[string]any { message = "任务执行失败" } } - if message != "" { + if message != "" && status != "success" { response["message"] = message } - if code := firstNonEmpty(task.ErrorCode, easyAIString(cleanResult["code"])); code != "" || status == "failed" { + if code := firstNonEmpty(task.ErrorCode, easyAIString(sourceResult["code"])); code != "" || status == "failed" { standard := publicTaskError(task) if code != "" && task.ErrorCode == "" { standard = publicerror.WithIDs(publicerror.FromFields(code, message, 0, false), task.RequestID, task.ID) @@ -268,19 +268,24 @@ func normalizeEasyAIOutputItems(task store.GatewayTask, items []any) []any { if len(output) == 0 { continue } - delete(output, "raw_data") normalizeEasyAIInlineMediaFields(output) mediaURL := easyAIOutputURL(output) - if mediaURL != "" { - output["url"] = mediaURL + if mediaURL == "" { + continue } - if strings.TrimSpace(easyAIString(output["type"])) == "" { - if outputType := easyAIOutputType(task, output, mediaURL); outputType != "" { - output["type"] = outputType + lightweight := map[string]any{"url": mediaURL} + for _, key := range []string{"type", "mime_type", "width", "height", "duration", "format", "seed", "revised_prompt"} { + if value, exists := output[key]; exists && value != nil { + lightweight[key] = value } } - normalized = append(normalized, output) + if strings.TrimSpace(easyAIString(lightweight["type"])) == "" { + if outputType := easyAIOutputType(task, output, mediaURL); outputType != "" { + lightweight["type"] = outputType + } + } + normalized = append(normalized, lightweight) } return normalized } @@ -492,13 +497,26 @@ func cloneEasyAIMap(source map[string]any) map[string]any { if len(source) == 0 { return map[string]any{} } - raw, err := json.Marshal(source) - if err != nil { - return map[string]any{} - } - var result map[string]any - if err := json.Unmarshal(raw, &result); err != nil { - return map[string]any{} + result := make(map[string]any, len(source)) + for key, value := range source { + result[key] = cloneEasyAIValue(value) } return result } + +func cloneEasyAIValue(value any) any { + switch typed := value.(type) { + case map[string]any: + return cloneEasyAIMap(typed) + case []any: + next := make([]any, len(typed)) + for index, item := range typed { + next[index] = cloneEasyAIValue(item) + } + return next + case []string: + return append([]string(nil), typed...) + default: + return value + } +} diff --git a/apps/api/internal/httpapi/easyai_compat_test.go b/apps/api/internal/httpapi/easyai_compat_test.go index 4555c4f..bab7adf 100644 --- a/apps/api/internal/httpapi/easyai_compat_test.go +++ b/apps/api/internal/httpapi/easyai_compat_test.go @@ -4,6 +4,9 @@ import ( "encoding/json" "net/http" "net/http/httptest" + "runtime" + "strings" + "sync" "testing" "time" @@ -11,6 +14,98 @@ import ( "github.com/easyai/easyai-ai-gateway/apps/api/internal/store" ) +func TestEasyAITaskResultURLResponseIsCompact(t *testing.T) { + task := store.GatewayTask{ + ID: "image-url-compact", Kind: "images.generations", Status: "succeeded", + Result: map[string]any{"data": []any{map[string]any{ + "type": "image", "url": "https://cdn.example/result.png", "mime_type": "image/png", + "provider_payload": strings.Repeat("x", 4096), "upload": map[string]any{"objectKey": "secret"}, + }}}, + } + runtime.GC() + var before runtime.MemStats + runtime.ReadMemStats(&before) + response := easyAITaskResultResponse(task) + payload, err := json.Marshal(response) + if err != nil { + t.Fatal(err) + } + var after runtime.MemStats + runtime.ReadMemStats(&after) + if len(payload) >= maxAsyncMediaResultResponseBytes { + t.Fatalf("URL response bytes=%d", len(payload)) + } + if allocated := after.TotalAlloc - before.TotalAlloc; allocated >= 1<<20 { + t.Fatalf("URL response allocated %d bytes", allocated) + } + if !json.Valid(payload) { + t.Fatal("URL response is not valid JSON") + } + if strings.Contains(string(payload), "provider_payload") || strings.Contains(string(payload), "objectKey") { + t.Fatalf("internal provider or object-storage fields leaked: %s", payload) + } +} + +func TestWriteAsyncMediaResultJSONSetsURLAndDeprecationHeaders(t *testing.T) { + recorder := httptest.NewRecorder() + writeAsyncMediaResultJSON(recorder, map[string]any{ + "status": "success", "data": []any{map[string]any{"url": "https://cdn.example/result.png"}}, + }, nil) + if recorder.Code != http.StatusOK { + t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String()) + } + if recorder.Header().Get("X-Gateway-Response-Format") != "url" || recorder.Header().Get("Deprecation") != "true" { + t.Fatalf("unexpected headers: %#v", recorder.Header()) + } +} + +func TestWriteAsyncMediaResultJSONRejectsOversizedPayload(t *testing.T) { + recorder := httptest.NewRecorder() + writeAsyncMediaResultJSON(recorder, map[string]any{"data": strings.Repeat("x", maxAsyncMediaResultResponseBytes)}, nil) + if recorder.Code != http.StatusInternalServerError || !strings.Contains(recorder.Body.String(), "result_response_too_large") { + t.Fatalf("status=%d body=%s", recorder.Code, recorder.Body.String()) + } +} + +func TestConcurrentURLResultResponsesRemainBounded(t *testing.T) { + task := store.GatewayTask{ + ID: "image-url-load", Kind: "images.generations", Status: "succeeded", + Result: map[string]any{"data": []any{map[string]any{ + "type": "image", "url": "https://cdn.example/result.png", "mime_type": "image/png", + }}}, + } + runtime.GC() + var before runtime.MemStats + runtime.ReadMemStats(&before) + + var wait sync.WaitGroup + errors := make(chan string, 1000) + for index := 0; index < 1000; index++ { + wait.Add(1) + go func() { + defer wait.Done() + payload, err := json.Marshal(easyAITaskResultResponse(task)) + if err != nil { + errors <- err.Error() + return + } + if len(payload) >= maxAsyncMediaResultResponseBytes { + errors <- "response exceeded URL-only size gate" + } + }() + } + wait.Wait() + close(errors) + for message := range errors { + t.Fatal(message) + } + var after runtime.MemStats + runtime.ReadMemStats(&after) + if allocated := after.TotalAlloc - before.TotalAlloc; allocated >= 256<<20 { + t.Fatalf("1000 concurrent URL responses allocated %d bytes", allocated) + } +} + func TestEasyAITaskAcceptedResponseKeepsGatewayFieldsAndAddsLegacyFields(t *testing.T) { task := store.GatewayTask{ ID: "task-accepted-1", @@ -119,8 +214,8 @@ func TestEasyAITaskResultResponseNormalizesMediaOutputs(t *testing.T) { t.Fatalf("output_content is not synchronized: %+v", got) } if item.wantCount == 0 { - if got["voice_id"] != "voice-test-1" { - t.Fatalf("voice clone fields were lost: %+v", got) + if got["voice_id"] != nil || got["cloned_voice"] != nil { + t.Fatalf("provider-specific fields leaked: %+v", got) } return } @@ -139,7 +234,7 @@ func TestEasyAITaskResultResponseNormalizesMediaOutputs(t *testing.T) { } } -func TestEasyAITaskResultResponsePreservesBase64ImageOutput(t *testing.T) { +func TestEasyAITaskResultResponseDropsBase64ImageOutput(t *testing.T) { base64Payload := "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAusB9Y9ZlB8AAAAASUVORK5CYII=" task := store.GatewayTask{ ID: "image-base64-1", Kind: "images.generations", Status: "succeeded", @@ -151,20 +246,8 @@ func TestEasyAITaskResultResponsePreservesBase64ImageOutput(t *testing.T) { got := easyAITaskResultResponse(task) data, _ := got["data"].([]any) - if len(data) != 1 { - t.Fatalf("unexpected base64 output count: %+v", got) - } - output, _ := data[0].(map[string]any) - if output["type"] != "image" || output["b64_json"] != base64Payload || output["mime_type"] != "image/png" { - t.Fatalf("base64 image output was not preserved: %+v", output) - } - if _, exposedAsURL := output["url"]; exposedAsURL { - t.Fatalf("base64 image must not be exposed as a URL: %+v", output) - } - outputContent, _ := got["output_content"].([]any) - content, _ := outputContent[0].(map[string]any) - if content["b64_json"] != base64Payload { - t.Fatalf("output_content lost base64 image: %+v", outputContent) + if len(data) != 0 { + t.Fatalf("Base64 output leaked into URL-only response: %+v", got) } urls, _ := got["output"].([]string) if len(urls) != 0 { @@ -172,7 +255,7 @@ func TestEasyAITaskResultResponsePreservesBase64ImageOutput(t *testing.T) { } } -func TestEasyAITaskResultResponseMovesImageDataURLToB64JSON(t *testing.T) { +func TestEasyAITaskResultResponseDropsImageDataURL(t *testing.T) { base64Payload := "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAusB9Y9ZlB8AAAAASUVORK5CYII=" task := store.GatewayTask{ ID: "image-data-url-1", Kind: "images.generations", Status: "succeeded", @@ -183,12 +266,8 @@ func TestEasyAITaskResultResponseMovesImageDataURLToB64JSON(t *testing.T) { got := easyAITaskResultResponse(task) data, _ := got["data"].([]any) - output, _ := data[0].(map[string]any) - if output["b64_json"] != base64Payload || output["mime_type"] != "image/png" || output["type"] != "image" { - t.Fatalf("image data URL was not normalized: %+v", output) - } - if _, exists := output["url"]; exists { - t.Fatalf("image data URL should move out of the URL field: %+v", output) + if len(data) != 0 { + t.Fatalf("data URL leaked into URL-only response: %+v", got) } } @@ -220,7 +299,7 @@ func TestEasyAITaskResultResponseForwardsSafeUpstreamParameterMessage(t *testing } } -func TestEasyAITaskResultResponsePreservesAudioDataURLAsContent(t *testing.T) { +func TestEasyAITaskResultResponseDropsAudioDataURL(t *testing.T) { base64Payload := "SUQzBAAAAAAAI1RTU0UAAAAPAAADTGF2ZjYwLjMuMTAwAAAAAAAAAAAAAAD/" task := store.GatewayTask{ ID: "audio-data-url-1", Kind: "speech.generations", Status: "succeeded", @@ -231,14 +310,8 @@ func TestEasyAITaskResultResponsePreservesAudioDataURLAsContent(t *testing.T) { got := easyAITaskResultResponse(task) data, _ := got["data"].([]any) - output, _ := data[0].(map[string]any) - if output["content"] != "data:audio/mpeg;base64,"+base64Payload || - output["mime_type"] != "audio/mpeg" || - output["type"] != "audio" { - t.Fatalf("audio data URL was not normalized to inline content: %+v", output) - } - if _, exists := output["audio_url"]; exists { - t.Fatalf("audio data URL should move out of the URL field: %+v", output) + if len(data) != 0 { + t.Fatalf("audio data URL leaked into URL-only response: %+v", got) } urls, _ := got["output"].([]string) if len(urls) != 0 { diff --git a/apps/api/internal/httpapi/handlers.go b/apps/api/internal/httpapi/handlers.go index 347451c..54e9583 100644 --- a/apps/api/internal/httpapi/handlers.go +++ b/apps/api/internal/httpapi/handlers.go @@ -2,6 +2,7 @@ package httpapi import ( "context" + "encoding/base64" "encoding/json" "errors" "fmt" @@ -1166,6 +1167,10 @@ func (s *Server) createTask(kind string, compatible bool) http.Handler { return } } + if err := validateMediaResponseFormat(kind, body); err != nil { + writeTaskError(http.StatusBadRequest, err.Error(), map[string]any{"param": "response_format"}, "invalid_parameter") + return + } requestedModel := requestModelName(body) model := canonicalTaskModelName(kind, requestedModel) if model == "" { @@ -1550,6 +1555,12 @@ func writeProtocolCompatibleTaskResponse(runCtx context.Context, w http.Response return } + releaseInlineResponse, limitErr := acquireSynchronousInlineResponseSlot(runCtx, kind, task.Request) + if limitErr != nil { + writeProtocolError(w, targetProtocol, statusFromRunError(limitErr), runErrorMessage(limitErr), runErrorDetails(limitErr), runErrorCode(limitErr)) + return + } + defer releaseInlineResponse() result, runErr := executor.Execute(runCtx, task, user) if runErr != nil { if !requestStillConnected(r) { @@ -1566,13 +1577,107 @@ func writeProtocolCompatibleTaskResponse(runCtx context.Context, w http.Response if !requestStillConnected(r) { return } - if wireResponseMatches(result.Wire, targetProtocol) { + if synchronousInlineResponseRequested(kind, task.Request) { + rawBytes := inlineMediaDecodedSize(result.Output) + if rawBytes > maxSynchronousInlineResponseBytes { + writeProtocolError(w, targetProtocol, http.StatusRequestEntityTooLarge, "synchronous Base64 response exceeds the 20 MiB limit", map[string]any{ + "task_id": result.Task.ID, + "query_url": "/api/v1/ai/result/" + result.Task.ID, + "limit_bytes": maxSynchronousInlineResponseBytes, + "actual_bytes": rawBytes, + "response_format": "url", + }, "response_format_too_large") + return + } + if rawBytes > 0 { + w.Header().Set("X-Gateway-Response-Format", "b64_json") + } else { + w.Header().Set("X-Gateway-Response-Format", "url") + } + } else if mediaResultKind(kind) { + w.Header().Set("X-Gateway-Response-Format", "url") + } + if !mediaResultKind(kind) && wireResponseMatches(result.Wire, targetProtocol) { writeWireResponse(w, result.Wire) return } writeJSON(w, http.StatusOK, easyAISynchronousTaskResponse(result.Task, result.Output)) } +const maxSynchronousInlineResponseBytes = runner.MaxSynchronousInlineResponseBytes + +var synchronousInlineResponseSlots = make(chan struct{}, 2) + +func acquireSynchronousInlineResponseSlot(ctx context.Context, kind string, request map[string]any) (func(), error) { + if !synchronousInlineResponseRequested(kind, request) { + return func() {}, nil + } + select { + case synchronousInlineResponseSlots <- struct{}{}: + return func() { <-synchronousInlineResponseSlots }, nil + case <-ctx.Done(): + return func() {}, &clients.ClientError{Code: "response_format_capacity_timeout", Message: "synchronous Base64 response capacity wait timed out", StatusCode: http.StatusServiceUnavailable, Retryable: true} + } +} + +func synchronousInlineResponseRequested(kind string, request map[string]any) bool { + if !mediaResultKind(kind) { + return false + } + value, _ := request["response_format"].(string) + return strings.EqualFold(strings.TrimSpace(value), "b64_json") +} + +func mediaResultKind(kind string) bool { + return strings.HasPrefix(kind, "images.") || strings.HasPrefix(kind, "videos.") || + kind == "song.generations" || kind == "music.generations" || kind == "speech.generations" +} + +func inlineMediaDecodedSize(value any) int64 { + switch typed := value.(type) { + case map[string]any: + var total int64 + for key, item := range typed { + normalized := strings.ToLower(strings.ReplaceAll(strings.TrimSpace(key), "-", "_")) + if raw, ok := item.(string); ok && (normalized == "b64_json" || strings.Contains(normalized, "base64")) { + total += base64DecodedSize(raw) + continue + } + total += inlineMediaDecodedSize(item) + } + return total + case []any: + var total int64 + for _, item := range typed { + total += inlineMediaDecodedSize(item) + } + return total + case string: + raw := strings.TrimSpace(typed) + if strings.HasPrefix(strings.ToLower(raw), "data:") { + if comma := strings.IndexByte(raw, ','); comma >= 0 { + return base64DecodedSize(raw[comma+1:]) + } + } + } + return 0 +} + +func base64DecodedSize(value string) int64 { + value = strings.TrimSpace(value) + if value == "" { + return 0 + } + padding := 0 + if strings.HasSuffix(value, "=") { + padding++ + } + if strings.HasSuffix(value, "==") { + padding++ + } + return int64(base64.StdEncoding.DecodedLen(len(value)) - padding) +} + func gatewayAPIV1Request(r *http.Request) bool { return r != nil && strings.HasPrefix(r.URL.Path, "/api/v1/") } @@ -1583,6 +1688,27 @@ func streamIncludeUsage(body map[string]any) bool { return includeUsage } +func validateMediaResponseFormat(kind string, body map[string]any) error { + if !strings.HasPrefix(kind, "images.") { + return nil + } + value, exists := body["response_format"] + if !exists || value == nil { + return nil + } + format, ok := value.(string) + if !ok { + return errors.New("response_format must be url or b64_json") + } + switch strings.ToLower(strings.TrimSpace(format)) { + case "url", "b64_json": + body["response_format"] = strings.ToLower(strings.TrimSpace(format)) + return nil + default: + return errors.New("response_format must be url or b64_json") + } +} + func asyncRequest(r *http.Request) bool { value := strings.TrimSpace(strings.ToLower(r.Header.Get("x-async"))) return value == "1" || value == "true" || value == "yes" || value == "on" diff --git a/apps/api/internal/httpapi/openapi_models.go b/apps/api/internal/httpapi/openapi_models.go index c612eae..4f333a1 100644 --- a/apps/api/internal/httpapi/openapi_models.go +++ b/apps/api/internal/httpapi/openapi_models.go @@ -295,27 +295,29 @@ type TaskNextLinks struct { } type EasyAIGeneratedResponse struct { - Status string `json:"status" example:"success" enums:"submitted,process,success,failed"` - TaskID string `json:"task_id" example:"9f4d8f3d-5f5f-4bb7-a4be-344a9f930e25"` - CamelTaskID string `json:"taskId,omitempty" example:"9f4d8f3d-5f5f-4bb7-a4be-344a9f930e25"` - UpstreamTaskID string `json:"upstream_task_id,omitempty" example:"provider-task-123"` - QueryURL string `json:"query_url,omitempty" example:"/api/v1/ai/result/9f4d8f3d-5f5f-4bb7-a4be-344a9f930e25"` - Created int64 `json:"created" example:"1784772000000"` - Message string `json:"message,omitempty"` - Code string `json:"code,omitempty"` - Data []EasyAIMediaOutput `json:"data"` - Output []string `json:"output"` - OutputContent []EasyAIMediaOutput `json:"output_content"` - Cancellable bool `json:"cancellable"` - Submitted bool `json:"submitted"` - Result map[string]interface{} `json:"result,omitempty"` - Usage map[string]interface{} `json:"usage,omitempty"` + Status string `json:"status" example:"success" enums:"submitted,process,success,failed"` + TaskID string `json:"task_id" example:"9f4d8f3d-5f5f-4bb7-a4be-344a9f930e25"` + CamelTaskID string `json:"taskId,omitempty" example:"9f4d8f3d-5f5f-4bb7-a4be-344a9f930e25"` + UpstreamTaskID string `json:"upstream_task_id,omitempty" example:"provider-task-123"` + QueryURL string `json:"query_url,omitempty" example:"/api/v1/ai/result/9f4d8f3d-5f5f-4bb7-a4be-344a9f930e25"` + Created int64 `json:"created" example:"1784772000000"` + Message string `json:"message,omitempty"` + Code string `json:"code,omitempty"` + Data []EasyAIMediaOutput `json:"data"` + Output []string `json:"output"` + // OutputContent is a one-release URL-only compatibility alias for Data. + OutputContent []EasyAIMediaOutput `json:"output_content"` + Cancellable bool `json:"cancellable"` + Submitted bool `json:"submitted"` + // Result is a one-release URL-only compatibility alias and is deprecated. + Result map[string]interface{} `json:"result,omitempty"` + Usage map[string]interface{} `json:"usage,omitempty"` } type EasyAIMediaOutput struct { Type string `json:"type,omitempty" example:"video" enums:"image,video,audio,file,text"` URL string `json:"url,omitempty" example:"https://cdn.example.com/output.mp4"` - // B64JSON is retained when result media transfer is disabled. + // B64JSON is available only to bounded synchronous OpenAI image responses. B64JSON string `json:"b64_json,omitempty"` ImageURL string `json:"image_url,omitempty"` VideoURL string `json:"video_url,omitempty"` diff --git a/apps/api/internal/httpapi/public_errors.go b/apps/api/internal/httpapi/public_errors.go index fe81a92..54d3a49 100644 --- a/apps/api/internal/httpapi/public_errors.go +++ b/apps/api/internal/httpapi/public_errors.go @@ -22,6 +22,11 @@ func publicGatewayTask(task store.GatewayTask) store.GatewayTask { task.Attempts = append([]store.TaskAttempt(nil), task.Attempts...) for index := range task.Attempts { attempt := &task.Attempts[index] + // User-facing task details expose the canonical task result. Attempt + // snapshots can duplicate large input media or retain raw provider output, + // so keep them available only on dedicated administrative surfaces. + attempt.RequestSnapshot = nil + attempt.ResponseSnapshot = nil if attempt.ErrorCode == "" && attempt.ErrorMessage == "" { continue } diff --git a/apps/api/internal/httpapi/public_errors_test.go b/apps/api/internal/httpapi/public_errors_test.go index 3512788..994ebf1 100644 --- a/apps/api/internal/httpapi/public_errors_test.go +++ b/apps/api/internal/httpapi/public_errors_test.go @@ -66,7 +66,7 @@ func TestPublicHTTPErrorIncludesSafeUpstreamParameterMessage(t *testing.T) { } } -func TestPublicGatewayTaskSanitizesCopyAndKeepsRawAuditFields(t *testing.T) { +func TestPublicGatewayTaskSanitizesCopyAndKeepsRawAuditFieldsInternal(t *testing.T) { rawMessage := `404 page not found: {"privateProject":"secret"}` task := store.GatewayTask{ ID: "task-1", @@ -74,11 +74,13 @@ func TestPublicGatewayTaskSanitizesCopyAndKeepsRawAuditFields(t *testing.T) { ErrorCode: "http_404", ErrorMessage: rawMessage, Attempts: []store.TaskAttempt{{ - AttemptNo: 1, - Status: "failed", - StatusCode: http.StatusNotFound, - ErrorCode: "http_404", - ErrorMessage: rawMessage, + AttemptNo: 1, + Status: "failed", + StatusCode: http.StatusNotFound, + ErrorCode: "http_404", + ErrorMessage: rawMessage, + RequestSnapshot: map[string]any{"image_base64": "private-input"}, + ResponseSnapshot: map[string]any{"provider_payload": "private-output"}, }}, } @@ -89,9 +91,15 @@ func TestPublicGatewayTaskSanitizesCopyAndKeepsRawAuditFields(t *testing.T) { if strings.Contains(public.ErrorMessage, "secret") || public.Attempts[0].ErrorMessage == rawMessage { t.Fatalf("public task leaked upstream details: %+v", public) } + if public.Attempts[0].RequestSnapshot != nil || public.Attempts[0].ResponseSnapshot != nil { + t.Fatalf("public task leaked attempt snapshots: %+v", public.Attempts[0]) + } if task.ErrorCode != "http_404" || task.ErrorMessage != rawMessage || task.Attempts[0].ErrorCode != "http_404" || task.Attempts[0].ErrorMessage != rawMessage { t.Fatalf("public conversion mutated raw audit fields: %+v", task) } + if task.Attempts[0].RequestSnapshot == nil || task.Attempts[0].ResponseSnapshot == nil { + t.Fatalf("public conversion mutated internal snapshots: %+v", task.Attempts[0]) + } } func TestPublicGatewayTaskForwardsSafeUpstreamParameterMessage(t *testing.T) { diff --git a/apps/api/internal/httpapi/response.go b/apps/api/internal/httpapi/response.go index 8ac7c99..66a35f5 100644 --- a/apps/api/internal/httpapi/response.go +++ b/apps/api/internal/httpapi/response.go @@ -9,12 +9,43 @@ import ( "github.com/easyai/easyai-ai-gateway/apps/api/internal/publicerror" ) +const maxAsyncMediaResultResponseBytes = 64 << 10 + func writeJSON(w http.ResponseWriter, status int, value any) { w.Header().Set("Content-Type", "application/json; charset=utf-8") w.WriteHeader(status) _ = json.NewEncoder(w).Encode(value) } +func writeAsyncMediaResultJSON(w http.ResponseWriter, value any, observer interface { + ObserveResultDelivery(string, int64) +}) { + payload, err := json.Marshal(value) + if err != nil { + writeError(w, http.StatusInternalServerError, "encode task result failed", "internal_error") + return + } + if len(payload) > maxAsyncMediaResultResponseBytes { + if observer != nil { + observer.ObserveResultDelivery("poll_oversized", int64(len(payload))) + } + writeErrorWithDetails(w, http.StatusInternalServerError, "task result response exceeds the URL-only size limit", map[string]any{ + "limit_bytes": maxAsyncMediaResultResponseBytes, + "actual_bytes": len(payload), + }, "result_response_too_large") + return + } + w.Header().Set("Content-Type", "application/json; charset=utf-8") + w.Header().Set("X-Gateway-Response-Format", "url") + w.Header().Set("Deprecation", "true") + w.Header().Set("X-Gateway-Deprecated-Fields", "output_content,result") + w.WriteHeader(http.StatusOK) + _, _ = w.Write(append(payload, '\n')) + if observer != nil { + observer.ObserveResultDelivery("poll_success", int64(len(payload)+1)) + } +} + func writeError(w http.ResponseWriter, status int, message string, codes ...string) { writeErrorWithDetails(w, status, message, nil, codes...) } diff --git a/apps/api/internal/httpapi/volces_compat_handlers.go b/apps/api/internal/httpapi/volces_compat_handlers.go index 6cea4de..42094b1 100644 --- a/apps/api/internal/httpapi/volces_compat_handlers.go +++ b/apps/api/internal/httpapi/volces_compat_handlers.go @@ -185,6 +185,8 @@ func (s *Server) createLegacyVolcesVideoGeneration(w http.ResponseWriter, r *htt // @Security BearerAuth // @Param taskID path string true "任务 ID" // @Success 200 {object} EasyAIGeneratedResponse +// @Header 200 {string} X-Gateway-Response-Format "实际结果格式，异步轮询固定为 url" +// @Header 200 {string} Deprecation "output_content 与 result 兼容别名的废弃标记" // @Failure 404 {object} EasyAIGeneratedResponse // @Failure 410 {object} EasyAIGeneratedResponse // @Failure 503 {object} ErrorEnvelope @@ -214,7 +216,7 @@ func (s *Server) getEasyAITaskResult(w http.ResponseWriter, r *http.Request) { writeEasyAIAsyncError(w, statusFromRunError(err), err.Error(), nil, clients.ErrorCode(err)) return } - writeJSON(w, http.StatusOK, easyAITaskResultResponse(task)) + writeAsyncMediaResultJSON(w, easyAITaskResultResponse(task), s.billingMetrics) } func (s *Server) createVolcesCompatibleTask(r *http.Request, user *auth.User, body map[string]any) (store.GatewayTask, error) { diff --git a/apps/api/internal/publicerror/public_error.go b/apps/api/internal/publicerror/public_error.go index 57fa1a5..1db9dd4 100644 --- a/apps/api/internal/publicerror/public_error.go +++ b/apps/api/internal/publicerror/public_error.go @@ -92,6 +92,8 @@ func mappedError(code string, message string, status int, retryable bool) (Error break } return newError("upstream_unavailable", "The upstream service is temporarily unavailable.", "upstream", http.StatusServiceUnavailable, true, "retry"), true + case "invalid_upstream_result": + return newError("invalid_upstream_result", "The upstream service returned an invalid media result.", "upstream", http.StatusBadGateway, false, "contact_support"), true case "upload_invalid_response", "invalid_response", "response_too_large", "invalid_upstream_response", "upstream_invalid_response": return newError("upstream_invalid_response", "The upstream service returned an invalid or incomplete response.", "upstream", http.StatusBadGateway, true, "retry"), true case "invalid_api_key", "authentication_error", "auth_failed", "missing_credentials", "upstream_auth_failed": @@ -100,6 +102,16 @@ func mappedError(code string, message string, status int, retryable bool) (Error return newError("storage_write_failed", "The media asset could not be written to object storage.", "storage", http.StatusServiceUnavailable, true, "retry"), true case "storage_read_failed", "upload_source_fetch_failed", "upload_source_read_failed": return newError("storage_read_failed", "The media asset could not be read from object storage.", "storage", http.StatusServiceUnavailable, true, "retry"), true + case "result_url_unavailable": + return newError("result_url_unavailable", "The generated result URL is temporarily unavailable.", "storage", http.StatusServiceUnavailable, true, "retry"), true + case "result_materialization_required": + return newError("result_materialization_required", "The generated result is being migrated to URL storage.", "storage", http.StatusServiceUnavailable, true, "retry"), true + case "response_format_too_large": + return newError("response_format_too_large", "The generated media is too large for a synchronous Base64 response. Use the task result URL instead.", "request", http.StatusRequestEntityTooLarge, false, "use_result_url"), true + case "response_format_capacity_timeout": + return newError("response_format_capacity_timeout", "Synchronous Base64 response capacity is temporarily unavailable.", "gateway", http.StatusServiceUnavailable, true, "retry"), true + case "result_response_too_large": + return newError("result_response_too_large", "The generated result response exceeded the URL-only size limit.", "gateway", http.StatusInternalServerError, false, "contact_support"), true case "binary_result_expired", "result_expired": return newError("result_expired", "The generated result has expired and must be submitted again.", "storage", http.StatusGone, false, "resubmit"), true case "result_unavailable", "request_asset_expired": diff --git a/apps/api/internal/runner/binary_results.go b/apps/api/internal/runner/binary_results.go index a2b2748..850c8be 100644 --- a/apps/api/internal/runner/binary_results.go +++ b/apps/api/internal/runner/binary_results.go @@ -8,6 +8,7 @@ import ( "errors" "fmt" "io" + "net/http" "os" "path/filepath" "sort" @@ -21,6 +22,9 @@ import ( ) const ( + // MaxSynchronousInlineResponseBytes bounds the raw media payload restored for + // an explicit synchronous OpenAI-compatible b64_json response. + MaxSynchronousInlineResponseBytes = int64(20 << 20) localBinaryResultDirName = "results" localBinaryPlaceholderPrefix = "[GatewayBinary:v1;" localBinaryGenericBase64MinLength = 4096 @@ -96,6 +100,225 @@ func (s *Service) MaterializeTaskResultForStorage(ctx context.Context, taskID st return next, hadInline && !TaskResultHasInlineBinary(next), nil } +// MigrateTaskResultToURLs materializes historical inline payloads and rewrites +// legacy assetRef wrappers into the canonical URL-plus-upload storage shape. +func (s *Service) MigrateTaskResultToURLs(ctx context.Context, taskID string, result map[string]any) (map[string]any, bool, error) { + next := result + changed := false + if TaskResultHasLocalPlaceholder(next) { + restored, restoredChanged, err := s.restoreLocalPlaceholderValue(ctx, taskID, next, 0) + if err != nil { + return nil, false, err + } + mapped, ok := restored.(map[string]any) + if !ok { + return nil, false, &clients.ClientError{Code: "result_binary_not_materialized", Message: "generated result is not a JSON object", StatusCode: http.StatusInternalServerError} + } + next = mapped + changed = restoredChanged + } + if TaskResultHasInlineBinary(next) { + materialized, materializedChanged, err := s.MaterializeTaskResultForStorage(ctx, taskID, next) + if err != nil { + return nil, false, err + } + next = materialized + changed = materializedChanged + } + rewritten, rewrittenChanged := migrateStoredResultURLValue(next, 0) + mapped, ok := rewritten.(map[string]any) + if !ok { + return nil, false, &clients.ClientError{Code: "result_binary_not_materialized", Message: "generated result is not a JSON object", StatusCode: http.StatusInternalServerError} + } + return mapped, changed || rewrittenChanged, nil +} + +// TaskResultNeedsURLMigration reports whether a stored result contains inline +// binary, a local placeholder, or the legacy assetRef representation. +func TaskResultNeedsURLMigration(result map[string]any) bool { + if TaskResultHasInlineBinary(result) { + return true + } + return storedResultURLValueNeedsMigration(result, 0) +} + +// TaskResultHasLocalPlaceholder reports whether a historical result still +// references a node-local GatewayBinary file. +func TaskResultHasLocalPlaceholder(result map[string]any) bool { + return localBinaryResultHasPlaceholders(result) +} + +func (s *Service) restoreLocalPlaceholderValue(ctx context.Context, taskID string, value any, depth int) (any, bool, error) { + if err := ctx.Err(); err != nil { + return nil, false, err + } + if depth >= localBinaryMaxDepth { + return nil, false, &clients.ClientError{Code: "binary_result_corrupted", Message: "stored result exceeds the maximum JSON depth", StatusCode: http.StatusInternalServerError} + } + switch typed := value.(type) { + case map[string]any: + next := make(map[string]any, len(typed)) + changed := false + for key, childValue := range typed { + child, childChanged, err := s.restoreLocalPlaceholderValue(ctx, taskID, childValue, depth+1) + if err != nil { + return nil, false, err + } + next[key] = child + changed = changed || childChanged + } + if !changed { + return value, false, nil + } + return next, true, nil + case []any: + next := make([]any, len(typed)) + changed := false + for index, childValue := range typed { + child, childChanged, err := s.restoreLocalPlaceholderValue(ctx, taskID, childValue, depth+1) + if err != nil { + return nil, false, err + } + next[index] = child + changed = changed || childChanged + } + if !changed { + return value, false, nil + } + return next, true, nil + case string: + descriptor, ok := parseLocalBinaryPlaceholder(typed) + if !ok { + return value, false, nil + } + payload, err := s.readLocalBinaryResult(taskID, descriptor) + if err != nil { + return nil, false, err + } + encoded := base64.StdEncoding.EncodeToString(payload) + if descriptor.Encoding == "data-uri" { + return "data:" + descriptor.ContentType + ";base64," + encoded, true, nil + } + return encoded, true, nil + default: + return value, false, nil + } +} + +func storedResultURLValueNeedsMigration(value any, depth int) bool { + if depth >= localBinaryMaxDepth { + return true + } + switch typed := value.(type) { + case map[string]any: + if _, ok := typed["assetRef"].(map[string]any); ok { + return true + } + for _, child := range typed { + if storedResultURLValueNeedsMigration(child, depth+1) { + return true + } + } + case []any: + for _, child := range typed { + if storedResultURLValueNeedsMigration(child, depth+1) { + return true + } + } + case string: + _, ok := parseLocalBinaryPlaceholder(typed) + return ok + } + return false +} + +func migrateStoredResultURLValue(value any, depth int) (any, bool) { + if depth >= localBinaryMaxDepth { + return value, false + } + switch typed := value.(type) { + case map[string]any: + if asset, ok := generatedResultAssetReference(typed); ok { + upload, _ := typed["upload"].(map[string]any) + if upload == nil { + upload = uploadMetadataFromLegacyAsset(asset) + } + accessURL := firstNonEmptyString(stringFromAny(upload["url"]), asset.URL) + next := map[string]any{ + "url": accessURL, + "upload": upload, + "assetStorage": map[string]any{ + "scene": store.FileStorageSceneImageResult, + "source": stringFromAny(typed["assetStorage"].(map[string]any)["source"]), + "strategy": "migrate_asset_ref", + "contentType": asset.ContentType, + }, + } + for _, key := range []string{"type", "mime_type", "width", "height", "duration", "format", "seed", "revised_prompt"} { + if item, exists := typed[key]; exists { + next[key] = item + } + } + if stringFromAny(next["mime_type"]) == "" && asset.ContentType != "" { + next["mime_type"] = asset.ContentType + } + return next, true + } + next := make(map[string]any, len(typed)) + changed := false + for key, childValue := range typed { + child, childChanged := migrateStoredResultURLValue(childValue, depth+1) + if resultCanonicalInlineField(key) { + if item, ok := child.(map[string]any); ok && stringFromAny(item["url"]) != "" { + mergeProjectedResultURL(next, item) + changed = true + continue + } + } + next[key] = child + changed = changed || childChanged + } + if !changed { + return value, false + } + return next, true + case []any: + next := make([]any, len(typed)) + changed := false + for index, childValue := range typed { + child, childChanged := migrateStoredResultURLValue(childValue, depth+1) + next[index] = child + changed = changed || childChanged + } + if !changed { + return value, false + } + return next, true + default: + return value, false + } +} + +func uploadMetadataFromLegacyAsset(asset store.RequestAsset) map[string]any { + upload := map[string]any{ + "url": asset.URL, + "objectKey": asset.ObjectKey, + "contentType": asset.ContentType, + "size": asset.ByteSize, + "sha256": asset.SHA256, + "accessScope": asset.AccessScope, + "storageChannel": map[string]any{ + "id": asset.StorageChannelID, + "channelKey": asset.StorageChannelKey, + "provider": asset.StorageProvider, + }, + } + if asset.ExpiresAt != nil { + upload["objectExpiresAt"] = asset.ExpiresAt.Format(time.RFC3339) + } + return upload +} + // CompactExpiredTaskResultForStorage keeps the historical maintenance API but // now uses the same object-storage path as live results. New GatewayBinary // placeholders are never created; their parser remains read-only compatibility. @@ -325,8 +548,9 @@ func verifyLocalBinaryFile(path string, expectedHash string, expectedSize int64) return nil } -// HydrateTaskResult restores placeholders from verified local files. It is only -// used by result/detail/replay endpoints, never by task lists or callbacks. +// HydrateTaskResult restores verified stored payloads only for bounded, +// explicit synchronous inline responses. Asynchronous readers use +// ProjectTaskResultURLs and never call this path. func (s *Service) HydrateTaskResult(ctx context.Context, taskID string, result map[string]any) (map[string]any, error) { next, changed, err := s.hydrateLocalBinaryValue(ctx, taskID, result, 0) if err != nil { @@ -342,6 +566,244 @@ func (s *Service) HydrateTaskResult(ctx context.Context, taskID string, result m return mapped, nil } +// SynchronousInlineResultBytes returns the stored raw byte count that the +// synchronous hydration path would read. It deliberately inspects metadata +// only, so callers can reject oversized responses before any object GET or +// local file read occurs. +func SynchronousInlineResultBytes(result map[string]any) int64 { + return synchronousInlineResultValueBytes(result, 0) +} + +func synchronousInlineResultValueBytes(value any, depth int) int64 { + if depth >= localBinaryMaxDepth { + return 0 + } + switch typed := value.(type) { + case map[string]any: + if asset, _, ok := generatedResultUploadReference(typed); ok { + storage, _ := typed["assetStorage"].(map[string]any) + if resultCanonicalInlineField(stringFromAny(storage["source"])) { + return asset.ByteSize + } + } + if asset, ok := generatedResultAssetReference(typed); ok { + return asset.ByteSize + } + var total int64 + for _, child := range typed { + total += synchronousInlineResultValueBytes(child, depth+1) + if total > MaxSynchronousInlineResponseBytes { + return total + } + } + return total + case []any: + var total int64 + for _, child := range typed { + total += synchronousInlineResultValueBytes(child, depth+1) + if total > MaxSynchronousInlineResponseBytes { + return total + } + } + return total + case string: + if descriptor, ok := parseLocalBinaryPlaceholder(typed); ok { + return descriptor.Size + } + } + return 0 +} + +// ProjectTaskResultURLs returns the public, URL-only representation of a +// stored task result. It may refresh a signed object-storage URL, but it never +// reads object bytes or restores historical inline binary payloads. +func (s *Service) ProjectTaskResultURLs(ctx context.Context, taskID string, result map[string]any) (map[string]any, error) { + next, changed, err := s.projectTaskResultURLValue(ctx, taskID, result, 0) + if err != nil { + return nil, err + } + if !changed { + return result, nil + } + mapped, ok := next.(map[string]any) + if !ok { + return nil, &clients.ClientError{Code: "binary_result_corrupted", Message: "stored result is not a JSON object", StatusCode: http.StatusInternalServerError} + } + return mapped, nil +} + +func (s *Service) projectTaskResultURLValue(ctx context.Context, taskID string, value any, depth int) (any, bool, error) { + if err := ctx.Err(); err != nil { + return nil, false, err + } + if depth >= localBinaryMaxDepth { + return nil, false, &clients.ClientError{Code: "binary_result_corrupted", Message: "stored result exceeds the maximum JSON depth", StatusCode: http.StatusInternalServerError} + } + switch typed := value.(type) { + case map[string]any: + if asset, _, ok := generatedResultUploadReference(typed); ok { + accessURL, err := s.resultAssetAccessURL(ctx, asset) + if err != nil { + return nil, false, err + } + next := publicResultURLItem(typed, accessURL, asset.ContentType) + return next, true, nil + } + if asset, ok := generatedResultAssetReference(typed); ok { + accessURL, err := s.resultAssetAccessURL(ctx, asset) + if err != nil { + return nil, false, err + } + next := publicResultURLItem(typed, accessURL, asset.ContentType) + return next, true, nil + } + if _, _, ok := localBufferObjectBytes(typed); ok { + return nil, false, resultMaterializationRequired(taskID) + } + next := make(map[string]any, len(typed)) + changed := false + for key, childValue := range typed { + if resultInternalField(key) { + changed = true + continue + } + child, childChanged, err := s.projectTaskResultURLValue(ctx, taskID, childValue, depth+1) + if err != nil { + return nil, false, err + } + if resultCanonicalInlineField(key) { + if item, ok := child.(map[string]any); ok && stringFromAny(item["url"]) != "" { + mergeProjectedResultURL(next, item) + changed = true + continue + } + if raw, ok := child.(string); ok { + if requestAssetStringIsHTTPURL(raw) { + next["url"] = strings.TrimSpace(raw) + changed = true + continue + } + if strings.TrimSpace(raw) != "" { + return nil, false, resultMaterializationRequired(taskID) + } + } + } + if localBinaryKey(key) { + switch inline := child.(type) { + case map[string]any: + if stringFromAny(inline["url"]) != "" { + mergeProjectedResultURL(next, inline) + changed = true + continue + } + case string: + if strings.TrimSpace(inline) != "" { + return nil, false, resultMaterializationRequired(taskID) + } + case []any: + if len(inline) > 0 { + return nil, false, resultMaterializationRequired(taskID) + } + } + } + next[key] = child + changed = changed || childChanged + } + if !changed { + return value, false, nil + } + return next, true, nil + case []any: + next := make([]any, len(typed)) + changed := false + for index, childValue := range typed { + child, childChanged, err := s.projectTaskResultURLValue(ctx, taskID, childValue, depth+1) + if err != nil { + return nil, false, err + } + next[index] = child + changed = changed || childChanged + } + if !changed { + return value, false, nil + } + return next, true, nil + case []byte: + if len(typed) > 0 { + return nil, false, resultMaterializationRequired(taskID) + } + return value, false, nil + case string: + if _, ok := parseLocalBinaryPlaceholder(typed); ok { + return nil, false, resultMaterializationRequired(taskID) + } + if strings.HasPrefix(strings.ToLower(strings.TrimSpace(typed)), "data:") { + return nil, false, resultMaterializationRequired(taskID) + } + return value, false, nil + default: + return value, false, nil + } +} + +func (s *Service) resultAssetAccessURL(ctx context.Context, asset store.RequestAsset) (string, error) { + accessURL, err := s.requestAssetAccessURL(ctx, asset) + if err != nil || strings.TrimSpace(accessURL) == "" { + message := "stored result URL is unavailable" + if err != nil { + message = err.Error() + } + return "", &clients.ClientError{Code: "result_url_unavailable", Message: message, StatusCode: http.StatusServiceUnavailable, Retryable: true} + } + return accessURL, nil +} + +func resultMaterializationRequired(taskID string) error { + return &clients.ClientError{ + Code: "result_materialization_required", + Message: "stored result must be migrated to object storage before it can be returned", + Details: map[string]any{"task_id": strings.TrimSpace(taskID)}, + StatusCode: http.StatusServiceUnavailable, + Retryable: true, + } +} + +func publicResultURLItem(source map[string]any, accessURL string, contentType string) map[string]any { + next := map[string]any{"url": accessURL} + for _, key := range []string{"type", "mime_type", "mimeType", "content_type", "contentType", "width", "height", "duration", "format", "seed", "revised_prompt"} { + if value, ok := source[key]; ok && value != nil { + next[key] = value + } + } + if stringFromAny(next["mime_type"]) == "" && strings.TrimSpace(contentType) != "" { + next["mime_type"] = strings.TrimSpace(contentType) + } + return next +} + +func mergeProjectedResultURL(target map[string]any, item map[string]any) { + for key, value := range item { + if _, exists := target[key]; !exists || key == "url" { + target[key] = value + } + } +} + +func resultInternalField(key string) bool { + switch normalizeLocalBinaryKey(key) { + case "assetref", "assetstorage", "upload", "raw", "rawdata", "rawresponse", "providerpayload", "providerresponse", "providerraw", "upstreamresponse", "thinkingbytes", "thoughtsignature", "signaturebuffer": + return true + default: + return false + } +} + +func resultCanonicalInlineField(key string) bool { + normalized := normalizeLocalBinaryKey(key) + return normalized == "b64json" || normalized == "base64" || normalized == "b64" || + normalized == "datauri" || strings.Contains(normalized, "base64") || strings.HasSuffix(normalized, "b64") +} + func (s *Service) hydrateLocalBinaryValue(ctx context.Context, taskID string, value any, depth int) (any, bool, error) { if err := ctx.Err(); err != nil { return nil, false, err @@ -374,6 +836,31 @@ func (s *Service) hydrateLocalBinaryValue(ctx context.Context, taskID string, va } typed = next refreshedAccessURL = true + storage, _ := typed["assetStorage"].(map[string]any) + sourceKey := strings.TrimSpace(stringFromAny(storage["source"])) + if resultCanonicalInlineField(sourceKey) && asset.SHA256 != "" && asset.ByteSize > 0 { + payload, contentType, err := s.readGeneratedResultAsset(ctx, asset) + if err != nil { + return nil, false, err + } + encoded := base64.StdEncoding.EncodeToString(payload) + inline := make(map[string]any, len(typed)) + for key, item := range typed { + if resultInternalField(key) || key == "url" || key == "image_url" || key == "video_url" || key == "audio_url" { + continue + } + inline[key] = item + } + if generatedResultAssetUsesDataURI(typed) { + inline[sourceKey] = "data:" + contentType + ";base64," + encoded + } else { + inline[sourceKey] = encoded + } + if stringFromAny(inline["mime_type"]) == "" { + inline["mime_type"] = contentType + } + return inline, true, nil + } } if ref, ok := generatedResultAssetReference(typed); ok { payload, contentType, err := s.readGeneratedResultAsset(ctx, ref) @@ -389,6 +876,10 @@ func (s *Service) hydrateLocalBinaryValue(ctx context.Context, taskID string, va next := make(map[string]any, len(typed)) changed := refreshedAccessURL for key, childValue := range typed { + if resultInternalField(key) { + changed = true + continue + } child, childChanged, err := s.hydrateLocalBinaryValue(ctx, taskID, childValue, depth+1) if err != nil { return nil, false, err @@ -446,6 +937,9 @@ func generatedResultUploadReference(value map[string]any) (store.RequestAsset, m return store.RequestAsset{}, nil, false } return store.RequestAsset{ + SHA256: strings.ToLower(strings.TrimSpace(stringFromAny(upload["sha256"]))), + ContentType: stringFromAny(upload["contentType"]), + ByteSize: int64(floatFromAny(upload["size"])), URL: stringFromAny(upload["url"]), StorageProvider: stringFromAny(channel["provider"]), StorageChannelID: stringFromAny(channel["id"]), diff --git a/apps/api/internal/runner/binary_results_test.go b/apps/api/internal/runner/binary_results_test.go index 0bc9de1..a24031f 100644 --- a/apps/api/internal/runner/binary_results_test.go +++ b/apps/api/internal/runner/binary_results_test.go @@ -8,6 +8,7 @@ import ( "encoding/hex" "encoding/json" "errors" + "fmt" "net/http" "net/http/httptest" "os" @@ -209,6 +210,181 @@ func TestHydrateGeneratedResultRefreshesPrivateObjectURL(t *testing.T) { } } +func TestHydrateCanonicalUploadedResultForExplicitSynchronousBase64(t *testing.T) { + payload := []byte("canonical uploaded image") + digest := sha256.Sum256(payload) + getCount := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodGet { + getCount++ + _, _ = w.Write(payload) + return + } + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + cfg := config.Config{ + MediaOSSDirectEnabled: true, MediaOSSEndpoint: server.URL, MediaOSSBucket: "media-bucket", + MediaOSSAccessKeyID: "access-id", MediaOSSAccessKeySecret: "access-secret", MediaOSSObjectPrefix: "media", + } + service := &Service{cfg: cfg, directOSS: newDirectOSSUploader(cfg)} + result := map[string]any{"thinking_bytes": strings.Repeat("opaque", 1024), "thought_signature": "signature", "data": []any{map[string]any{ + "type": "image", "url": "https://expired.example/result.png", "mime_type": "image/png", + "assetStorage": map[string]any{"scene": store.FileStorageSceneImageResult, "source": "b64_json"}, + "upload": map[string]any{ + "url": "https://expired.example/result.png", "objectKey": "media/image_result/hash.png", "accessScope": "private", + "sha256": hex.EncodeToString(digest[:]), "size": len(payload), "contentType": "image/png", + "storageChannel": map[string]any{"channelKey": "environment-direct-oss", "provider": "aliyun_oss"}, + }, + }}} + projected, err := service.ProjectTaskResultURLs(t.Context(), "task-sync-b64", result) + if err != nil { + t.Fatal(err) + } + projectedItem := projected["data"].([]any)[0].(map[string]any) + if stringFromAny(projectedItem["url"]) == "" || getCount != 0 { + t.Fatalf("URL projection read object: item=%#v getCount=%d", projectedItem, getCount) + } + if projected["thinking_bytes"] != nil || projected["thought_signature"] != nil { + t.Fatalf("provider metadata leaked: %#v", projected) + } + + hydrated, err := service.HydrateTaskResult(t.Context(), "task-sync-b64", result) + if err != nil { + t.Fatal(err) + } + item := hydrated["data"].([]any)[0].(map[string]any) + if got := stringFromAny(item["b64_json"]); got != base64.StdEncoding.EncodeToString(payload) { + t.Fatalf("Base64=%q", got) + } + if item["url"] != nil || item["upload"] != nil || getCount != 1 { + t.Fatalf("unexpected hydrated item=%#v getCount=%d", item, getCount) + } + if hydrated["thinking_bytes"] != nil || hydrated["thought_signature"] != nil { + t.Fatalf("provider metadata leaked into synchronous response: %#v", hydrated) + } +} + +func TestSynchronousInlineResultBytesUsesMetadataBeforeObjectRead(t *testing.T) { + result := map[string]any{"data": []any{ + map[string]any{ + "assetStorage": map[string]any{"scene": store.FileStorageSceneImageResult, "source": "b64_json"}, + "upload": map[string]any{ + "objectKey": "media/image_result/large.png", "sha256": strings.Repeat("a", 64), + "size": MaxSynchronousInlineResponseBytes + 1, "contentType": "image/png", + "storageChannel": map[string]any{"channelKey": "environment-direct-oss"}, + }, + }, + }} + + if got := SynchronousInlineResultBytes(result); got != MaxSynchronousInlineResponseBytes+1 { + t.Fatalf("stored bytes=%d", got) + } +} + +func TestProjectTaskResultURLsConvertsLegacyAssetWithoutReadingObject(t *testing.T) { + payload := []byte("must never be downloaded") + digest := sha256.Sum256(payload) + getCount := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodGet { + getCount++ + } + http.Error(w, "object read is forbidden", http.StatusInternalServerError) + })) + defer server.Close() + + service := &Service{} + result := map[string]any{"data": []any{map[string]any{ + "b64_json": map[string]any{ + "assetRef": map[string]any{ + "sha256": hex.EncodeToString(digest[:]), "contentType": "image/png", "size": len(payload), "url": server.URL + "/result.png", + }, + "assetStorage": map[string]any{"scene": store.FileStorageSceneImageResult, "source": "b64_json"}, + }, + }}} + + projected, err := service.ProjectTaskResultURLs(t.Context(), "task-url-only", result) + if err != nil { + t.Fatal(err) + } + item := projected["data"].([]any)[0].(map[string]any) + if got := stringFromAny(item["url"]); got != server.URL+"/result.png" { + t.Fatalf("projected URL=%q", got) + } + if item["b64_json"] != nil || item["assetRef"] != nil || item["assetStorage"] != nil || item["upload"] != nil { + t.Fatalf("internal or inline fields leaked: %#v", item) + } + if getCount != 0 { + t.Fatalf("projector downloaded object %d time(s)", getCount) + } +} + +func TestProjectTaskResultURLsRejectsHistoricalInlinePayload(t *testing.T) { + service := &Service{} + _, err := service.ProjectTaskResultURLs(t.Context(), "task-inline", map[string]any{ + "data": []any{map[string]any{"b64_json": base64.StdEncoding.EncodeToString([]byte("inline"))}}, + }) + assertClientErrorCode(t, err, "result_materialization_required") +} + +func TestMigrateTaskResultToURLsRewritesLegacyAssetReference(t *testing.T) { + payload := []byte("legacy") + digest := sha256.Sum256(payload) + service := &Service{} + result := map[string]any{"data": []any{map[string]any{ + "b64_json": map[string]any{ + "assetRef": map[string]any{ + "sha256": hex.EncodeToString(digest[:]), "contentType": "image/png", "size": len(payload), "url": "https://cdn.example/result.png", + }, + "assetStorage": map[string]any{"scene": store.FileStorageSceneImageResult, "source": "b64_json"}, + }, + }}} + + migrated, changed, err := service.MigrateTaskResultToURLs(t.Context(), "task-legacy", result) + if err != nil { + t.Fatal(err) + } + if !changed || TaskResultNeedsURLMigration(migrated) { + t.Fatalf("legacy result was not fully migrated: %#v", migrated) + } + item := migrated["data"].([]any)[0].(map[string]any) + if stringFromAny(item["url"]) != "https://cdn.example/result.png" || item["upload"] == nil || item["b64_json"] != nil { + t.Fatalf("unexpected migrated result: %#v", item) + } +} + +func TestMigrateTaskResultToURLsUploadsActiveLocalPlaceholder(t *testing.T) { + payload := []byte("historical local image") + digest := sha256.Sum256(payload) + putCount := 0 + storageServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodPut { + putCount++ + w.WriteHeader(http.StatusOK) + return + } + http.Error(w, "unexpected method", http.StatusMethodNotAllowed) + })) + defer storageServer.Close() + + service := newLocalBinaryTestService(t) + service.directOSS = &directOSSUploader{ + endpoint: storageServer.URL, bucket: "bucket", accessKeyID: "access-id", accessKeySecret: "access-secret", objectPrefix: "media", + } + writeHistoricalLocalBinaryFixture(t, service, "task-local-migrate", payload) + placeholder := fmt.Sprintf("%ssha256=%s;bytes=%d;mime=image/png;encoding=base64]", localBinaryPlaceholderPrefix, hex.EncodeToString(digest[:]), len(payload)) + result := map[string]any{"data": []any{map[string]any{"b64_json": placeholder}}} + + migrated, changed, err := service.MigrateTaskResultToURLs(t.Context(), "task-local-migrate", result) + if err != nil { + t.Fatal(err) + } + if !changed || TaskResultNeedsURLMigration(migrated) || putCount != 1 { + t.Fatalf("placeholder migration changed=%t putCount=%d result=%#v", changed, putCount, migrated) + } +} + func TestHydrateLocalBinaryResultReturnsExpiredAndCorruptedErrors(t *testing.T) { service := newLocalBinaryTestService(t) service.cfg.LocalResultTTLHours = 1 diff --git a/apps/api/internal/runner/service.go b/apps/api/internal/runner/service.go index f322872..0744f46 100644 --- a/apps/api/internal/runner/service.go +++ b/apps/api/internal/runner/service.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "log/slog" + "net/http" "os" "regexp" "strconv" @@ -243,6 +244,13 @@ func (s *Service) observeObjectStorage(event string, provider string, bytes int, } } +func (s *Service) observeResultStorage(source string) { + observer, ok := s.billingMetrics.(interface{ ObserveResultStorage(string) }) + if ok { + observer.ObserveResultStorage(source) + } +} + func (s *Service) Execute(ctx context.Context, task store.GatewayTask, user *auth.User) (Result, error) { return s.execute(ctx, task, user, nil) } @@ -982,7 +990,7 @@ candidatesLoop: } walletReservationFinalized = true s.logger.Warn("task succeeded but billing requires manual review", "taskID", task.ID, "error_category", "billing_calculation_failed") - output, hydrateErr := s.HydrateTaskResult(ctx, task.ID, response.Result) + output, hydrateErr := s.taskResultForSynchronousResponse(ctx, task, response.Result) if hydrateErr != nil { return Result{Task: review}, hydrateErr } @@ -1075,7 +1083,7 @@ candidatesLoop: // time after the task is already durably complete. return Result{Task: finished, Output: response.Result}, nil } - output, hydrateErr := s.HydrateTaskResult(ctx, task.ID, response.Result) + output, hydrateErr := s.taskResultForSynchronousResponse(ctx, task, response.Result) if hydrateErr != nil { return Result{Task: finished}, hydrateErr } @@ -1272,6 +1280,33 @@ candidatesLoop: return Result{Task: failed, Output: failed.Result}, lastErr } +func (s *Service) taskResultForSynchronousResponse(ctx context.Context, task store.GatewayTask, result map[string]any) (map[string]any, error) { + if taskRequestsInlineMediaResult(task.Request) { + if rawBytes := SynchronousInlineResultBytes(result); rawBytes > MaxSynchronousInlineResponseBytes { + return nil, &clients.ClientError{ + Code: "response_format_too_large", + Message: "synchronous Base64 response exceeds the 20 MiB limit", + StatusCode: http.StatusRequestEntityTooLarge, + Retryable: false, + Details: map[string]any{ + "task_id": task.ID, + "query_url": "/api/v1/ai/result/" + task.ID, + "limit_bytes": MaxSynchronousInlineResponseBytes, + "actual_bytes": rawBytes, + "response_format": "url", + }, + } + } + return s.HydrateTaskResult(ctx, task.ID, result) + } + return s.ProjectTaskResultURLs(ctx, task.ID, result) +} + +func taskRequestsInlineMediaResult(request map[string]any) bool { + value, _ := request["response_format"].(string) + return strings.EqualFold(strings.TrimSpace(value), "b64_json") +} + func pricingCandidateKey(candidate store.RuntimeModelCandidate) string { return firstNonEmptyString(candidate.PlatformModelID, candidate.PlatformID+":"+candidate.ModelName) } diff --git a/apps/api/internal/runner/upload.go b/apps/api/internal/runner/upload.go index 6eb7d3a..73aefa1 100644 --- a/apps/api/internal/runner/upload.go +++ b/apps/api/internal/runner/upload.go @@ -50,6 +50,13 @@ type generatedAssetDecision struct { StripKeys []string } +type generatedAssetUploadResult struct { + upload map[string]any + contentType string + kind string + strategy string +} + type generatedInlineAsset struct { Bytes []byte ContentType string @@ -150,6 +157,9 @@ func (s *Service) uploadGeneratedAssets(ctx context.Context, taskID string, task return nil, err } decisions[index] = decision + if _, mediaURL := mediaURLSourceFromItem(item); mediaURL != "" && decision.URL == nil { + s.observeResultStorage("upstream_url") + } if decision.Inline != nil || decision.URL != nil { needsUpload = true } @@ -175,6 +185,7 @@ func (s *Service) uploadGeneratedAssets(ctx context.Context, taskID string, task next[key] = value } nextData := make([]any, 0, len(data)) + uploadCache := make(map[string]generatedAssetUploadResult) for index, rawItem := range data { item, _ := rawItem.(map[string]any) if item == nil { @@ -197,7 +208,15 @@ func (s *Service) uploadGeneratedAssets(ctx context.Context, taskID string, task var contentType string var err error if decision.Inline != nil { - upload, contentType, kind, strategy, err = s.uploadGeneratedAsset(ctx, taskID, decision.Inline, index, channels) + cacheKey := generatedInlineAssetCacheKey(decision.Inline) + if cached, ok := uploadCache[cacheKey]; ok { + upload, contentType, kind, strategy = cached.upload, cached.contentType, cached.kind, cached.strategy + } else { + upload, contentType, kind, strategy, err = s.uploadGeneratedAsset(ctx, taskID, decision.Inline, index, channels) + if err == nil { + uploadCache[cacheKey] = generatedAssetUploadResult{upload: upload, contentType: contentType, kind: kind, strategy: strategy} + } + } sourceKey = decision.Inline.SourceKey } else { upload, contentType, kind, strategy, err = s.uploadGeneratedURLAsset(ctx, taskID, decision.URL, index, channels, acceptanceEmulatorBaseURL) @@ -233,9 +252,6 @@ func (s *Service) uploadGeneratedAssets(ctx context.Context, taskID string, task if contentType != "" && stringFromAny(merged["mime_type"]) == "" { merged["mime_type"] = contentType } - if decision.Inline != nil && strings.TrimSpace(sourceKey) != "" { - merged[sourceKey] = generatedRawMediaReference(decision.Inline, upload, contentType, kind, strategy) - } } nextData = append(nextData, merged) } @@ -255,6 +271,15 @@ func (s *Service) uploadGeneratedAssets(ctx context.Context, taskID string, task return s.finalizeGeneratedAssets(ctx, taskID, taskKind, next, policy, channels, channelsLoaded, len(nextData)) } +func generatedInlineAssetCacheKey(asset *generatedInlineAsset) string { + if asset == nil { + return "" + } + digest := sha256.Sum256(asset.Bytes) + contentType := resolvedGeneratedAssetContentType(asset.ContentType, asset.Kind, asset.Bytes) + return hex.EncodeToString(digest[:]) + ":" + contentType +} + func generatedAssetUploadPolicyForAcceptanceRun(policy generatedAssetUploadPolicy, acceptanceRunID string) generatedAssetUploadPolicy { if strings.TrimSpace(acceptanceRunID) != "" { policy.UploadURLMedia = true @@ -274,7 +299,7 @@ func (s *Service) finalizeGeneratedAssets( ) (map[string]any, error) { redactGeneratedResultRawData(result) if !TaskResultHasInlineBinary(result) { - return result, nil + return canonicalStoredResultURLs(result), nil } next := result if policy.UploadInlineMedia { @@ -304,7 +329,7 @@ func (s *Service) finalizeGeneratedAssets( } } if !TaskResultHasInlineBinary(next) { - return next, nil + return canonicalStoredResultURLs(next), nil } diagnostics := taskResultInlineBinaryDiagnostics(next) if s.logger != nil { @@ -322,6 +347,18 @@ func (s *Service) finalizeGeneratedAssets( } } +func canonicalStoredResultURLs(result map[string]any) map[string]any { + next, changed := migrateStoredResultURLValue(result, 0) + if !changed { + return result + } + mapped, ok := next.(map[string]any) + if !ok { + return result + } + return mapped +} + func generatedRawValueHasInlineMedia(value any, key string, siblings map[string]any) bool { switch typed := value.(type) { case map[string]any: @@ -448,42 +485,9 @@ func generatedRawInlineMediaAsset(key string, value string, siblings map[string] } func generatedRawMediaReference(asset *generatedInlineAsset, upload map[string]any, contentType string, kind string, strategy string) map[string]any { - digest := sha256.Sum256(asset.Bytes) urlValue := stringFromAny(upload["url"]) - ref := map[string]any{ - "sha256": hex.EncodeToString(digest[:]), - "contentType": contentType, - "size": len(asset.Bytes), - } - if urlValue != "" { - ref["url"] = urlValue - } - if fileName := stringFromAny(upload["fileName"]); fileName != "" { - ref["fileName"] = fileName - } - if expiresAt := stringFromAny(upload["expiresAt"]); expiresAt != "" { - ref["expiresAt"] = expiresAt - } - if channel, ok := upload["storageChannel"].(map[string]any); ok { - if provider := stringFromAny(channel["provider"]); provider != "" { - ref["storageProvider"] = provider - } - if id := stringFromAny(channel["id"]); id != "" { - ref["storageChannelId"] = id - } - if key := stringFromAny(channel["channelKey"]); key != "" { - ref["storageChannelKey"] = key - } - } - if objectKey := stringFromAny(upload["objectKey"]); objectKey != "" { - ref["objectKey"] = objectKey - } - if accessScope := stringFromAny(upload["accessScope"]); accessScope != "" { - ref["accessScope"] = accessScope - } out := map[string]any{ - "assetRef": ref, - "upload": upload, + "upload": upload, "assetStorage": map[string]any{ "scene": store.FileStorageSceneImageResult, "source": asset.SourceKey, @@ -497,6 +501,9 @@ func generatedRawMediaReference(asset *generatedInlineAsset, upload map[string]a if kind != "" { out["type"] = kind } + if contentType != "" { + out["mime_type"] = contentType + } return out } @@ -713,6 +720,9 @@ func (s *Service) uploadGeneratedAsset(ctx context.Context, taskID string, asset return nil, "", "", "", &clients.ClientError{Code: "storage_write_failed", Message: "no enabled object storage channel", StatusCode: http.StatusServiceUnavailable, Retryable: true} } upload, err := s.uploadFileWithFailover(ctx, payload, channels) + if err == nil { + s.observeResultStorage("uploaded") + } return upload, contentType, kind, "upload_inline_media", err } @@ -734,6 +744,9 @@ func (s *Service) uploadGeneratedURLAsset(ctx context.Context, taskID string, as return nil, "", "", "", &clients.ClientError{Code: "storage_write_failed", Message: "no enabled object storage channel", StatusCode: http.StatusServiceUnavailable, Retryable: true} } upload, err := s.uploadFileWithFailover(ctx, uploadPayload, channels) + if err == nil { + s.observeResultStorage("uploaded") + } return upload, contentType, kind, "upload_url_media", err } @@ -984,7 +997,7 @@ func (s *Service) uploadFileWithFailover(ctx context.Context, payload FileUpload func storageFailureAllowsFailover(channel store.FileStorageChannel, err error) bool { code := strings.ToLower(strings.TrimSpace(clients.ErrorCode(err))) switch code { - case "upload_source_too_large", "upload_decode_failed", "invalid_multipart_file", "invalid_multipart_image", "invalid_multipart_audio": + case "upload_source_too_large", "upload_decode_failed", "invalid_upstream_result", "invalid_multipart_file", "invalid_multipart_image", "invalid_multipart_audio": return false } var clientErr *clients.ClientError @@ -1120,6 +1133,12 @@ func stripDataURLPrefix(value string) string { func generatedAssetDecisionForItem(taskKind string, item map[string]any, policy generatedAssetUploadPolicy) (generatedAssetDecision, error) { decision := generatedAssetDecision{} + for _, key := range mediaURLCandidateKeys() { + value := strings.TrimSpace(stringFromAny(item[key])) + if value != "" && strings.Contains(value, "://") && !mediaURLString(value) { + return decision, &clients.ClientError{Code: "invalid_upstream_result", Message: "generated media URL must use http or https", StatusCode: http.StatusBadGateway, Retryable: false} + } + } urlKey, mediaURL := mediaURLSourceFromItem(item) if mediaURL != "" { if !policy.UploadURLMedia { @@ -1225,7 +1244,7 @@ func inlineMediaPayloadFromString(value string, strictBase64 bool) ([]byte, stri } payload, err := decodeBase64Payload(encoded) if err != nil { - return nil, "", false, &clients.ClientError{Code: "upload_decode_failed", Message: err.Error(), Retryable: false} + return nil, "", false, &clients.ClientError{Code: "invalid_upstream_result", Message: err.Error(), StatusCode: http.StatusBadGateway, Retryable: false} } return payload, contentType, true, nil } @@ -1235,7 +1254,7 @@ func inlineMediaPayloadFromString(value string, strictBase64 bool) ([]byte, stri payload, err := decodeBase64Payload(raw) if err != nil { if strictBase64 { - return nil, "", false, &clients.ClientError{Code: "upload_decode_failed", Message: err.Error(), Retryable: false} + return nil, "", false, &clients.ClientError{Code: "invalid_upstream_result", Message: err.Error(), StatusCode: http.StatusBadGateway, Retryable: false} } return nil, "", false, nil } @@ -1245,7 +1264,7 @@ func inlineMediaPayloadFromString(value string, strictBase64 bool) ([]byte, stri func parseBase64DataURL(value string) (string, string, bool, error) { prefix, payload, ok := strings.Cut(value, ",") if !ok { - return "", "", false, &clients.ClientError{Code: "upload_decode_failed", Message: "invalid data URL media payload", Retryable: false} + return "", "", false, &clients.ClientError{Code: "invalid_upstream_result", Message: "invalid data URL media payload", StatusCode: http.StatusBadGateway, Retryable: false} } meta := strings.TrimPrefix(prefix, "data:") meta = strings.TrimPrefix(meta, "DATA:") @@ -1259,7 +1278,7 @@ func parseBase64DataURL(value string) (string, string, bool, error) { } } if !isBase64 { - return "", "", false, &clients.ClientError{Code: "upload_decode_failed", Message: "data URL media payload is not base64 encoded", Retryable: false} + return "", "", false, &clients.ClientError{Code: "invalid_upstream_result", Message: "data URL media payload is not base64 encoded", StatusCode: http.StatusBadGateway, Retryable: false} } return contentType, payload, true, nil } @@ -1456,10 +1475,11 @@ func mediaURLString(value string) bool { if strings.HasPrefix(lower, "data:") { return false } - return strings.HasPrefix(lower, "http://") || - strings.HasPrefix(lower, "https://") || - strings.HasPrefix(lower, "/") || - strings.Contains(lower, "://") + if strings.HasPrefix(lower, "/") { + return true + } + parsed, err := url.Parse(raw) + return err == nil && parsed.User == nil && parsed.Host != "" && (parsed.Scheme == "http" || parsed.Scheme == "https") } func mediaContentTypeFromItem(item map[string]any) string { diff --git a/apps/api/internal/runner/upload_test.go b/apps/api/internal/runner/upload_test.go index 46be563..9dede9e 100644 --- a/apps/api/internal/runner/upload_test.go +++ b/apps/api/internal/runner/upload_test.go @@ -55,6 +55,43 @@ func TestGeneratedAssetDecisionUploadsInlineImageBase64(t *testing.T) { } } +func TestUploadGeneratedAssetsReusesDuplicateSHAWithinTask(t *testing.T) { + putCount := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPut { + http.Error(w, "unexpected method", http.StatusMethodNotAllowed) + return + } + putCount++ + w.WriteHeader(http.StatusOK) + })) + defer server.Close() + service := &Service{} + service.directOSS = &directOSSUploader{ + endpoint: server.URL, bucket: "bucket", accessKeyID: "access-id", accessKeySecret: "access-secret", objectPrefix: "media", + } + payload := append([]byte{0x89, 'P', 'N', 'G', 0x0d, 0x0a, 0x1a, 0x0a}, bytes.Repeat([]byte{0}, 32)...) + encoded := base64.StdEncoding.EncodeToString(payload) + result := map[string]any{"data": []any{ + map[string]any{"b64_json": encoded, "mime_type": "image/png"}, + map[string]any{"b64_json": encoded, "mime_type": "image/png"}, + }} + + stored, err := service.uploadGeneratedAssets(t.Context(), "task-duplicate", "images.generations", "", result) + if err != nil { + t.Fatal(err) + } + items := stored["data"].([]any) + first := items[0].(map[string]any) + second := items[1].(map[string]any) + if putCount != 1 || stringFromAny(first["url"]) == "" || first["url"] != second["url"] { + t.Fatalf("duplicate SHA was not reused: puts=%d first=%#v second=%#v", putCount, first, second) + } + if TaskResultHasInlineBinary(stored) || first["b64_json"] != nil || second["b64_json"] != nil { + t.Fatalf("inline payload remained after deduplicated upload: %#v", stored) + } +} + func TestMediaResultMaterializationConcurrencyIsBounded(t *testing.T) { service := New(config.Config{MediaMaterializationConcurrency: 1}, nil, nil) result := map[string]any{ @@ -227,6 +264,15 @@ func TestGeneratedAssetDecisionUploadsURLWhenPolicyUploadAll(t *testing.T) { } } +func TestGeneratedAssetDecisionRejectsUnsupportedURLScheme(t *testing.T) { + _, err := generatedAssetDecisionForItem("images.generations", map[string]any{ + "url": "ftp://files.example/result.png", "type": "image", + }, defaultGeneratedAssetUploadPolicy()) + if clients.ErrorCode(err) != "invalid_upstream_result" { + t.Fatalf("unexpected error: %v", err) + } +} + func TestGeneratedAssetUploadPolicyFromName(t *testing.T) { tests := []struct { name string @@ -329,13 +375,15 @@ func TestFinalizeGeneratedAssetsUploadsNestedInlineBinaryUnderDefaultPolicy(t *t t.Fatal("finalized result still contains inline binary") } nested := finalized["provider_result"].(map[string]any) - reference, ok := nested["binary_data_base64"].(map[string]any) - if !ok { - t.Fatalf("nested binary should be replaced by an asset reference: %+v", nested) + if _, exists := nested["binary_data_base64"]; exists { + t.Fatalf("nested Base64 field was retained: %+v", nested) } - if urlValue := stringFromAny(reference["url"]); !strings.HasPrefix(urlValue, "https://cdn.example.com/media/image_result/") { + if urlValue := stringFromAny(nested["url"]); !strings.HasPrefix(urlValue, "https://cdn.example.com/media/image_result/") { t.Fatalf("unexpected object storage URL: %s", urlValue) } + if nested["upload"] == nil { + t.Fatalf("nested upload metadata was not retained: %+v", nested) + } } func TestFinalizeGeneratedAssetsMaterializesDetectedMediaAndKeepsOpaqueMetadata(t *testing.T) { @@ -368,7 +416,7 @@ func TestFinalizeGeneratedAssetsMaterializesDetectedMediaAndKeepsOpaqueMetadata( t.Fatalf("finalized result still contains generated media: %#v", finalized) } reference, ok := finalized["provider_payload"].(map[string]any) - if !ok || reference["assetRef"] == nil || reference["upload"] == nil { + if !ok || reference["url"] == nil || reference["upload"] == nil || reference["assetRef"] != nil { t.Fatalf("detected media was not objectified: %#v", finalized["provider_payload"]) } if finalized["thought_signature"] != opaque { @@ -407,9 +455,11 @@ func TestFinalizeGeneratedAssetsMaterializesGeminiImageData(t *testing.T) { } data := finalized["data"].([]any) item := data[0].(map[string]any) - reference, ok := item["b64_json"].(map[string]any) - if !ok || reference["assetRef"] == nil || reference["upload"] == nil { - t.Fatalf("Gemini b64_json was not replaced by an object reference: %#v", item) + if _, exists := item["b64_json"]; exists { + t.Fatalf("Gemini b64_json was retained: %#v", item) + } + if stringFromAny(item["url"]) == "" || item["upload"] == nil || item["assetRef"] != nil { + t.Fatalf("Gemini b64_json was not replaced by a URL result: %#v", item) } } @@ -474,7 +524,7 @@ func TestUploadGeneratedAudioFailsWithoutObjectStorage(t *testing.T) { } } -func TestUploadGeneratedRawMediaValueReplacesGeminiInlineDataWithAssetRef(t *testing.T) { +func TestUploadGeneratedRawMediaValueReplacesGeminiInlineDataWithURL(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) })) defer server.Close() service := &Service{} @@ -516,9 +566,9 @@ func TestUploadGeneratedRawMediaValueReplacesGeminiInlineDataWithAssetRef(t *tes if !ok { t.Fatalf("inlineData.data should be an asset reference, got %+v", inlineData["data"]) } - ref, _ := data["assetRef"].(map[string]any) - if ref["sha256"] == "" || ref["contentType"] != "image/png" || ref["size"] != len(payload) { - t.Fatalf("unexpected asset ref: %+v", ref) + upload, _ := data["upload"].(map[string]any) + if upload["sha256"] == "" || upload["contentType"] != "image/png" || upload["size"] != len(payload) || data["assetRef"] != nil { + t.Fatalf("unexpected URL storage metadata: %+v", data) } if urlValue := stringFromAny(data["url"]); !strings.HasPrefix(urlValue, "https://cdn.example.com/media/image_result/") || !strings.HasSuffix(urlValue, ".png") { t.Fatalf("unexpected raw media URL: %s", urlValue) @@ -528,7 +578,7 @@ func TestUploadGeneratedRawMediaValueReplacesGeminiInlineDataWithAssetRef(t *tes } } -func TestUploadGeneratedRawMediaValueReplacesBufferAndBytesWithAssetRefs(t *testing.T) { +func TestUploadGeneratedRawMediaValueReplacesBufferAndBytesWithURLs(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) })) defer server.Close() service := &Service{} @@ -551,7 +601,7 @@ func TestUploadGeneratedRawMediaValueReplacesBufferAndBytesWithAssetRefs(t *test next := uploaded.(map[string]any) for _, key := range []string{"buffer", "audio_bytes", "direct"} { item, ok := next[key].(map[string]any) - if !ok || item["assetRef"] == nil || item["upload"] == nil { + if !ok || item["url"] == nil || item["upload"] == nil || item["assetRef"] != nil { t.Fatalf("%s was not objectified: %#v", key, next[key]) } } diff --git a/apps/api/internal/securityevents/metrics.go b/apps/api/internal/securityevents/metrics.go index d474e70..0bc8748 100644 --- a/apps/api/internal/securityevents/metrics.go +++ b/apps/api/internal/securityevents/metrics.go @@ -109,6 +109,11 @@ type Metrics struct { storageChannelFailovers atomic.Uint64 storageAllChannelsFailed atomic.Uint64 storageObjectifiedBytes atomic.Uint64 + resultSourceUpstreamURL atomic.Uint64 + resultSourceUploaded atomic.Uint64 + resultPollResponses atomic.Uint64 + resultPollResponseBytes atomic.Uint64 + resultPollOversized atomic.Uint64 taskAdmissionWaitBuckets [11]atomic.Uint64 taskAdmissionWaitMicros atomic.Uint64 } @@ -345,6 +350,27 @@ func (m *Metrics) ObserveObjectStorage(event string, provider string, bytes int6 } } +func (m *Metrics) ObserveResultStorage(source string) { + switch strings.ToLower(strings.TrimSpace(source)) { + case "upstream_url": + m.resultSourceUpstreamURL.Add(1) + case "uploaded": + m.resultSourceUploaded.Add(1) + } +} + +func (m *Metrics) ObserveResultDelivery(event string, bytes int64) { + switch strings.ToLower(strings.TrimSpace(event)) { + case "poll_success": + m.resultPollResponses.Add(1) + if bytes > 0 { + m.resultPollResponseBytes.Add(uint64(bytes)) + } + case "poll_oversized": + m.resultPollOversized.Add(1) + } +} + func (m *Metrics) ObserveTaskAdmissionWait(wait time.Duration) { if wait < 0 { wait = 0 @@ -561,6 +587,13 @@ func (m *Metrics) Handler(provider MetricsSnapshotProvider, issuer, audience str plainCounter(w, "easyai_gateway_storage_channel_failovers_total", "Switches to the next configured storage channel.", m.storageChannelFailovers.Load()) plainCounter(w, "easyai_gateway_storage_all_channels_failed_total", "Storage writes where every eligible channel failed.", m.storageAllChannelsFailed.Load()) plainCounter(w, "easyai_gateway_storage_objectified_bytes_total", "Binary bytes successfully written through storage channels.", m.storageObjectifiedBytes.Load()) + outcomeCounters(w, "easyai_gateway_result_storage_total", "Generated media results by canonical storage source.", []outcomeValue{ + {"upstream_url", m.resultSourceUpstreamURL.Load()}, + {"uploaded", m.resultSourceUploaded.Load()}, + }) + plainCounter(w, "easyai_gateway_result_poll_responses_total", "Successful URL-only task result responses.", m.resultPollResponses.Load()) + plainCounter(w, "easyai_gateway_result_poll_response_bytes_total", "Bytes written by successful URL-only task result responses.", m.resultPollResponseBytes.Load()) + plainCounter(w, "easyai_gateway_result_poll_oversized_total", "Task result responses rejected by the URL-only size gate.", m.resultPollOversized.Load()) publicErrorCounters(w, publicerror.MetricSnapshot()) platformModelRateLimitUtilizationGauges(w, modelRateLimits) plainGauge(w, "easyai_gateway_postgres_pool_max_connections", "Maximum PostgreSQL connections in this process pool.", int64(postgresPool.MaxConnections)) diff --git a/apps/api/internal/securityevents/metrics_test.go b/apps/api/internal/securityevents/metrics_test.go index b551de8..2041d71 100644 --- a/apps/api/internal/securityevents/metrics_test.go +++ b/apps/api/internal/securityevents/metrics_test.go @@ -57,6 +57,10 @@ func TestMetricsExposeBoundedOutcomesAndState(t *testing.T) { metrics.ObserveObjectStorage("retry", "s3", 0, 0) metrics.ObserveObjectStorage("failover", "s3", 0, 0) metrics.ObserveObjectStorage("all_failed", "", 0, 0) + metrics.ObserveResultStorage("upstream_url") + metrics.ObserveResultStorage("uploaded") + metrics.ObserveResultDelivery("poll_success", 512) + metrics.ObserveResultDelivery("poll_oversized", 70<<10) metrics.ObserveAsyncWorkerResize("success") metrics.ObserveConcurrencyLeaseRenewal("success") metrics.ObserveConcurrencyLeaseRenewal("lost") @@ -117,6 +121,11 @@ func TestMetricsExposeBoundedOutcomesAndState(t *testing.T) { `easyai_gateway_storage_channel_failovers_total 1`, `easyai_gateway_storage_all_channels_failed_total 1`, `easyai_gateway_storage_objectified_bytes_total 128`, + `easyai_gateway_result_storage_total{outcome="upstream_url"} 1`, + `easyai_gateway_result_storage_total{outcome="uploaded"} 1`, + `easyai_gateway_result_poll_responses_total 1`, + `easyai_gateway_result_poll_response_bytes_total 512`, + `easyai_gateway_result_poll_oversized_total 1`, `easyai_gateway_platform_model_rate_limit_utilization{platform_model_id="platform-model-1",metric="concurrent"} 0.750000`, `easyai_gateway_async_worker_resizes_total{outcome="success"} 1`, `easyai_gateway_concurrency_lease_renewals_total{outcome="success"} 1`, diff --git a/deploy/kubernetes/production/application.yaml b/deploy/kubernetes/production/application.yaml index 341a069..f8b657c 100644 --- a/deploy/kubernetes/production/application.yaml +++ b/deploy/kubernetes/production/application.yaml @@ -48,6 +48,8 @@ spec: - secretRef: name: easyai-ai-gateway-runtime env: + - name: GOMEMLIMIT + value: 1536MiB - name: AI_GATEWAY_PROCESS_ROLE value: api - name: AI_GATEWAY_ASYNC_QUEUE_WORKER_ENABLED @@ -315,6 +317,8 @@ spec: - secretRef: name: easyai-ai-gateway-runtime env: + - name: GOMEMLIMIT + value: 1536MiB - name: AI_GATEWAY_PROCESS_ROLE value: api - name: AI_GATEWAY_ASYNC_QUEUE_WORKER_ENABLED