fix(qoder): bound ensureMeaningfulUpstream after 200 (#344)

changelog/unreleased/qoder-pat-meaningful-upstream-timeout.md

@@ -0,0 +1,9 @@
+### English
+
+- Qoder PAT in-process ChatStream now bounds the wait for a meaningful SSE data line after a 200 (reuses `QODER_STREAM_TTFB_MS`); heartbeat-only upstreams fail over with `upstream_empty` instead of hanging. (#344)
+- QwenWork ChatStream applies the same meaningful-upstream deadline for consistency.
+
+### 中文
+
+- Qoder PAT 进程内 ChatStream 在收到 200 后等待有意义 SSE 数据行时也会受 `QODER_STREAM_TTFB_MS` 约束；仅心跳的上游会以 `upstream_empty` 失败并切换，而不再挂起。（#344）
+- QwenWork ChatStream 同样加上有意义上游数据的等待上限，保持一致。

deploy/README.md

@@ -134,7 +134,7 @@ console; environment variables cannot replace it.
 | `QODER_MAX_RETRY_ACCOUNTS` | `4` | Maximum accounts attempted for one request (1-64) |
 | `QODER_SSE_DIAGNOSTIC_MODELS` | empty | Comma-separated Qoder model IDs for redacted SSE diagnostics in Runtime Logs; `*` enables all |
 | `QODER_WORKER_BASE_PORT` | `32100` | Internal child-runtime port range |
-| `QODER_STREAM_TTFB_MS` | `30000` | Qoder stream first-byte (response head) deadline in ms, Go side and worker; min `5000`; timeout fails over to the next account |
+| `QODER_STREAM_TTFB_MS` | `30000` | Qoder stream first-byte deadline in ms (response head + first meaningful SSE data line after 200), Go side and worker; min `5000`; timeout fails over to the next account |
 | `QODER_STREAM_TOTAL_MS` | `300000` | Qoder worker mid-stream stall timeout in ms: aborts only after this long with no new chunk (resets per chunk, not a wall-clock cap); never below the TTFB value |
 | `QODER_SIGNATURE_SECRET` | built-in | Override the COSY request-signing secret for in-process Qoder; leave unset unless upstream rotates it |
 | `QODER_PROXY_URL` | empty | Initial global outbound proxy: `http(s)://`, `direct`, or `none`; saved console settings take precedence |

internal/providers/qoder/direct.go

@@ -12,6 +12,7 @@ import (
 	"sort"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/caigee-cmd/cli2api/internal/accounts"
@@ -29,8 +30,9 @@ type Direct struct {
 	bridgeOf func(ctx context.Context, accountID string) (*Bridge, error)
 	http     *http.Client
 	now      func() time.Time
-	// ttfb bounds how long ChatStream waits for the upstream response head.
-	// Zero falls back to defaultFirstByteTimeout inside doWithFirstByteTimeout.
+	// ttfb bounds how long ChatStream waits for the upstream response head
+	// and, after a 200, for the first meaningful SSE data line
+	// (ensureMeaningfulUpstream). Zero falls back to defaultFirstByteTimeout.
 	ttfb time.Duration
 }
 
@@ -205,7 +207,7 @@ func (d *Direct) ChatStream(ctx context.Context, accountID string, req translate
 		resp.Body.Close()
 		return nil, providers.ResolvedChat{}, upstreamStatusError(resp.StatusCode, raw)
 	}
-	guarded, err := ensureMeaningfulUpstream(resp)
+	guarded, err := ensureMeaningfulUpstream(resp, d.ttfb)
 	if err != nil {
 		// ensureMeaningfulUpstream closes the body on failure; no client bytes
 		// have been written yet, so the executor can still fail over.
@@ -251,20 +253,60 @@ func emptyUpstreamError() error {
 // completion. Peek before the first client byte so the failure can still be
 // classified and retried on another account. On success the buffered bytes
 // are replayed ahead of the remaining body, preserving real-time streaming.
-func ensureMeaningfulUpstream(resp *http.Response) (*http.Response, error) {
+//
+// timeout bounds how long to wait for a meaningful data line after the
+// response head arrives. Zero falls back to defaultFirstByteTimeout so the
+// bound matches QODER_STREAM_TTFB_MS / Direct.ttfb. A stuck heartbeat-only
+// upstream closes the body and returns emptyUpstreamError so the executor
+// can fail over instead of hanging forever (doWithFirstByteTimeout only
+// covers headers).
+func ensureMeaningfulUpstream(resp *http.Response, timeout time.Duration) (*http.Response, error) {
+	if resp == nil || resp.Body == nil {
+		return nil, emptyUpstreamError()
+	}
+	if timeout <= 0 {
+		timeout = defaultFirstByteTimeout
+	}
+
+	// Close at most once: the timeout timer and the read-error path both need
+	// to finish the body, but cancelOnCloseBody.Close is not concurrency-safe.
+	// Success stops the timer before returning the live peeked body; if the
+	// closer already won the race, treat it as emptyUpstreamError rather than
+	// handing the caller a half-closed stream.
+	var closeOnce sync.Once
+	closeBody := func() {
+		closeOnce.Do(func() { _ = resp.Body.Close() })
+	}
+
+	var expired atomic.Bool
+	timer := time.AfterFunc(timeout, func() {
+		expired.Store(true)
+		closeBody()
+	})
+	stopTimer := func() { _ = timer.Stop() }
+
 	br := bufio.NewReader(resp.Body)
 	var pending bytes.Buffer
 	for {
 		line, readErr := br.ReadString('\n')
 		if line != "" {
 			pending.WriteString(line)
 			if data, ok := strings.CutPrefix(strings.TrimSpace(line), "data:"); ok && meaningfulDataLine(strings.TrimSpace(data)) {
+				stopTimer()
+				if expired.Load() {
+					closeBody()
+					return nil, emptyUpstreamError()
+				}
 				resp.Body = peekedBody{Reader: io.MultiReader(&pending, br), Closer: resp.Body}
 				return resp, nil
 			}
 		}
 		if readErr != nil {
-			resp.Body.Close()
+			stopTimer()
+			closeBody()
+			if expired.Load() {
+				return nil, emptyUpstreamError()
+			}
 			if errors.Is(readErr, io.EOF) {
 				return nil, emptyUpstreamError()
 			}

internal/providers/qoder/direct_test.go

@@ -245,3 +245,133 @@ func TestDirectChatStreamSalvagesTruncatedStreamWithCachedUsage(t *testing.T) {
 		t.Fatalf("missing cached_tokens in terminal chunk: %s", text)
 	}
 }
+
+// Heartbeat-only upstream that never becomes meaningful must time out with
+// upstream_empty so the executor can fail over. Without the bound,
+// ensureMeaningfulUpstream would hang forever after a 200 (issue #344).
+func TestDirectChatStreamTimesOutHeartbeatOnlyUpstream(t *testing.T) {
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		w.WriteHeader(http.StatusOK)
+		flusher, _ := w.(http.Flusher)
+		heartbeat := "data: {\"body\":{\"choices\":[{\"delta\":{},\"finish_reason\":null}]}}\n\n"
+		for {
+			select {
+			case <-r.Context().Done():
+				return
+			default:
+				_, _ = io.WriteString(w, heartbeat)
+				if flusher != nil {
+					flusher.Flush()
+				}
+				time.Sleep(20 * time.Millisecond)
+			}
+		}
+	}))
+	defer upstream.Close()
+	direct := newDirectFor(t, upstream)
+	direct.ttfb = 80 * time.Millisecond
+
+	started := time.Now()
+	resp, _, err := direct.ChatStream(context.Background(), "acc", translate.ChatRequest{Model: "auto"})
+	elapsed := time.Since(started)
+	if err == nil {
+		resp.Body.Close()
+		t.Fatal("expected timeout error for heartbeat-only upstream")
+	}
+	var providerErr *providers.Error
+	if !errors.As(err, &providerErr) || providerErr.Code != "upstream_empty" {
+		t.Fatalf("err = %#v", err)
+	}
+	if resp != nil {
+		t.Fatalf("resp = %v, want nil", resp)
+	}
+	// Must fail within the bound (plus small slack), not hang for minutes.
+	if elapsed > 2*time.Second {
+		t.Fatalf("took %s, want bound near ttfb", elapsed)
+	}
+}
+
+// A meaningful data line that arrives within the deadline must succeed, and
+// the peeked heartbeat + content bytes must still be replayed to the client.
+func TestDirectChatStreamPreservesBytesAfterHeartbeatThenContent(t *testing.T) {
+	upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.Header().Set("Content-Type", "text/event-stream")
+		w.WriteHeader(http.StatusOK)
+		flusher, _ := w.(http.Flusher)
+		_, _ = io.WriteString(w, "data: {\"body\":{\"choices\":[{\"delta\":{},\"finish_reason\":null}]}}\n\n")
+		if flusher != nil {
+			flusher.Flush()
+		}
+		time.Sleep(30 * time.Millisecond)
+		_, _ = io.WriteString(w, "data: {\"body\":{\"choices\":[{\"delta\":{\"content\":\"OK\"},\"finish_reason\":\"stop\"}]}}\n\n")
+		if flusher != nil {
+			flusher.Flush()
+		}
+	}))
+	defer upstream.Close()
+	direct := newDirectFor(t, upstream)
+	direct.ttfb = 500 * time.Millisecond
+
+	resp, _, err := direct.ChatStream(context.Background(), "acc", translate.ChatRequest{Model: "auto"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer resp.Body.Close()
+	body, _ := io.ReadAll(resp.Body)
+	text := string(body)
+	if !strings.Contains(text, `"content":"OK"`) {
+		t.Fatalf("meaningful content lost: %s", text)
+	}
+	if !strings.Contains(text, `"finish_reason":"stop"`) {
+		t.Fatalf("finish_reason lost: %s", text)
+	}
+}
+
+func TestEnsureMeaningfulUpstreamTimesOutHeartbeatOnly(t *testing.T) {
+	pr, pw := io.Pipe()
+	go func() {
+		defer pw.Close()
+		heartbeat := "data: {\"body\":{\"choices\":[{\"delta\":{},\"finish_reason\":null}]}}\n\n"
+		for i := 0; i < 100; i++ {
+			if _, err := io.WriteString(pw, heartbeat); err != nil {
+				return
+			}
+			time.Sleep(15 * time.Millisecond)
+		}
+	}()
+	resp := &http.Response{StatusCode: http.StatusOK, Body: pr}
+	started := time.Now()
+	out, err := ensureMeaningfulUpstream(resp, 60*time.Millisecond)
+	elapsed := time.Since(started)
+	if err == nil {
+		out.Body.Close()
+		t.Fatal("expected timeout")
+	}
+	var providerErr *providers.Error
+	if !errors.As(err, &providerErr) || providerErr.Code != "upstream_empty" {
+		t.Fatalf("err = %#v", err)
+	}
+	if out != nil {
+		t.Fatalf("out = %v, want nil", out)
+	}
+	if elapsed > 2*time.Second {
+		t.Fatalf("took %s", elapsed)
+	}
+}
+
+func TestEnsureMeaningfulUpstreamPreservesBufferedBytes(t *testing.T) {
+	body := "data: {\"body\":{\"choices\":[{\"delta\":{},\"finish_reason\":null}]}}\n\n" +
+		"data: {\"body\":{\"choices\":[{\"delta\":{\"content\":\"hi\"},\"finish_reason\":\"stop\"}]}}\n\n" +
+		"data: {\"body\":{\"choices\":[{\"delta\":{\"content\":\" more\"}}]}}\n\n"
+	resp := &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader(body))}
+	out, err := ensureMeaningfulUpstream(resp, time.Second)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer out.Body.Close()
+	got, _ := io.ReadAll(out.Body)
+	if string(got) != body {
+		t.Fatalf("peeked replay mismatch:\n got %q\nwant %q", got, body)
+	}
+}

internal/providers/qwenwork/client.go

@@ -680,7 +680,7 @@ func (c *Client) ChatStream(ctx context.Context, accountID string, req translate
 		resp.Body.Close()
 		return nil, providers.ResolvedChat{}, classifiedError(resp.StatusCode, raw)
 	}
-	guarded, err := ensureMeaningfulUpstream(resp)
+	guarded, err := ensureMeaningfulUpstream(resp, defaultMeaningfulUpstreamTimeout)
 	if err != nil {
 		// ensureMeaningfulUpstream closes the body on failure; no client
 		// bytes have been written yet, so the executor can still fail over.

internal/providers/qwenwork/sse.go

@@ -11,6 +11,8 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/caigee-cmd/cli2api/internal/accounts"
@@ -590,11 +592,40 @@ func emptyUpstreamError() *providers.Error {
 	}
 }
 
+// defaultMeaningfulUpstreamTimeout bounds how long ChatStream waits after a
+// 200 for a meaningful SSE data line. Matches qoder's defaultFirstByteTimeout
+// so heartbeat-only upstreams fail over instead of hanging the client.
+const defaultMeaningfulUpstreamTimeout = 30 * time.Second
+
 // ensureMeaningfulUpstream peeks the upstream SSE body before anything is
 // relayed to the client, so an empty or error-only upstream can still fail
 // over to another account. On success the buffered bytes are replayed ahead
 // of the remaining body, preserving real-time streaming.
-func ensureMeaningfulUpstream(resp *http.Response) (*http.Response, error) {
+//
+// timeout bounds the wait for a meaningful data line after the response head.
+// Zero falls back to defaultMeaningfulUpstreamTimeout. A stuck heartbeat-only
+// upstream closes the body and returns emptyUpstreamError so the executor can
+// fail over (same hang shape as qoder PAT in-process ChatStream).
+func ensureMeaningfulUpstream(resp *http.Response, timeout time.Duration) (*http.Response, error) {
+	if resp == nil || resp.Body == nil {
+		return nil, emptyUpstreamError()
+	}
+	if timeout <= 0 {
+		timeout = defaultMeaningfulUpstreamTimeout
+	}
+
+	var closeOnce sync.Once
+	closeBody := func() {
+		closeOnce.Do(func() { _ = resp.Body.Close() })
+	}
+
+	var expired atomic.Bool
+	timer := time.AfterFunc(timeout, func() {
+		expired.Store(true)
+		closeBody()
+	})
+	stopTimer := func() { _ = timer.Stop() }
+
 	br := bufio.NewReader(resp.Body)
 	var pending bytes.Buffer
 	for {
@@ -603,11 +634,16 @@ func ensureMeaningfulUpstream(resp *http.Response) (*http.Response, error) {
 			pending.WriteString(line)
 			if data, ok := strings.CutPrefix(strings.TrimSpace(line), "data:"); ok {
 				if frame, meaningful := meaningfulDataLine(strings.TrimSpace(data)); meaningful {
+					stopTimer()
+					if expired.Load() {
+						closeBody()
+						return nil, emptyUpstreamError()
+					}
 					if frame.errorOnly() {
 						// Nothing has reached the client yet: return the
 						// classified error so the executor cools this account
 						// (quota, auth, model) and fails over.
-						resp.Body.Close()
+						closeBody()
 						return nil, frame.providerError()
 					}
 					resp.Body = peekedBody{Reader: io.MultiReader(&pending, br), Closer: resp.Body}
@@ -616,7 +652,11 @@ func ensureMeaningfulUpstream(resp *http.Response) (*http.Response, error) {
 			}
 		}
 		if readErr != nil {
-			resp.Body.Close()
+			stopTimer()
+			closeBody()
+			if expired.Load() {
+				return nil, emptyUpstreamError()
+			}
 			if errors.Is(readErr, io.EOF) {
 				return nil, emptyUpstreamError()
 			}

internal/providers/qwenwork/sse_meaningful_timeout_test.go

@@ -0,0 +1,63 @@
+package qwenwork
+
+import (
+	"errors"
+	"io"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/caigee-cmd/cli2api/internal/providers"
+)
+
+// Heartbeat-only forever stream must time out with upstream_empty (same hang
+// shape as qoder PAT in-process ChatStream; issue #344).
+func TestEnsureMeaningfulUpstreamTimesOutHeartbeatOnly(t *testing.T) {
+	pr, pw := io.Pipe()
+	go func() {
+		defer pw.Close()
+		// Empty body choices are not meaningful for qwenwork either.
+		heartbeat := "data: {\"headers\":{},\"body\":\"{\\\"choices\\\":[{\\\"delta\\\":{},\\\"finish_reason\\\":null}]}\",\"statusCodeValue\":200}\n\n"
+		for i := 0; i < 100; i++ {
+			if _, err := io.WriteString(pw, heartbeat); err != nil {
+				return
+			}
+			time.Sleep(15 * time.Millisecond)
+		}
+	}()
+	resp := &http.Response{StatusCode: http.StatusOK, Body: pr}
+	started := time.Now()
+	out, err := ensureMeaningfulUpstream(resp, 60*time.Millisecond)
+	elapsed := time.Since(started)
+	if err == nil {
+		out.Body.Close()
+		t.Fatal("expected timeout")
+	}
+	var providerErr *providers.Error
+	if !errors.As(err, &providerErr) || providerErr.Code != "upstream_empty" {
+		t.Fatalf("err = %#v", err)
+	}
+	if out != nil {
+		t.Fatalf("out = %v, want nil", out)
+	}
+	if elapsed > 2*time.Second {
+		t.Fatalf("took %s", elapsed)
+	}
+}
+
+func TestEnsureMeaningfulUpstreamPreservesBufferedBytes(t *testing.T) {
+	body := "data: {\"headers\":{},\"body\":\"{\\\"choices\\\":[{\\\"delta\\\":{},\\\"finish_reason\\\":null}]}\",\"statusCodeValue\":200}\n\n" +
+		"data: {\"headers\":{},\"body\":\"{\\\"choices\\\":[{\\\"delta\\\":{\\\"content\\\":\\\"hi\\\"},\\\"finish_reason\\\":\\\"stop\\\"}]}\",\"statusCodeValue\":200}\n\n" +
+		"data: {\"headers\":{},\"body\":\"{\\\"choices\\\":[{\\\"delta\\\":{\\\"content\\\":\\\" more\\\"}}]}\",\"statusCodeValue\":200}\n\n"
+	resp := &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader(body))}
+	out, err := ensureMeaningfulUpstream(resp, time.Second)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer out.Body.Close()
+	got, _ := io.ReadAll(out.Body)
+	if string(got) != body {
+		t.Fatalf("peeked replay mismatch:\n got %q\nwant %q", got, body)
+	}
+}