diff --git a/open-sse/utils/stream.ts b/open-sse/utils/stream.ts index 9d358ecd7f..cd2b16863e 100644 --- a/open-sse/utils/stream.ts +++ b/open-sse/utils/stream.ts @@ -1334,6 +1334,7 @@ export function createSSEStream(options: StreamOptions = {}) { } catch {} } clearIdleTimer(); + trackPendingRequest(model, provider, connectionId, false); controller.error(new Error(failurePayload.message || "Upstream failure")); return; } diff --git a/tests/unit/stream-utils.test.ts b/tests/unit/stream-utils.test.ts index 50061e2ac3..e552551ebe 100644 --- a/tests/unit/stream-utils.test.ts +++ b/tests/unit/stream-utils.test.ts @@ -972,3 +972,63 @@ test("createRequestLogger caps retained stream chunk item count", async () => { "[stream chunk log truncated after 2 chunks]", ]); }); + +// T-VERIFY: passthrough mode failure decrements pending requests +// Regression test for missing trackPendingRequest(false) on passthrough failure +import { getPendingRequests, clearPendingRequests } from "../../src/lib/usage/usageHistory.ts"; + +test("createSSEStream passthrough mode decrements pending requests on failure", async () => { + // Clear any existing pending requests first + clearPendingRequests(); + const initial = getPendingRequests(); + assert.equal(Object.keys(initial.byModel).length, 0, "should start with no pending requests"); + + let failurePayload = null; + const testProvider = "openai-compatible-test-failure"; + const testModel = "gpt-test"; + const testConnectionId = "test-conn-123"; + + await assert.rejects( + readTransformed( + [ + `data: ${JSON.stringify({ + type: "response.failed", + response: { + id: "resp_failed_test", + object: "response", + model: testModel, + status: "failed", + error: { + code: "test_failure", + message: "Test failure for pending request tracking", + }, + }, + })}\n\n`, + ], + { + mode: "passthrough", + sourceFormat: FORMATS.OPENAI_RESPONSES, + provider: testProvider, + model: testModel, + connectionId: testConnectionId, + body: { input: "hello" }, + onFailure(payload) { + failurePayload = payload; + }, + } + ), + /Test failure|Upstream failure/ + ); + + assert.ok(failurePayload, "should report the stream failure"); + + // Verify pending requests are properly decremented after failure + const pending = getPendingRequests(); + const modelKey = `${testModel} (${testProvider})`; + const count = pending.byModel[modelKey] || 0; + assert.equal( + count, + 0, + `pending request count for ${modelKey} should be 0 after failure, got ${count}` + ); +});