fix(qoder): fail over on streaming first-byte timeout and watchdog stuck workers

changelog/unreleased/worker-stream-timeout.md

@@ -0,0 +1,11 @@
+### English
+
+- Stop streaming chat from hanging on a stuck upstream. A Qoder stream request whose first byte never arrives (the symptom was `qwen3.8-flash` requests with only an entry log line and no `chat reasoning resolved`) now fails fast after a 30-second TTFB deadline, classifies as `first_byte_timeout` / `unavailable` with failover, and is retried on the next eligible account instead of hanging the client until it gives up. The deadline is on the response head only — once headers arrive, the SSE body streams under the request's normal lifetime, so long thinking-model output is not cut off.
+- Give the worker daemon its own upstream timeout. The worker's `fetch` to the Qoder gateway previously had no `AbortSignal`, so a stalled upstream held the worker's in-flight slot forever. It now aborts on a 30-second first-byte deadline, plus a mid-stream stall watchdog that fires only after 300 seconds of silence between chunks (it resets on every chunk, so a long but steady thinking stream is never truncated), returning a stable `first_byte_timeout` / `stream_total_timeout` error so the Go side can classify and fail over. Both are tunable via `QODER_STREAM_TTFB_MS` / `QODER_STREAM_TOTAL_MS`.
+- Add a worker liveness watchdog to the maintenance loop. Health probes previously ran only at boot plus on manual console refresh, so a worker that stayed "alive" as a process but stopped serving `/health` was never noticed and kept being routed to. Every 30-second tick now issues one lightweight `/health` round-trip per worker account; three consecutive failures (~90s grace, enough to ride out a transient blip) restart the worker through the existing recovery path with backoff. A worker that answers but is still starting, rewarming, or waiting for login (`ready=false`) is treated as alive, so warming accounts are not rebooted. In-process accounts (WorkBuddy / Trae / Devin) are skipped and unaffected.
+
+### 中文
+
+- 流式 chat 不再因上游卡死而无响应地挂起。Qoder 流式请求如果首字节迟迟不到（现象是 `qwen3.8-flash` 请求只有入口日志、没有 `chat reasoning resolved` 出口日志），现在会在 30 秒 TTFB 阈值后快速失败，归类为 `first_byte_timeout` / `unavailable` 并触发 failover，切到下一个可用账号重试，而不是把客户端挂到自行超时。该超时只针对响应头：一旦头部到达，SSE body 回到请求原本的生命周期，长思考模型的输出不会被截断。
+- 给 worker 守护进程加上自己的上游超时。worker 对 Qoder 网关的 `fetch` 此前没有 `AbortSignal`，上游停滞时会永久占用 worker 的并发槽。现在它在 30 秒首字节超时，以及"连续 300 秒没有新数据"的中途停滞看门狗（每收到一个 chunk 就重置，所以长但持续输出的思考流不会被截断）任一触发时 abort，返回稳定的 `first_byte_timeout` / `stream_total_timeout` 错误，让 Go 侧分类并 failover。两者均可通过 `QODER_STREAM_TTFB_MS` / `QODER_STREAM_TOTAL_MS` 调整。
+- 给维护循环加 worker liveness 看门狗。健康探针此前只在启动时和手动控制台刷新时跑，所以一个"进程还活着但 `/health` 不再响应"的 worker 永远不会被察觉，还会继续被路由命中。现在每 30 秒维护周期对每个 worker 账号发一次轻量 `/health` 探针；连续 3 次失败（约 90 秒容忍窗口，足以扛过瞬时抖动）就通过既有的恢复路径按退避重启该 worker。对 `/health` 有响应但仍在启动、rewarm 或等待登录（`ready=false`）的 worker 视为存活，不会误杀。进程内账号（WorkBuddy / Trae / Devin）被跳过，不受影响。

internal/executor/classify_test.go

@@ -320,3 +320,33 @@ func TestClassifyErrorUpstreamEmptyNoCooldown(t *testing.T) {
 		t.Errorf("expected generic unavailable to keep cooldown, got %v", generic.Cooldown)
 	}
 }
+
+// A streaming first-byte timeout is an upstream hang: the client request is
+// still alive, the upstream just never sent a response head. It must classify
+// as upstream_unavailable with failover so the next eligible account serves
+// the turn, plus a short cooldown to shed a stuck worker for a few seconds.
+func TestClassifyErrorFirstByteTimeoutFailsOver(t *testing.T) {
+	err := fmt.Errorf("%w: no response head within 1m0s", providers.ErrFirstByteTimeout)
+	got := ClassifyError(err)
+	if got.Kind != accounts.KindUnavailable {
+		t.Fatalf("Kind=%s, want %s", got.Kind, accounts.KindUnavailable)
+	}
+	if !got.Failover {
+		t.Fatalf("Failover=false, want true")
+	}
+	if got.Cooldown != 10*time.Second || got.RetryAfter != 10*time.Second {
+		t.Fatalf("Cooldown=%s RetryAfter=%s, want 10s", got.Cooldown, got.RetryAfter)
+	}
+	if got.Code != "first_byte_timeout" {
+		t.Fatalf("Code=%q, want first_byte_timeout", got.Code)
+	}
+}
+
+// A plain client cancel must not be reclassified as a first-byte timeout; the
+// executor skips failover and cooldown for cancellations.
+func TestClassifyErrorContextCanceledStaysCanceled(t *testing.T) {
+	got := ClassifyError(context.Canceled)
+	if got.Kind != accounts.KindCanceled || got.Failover {
+		t.Fatalf("got Kind=%s Failover=%v, want %s/false", got.Kind, got.Failover, accounts.KindCanceled)
+	}
+}

internal/executor/error.go

@@ -3,6 +3,7 @@ package executor
 import (
 	"context"
 	"errors"
+	"net/http"
 	"strings"
 	"time"
 
@@ -88,6 +89,17 @@ func ClassifyError(err error) Classified {
 	if errors.As(err, &execErr) && execErr != nil {
 		return execErr.Classified
 	}
+	// A streaming first-byte timeout is an upstream hang, not a client cancel:
+	// the parent request is still alive, the upstream just never answered. Cool
+	// the account briefly and fail over so the next eligible account serves the
+	// turn instead of hanging until the client gives up.
+	if errors.Is(err, providers.ErrFirstByteTimeout) {
+		return Classified{
+			Kind: accounts.KindUnavailable, Status: http.StatusBadGateway,
+			Failover: true, Cooldown: 10 * time.Second, RetryAfter: 10 * time.Second,
+			Code: "first_byte_timeout", Message: err.Error(),
+		}
+	}
 	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
 		return Classified{
 			Kind: accounts.KindCanceled, Status: 499, Failover: false,

internal/providers/interfaces.go

@@ -15,6 +15,14 @@ import (
 // implicit support.
 var ErrUnsupported = errors.New("provider capability unsupported")
 
+// ErrFirstByteTimeout is returned by streaming chat adapters when the upstream
+// did not send the response head (headers + first byte) within the TTFB
+// deadline. The executor classifies it as upstream_unavailable with failover so
+// the next eligible account is tried, instead of hanging on a stuck upstream.
+// It must not be wrapped with context.DeadlineExceeded, or the executor would
+// re-classify it as a client cancellation and skip failover.
+var ErrFirstByteTimeout = errors.New("first byte timeout")
+
 type ModelCapabilities struct {
 	ContextWindow    int `json:"context_window,omitempty"`
 	ContextWindowMax int `json:"context_window_max,omitempty"`

internal/providers/qoder/adapter.go

@@ -35,20 +35,26 @@ type Client struct {
 	loginTimeout  time.Duration
 	loginInterval time.Duration
 
+	// firstByteTimeout bounds how long the worker-path ChatStream waits for the
+	// upstream response head before failing over. Resolved from
+	// QODER_STREAM_TTFB_MS at construction; zero falls back to the default.
+	firstByteTimeout time.Duration
+
 	// direct is the in-process chat path. When set, chat does not enter the
 	// per-account Node worker. Login, catalog, and quota still use that worker.
 	direct *Direct
 }
 
 func NewClient() *Client {
 	return &Client{
-		modelsHTTP:    &http.Client{Timeout: 15 * time.Second},
-		healthHTTP:    &http.Client{Timeout: 2 * time.Second},
-		quotaHTTP:     &http.Client{Timeout: 5 * time.Second},
-		adminHTTP:     &http.Client{Timeout: 120 * time.Second},
-		chatHTTP:      &http.Client{Timeout: 120 * time.Second},
-		loginTimeout:  90 * time.Second,
-		loginInterval: 200 * time.Millisecond,
+		modelsHTTP:       &http.Client{Timeout: 15 * time.Second},
+		healthHTTP:       &http.Client{Timeout: 2 * time.Second},
+		quotaHTTP:        &http.Client{Timeout: 5 * time.Second},
+		adminHTTP:        &http.Client{Timeout: 120 * time.Second},
+		chatHTTP:         &http.Client{Timeout: 120 * time.Second},
+		loginTimeout:     90 * time.Second,
+		loginInterval:    200 * time.Millisecond,
+		firstByteTimeout: resolveFirstByteTimeout(),
 	}
 }
 
@@ -537,6 +543,7 @@ func (c *Client) ChatStream(ctx context.Context, accountID string, req translate
 	}
 	c.mu.RLock()
 	httpClient := c.chatHTTP
+	firstByte := c.firstByteTimeout
 	c.mu.RUnlock()
 	if httpClient == nil {
 		httpClient = http.DefaultClient
@@ -546,9 +553,9 @@ func (c *Client) ChatStream(ctx context.Context, accountID string, req translate
 		cloned.Timeout = 0
 		httpClient = &cloned
 	}
-	resp, err := httpClient.Do(httpReq)
+	resp, err := doWithFirstByteTimeout(httpClient, httpReq, firstByte)
 	if err != nil {
-		return nil, providers.ResolvedChat{}, TransportError{Err: err}
+		return nil, providers.ResolvedChat{}, err
 	}
 	return resp, resolved, nil
 }

internal/providers/qoder/direct.go

@@ -24,6 +24,9 @@ type Direct struct {
 	region func(ctx context.Context, accountID string) (string, 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 time.Duration
 }
 
 func NewDirect(store CredentialStore, region func(context.Context, string) (string, error)) *Direct {
@@ -32,6 +35,7 @@ func NewDirect(store CredentialStore, region func(context.Context, string) (stri
 		region: region,
 		http:   &http.Client{},
 		now:    time.Now,
+		ttfb:   resolveFirstByteTimeout(),
 	}
 }
 
@@ -128,7 +132,7 @@ func (d *Direct) ChatStream(ctx context.Context, accountID string, req translate
 	if err != nil {
 		return nil, providers.ResolvedChat{}, err
 	}
-	resp, err := d.http.Do(prepared.request)
+	resp, err := doWithFirstByteTimeout(d.http, prepared.request, d.ttfb)
 	if err != nil {
 		return nil, providers.ResolvedChat{}, err
 	}

internal/providers/qoder/first_byte_timeout_test.go

@@ -0,0 +1,110 @@
+package qoder
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/caigee-cmd/cli2api/internal/providers"
+)
+
+// A stuck upstream that never sends the response head must surface as
+// ErrFirstByteTimeout so the executor can fail the attempt over instead of
+// hanging the client until it gives up. The production deadline is shrunk for
+// the test so the assertion runs in milliseconds.
+func TestDoWithFirstByteTimeoutHangsAsFirstByteTimeout(t *testing.T) {
+	// Handler blocks forever: it accepts the body but never writes a response.
+	hung := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		<-r.Context().Done()
+	}))
+	t.Cleanup(hung.Close)
+
+	req, err := http.NewRequestWithContext(context.Background(), http.MethodPost, hung.URL, nil)
+	if err != nil {
+		t.Fatalf("NewRequest: %v", err)
+	}
+	client := &http.Client{Timeout: 0} // mimic streaming client: no whole-call timeout
+	_, err = doWithFirstByteTimeout(client, req, 50*time.Millisecond)
+	if !errors.Is(err, providers.ErrFirstByteTimeout) {
+		t.Fatalf("want ErrFirstByteTimeout, got %v", err)
+	}
+}
+
+// When the parent context is already cancelled the helper must propagate the
+// plain transport error instead of masking it as a first-byte timeout; the
+// executor classifies client cancels as KindCanceled (no failover, no cooldown).
+func TestDoWithFirstByteTimeoutParentCancelIsNotFirstByteTimeout(t *testing.T) {
+	hung := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		<-r.Context().Done()
+	}))
+	t.Cleanup(hung.Close)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, hung.URL, nil)
+	if err != nil {
+		t.Fatalf("NewRequest: %v", err)
+	}
+	client := &http.Client{Timeout: 0}
+
+	// Cancel before the call so parent.Err() is non-nil at the point of failure.
+	cancel()
+	_, err = doWithFirstByteTimeout(client, req, 50*time.Millisecond)
+	if errors.Is(err, providers.ErrFirstByteTimeout) {
+		t.Fatalf("parent-cancelled call must not surface as ErrFirstByteTimeout, got %v", err)
+	}
+	if err == nil {
+		t.Fatalf("want non-nil error for cancelled parent context")
+	}
+}
+
+// A well-behaved upstream that returns a response head quickly must not be
+// cut off by the TTFB timer, and the whole multi-chunk body must stay readable
+// to EOF afterwards. This drains every chunk rather than doing a single buffered
+// Read: a cancel-on-return bug only shows up once the stream continues past the
+// first read, which is exactly the "dies right after headers" symptom.
+func TestDoWithFirstByteTimeoutStreamsBodyToEOF(t *testing.T) {
+	const chunks = 4
+	ok := 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)
+		for i := 0; i < chunks; i++ {
+			_, _ = w.Write([]byte("data: chunk\n\n"))
+			if flusher != nil {
+				flusher.Flush()
+			}
+			// Space the chunks out so the first-byte deadline would fire if the
+			// timer were not stopped and the context were cancelled on return.
+			time.Sleep(20 * time.Millisecond)
+		}
+	}))
+	t.Cleanup(ok.Close)
+
+	req, err := http.NewRequestWithContext(context.Background(), http.MethodPost, ok.URL, nil)
+	if err != nil {
+		t.Fatalf("NewRequest: %v", err)
+	}
+	client := &http.Client{Timeout: 0}
+	// Short TTFB deadline: the head arrives immediately, so a correct
+	// implementation stops the timer and never cancels the body.
+	resp, err := doWithFirstByteTimeout(client, req, 50*time.Millisecond)
+	if err != nil {
+		t.Fatalf("want nil err for fast head, got %v", err)
+	}
+	defer resp.Body.Close()
+
+	var got bytes.Buffer
+	if _, err := io.Copy(&got, resp.Body); err != nil {
+		t.Fatalf("drain body to EOF: %v (read %q)", err, got.String())
+	}
+	want := strings.Repeat("data: chunk\n\n", chunks)
+	if got.String() != want {
+		t.Fatalf("body = %q, want %q", got.String(), want)
+	}
+}

internal/providers/qoder/worker.go

@@ -8,11 +8,14 @@ import (
 	"fmt"
 	"io"
 	"net/http"
+	"os"
+	"strconv"
 	"strings"
 	"time"
 
 	"github.com/caigee-cmd/cli2api/internal/accounts"
 	"github.com/caigee-cmd/cli2api/internal/endpoint"
+	"github.com/caigee-cmd/cli2api/internal/providers"
 )
 
 var (
@@ -49,6 +52,101 @@ func (e TransportError) Error() string {
 
 func (e TransportError) Unwrap() error { return e.Err }
 
+// defaultFirstByteTimeout caps how long a streaming chat request will wait for
+// the upstream response head (status + headers). Once the head arrives, the
+// timer is stopped and the SSE body streams under the parent context's lifetime,
+// so long thinking-model output is not cut off. A stuck upstream that never
+// sends a first byte surfaces as ErrFirstByteTimeout so the executor can fail
+// over to the next account instead of hanging forever. 30s covers slow
+// cold-starts and thinking-model first-token latency while keeping a
+// stuck-upstream failover inside one round-trip of the client's patience.
+//
+// The effective value is resolved once at construction (see
+// resolveFirstByteTimeout) and can be tuned per deployment with
+// QODER_STREAM_TTFB_MS, mirroring the worker daemon's own knob.
+const defaultFirstByteTimeout = 30 * time.Second
+
+// resolveFirstByteTimeout reads QODER_STREAM_TTFB_MS, falling back to
+// defaultFirstByteTimeout for an absent or invalid value. The floor matches the
+// worker daemon's so the two deadlines cannot be configured to disagree wildly.
+func resolveFirstByteTimeout() time.Duration {
+	raw := strings.TrimSpace(os.Getenv("QODER_STREAM_TTFB_MS"))
+	if raw == "" {
+		return defaultFirstByteTimeout
+	}
+	ms, err := strconv.Atoi(raw)
+	if err != nil || ms <= 0 {
+		return defaultFirstByteTimeout
+	}
+	d := time.Duration(ms) * time.Millisecond
+	if d < 5*time.Second {
+		return 5 * time.Second
+	}
+	return d
+}
+
+// doWithFirstByteTimeout sends req with client, returning a response once the
+// response head is received. If the head does not arrive within timeout it
+// returns an error wrapping providers.ErrFirstByteTimeout; if the parent
+// context was already cancelled it surfaces that error unchanged so the
+// executor classifies it as a client cancellation rather than a timeout.
+//
+// The TTFB context must NOT be cancelled once the head arrives: Go's HTTP
+// client ties the response body to the request context, so cancelling would
+// kill the body mid-stream (the SSE read then fails with "context canceled"
+// after the head was already delivered). Only the error path cancels; on
+// success the deadline goroutine is stopped and the body lives under the
+// parent context, with the response body's Close as the eventual release.
+func doWithFirstByteTimeout(client *http.Client, req *http.Request, timeout time.Duration) (*http.Response, error) {
+	if timeout <= 0 {
+		timeout = defaultFirstByteTimeout
+	}
+	parent := req.Context()
+	ttfbCtx, cancelTTFB := context.WithCancel(parent)
+	timer := time.AfterFunc(timeout, cancelTTFB)
+	resp, err := client.Do(req.WithContext(ttfbCtx))
+	if err == nil {
+		// Head arrived: stop the timer but keep ttfbCtx alive so the body
+		// remains readable. Tie the cancel to Body.Close so the derived context
+		// is released exactly when the caller is done with the stream, while
+		// the parent context still governs its lifetime (client disconnect /
+		// shutdown). Cancelling here instead would kill the body mid-stream.
+		timer.Stop()
+		resp.Body = &cancelOnCloseBody{ReadCloser: resp.Body, cancel: cancelTTFB}
+		return resp, nil
+	}
+	timer.Stop()
+	// TTFB fired only when the parent context is still alive: a parent cancel
+	// (client disconnect, shutdown) must stay a plain cancel so the executor
+	// does not treat it as an upstream failure worth cooling down. Decide this
+	// before cancelTTFB below, which would otherwise make ttfbCtx.Err() non-nil
+	// regardless of the real cause.
+	timedOut := parent.Err() == nil && ttfbCtx.Err() != nil && !errors.Is(parent.Err(), context.Canceled)
+	cancelTTFB()
+	if timedOut {
+		return nil, fmt.Errorf("%w: no response head within %s", providers.ErrFirstByteTimeout, timeout)
+	}
+	return nil, TransportError{Err: err}
+}
+
+// cancelOnCloseBody releases the TTFB-derived context when the caller closes
+// the response body. The body itself is tied to that context by Go's HTTP
+// client, so cancelling earlier would truncate the stream; cancelling on Close
+// frees the timer goroutine's context exactly when the stream is finished.
+type cancelOnCloseBody struct {
+	io.ReadCloser
+	cancel context.CancelFunc
+}
+
+func (b *cancelOnCloseBody) Close() error {
+	err := b.ReadCloser.Close()
+	if b.cancel != nil {
+		b.cancel()
+		b.cancel = nil
+	}
+	return err
+}
+
 type HTTPStatusError struct {
 	Op     string
 	Status int