@@ -20,27 +20,58 @@ const {
2020 resolveProviderRequestPolicyConfigMock,
2121 shouldUseEnvHttpProxyForUrlMock,
2222 withTrustedEnvProxyGuardedFetchModeMock,
23-} = vi.hoisted(() => ({
24-buildProviderRequestDispatcherPolicyMock: vi.fn<
25-(_request?: unknown) => { mode: "direct" } | undefined
26->(() => undefined),
27-fetchWithSsrFGuardMock: vi.fn(),
28-ensureModelProviderLocalServiceMock: vi.fn(),
29-mergeModelProviderRequestOverridesMock: vi.fn((current, overrides) => ({
30- ...current,
31- ...overrides,
32-})),
33-resolveProviderRequestPolicyConfigMock: vi.fn<() => ProviderRequestPolicyConfigMockResult>(
34-() => ({
35-allowPrivateNetwork: false,
36-}),
37-),
38-shouldUseEnvHttpProxyForUrlMock: vi.fn(() => false),
39-withTrustedEnvProxyGuardedFetchModeMock: vi.fn((params: Record<string, unknown>) => ({
40- ...params,
41-mode: "trusted_env_proxy",
42-})),
43-}));
23+ managedStreamCleanupRegistrations,
24+} = vi.hoisted(() => {
25+const managedStreamCleanupRegistrations: Array<{
26+callback: (held: { finalize: () => Promise<void> }) => void;
27+held: { finalize: () => Promise<void> };
28+token: object;
29+}> = [];
30+31+class MockFinalizationRegistry {
32+constructor(private callback: (held: { finalize: () => Promise<void> }) => void) {}
33+34+register(_target: object, held: { finalize: () => Promise<void> }, token?: object) {
35+managedStreamCleanupRegistrations.push({
36+callback: this.callback,
37+ held,
38+token: token ?? {},
39+});
40+}
41+42+unregister(token: object) {
43+const index = managedStreamCleanupRegistrations.findIndex((entry) => entry.token === token);
44+if (index >= 0) {
45+managedStreamCleanupRegistrations.splice(index, 1);
46+}
47+}
48+}
49+50+vi.stubGlobal("FinalizationRegistry", MockFinalizationRegistry);
51+52+return {
53+buildProviderRequestDispatcherPolicyMock: vi.fn<
54+(_request?: unknown) => { mode: "direct" } | undefined
55+>(() => undefined),
56+fetchWithSsrFGuardMock: vi.fn(),
57+ensureModelProviderLocalServiceMock: vi.fn(),
58+mergeModelProviderRequestOverridesMock: vi.fn((current, overrides) => ({
59+ ...current,
60+ ...overrides,
61+})),
62+resolveProviderRequestPolicyConfigMock: vi.fn<() => ProviderRequestPolicyConfigMockResult>(
63+() => ({
64+allowPrivateNetwork: false,
65+}),
66+),
67+shouldUseEnvHttpProxyForUrlMock: vi.fn(() => false),
68+withTrustedEnvProxyGuardedFetchModeMock: vi.fn((params: Record<string, unknown>) => ({
69+ ...params,
70+mode: "trusted_env_proxy",
71+})),
72+ managedStreamCleanupRegistrations,
73+};
74+});
44754576vi.mock("../infra/net/fetch-guard.js", () => ({
4677fetchWithSsrFGuard: fetchWithSsrFGuardMock,
@@ -82,6 +113,7 @@ function latestTrustedEnvProxyParams(): Record<string, unknown> {
8211383114describe("buildGuardedModelFetch", () => {
84115beforeEach(() => {
116+managedStreamCleanupRegistrations.length = 0;
85117fetchWithSsrFGuardMock.mockReset().mockResolvedValue({
86118response: new Response("ok", { status: 200 }),
87119finalUrl: "https://api.openai.com/v1/responses",
@@ -151,6 +183,49 @@ describe("buildGuardedModelFetch", () => {
151183await vi.waitFor(() => expect(release).toHaveBeenCalledTimes(1));
152184});
153185186+it("releases guarded fetch slots when streamed bodies are abandoned", async () => {
187+const release = vi.fn(async () => undefined);
188+const encoder = new TextEncoder();
189+fetchWithSsrFGuardMock.mockResolvedValue({
190+response: new Response(
191+new ReadableStream<Uint8Array>({
192+start(controller) {
193+controller.enqueue(encoder.encode("chunk-1"));
194+controller.enqueue(encoder.encode("chunk-2"));
195+},
196+}),
197+{ status: 200 },
198+),
199+finalUrl: "https://api.anthropic.com/v1/messages",
200+ release,
201+});
202+const model = {
203+id: "claude-sonnet-4-6",
204+provider: "anthropic",
205+api: "anthropic-messages",
206+baseUrl: "https://api.anthropic.com",
207+} as unknown as Model<"anthropic-messages">;
208+209+const fetcher = buildGuardedModelFetch(model, undefined, { sanitizeSse: false });
210+let response = await fetcher("https://api.anthropic.com/v1/messages", {
211+method: "POST",
212+headers: { "content-type": "application/json" },
213+body: '{"stream":true}',
214+});
215+const reader = response.body?.getReader();
216+expect(reader).toBeDefined();
217+const firstChunk = await reader?.read();
218+expect(firstChunk?.done).toBe(false);
219+220+response = undefined as unknown as Response;
221+const registration = managedStreamCleanupRegistrations.at(-1);
222+expect(registration).toBeDefined();
223+await registration?.held.finalize();
224+225+expect(release).toHaveBeenCalledTimes(1);
226+expect(managedStreamCleanupRegistrations).toHaveLength(0);
227+});
228+154229it("passes model request headers to local service health probes", async () => {
155230const model = {
156231id: "deepseek-v4-flash",
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。