fix: retry verified mislabeled SSE disconnect

classify.go

@@ -224,7 +224,11 @@ func classifySSEError(data string) *failure {
 		return &failure{transient: true, status: 429, atype: "rate_limit_error", code: "sse_rate_limit", message: e.Error.Message}
 	case "api_error", "timeout_error", "":
 		return &failure{transient: true, status: 502, atype: "api_error", code: "sse_api_error", message: e.Error.Message}
-	case "invalid_request_error": // request-shape: can never succeed, surface it
+	case "invalid_request_error":
+		// Only this verified transport failure overrides the request-error label.
+		if e.Error.Message == "stream error: stream disconnected before completion: stream closed before response.completed" {
+			return &failure{transient: true, status: 502, atype: "api_error", code: "sse_stream_disconnect", message: e.Error.Message}
+		}
 		return &failure{transient: false, status: 400, atype: e.Error.Type, code: "sse_request_shape", message: e.Error.Message}
 	default: // authentication_error, permission_error, not_found_error, unknown -> retry
 		return &failure{transient: true, status: 502, atype: "api_error", code: "sse_retryable", message: e.Error.Message}

docs/TROUBLESHOOTING.md

@@ -76,6 +76,18 @@ specific `code` is `request_timeout`, `rate_limit_exceeded`, or
 example, `request_timeout` with “stream disconnected before completion” remains
 retryable instead of becoming an invalid-request failure.
 
+For Anthropic SSE `event: error`, the proxy also recognizes one verified transport
+disconnect mislabeled `invalid_request_error`. The complete message must be exactly:
+
+```text
+stream error: stream disconnected before completion: stream closed before response.completed
+```
+
+This failure is classified as `sse_stream_disconnect`. Before commit, failed
+output is discarded and the existing local and client retry budgets apply, within
+the current deadline and buffering window. After commit, the stream ends without
+transparent replay. This exception does not change retry limits or timeouts.
+
 If retries continue, check the gateway's credentials, account state, availability,
 and rate limits. A permanently bad key or exhausted balance will not improve
 without intervention, even though the proxy allows time for recovery. Generic
@@ -108,9 +120,11 @@ are returned with `X-Should-Retry: false`; terminal Responses errors use HTTP
 
 Anthropic HTTP errors use recognized message signatures. A bare HTTP
 `invalid_request_error` without one can still retry; an Anthropic in-stream
-`invalid_request_error` is terminal. Responses classification uses error codes,
-types, and recognized messages, with known transient codes taking precedence over
-generic invalid-request types. Gateway retry headers override HTTP heuristics.
+`invalid_request_error` is terminal except for the exact transport-disconnect
+signature above. Invalid schemas, malformed tool messages, and context-limit
+failures remain terminal. Responses classification uses error codes, types, and
+recognized messages, with known transient codes taking precedence over generic
+invalid-request types. Gateway retry headers override HTTP heuristics.
 
 ### Codex context compaction
 

e2e_test.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"bufio"
 	"bytes"
 	"io"
 	"log"
@@ -395,6 +396,125 @@ func TestE2ETransactionalLocalRetryTruncateThenSuccess(t *testing.T) {
 	}
 }
 
+func TestE2ESSEDisconnectRetryBudget(t *testing.T) {
+	for _, tc := range []struct {
+		name                      string
+		retries                   int
+		recover                   bool
+		backoff, window, deadline time.Duration
+		attempts, status          int
+	}{
+		{name: "recovered", retries: 1, recover: true, attempts: 2, status: 200},
+		{name: "local_disabled", recover: true, attempts: 1, status: 503},
+		{name: "local_exhausted", retries: 1, attempts: 2, status: 503},
+		{name: "deadline_insufficient", retries: 1, recover: true, backoff: time.Second, deadline: time.Second, attempts: 1, status: 503},
+		{name: "window_insufficient", retries: 1, recover: true, backoff: time.Second, window: 500 * time.Millisecond, attempts: 1, status: 503},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			var hits atomic.Int32
+			prefix := strings.Replace(goodStream[:strings.Index(goodStream, "event: content_block_stop")], "Hi", "discarded-attempt", 1)
+			up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				w.Header().Set("Content-Type", "text/event-stream")
+				if hits.Add(1) > 1 && tc.recover {
+					io.WriteString(w, goodStream)
+					return
+				}
+				io.WriteString(w, prefix+mislabeledDisconnectStream)
+			}))
+			defer up.Close()
+			setupForTest(up.URL)
+			t.Cleanup(func() { cfg = testConfig() })
+			cfg.txLocalRetries = tc.retries
+			cfg.localBackoffCap = tc.backoff
+			if tc.window > 0 {
+				cfg.keepaliveMs = tc.window
+			}
+			if tc.deadline > 0 {
+				cfg.maxRequestDur = tc.deadline
+			}
+
+			rec := doStream(`{"stream":true,"model":"m"}`)
+			body := rec.Body.String()
+			if rec.Code != tc.status || int(hits.Load()) != tc.attempts {
+				t.Fatalf("want status=%d attempts=%d, got status=%d attempts=%d body=%s", tc.status, tc.attempts, rec.Code, hits.Load(), body)
+			}
+			if strings.Contains(body, "discarded-attempt") {
+				t.Fatalf("failed attempt leaked output: %s", body)
+			}
+			if tc.status == 200 {
+				if rec.Header().Get("X-Steady-Proxy-Mode") != "buffered" || !strings.Contains(body, "message_stop") || !strings.Contains(body, "Hi") || strings.Contains(body, "event: error") {
+					t.Fatalf("recovery did not deliver a complete buffered response: headers=%v body=%s", rec.Header(), body)
+				}
+			} else if rec.Header().Get("X-Should-Retry") != "true" || rec.Header().Get("X-Steady-Proxy-Reason") != "sse_stream_disconnect" || !strings.Contains(body, `"type":"api_error"`) {
+				t.Fatalf("uncommitted disconnect must retain the client retry signal: headers=%v body=%s", rec.Header(), body)
+			}
+		})
+	}
+}
+
+func TestE2ESSEDisconnectAfterCommitDoesNotReplay(t *testing.T) {
+	var hits atomic.Int32
+	releaseError := make(chan struct{}, 1)
+	prefix := strings.Replace(goodStream[:strings.Index(goodStream, "event: content_block_stop")], "Hi", "committed-attempt", 1)
+	up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		if hits.Add(1) > 1 {
+			io.WriteString(w, goodStream)
+			return
+		}
+		io.WriteString(w, prefix)
+		w.(http.Flusher).Flush()
+		select {
+		case <-releaseError:
+			io.WriteString(w, mislabeledDisconnectStream)
+		case <-r.Context().Done():
+		}
+	}))
+	defer up.Close()
+	setupForTest(up.URL)
+	t.Cleanup(func() { cfg = testConfig() })
+	cfg.keepaliveMs = 20 * time.Millisecond
+	cfg.txLocalRetries = 2
+	cfg.localBackoffCap = 0
+	proxy := httptest.NewServer(http.HandlerFunc(handle))
+	defer proxy.Close()
+	downstream := &http.Client{Timeout: 3 * time.Second}
+	resp, err := downstream.Post(proxy.URL+"/v1/messages", "application/json", strings.NewReader(`{"stream":true,"model":"m"}`))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != 200 || resp.Header.Get("X-Steady-Proxy-Mode") != "live" {
+		t.Fatalf("want committed live response, got status=%d headers=%v", resp.StatusCode, resp.Header)
+	}
+	reader := bufio.NewReader(resp.Body)
+	var received strings.Builder
+	for {
+		line, err := reader.ReadString('\n')
+		if err != nil {
+			t.Fatalf("reading committed content: %v", err)
+		}
+		received.WriteString(line)
+		if strings.Contains(line, "committed-attempt") {
+			break
+		}
+	}
+	// Release the error only after the client has observed actual response content.
+	releaseError <- struct{}{}
+	rest, err := io.ReadAll(reader)
+	if err != nil {
+		t.Fatal(err)
+	}
+	received.Write(rest)
+	body := received.String()
+	if hits.Load() != 1 {
+		t.Fatalf("committed failure replayed upstream: attempts=%d body=%s", hits.Load(), body)
+	}
+	if strings.Contains(body, "message_stop") || strings.Contains(body, "event: error") || strings.Contains(body, streamDisconnectMessage) || strings.Contains(body, `"text":"Hi"`) {
+		t.Fatalf("committed failure must end the partial stream without an error or replacement response: %s", body)
+	}
+}
+
 func TestE2ETransactionalLocalRetryDoesNotRetryPermanentShape(t *testing.T) {
 	var hits atomic.Int32
 	up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

sse_test.go

@@ -31,6 +31,10 @@ data: {"type":"message_stop"}
 
 `
 
+const streamDisconnectMessage = "stream error: stream disconnected before completion: stream closed before response.completed"
+
+const mislabeledDisconnectStream = "event: error\ndata: {\"type\":\"error\",\"error\":{\"type\":\"invalid_request_error\",\"message\":\"" + streamDisconnectMessage + "\"}}\n\n"
+
 func init() { cfg = testConfig() }
 
 func capture(t *testing.T, s string) (*httptest.ResponseRecorder, *failure) {
@@ -482,11 +486,46 @@ func TestCaptureSSEErrorOverloaded(t *testing.T) {
 	}
 }
 
-func TestCaptureSSEErrorPermanent(t *testing.T) {
-	s := "event: error\ndata: {\"type\":\"error\",\"error\":{\"type\":\"invalid_request_error\",\"message\":\"bad\"}}\n\n"
-	_, f := capture(t, s)
-	if f == nil || f.transient {
-		t.Fatalf("want permanent failure, got %+v", f)
+func TestCaptureSSEErrorClassification(t *testing.T) {
+	useDefaultConfig(t)
+	for _, tc := range []struct {
+		name, typ, message string
+		transient          bool
+	}{
+		{"mislabeled_disconnect", "invalid_request_error", streamDisconnectMessage, true},
+		{"typed_disconnect", "timeout_error", streamDisconnectMessage, true},
+		{"unknown_request_error", "invalid_request_error", "bad", false},
+		{"invalid_schema", "invalid_request_error", "invalid tool schema", false},
+		{"malformed_tool_message", "invalid_request_error", "messages.0: unexpected tool_use_id in tool_result", false},
+		{"context_limit", "invalid_request_error", "prompt is too long for context", false},
+		{"stream_validation", "invalid_request_error", "invalid tool schema for stream", false},
+		{"stream_error_validation", "invalid_request_error", "stream error: invalid tool schema", false},
+		{"partial_disconnect", "invalid_request_error", "stream disconnected before completion", false},
+		{"different_stream_error", "invalid_request_error", "stream error: stream disconnected before completion: invalid tool schema", false},
+		{"disconnect_with_validation", "invalid_request_error", streamDisconnectMessage + ": invalid tool schema", false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			data, err := json.Marshal(map[string]any{
+				"type": "error", "error": map[string]string{"type": tc.typ, "message": tc.message},
+			})
+			if err != nil {
+				t.Fatal(err)
+			}
+			rec, f := capture(t, "event: error\ndata: "+string(data)+"\n\n")
+			if f == nil || f.transient != tc.transient || f.message != tc.message {
+				t.Fatalf("want transient=%v with original message, got %+v", tc.transient, f)
+			}
+			status, typ := http.StatusBadRequest, "invalid_request_error"
+			if tc.transient {
+				status, typ = http.StatusBadGateway, "api_error"
+			}
+			if f.status != status || f.atype != typ {
+				t.Fatalf("want status=%d type=%s, got %+v", status, typ, f)
+			}
+			if rec.Body.Len() != 0 {
+				t.Fatalf("failed attempt leaked output: %s", rec.Body.String())
+			}
+		})
 	}
 }