diff --git a/forge-core/llm/providers/responses.go b/forge-core/llm/providers/responses.go index f0685c9..b37eb30 100644 --- a/forge-core/llm/providers/responses.go +++ b/forge-core/llm/providers/responses.go @@ -372,6 +372,11 @@ func (c *ResponsesClient) readStream(r io.Reader, ch chan<- llm.StreamDelta) { pendingFCs := make(map[int]*pendingFC) scanner := bufio.NewScanner(r) + // A single SSE `data:` frame can carry the full terminal response — + // `response.completed` embeds the entire output[] + usage — which + // overruns bufio.Scanner's default 64KB line cap on a large answer and + // aborts the stream mid-parse. Give it generous headroom. + scanner.Buffer(make([]byte, 0, 64*1024), 8*1024*1024) var currentEvent string for scanner.Scan() { @@ -387,7 +392,24 @@ func (c *ResponsesClient) readStream(r io.Reader, ch chan<- llm.StreamDelta) { continue } - switch currentEvent { + // Determine the event type. The OpenAI Responses API sends BOTH an + // SSE `event:` line and a `type` field inside the data payload, but + // `event:` is optional per the SSE spec and some gateways (e.g. the + // Bedrock openai-sigv4 shim fronting this provider) omit it, carrying + // the type only in the JSON. Dispatching on the `event:` line alone + // drops EVERY frame from such a server — empty content, zero usage, + // empty finish reason. Prefer the payload `type` (authoritative and + // present in both dialects); fall back to the `event:` line when the + // payload carries no type. + eventType := currentEvent + var typed struct { + Type string `json:"type"` + } + if json.Unmarshal([]byte(data), &typed) == nil && typed.Type != "" { + eventType = typed.Type + } + + switch eventType { case "response.output_text.delta": var ev streamTextDelta if err := json.Unmarshal([]byte(data), &ev); err != nil { diff --git a/forge-core/llm/providers/responses_test.go b/forge-core/llm/providers/responses_test.go index 03d9318..07e5a1a 100644 --- a/forge-core/llm/providers/responses_test.go +++ b/forge-core/llm/providers/responses_test.go @@ -63,3 +63,70 @@ func TestResponsesClient_NoOrgIDHeader(t *testing.T) { t.Errorf("expected no OpenAI-Organization header, got %q", gotHeader) } } + +// TestResponsesClient_DispatchesOnPayloadType_NoEventLines is the regression +// for the Bedrock openai-sigv4 gateway: it streams spec-legal SSE with NO +// `event:` lines, carrying the event type only as a `type` field inside each +// `data:` payload. The old parser routed solely on the `event:` line, so +// `currentEvent` stayed "" and EVERY frame was dropped — empty content, zero +// usage, empty finish reason (the field-observed failure). The parser must +// fall back to the payload `type` and recover both text and usage. +func TestResponsesClient_DispatchesOnPayloadType_NoEventLines(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + // No `event:` lines — type lives inside the JSON, exactly as the + // gateway sends it. + _, _ = w.Write([]byte("data: {\"output_index\":0,\"content_index\":0,\"delta\":\"Hello\",\"type\":\"response.output_text.delta\"}\n\n")) + _, _ = w.Write([]byte("data: {\"output_index\":0,\"content_index\":0,\"delta\":\" world\",\"type\":\"response.output_text.delta\"}\n\n")) + _, _ = w.Write([]byte("data: {\"response\":{\"id\":\"resp-1\",\"status\":\"completed\",\"output\":[{\"type\":\"message\",\"role\":\"assistant\",\"content\":[{\"type\":\"output_text\",\"text\":\"Hello world\"}]}],\"usage\":{\"input_tokens\":76,\"output_tokens\":73,\"total_tokens\":149}},\"type\":\"response.completed\"}\n\n")) + })) + defer srv.Close() + + client := NewResponsesClient(llm.ClientConfig{ + APIKey: "sk-test", + Model: "openai.gpt-5.4", + BaseURL: srv.URL, + }) + + resp, err := client.Chat(context.Background(), &llm.ChatRequest{ + Messages: []llm.ChatMessage{{Role: llm.RoleUser, Content: "hi"}}, + }) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if resp.Message.Content != "Hello world" { + t.Errorf("content = %q, want %q — text deltas dropped (no event: lines)", resp.Message.Content, "Hello world") + } + if resp.Usage.InputTokens != 76 || resp.Usage.OutputTokens != 73 || resp.Usage.TotalTokens != 149 { + t.Errorf("usage = %+v, want in=76 out=73 total=149 — response.completed frame dropped", resp.Usage) + } + if resp.FinishReason != "stop" { + t.Errorf("finish_reason = %q, want \"stop\" — the empty value was the field symptom", resp.FinishReason) + } +} + +// TestResponsesClient_EventLineDialectStillWorks guards backward compatibility: +// real OpenAI sends BOTH an `event:` line and a payload `type`. The payload-type +// dispatch must not regress that dialect. +func TestResponsesClient_EventLineDialectStillWorks(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + _, _ = w.Write([]byte("event: response.output_text.delta\ndata: {\"output_index\":0,\"content_index\":0,\"delta\":\"ok\",\"type\":\"response.output_text.delta\"}\n\n")) + _, _ = w.Write([]byte("event: response.completed\ndata: {\"response\":{\"id\":\"resp-2\",\"status\":\"completed\",\"output\":[],\"usage\":{\"input_tokens\":3,\"output_tokens\":2,\"total_tokens\":5}},\"type\":\"response.completed\"}\n\n")) + })) + defer srv.Close() + + client := NewResponsesClient(llm.ClientConfig{APIKey: "sk-test", Model: "gpt-4o", BaseURL: srv.URL}) + resp, err := client.Chat(context.Background(), &llm.ChatRequest{ + Messages: []llm.ChatMessage{{Role: llm.RoleUser, Content: "hi"}}, + }) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if resp.Message.Content != "ok" { + t.Errorf("content = %q, want \"ok\"", resp.Message.Content) + } + if resp.Usage.TotalTokens != 5 { + t.Errorf("usage total = %d, want 5", resp.Usage.TotalTokens) + } +}