diff --git a/src/app/v1/_lib/proxy/forwarder.ts b/src/app/v1/_lib/proxy/forwarder.ts index 84bc26942..63b7473a1 100644 --- a/src/app/v1/_lib/proxy/forwarder.ts +++ b/src/app/v1/_lib/proxy/forwarder.ts @@ -116,6 +116,11 @@ import { setDeferredStreamingFinalization, } from "./stream-finalization"; import { mapProviderTypeToFamily } from "./stream-gate/frame-classifier"; +import { + getStreamGateResponsePolicy, + inheritStreamGateResponsePolicy, + setStreamGateResponsePolicy, +} from "./stream-gate/response-policy"; import { concatChunks, resolveStreamGateCaps, @@ -1734,6 +1739,8 @@ export class ProxyForwarder { family: gateFamily, providerId: currentProvider.id, providerName: currentProvider.name, + allowTerminalOnlyCommit: + getStreamGateResponsePolicy(response)?.allowTerminalOnlyCommit, ...resolveStreamGateCaps(), // 首字节到达即清除首字节计时器，保持「首字节超时」的原始语义—— // 思考型模型可在首个内容帧前长时间输出中性帧，不应触发该计时器 @@ -1811,7 +1818,7 @@ export class ProxyForwarder { : {}), }); - streamingResponse = new Response( + const gatedResponse = new Response( ProxyForwarder.buildBufferedPrefixStream(gate.prefixChunks, gateReader), { status: response.status, @@ -1819,6 +1826,8 @@ export class ProxyForwarder { headers: response.headers, } ); + inheritStreamGateResponsePolicy(response, gatedResponse); + streamingResponse = gatedResponse; } } @@ -3545,6 +3554,11 @@ export class ProxyForwarder { if ("response" in wsResult) { responsesWsResponse = wsResult.response; + if (requestBodyJson.generate === false) { + setStreamGateResponsePolicy(responsesWsResponse, { + allowTerminalOnlyCommit: true, + }); + } logger.info("ProxyForwarder: Upstream Responses WebSocket connected", { providerId: provider.id, providerName: provider.name, @@ -4788,6 +4802,8 @@ export class ProxyForwarder { family: hedgeGateFamily, providerId: attempt.provider.id, providerName: attempt.provider.name, + allowTerminalOnlyCommit: + getStreamGateResponsePolicy(response)?.allowTerminalOnlyCommit, ...resolveStreamGateCaps(), // 首字节时刻先挂在 attempt 上，由 commitWinner 决定是否记为 session TTFB onFirstByte: () => { @@ -5255,6 +5271,7 @@ export class ProxyForwarder { headers: attempt.response.headers, } ); + inheritStreamGateResponsePolicy(attempt.response, response); settleSuccess(response); }; @@ -6183,16 +6200,16 @@ export class ProxyForwarder { providerSessionRefRetainOnSuccess: attempt.providerSessionRefRetainOnSuccess, }); leaseTransferred = true; - resolveResult?.({ - response: new Response( - ProxyForwarder.buildBufferedPrefixStream(attempt.chunks, attempt.reader), - { - status: attempt.response.status, - statusText: attempt.response.statusText, - headers: attempt.response.headers, - } - ), - }); + const response = new Response( + ProxyForwarder.buildBufferedPrefixStream(attempt.chunks, attempt.reader), + { + status: attempt.response.status, + statusText: attempt.response.statusText, + headers: attempt.response.headers, + } + ); + inheritStreamGateResponsePolicy(attempt.response, response); + resolveResult?.({ response }); }; const clearCapturedStickyBinding = async (cooldownTtlSeconds: number): Promise => { diff --git a/src/app/v1/_lib/proxy/response-handler.ts b/src/app/v1/_lib/proxy/response-handler.ts index d06c75c65..edb24a2b6 100644 --- a/src/app/v1/_lib/proxy/response-handler.ts +++ b/src/app/v1/_lib/proxy/response-handler.ts @@ -75,6 +75,7 @@ import { peekDeferredStreamingFinalization, } from "./stream-finalization"; import { mapProviderTypeToFamily } from "./stream-gate/frame-classifier"; +import { getStreamGateResponsePolicy } from "./stream-gate/response-policy"; import { createShadowGateObserver, resolveStreamGateMode } from "./stream-gate/stream-content-gate"; import { createStreamProtocolObserver, @@ -3567,6 +3568,7 @@ export class ProxyResponseHandler { family, providerId: provider.id, providerName: provider.name, + allowTerminalOnlyCommit: getStreamGateResponsePolicy(response)?.allowTerminalOnlyCommit, }); })(); @@ -4761,6 +4763,7 @@ export class ProxyResponseHandler { family, providerId: provider.id, providerName: provider.name, + allowTerminalOnlyCommit: getStreamGateResponsePolicy(response)?.allowTerminalOnlyCommit, }); })(); diff --git a/src/app/v1/_lib/proxy/stream-gate/response-policy.ts b/src/app/v1/_lib/proxy/stream-gate/response-policy.ts new file mode 100644 index 000000000..1776472d7 --- /dev/null +++ b/src/app/v1/_lib/proxy/stream-gate/response-policy.ts @@ -0,0 +1,23 @@ +export interface StreamGateResponsePolicy { + allowTerminalOnlyCommit: boolean; +} + +const responsePolicies = new WeakMap(); + +export function setStreamGateResponsePolicy( + response: Response, + policy: StreamGateResponsePolicy +): void { + responsePolicies.set(response, policy); +} + +export function getStreamGateResponsePolicy( + response: Response +): StreamGateResponsePolicy | undefined { + return responsePolicies.get(response); +} + +export function inheritStreamGateResponsePolicy(source: Response, target: Response): void { + const policy = responsePolicies.get(source); + if (policy) responsePolicies.set(target, policy); +} diff --git a/src/app/v1/_lib/proxy/stream-gate/stream-content-gate.ts b/src/app/v1/_lib/proxy/stream-gate/stream-content-gate.ts index e286306a4..2d909a770 100644 --- a/src/app/v1/_lib/proxy/stream-gate/stream-content-gate.ts +++ b/src/app/v1/_lib/proxy/stream-gate/stream-content-gate.ts @@ -125,6 +125,8 @@ export interface StreamGateOptions extends StreamGateCaps { family: ProtocolFamily; providerId: number; providerName: string; + /** 允许已显式标记的请求以干净终止帧作为有效的 precommit 提交点 */ + allowTerminalOnlyCommit?: boolean; /** 首个非空上游 chunk 到达时回调一次（调用方用于清除首字节计时器，恢复其原始语义） */ onFirstByte?: () => void; /** 门控等待期的读间隔静默上限（毫秒；<=0 或未设不启用），对齐提交后 response-handler 的静默超时 */ @@ -225,6 +227,9 @@ export async function runStreamContentGate( } if (verdict === "error") return failure("gate_error", frame.data); if (verdict === "malformed") return failure("decode_error", frame.data); + if (verdict === "terminal" && options.allowTerminalOnlyCommit) { + return commit(frame.eventName, true); + } } return failure("empty_stream"); } @@ -255,6 +260,9 @@ export async function runStreamContentGate( return failure("decode_error", frame.data); } if (verdict === "terminal") { + if (options.allowTerminalOnlyCommit) { + return commit(frame.eventName, false); + } // 干净终止先于任何内容 = 空流 return failure("empty_stream", frame.data); } @@ -332,6 +340,7 @@ export function createShadowGateObserver(context: { family: ProtocolFamily; providerId: number; providerName: string; + allowTerminalOnlyCommit?: boolean; }): ShadowGateObserver { const parser = new SseFrameParser(); const verdictCounts: Record = { @@ -354,17 +363,27 @@ export function createShadowGateObserver(context: { for (const frame of parser.push(chunk)) { const verdict = classifyFrame(context.family, frame.eventName, frame.data); verdictCounts[verdict]++; - if (verdict === "content" || verdict === "error" || verdict === "malformed") { + if ( + verdict === "content" || + verdict === "error" || + verdict === "malformed" || + verdict === "terminal" + ) { reported = true; + const enforcementDecision = + verdict === "content" || + (verdict === "terminal" && context.allowTerminalOnlyCommit === true) + ? "wouldCommit" + : "wouldReject"; logger.info("StreamGate[shadow]: first decisive frame observed", { providerId: context.providerId, providerName: context.providerName, family: context.family, decisiveVerdict: verdict, + enforcementDecision, // 现状「首非空字节即提交」与门控「首有效内容才提交」的判定分歧： - // divergent=true 表示门控会推迟提交（中性前缀）或触发 failover（error/malformed） - divergent: - verdict !== "content" || verdictCounts.neutral + verdictCounts.terminal > 0, + // divergent=true 表示门控会推迟提交（中性前缀），或按当前策略拒绝该终态 + divergent: enforcementDecision === "wouldReject" || verdictCounts.neutral > 0, firstContentLagMs: firstByteAt === null ? null : Date.now() - firstByteAt, verdictCounts: { ...verdictCounts }, }); diff --git a/tests/unit/proxy/proxy-forwarder-hedge-first-byte.test.ts b/tests/unit/proxy/proxy-forwarder-hedge-first-byte.test.ts index 02cab8a5f..771746874 100644 --- a/tests/unit/proxy/proxy-forwarder-hedge-first-byte.test.ts +++ b/tests/unit/proxy/proxy-forwarder-hedge-first-byte.test.ts @@ -175,6 +175,10 @@ import { ProxyForwarder } from "@/app/v1/_lib/proxy/forwarder"; import { ModelRedirector } from "@/app/v1/_lib/proxy/model-redirector"; import { ProxySession } from "@/app/v1/_lib/proxy/session"; import { peekDeferredStreamingFinalization } from "@/app/v1/_lib/proxy/stream-finalization"; +import { + getStreamGateResponsePolicy, + setStreamGateResponsePolicy, +} from "@/app/v1/_lib/proxy/stream-gate/response-policy"; import { DbPoolAdmissionError } from "@/drizzle/admitted-client"; import { logger } from "@/lib/logger"; import type { Provider } from "@/types/provider"; @@ -1286,6 +1290,29 @@ describe("ProxyForwarder - first-byte hedge scheduling", () => { } }); + test("hedge winner response wrapper preserves the stream gate response policy", async () => { + const provider = createProvider({ id: 1, name: "ws-prewarm" }); + const session = createSession(); + setProviderWithSessionRef(session, provider); + + const upstreamResponse = new Response( + 'data: {"type":"content_block_delta","delta":{"text":"winner"}}\n\n', + { headers: { "content-type": "text/event-stream" } } + ); + setStreamGateResponsePolicy(upstreamResponse, { allowTerminalOnlyCommit: true }); + vi.spyOn( + ProxyForwarder as unknown as { + doForward: (...args: unknown[]) => Promise; + }, + "doForward" + ).mockResolvedValueOnce(upstreamResponse); + + const response = await ProxyForwarder.send(session); + + expect(await response.text()).toContain('"text":"winner"'); + expect(getStreamGateResponsePolicy(response)).toEqual({ allowTerminalOnlyCommit: true }); + }); + test("hedge skips provider when concurrent session acquire is rejected", async () => { vi.useFakeTimers(); diff --git a/tests/unit/proxy/stream-gate-content-gate.test.ts b/tests/unit/proxy/stream-gate-content-gate.test.ts index 56140dd01..8a1271f06 100644 --- a/tests/unit/proxy/stream-gate-content-gate.test.ts +++ b/tests/unit/proxy/stream-gate-content-gate.test.ts @@ -1,10 +1,17 @@ -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import { EmptyResponseError, ProxyError } from "@/app/v1/_lib/proxy/errors"; +import { + getStreamGateResponsePolicy, + inheritStreamGateResponsePolicy, + setStreamGateResponsePolicy, +} from "@/app/v1/_lib/proxy/stream-gate/response-policy"; import { concatChunks, + createShadowGateObserver, runStreamContentGate, StreamPrecommitError, } from "@/app/v1/_lib/proxy/stream-gate/stream-content-gate"; +import { logger } from "@/lib/logger"; const encoder = new TextEncoder(); @@ -254,6 +261,72 @@ describe("runStreamContentGate", () => { expect(new TextDecoder().decode(rest.value)).toBe(completed); }); + it("openai-responses: an unmarked terminal-only response remains an empty stream", async () => { + const completed = + 'event: response.completed\ndata: {"type":"response.completed","response":{"status":"completed"}}\n\n'; + const result = await runStreamContentGate(readerFromChunks([completed]), { + ...GATE_OPTIONS, + family: "openai-responses", + }); + + expect(result.committed).toBe(false); + if (result.committed) return; + expect((result.error as StreamPrecommitError).gateReason).toBe("empty_stream"); + }); + + it.each([ + { suffix: "with a frame delimiter", terminalSuffix: "\n\n", readerDone: false }, + { suffix: "when flushed at EOF", terminalSuffix: "", readerDone: true }, + ])("openai-responses: allowed terminal-only response commits $suffix", async (testCase) => { + const completed = `event: response.completed\ndata: {"type":"response.completed","response":{"status":"completed"}}${testCase.terminalSuffix}`; + const result = await runStreamContentGate(readerFromChunks([completed]), { + ...GATE_OPTIONS, + family: "openai-responses", + allowTerminalOnlyCommit: true, + }); + + expect(result.committed).toBe(true); + if (!result.committed) return; + expect(await drainPrefix(result.prefixChunks)).toBe(completed); + expect(result.readerDone).toBe(testCase.readerDone); + }); + + it("openai-responses: allowed policy still rejects an error frame", async () => { + const error = + 'event: error\ndata: {"type":"error","code":"server_error","message":"failed"}\n\n'; + const result = await runStreamContentGate(readerFromChunks([error]), { + ...GATE_OPTIONS, + family: "openai-responses", + allowTerminalOnlyCommit: true, + }); + + expect(result.committed).toBe(false); + if (result.committed) return; + expect((result.error as StreamPrecommitError).gateReason).toBe("gate_error"); + }); + + it("openai-responses: allowed policy still rejects malformed frames and bare EOF", async () => { + const malformed = await runStreamContentGate(readerFromChunks(["data: {broken\n\n"]), { + ...GATE_OPTIONS, + family: "openai-responses", + allowTerminalOnlyCommit: true, + }); + expect(malformed.committed).toBe(false); + if (!malformed.committed) { + expect((malformed.error as StreamPrecommitError).gateReason).toBe("decode_error"); + } + + const bareEof = await runStreamContentGate(readerFromChunks([]), { + ...GATE_OPTIONS, + family: "openai-responses", + allowTerminalOnlyCommit: true, + }); + expect(bareEof.committed).toBe(false); + if (!bareEof.committed) { + expect((bareEof.error as StreamPrecommitError).gateReason).toBe("empty_stream"); + } + }); + it("openai-responses: commits compaction carried only by response.completed", async () => { const completed = 'event: response.completed\ndata: {"type":"response.completed","response":{"status":"completed","output":[{"type":"compaction","encrypted_content":"opaque-state"}]}}\n\n'; @@ -300,6 +373,59 @@ describe("runStreamContentGate", () => { }); }); +describe("stream gate response policy", () => { + it("stores policy internally and inherits it without exposing response headers", () => { + const source = new Response(null, { headers: { "x-source": "1" } }); + const target = new Response(null, { headers: { "x-target": "1" } }); + const sourceHeadersBefore = [...source.headers]; + const targetHeadersBefore = [...target.headers]; + + setStreamGateResponsePolicy(source, { allowTerminalOnlyCommit: true }); + inheritStreamGateResponsePolicy(source, target); + + expect(getStreamGateResponsePolicy(source)).toEqual({ allowTerminalOnlyCommit: true }); + expect(getStreamGateResponsePolicy(target)).toEqual({ allowTerminalOnlyCommit: true }); + expect([...source.headers]).toEqual(sourceHeadersBefore); + expect([...target.headers]).toEqual(targetHeadersBefore); + }); + + it("does not add a policy when the source is unmarked", () => { + const target = new Response(); + inheritStreamGateResponsePolicy(new Response(), target); + expect(getStreamGateResponsePolicy(target)).toBeUndefined(); + }); +}); + +describe("shadow gate terminal telemetry", () => { + it.each([ + { allowTerminalOnlyCommit: true, enforcementDecision: "wouldCommit" }, + { allowTerminalOnlyCommit: false, enforcementDecision: "wouldReject" }, + ])("reports $enforcementDecision for a terminal frame", (testCase) => { + const info = vi.spyOn(logger, "info").mockImplementation(() => undefined); + const observer = createShadowGateObserver({ + family: "openai-responses", + providerId: 7, + providerName: "test-provider", + allowTerminalOnlyCommit: testCase.allowTerminalOnlyCommit, + }); + + observer.observe( + encoder.encode( + 'event: response.completed\ndata: {"type":"response.completed","response":{"status":"completed"}}\n\n' + ) + ); + + expect(info).toHaveBeenCalledWith( + "StreamGate[shadow]: first decisive frame observed", + expect.objectContaining({ + decisiveVerdict: "terminal", + enforcementDecision: testCase.enforcementDecision, + }) + ); + info.mockRestore(); + }); +}); + describe("concatChunks", () => { it("returns null for empty, identity for single, concatenation for many", () => { expect(concatChunks([])).toBeNull(); diff --git a/tests/unit/proxy/stream-gate-forwarder-integration.test.ts b/tests/unit/proxy/stream-gate-forwarder-integration.test.ts index 23d4beb42..1e358dd73 100644 --- a/tests/unit/proxy/stream-gate-forwarder-integration.test.ts +++ b/tests/unit/proxy/stream-gate-forwarder-integration.test.ts @@ -1,5 +1,9 @@ import { beforeEach, describe, expect, test, vi } from "vitest"; import { resolveEndpointPolicy } from "@/app/v1/_lib/proxy/endpoint-policy"; +import { + getStreamGateResponsePolicy, + setStreamGateResponsePolicy, +} from "@/app/v1/_lib/proxy/stream-gate/response-policy"; /** * F1 流式内容门控（stream content gate）在 ProxyForwarder 顺序路径中的接线集成测试。 @@ -224,6 +228,11 @@ const OPENAI_RESPONSES_WINNER_FRAMES = [ }), ]; +const OPENAI_RESPONSES_TERMINAL_ONLY_FRAME = sseFrame("response.completed", { + type: "response.completed", + response: { id: "resp_prewarm", status: "completed" }, +}); + const VALID_OPENAI_RESPONSES_STREAMS = [ { name: "terminal compaction output", @@ -569,6 +578,29 @@ describe("F1 stream content gate x ProxyForwarder sequential path", () => { expect(mocks.recordFailure).not.toHaveBeenCalled(); }); + test("标记的 Responses WebSocket 预热终止帧提交，并在顺序包装后保留策略", async () => { + const provider = createProvider({ id: 1, name: "ws-prewarm", providerType: "codex" }); + const session = createSession(); + session.setProvider(provider); + Object.assign(session, { + requestUrl: new URL("https://example.com/v1/responses"), + originalFormat: "response", + endpointPolicy: resolveEndpointPolicy("/v1/responses"), + }); + + const upstreamResponse = createSseResponse([OPENAI_RESPONSES_TERMINAL_ONLY_FRAME]); + setStreamGateResponsePolicy(upstreamResponse, { allowTerminalOnlyCommit: true }); + const doForward = spyOnDoForward(); + doForward.mockResolvedValueOnce(upstreamResponse); + + const response = await ProxyForwarder.send(session); + + expect(await response.text()).toBe(OPENAI_RESPONSES_TERMINAL_ONLY_FRAME); + expect(getStreamGateResponsePolicy(response)).toEqual({ allowTerminalOnlyCommit: true }); + expect(doForward).toHaveBeenCalledTimes(1); + expect(mocks.recordFailure).not.toHaveBeenCalled(); + }); + test("terminal-only 流（message_stop 即终止）按 empty_stream 失败并切换供应商", async () => { const provider1 = createProvider({ id: 1, name: "gate-p1" }); const provider2 = createProvider({ id: 2, name: "gate-p2" });