diff --git a/.trellis/tasks/04-19-04-19-proxy-sse-minimal-fix/prd.md b/.trellis/tasks/04-19-04-19-proxy-sse-minimal-fix/prd.md new file mode 100644 index 0000000..4b0b1d6 --- /dev/null +++ b/.trellis/tasks/04-19-04-19-proxy-sse-minimal-fix/prd.md @@ -0,0 +1,26 @@ +# Proxy SSE Minimal Fix + +## Summary + +The local Responses proxy can report `upstream_status=200` while still handing OpenCode a broken stream in two narrow cases inside `internal/proxy/server.go`: + +- the upstream returns `200` with `text/event-stream` but closes before sending any bytes +- the upstream sends one SSE chunk and then stays silent longer than `streamIdleTimeout`, causing the proxy to terminate a still-valid SSE stream + +This task applies the smallest possible fix in the proxy layer only, without changing CLI, desktop, app, config, or public interfaces. + +## Requirements + +- Keep the existing `firstByteTimeout` behavior. +- Treat `readFirstChunk()` returning `io.EOF` with zero bytes as a retryable pre-first-byte failure. +- Do not write downstream `200` for that empty first-read case. +- Preserve existing failover behavior for transport errors and `429`/`5xx` responses. +- After the first chunk succeeds, bypass `streamIdleTimeout` only for `text/event-stream` responses. +- Keep non-SSE responses on the existing idle-timeout path. +- Add regression coverage for empty-200-SSE failover and long-idle SSE continuation. + +## Non-Goals + +1. No proxy refactor. +2. No new config flags. +3. No changes outside `internal/proxy/server.go` and `internal/proxy/server_test.go`. diff --git a/.trellis/tasks/04-19-04-19-proxy-sse-minimal-fix/task.json b/.trellis/tasks/04-19-04-19-proxy-sse-minimal-fix/task.json new file mode 100644 index 0000000..09c84a1 --- /dev/null +++ b/.trellis/tasks/04-19-04-19-proxy-sse-minimal-fix/task.json @@ -0,0 +1,65 @@ +{ + "id": "04-19-proxy-sse-minimal-fix", + "name": "04-19-proxy-sse-minimal-fix", + "title": "Fix proxy SSE minimal failure handling", + "description": "Minimal proxy fix for empty 200 SSE failover and SSE idle timeout handling", + "status": "completed", + "dev_type": "feature", + "scope": "proxy", + "package": null, + "priority": "P1", + "creator": "OpenCode", + "assignee": "OpenCode", + "createdAt": "2026-04-19", + "completedAt": "2026-04-19", + "branch": "master", + "base_branch": "master", + "worktree_path": null, + "current_phase": 6, + "next_action": [ + { + "phase": 1, + "action": "brainstorm" + }, + { + "phase": 2, + "action": "research" + }, + { + "phase": 3, + "action": "implement" + }, + { + "phase": 4, + "action": "check" + }, + { + "phase": 5, + "action": "update-spec" + }, + { + "phase": 6, + "action": "record-session" + } + ], + "commit": null, + "pr_url": null, + "subtasks": [], + "children": [], + "parent": null, + "relatedFiles": [ + ".trellis/tasks/04-19-04-19-proxy-sse-minimal-fix/prd.md", + "internal/proxy/server.go", + "internal/proxy/server_test.go" + ], + "notes": "Applied a minimal SSE reliability fix in the local Responses proxy. Empty `200 text/event-stream` first reads are now treated as retryable pre-first-byte failures so failover can continue before any downstream status is written. After the first SSE chunk arrives, the proxy no longer applies `streamIdleTimeout` to later SSE reads, while non-SSE responses keep the existing timeout behavior. Added regression coverage for empty-200 SSE failover and long-idle SSE continuation, and verified with `go test ./internal/proxy`.", + "meta": { + "tests": [ + "go test ./internal/proxy" + ], + "regressions": [ + "empty 200 SSE failover", + "SSE bypasses idle timeout after first chunk" + ] + } +} diff --git a/internal/proxy/server.go b/internal/proxy/server.go index 9064c70..a55c651 100644 --- a/internal/proxy/server.go +++ b/internal/proxy/server.go @@ -8,6 +8,7 @@ import ( "fmt" "io" "log" + "mime" "net" "net/http" "strings" @@ -296,13 +297,18 @@ func (s *Server) tryOnce( } if errors.Is(firstErr, io.EOF) { if len(firstChunk) == 0 { - firstChunk = nil + return false, true, fmt.Errorf("upstream closed before first byte"), nil } } else { return false, true, fmt.Errorf("upstream first read: %w", firstErr), nil } } + isEventStream := false + if mediaType, _, parseErr := mime.ParseMediaType(resp.Header.Get("Content-Type")); parseErr == nil { + isEventStream = mediaType == "text/event-stream" + } + s.logger.Printf("alias=%s attempt=%d provider=%s remote_model=%s upstream_status=%d", aliasName, attempt, provider.ID, target.Model, resp.StatusCode) s.writeDebugHeaders(w, aliasName, provider.ID, target.Model, attempt, failoverCount) copyResponseHeaders(w.Header(), resp.Header) @@ -318,7 +324,15 @@ func (s *Server) tryOnce( } buf := make([]byte, 16<<10) for { - n, rerr := readChunkWithTimeout(resp.Body, buf, streamIdleTimeout) + var ( + n int + rerr error + ) + if isEventStream { + n, rerr = resp.Body.Read(buf) + } else { + n, rerr = readChunkWithTimeout(resp.Body, buf, streamIdleTimeout) + } if n > 0 { if _, werr := w.Write(buf[:n]); werr != nil { return true, false, werr, nil @@ -331,6 +345,7 @@ func (s *Server) tryOnce( if errors.Is(rerr, io.EOF) { return true, false, nil, nil } + s.logger.Printf("alias=%s attempt=%d provider=%s remote_model=%s upstream body read failed after response start: %v", aliasName, attempt, provider.ID, target.Model, rerr) return true, false, rerr, nil } } diff --git a/internal/proxy/server_test.go b/internal/proxy/server_test.go index 97ea718..0632e7d 100644 --- a/internal/proxy/server_test.go +++ b/internal/proxy/server_test.go @@ -107,6 +107,100 @@ func TestHandleResponsesFailsOverOn429(t *testing.T) { } } +func TestHandleResponsesFailsOverOnEmptySSE200(t *testing.T) { + t.Parallel() + + first := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + })) + defer first.Close() + + second := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("data: ok\n\n")) + })) + defer second.Close() + + srv := New(&config.Config{ + Server: config.Server{APIKey: config.DefaultLocalAPIKey}, + Providers: []config.Provider{ + {ID: "p1", BaseURL: first.URL + "/v1"}, + {ID: "p2", BaseURL: second.URL + "/v1"}, + }, + Aliases: []config.Alias{{ + Alias: "gpt-5.4", + Enabled: true, + Targets: []config.Target{{Provider: "p1", Model: "up-1", Enabled: true}, {Provider: "p2", Model: "up-2", Enabled: true}}, + }}, + }) + + req := httptest.NewRequest(http.MethodPost, "/v1/responses", strings.NewReader(`{"model":"gpt-5.4","stream":true}`)) + req.Header.Set("Authorization", "Bearer "+config.DefaultLocalAPIKey) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "text/event-stream") + rr := httptest.NewRecorder() + + srv.handleResponses(rr, req) + + if rr.Code != http.StatusOK { + t.Fatalf("status = %d, want %d", rr.Code, http.StatusOK) + } + if body := rr.Body.String(); body != "data: ok\n\n" { + t.Fatalf("body = %q, want SSE payload from second upstream", body) + } + if got := rr.Header().Get("X-OCSWITCH-Attempt"); got != "2" { + t.Fatalf("X-OCSWITCH-Attempt = %q, want 2", got) + } +} + +func TestHandleResponsesSSEBypassesIdleTimeout(t *testing.T) { + oldTimeout := streamIdleTimeout + streamIdleTimeout = 30 * time.Millisecond + defer func() { streamIdleTimeout = oldTimeout }() + + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("data: first\n\n")) + if flusher, ok := w.(http.Flusher); ok { + flusher.Flush() + } + time.Sleep(90 * time.Millisecond) + _, _ = w.Write([]byte("data: second\n\n")) + if flusher, ok := w.(http.Flusher); ok { + flusher.Flush() + } + })) + defer upstream.Close() + + srv := New(&config.Config{ + Server: config.Server{APIKey: config.DefaultLocalAPIKey}, + Providers: []config.Provider{{ID: "p1", BaseURL: upstream.URL + "/v1"}}, + Aliases: []config.Alias{{ + Alias: "gpt-5.4", + Enabled: true, + Targets: []config.Target{{Provider: "p1", Model: "up-1", Enabled: true}}, + }}, + }) + + req := httptest.NewRequest(http.MethodPost, "/v1/responses", strings.NewReader(`{"model":"gpt-5.4","stream":true}`)) + req.Header.Set("Authorization", "Bearer "+config.DefaultLocalAPIKey) + req.Header.Set("Content-Type", "application/json") + req.Header.Set("Accept", "text/event-stream") + rr := httptest.NewRecorder() + + srv.handleResponses(rr, req) + + if rr.Code != http.StatusOK { + t.Fatalf("status = %d, want %d", rr.Code, http.StatusOK) + } + if body := rr.Body.String(); body != "data: first\n\ndata: second\n\n" { + t.Fatalf("body = %q, want both SSE chunks", body) + } +} + func TestHandleResponsesDoesNotFailOverOn400(t *testing.T) { t.Parallel()