fix: terminate invalid provider streams

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()

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")

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));
   }

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: {