From e945eff09141b8d875d724b43ea8de2d71d3060f Mon Sep 17 00:00:00 2001 From: WuYongJie Date: Tue, 11 Aug 2026 01:24:10 +0800 Subject: [PATCH] fix(bom-costing): accept completed json streams --- .../bom-costing-multimodal-gateway.test.ts | 114 +++++++++++++- .../bom-costing-multimodal-gateway.ts | 144 +++++++++++++++++- 2 files changed, 252 insertions(+), 6 deletions(-) diff --git a/furniture-3d-system/src/server/services/bom-costing/bom-costing-multimodal-gateway.test.ts b/furniture-3d-system/src/server/services/bom-costing/bom-costing-multimodal-gateway.test.ts index bfa285d0..48ea4c42 100644 --- a/furniture-3d-system/src/server/services/bom-costing/bom-costing-multimodal-gateway.test.ts +++ b/furniture-3d-system/src/server/services/bom-costing/bom-costing-multimodal-gateway.test.ts @@ -52,7 +52,7 @@ describe('BOM 多模态网关', () => { start(streamController) { controller = streamController streamController.enqueue(encoder.encode( - 'data: {"id":"call-1","model":"gpt-5.6-terra","choices":[{"delta":{"content":"{\\"ok\\":true}"},"finish_reason":null}]}\n\n' + + 'data: {"id":"call-1","model":"gpt-5.6-terra","choices":[{"delta":{"content":"not-json"},"finish_reason":null}]}\n\n' + 'data: [DONE]\n\n', )) }, @@ -80,7 +80,7 @@ describe('BOM 多模态网关', () => { timeoutMs: 5_000, }) - expect(result.content).toBe('{"ok":true}') + expect(result.content).toBe('not-json') expect(result.finishReason).toBeNull() expect(cancelReason).toBe('completed') expect(controller).not.toBeNull() @@ -124,6 +124,116 @@ describe('BOM 多模态网关', () => { expect(cancelReason).toBe('completed') }) + it('收到完整 JSON 后立即结束，兼容既不发送结束原因也不发送 DONE 的上游', async () => { + let cancelReason: unknown + const encoder = new TextEncoder() + const keepAliveStream = new ReadableStream({ + start(streamController) { + streamController.enqueue(encoder.encode( + 'data: {"id":"call-json","model":"gpt-5.6-terra","choices":[{"delta":{"content":"{\\"product\\":{\\"name\\":\\"衣柜\\"},\\"skus\\":[]}"},"finish_reason":null}]}\n\n', + )) + }, + cancel(reason) { + cancelReason = reason + }, + }) + jest.spyOn(global, 'fetch').mockResolvedValue( + new Response(keepAliveStream, { + status: 200, + headers: { 'Content-Type': 'text/event-stream' }, + }), + ) + + const result = await callBomCostingMultimodalModel({ + config: { + baseUrl: 'https://ai.example.com/v1', + apiKey: 'test-key', + model: 'gpt-5.6-terra', + }, + operation: 'SYNTHESIS', + systemPrompt: 'test', + userContent: 'test', + maxTokens: 100, + timeoutMs: 5_000, + }) + + expect(result.content).toBe('{"product":{"name":"衣柜"},"skus":[]}') + expect(result.finishReason).toBeNull() + expect(result.completedFromJsonPayload).toBe(true) + expect(cancelReason).toBe('completed') + }) + + it('截断 JSON 不会被当成完整结果，仍按超时进入恢复流程', async () => { + const encoder = new TextEncoder() + const truncatedStream = new ReadableStream({ + start(streamController) { + streamController.enqueue(encoder.encode( + 'data: {"choices":[{"delta":{"content":"{\\"product\\":{\\"name\\":\\"衣柜\\"}"},"finish_reason":null}]}\n\n', + )) + }, + }) + jest.spyOn(global, 'fetch').mockResolvedValue( + new Response(truncatedStream, { + status: 200, + headers: { 'Content-Type': 'text/event-stream' }, + }), + ) + + await expect(callBomCostingMultimodalModel({ + config: { + baseUrl: 'https://ai.example.com/v1', + apiKey: 'test-key', + model: 'gpt-5.6-terra', + }, + operation: 'SYNTHESIS', + systemPrompt: 'test', + userContent: 'test', + maxTokens: 100, + timeoutMs: 25, + })).rejects.toThrow('多模态 BOM 分析超时') + }) + + it('上游拒绝请求时记录脱敏后的错误类型和参数，不记录凭据或图片内容', async () => { + const warning = jest.spyOn(console, 'warn').mockImplementation(() => undefined) + jest.spyOn(global, 'fetch').mockResolvedValue( + new Response(JSON.stringify({ + error: { + code: 'invalid_request_error', + type: 'invalid_request_error', + param: 'messages[1].content', + message: 'Invalid image at https://secret.example.com/a.png Bearer abc123', + }, + }), { + status: 400, + headers: { 'Content-Type': 'application/json' }, + }), + ) + + await expect(callBomCostingMultimodalModel({ + config: { + baseUrl: 'https://ai.example.com/v1', + apiKey: 'test-key', + model: 'gpt-5.6-terra', + }, + operation: 'SYNTHESIS', + systemPrompt: 'test', + userContent: 'test', + maxTokens: 100, + timeoutMs: 5_000, + })).rejects.toThrow('多模态分析服务返回 400') + + const rejectionLog = warning.mock.calls.find((call) => ( + call[0] === '[BOM costing] multimodal upstream rejected request' + ))?.[1] as Record | undefined + expect(rejectionLog).toMatchObject({ + status: 400, + upstreamErrorCode: 'invalid_request_error', + upstreamErrorType: 'invalid_request_error', + upstreamErrorParam: 'messages[1].content', + upstreamErrorMessage: 'Invalid image at [REDACTED_URL] Bearer [REDACTED]', + }) + }) + it('规格汇总遇到网络超时时立即交由上层拆分，不白白重试整批', async () => { const networkTimeout = Object.assign( new Error('socket timed out'), diff --git a/furniture-3d-system/src/server/services/bom-costing/bom-costing-multimodal-gateway.ts b/furniture-3d-system/src/server/services/bom-costing/bom-costing-multimodal-gateway.ts index 35b54148..03e5055c 100644 --- a/furniture-3d-system/src/server/services/bom-costing/bom-costing-multimodal-gateway.ts +++ b/furniture-3d-system/src/server/services/bom-costing/bom-costing-multimodal-gateway.ts @@ -27,6 +27,7 @@ export interface BomCostingMultimodalCallResult { completionTokens: number | null durationMs: number finishReason: string | null + completedFromJsonPayload?: boolean } export type BomCostingMultimodalOperation = 'IMAGE_FACTS' | 'SYNTHESIS' @@ -54,6 +55,7 @@ const BOM_COSTING_RETRYABLE_TRANSPORT_CODES = new Set([ ]) const BOM_COSTING_SYNTHESIS_MAX_ATTEMPTS = 2 const BOM_COSTING_SYNTHESIS_RETRY_DELAY_MS = 1_000 +const BOM_COSTING_COMPLETE_JSON_SETTLE_MS = 75 export function readBomCostingBoundedInteger( value: string | undefined, @@ -213,6 +215,75 @@ function readUpstreamRequestId(headers: Headers): string | null { return null } +interface BomCostingUpstreamErrorSummary { + code: string | null + type: string | null + param: string | null + message: string | null +} + +function sanitizeUpstreamErrorValue(value: unknown, maximum: number) { + if (typeof value !== 'string') return null + const sanitized = value + .replace(/Bearer\s+[A-Za-z0-9._~+/=-]+/gi, 'Bearer [REDACTED]') + .replace(/data:[^;,\s]+;base64,[A-Za-z0-9+/=]+/gi, '[REDACTED_DATA_URL]') + .replace(/https?:\/\/\S+/gi, '[REDACTED_URL]') + .replace(/[\r\n\t]+/g, ' ') + .trim() + return sanitized ? sanitized.slice(0, maximum) : null +} + +async function readUpstreamErrorSummary( + response: Response, +): Promise { + const fallback: BomCostingUpstreamErrorSummary = { + code: null, + type: null, + param: null, + message: null, + } + const body = await response.text().catch(() => '') + if (!body.trim()) return fallback + + try { + const parsed = JSON.parse(body) as { + error?: { + code?: unknown + type?: unknown + param?: unknown + message?: unknown + } + code?: unknown + type?: unknown + param?: unknown + message?: unknown + } + const error = parsed.error ?? parsed + return { + code: sanitizeUpstreamErrorValue(error.code, 120), + type: sanitizeUpstreamErrorValue(error.type, 120), + param: sanitizeUpstreamErrorValue(error.param, 160), + message: sanitizeUpstreamErrorValue(error.message, 500), + } + } catch { + return { + ...fallback, + message: sanitizeUpstreamErrorValue(body, 500), + } + } +} + +function isCompleteJsonObjectPayload(content: string) { + const trimmed = content.trim() + if (!trimmed.startsWith('{') || !trimmed.endsWith('}')) return false + try { + const parsed = JSON.parse(trimmed) + return typeof parsed === 'object' && parsed !== null && !Array.isArray(parsed) + } catch { + return false + } +} + async function readStreamingChatCompletion( response: Response, options: { signal?: AbortSignal } = {}, @@ -250,7 +321,37 @@ async function readStreamingChatCompletion( let completionTokens: number | null = null let finishReason: string | null = null let receivedDoneMarker = false - const receivedCompletionSignal = () => receivedDoneMarker || finishReason !== null + let receivedCompleteJsonPayload = false + let acceptedCompleteJsonPayload = false + const receivedProtocolCompletionSignal = () => ( + receivedDoneMarker || finishReason !== null + ) + const receivedCompletionSignal = () => ( + receivedProtocolCompletionSignal() || acceptedCompleteJsonPayload + ) + + const readNextChunk = async () => { + if (!receivedCompleteJsonPayload) { + return { settled: false as const, result: await reader.read() } + } + + return new Promise< + | { settled: true; result?: never } + | { settled: false; result: ReadableStreamReadResult } + >((resolve, reject) => { + const timer = setTimeout(() => { + acceptedCompleteJsonPayload = true + resolve({ settled: true }) + }, BOM_COSTING_COMPLETE_JSON_SETTLE_MS) + void reader.read().then((result) => { + clearTimeout(timer) + resolve({ settled: false, result }) + }, (error: unknown) => { + clearTimeout(timer) + reject(error) + }) + }) + } const consumeLine = (line: string) => { const trimmed = line.trim() @@ -286,7 +387,16 @@ async function readStreamingChatCompletion( if (chunk.id?.trim()) callId = chunk.id.trim() if (chunk.model?.trim()) model = chunk.model.trim() const delta = chunk.choices?.[0]?.delta?.content - if (typeof delta === 'string') content += delta + if (typeof delta === 'string') { + content += delta + // response_format=json_object guarantees a single top-level object. A + // few OpenAI-compatible gateways keep the SSE connection open and omit + // both finish_reason and [DONE]. Once the accumulated top-level object + // parses successfully, the model answer is complete; cancelling here + // avoids discarding valid work at the transport timeout. Truncated JSON + // never passes this check and still follows the normal retry path. + receivedCompleteJsonPayload = isCompleteJsonObjectPayload(content) + } const chunkFinishReason = chunk.choices?.[0]?.finish_reason if (typeof chunkFinishReason === 'string' && chunkFinishReason.trim()) { finishReason = chunkFinishReason.trim() @@ -301,7 +411,9 @@ async function readStreamingChatCompletion( try { while (!receivedCompletionSignal()) { - const { done, value } = await reader.read() + const next = await readNextChunk() + if (next.settled) break + const { done, value } = next.result throwIfAborted() if (done) break buffer += decoder.decode(value, { stream: true }) @@ -309,7 +421,7 @@ async function readStreamingChatCompletion( buffer = lines.pop() ?? '' for (const line of lines) { consumeLine(line) - if (receivedCompletionSignal()) break + if (receivedProtocolCompletionSignal()) break } } if (!receivedCompletionSignal()) buffer += decoder.decode() @@ -324,6 +436,8 @@ async function readStreamingChatCompletion( pendingBufferChars: buffer.length, receivedDoneMarker, finishReason, + receivedCompleteJsonPayload, + acceptedCompleteJsonPayload, }, configurable: true, }) @@ -349,6 +463,8 @@ async function readStreamingChatCompletion( promptTokens, completionTokens, finishReason, + completedFromJsonPayload: acceptedCompleteJsonPayload && + !receivedDoneMarker && finishReason === null, } } @@ -526,12 +642,17 @@ export async function callBomCostingMultimodalModel(input: { } if (!response.ok) { + const upstreamError = await readUpstreamErrorSummary(response) console.warn('[BOM costing] multimodal upstream rejected request', { operation: input.operation, model: input.config.model, attempt, status: response.status, upstreamRequestId: readUpstreamRequestId(response.headers), + upstreamErrorCode: upstreamError.code, + upstreamErrorType: upstreamError.type, + upstreamErrorParam: upstreamError.param, + upstreamErrorMessage: upstreamError.message, durationMs: Date.now() - startedAt, }) if ( @@ -569,6 +690,8 @@ export async function callBomCostingMultimodalModel(input: { pendingBufferChars?: number receivedDoneMarker?: boolean finishReason?: string | null + receivedCompleteJsonPayload?: boolean + acceptedCompleteJsonPayload?: boolean } }).bomStreamMetrics : undefined @@ -583,6 +706,10 @@ export async function callBomCostingMultimodalModel(input: { pendingBufferChars: streamMetrics?.pendingBufferChars ?? null, receivedDoneMarker: streamMetrics?.receivedDoneMarker ?? false, finishReason: streamMetrics?.finishReason ?? null, + receivedCompleteJsonPayload: + streamMetrics?.receivedCompleteJsonPayload ?? false, + acceptedCompleteJsonPayload: + streamMetrics?.acceptedCompleteJsonPayload ?? false, }) if ( attempt < maxAttempts && @@ -620,6 +747,15 @@ export async function callBomCostingMultimodalModel(input: { '多模态分析服务没有返回可读取的结果', ) } + if (result.completedFromJsonPayload) { + console.info('[BOM costing] accepted complete JSON before stream terminator', { + operation: input.operation, + model: input.config.model, + attempt, + responseContentChars: result.content.length, + durationMs: Date.now() - startedAt, + }) + } return { ...result, model: result.model || input.config.model,