diff --git a/apps/workshop/homarr_provider.go b/apps/workshop/homarr_provider.go index 0c79af62d..8129d5130 100644 --- a/apps/workshop/homarr_provider.go +++ b/apps/workshop/homarr_provider.go @@ -318,7 +318,9 @@ func (provider *homarrProvider) chat(event *core.RequestEvent) error { if requestBody.Stream { event.Response.Header().Set("Content-Type", contentType) event.Response.WriteHeader(upstreamResponse.StatusCode) - _, _ = copyAliasedSSE(flushingWriter{event.Response}, upstreamResponse.Body, maxChatResponseBytes) + if _, streamErr := copyAliasedSSE(flushingWriter{event.Response}, upstreamResponse.Body, maxChatResponseBytes); streamErr != nil { + _, _ = io.WriteString(flushingWriter{event.Response}, streamErrorFrame()) + } return nil } @@ -333,6 +335,16 @@ func (provider *homarrProvider) chat(event *core.RequestEvent) error { return event.Blob(upstreamResponse.StatusCode, contentType, responseBody) } +func streamErrorFrame() string { + payload, _ := json.Marshal(map[string]any{ + "error": map[string]any{ + "message": "The model endpoint returned an invalid response.", + "type": "homarr_provider_error", + }, + }) + return "data: " + string(payload) + "\n\ndata: [DONE]\n\n" +} + func (provider *homarrProvider) acquireRequest(userID string) bool { provider.inFlightMutex.Lock() defer provider.inFlightMutex.Unlock() diff --git a/apps/workshop/homarr_provider_test.go b/apps/workshop/homarr_provider_test.go index dcdb6df9a..d94bdffe7 100644 --- a/apps/workshop/homarr_provider_test.go +++ b/apps/workshop/homarr_provider_test.go @@ -173,8 +173,14 @@ func TestValidateProviderInput(t *testing.T) { func TestProviderInFlightLimits(t *testing.T) { provider := &homarrProvider{inFlightByUser: make(map[string]int)} - if !provider.acquireRequest("one") || !provider.acquireRequest("one") || provider.acquireRequest("one") { - t.Fatal("expected the per-user in-flight limit to allow two requests") + if !provider.acquireRequest("one") { + t.Fatal("expected the first user request to be admitted") + } + if !provider.acquireRequest("one") { + t.Fatal("expected the second user request to be admitted") + } + if provider.acquireRequest("one") { + t.Fatal("expected the third user request to be rejected") } provider.releaseRequest("one") provider.releaseRequest("one") diff --git a/apps/workshop/tests/openrouter-mock.mjs b/apps/workshop/tests/openrouter-mock.mjs index 8cd496ae8..39fe89bf7 100644 --- a/apps/workshop/tests/openrouter-mock.mjs +++ b/apps/workshop/tests/openrouter-mock.mjs @@ -81,6 +81,11 @@ const server = createServer(async (request, response) => { json(response, 503, { error: { message: "Simulated upstream failure" } }); return; } + if (body.messages?.some((message) => message.content === "malformed stream")) { + response.writeHead(200, { "content-type": "text/event-stream" }); + response.end('data: {"model":\n\n'); + return; + } if (body.messages?.some((message) => message.content === "in-flight probe")) { await new Promise((resolve) => setTimeout(resolve, 500)); } diff --git a/apps/workshop/tests/workshop.integration.mjs b/apps/workshop/tests/workshop.integration.mjs index 0fd15c49e..91baed468 100644 --- a/apps/workshop/tests/workshop.integration.mjs +++ b/apps/workshop/tests/workshop.integration.mjs @@ -278,6 +278,24 @@ if ( ) { throw new Error("The Homarr provider did not preserve the upstream SSE stream"); } +const malformedStreamResponse = await fetch(`${baseUrl}/api/ai/v1/chat/completions`, { + method: "POST", + headers: { "content-type": "application/json", ...concurrencySession.headers }, + body: JSON.stringify({ + model: "homarr/model", + messages: [{ role: "user", content: "malformed stream" }], + stream: true, + }), +}); +const malformedStreamText = await malformedStreamResponse.text(); +if ( + !malformedStreamResponse.ok || + !malformedStreamText.includes('"type":"homarr_provider_error"') || + malformedStreamText.includes('data: {"model":') || + !malformedStreamText.endsWith("data: [DONE]\n\n") +) { + throw new Error(`Malformed upstream streams did not terminate safely: ${malformedStreamText}`); +} const toolResponse = await fetch(`${baseUrl}/api/ai/v1/chat/completions`, { method: "POST", headers: {