fix: 实时跟踪当前并发（Acquire/Release）

changelog/unreleased/live-inflight-tracking.md

@@ -0,0 +1,7 @@
+### English
+
+- Current concurrency now tracks live proxy requests (Acquire/Release on the executor). `/api/live` and the sidebar update within a few seconds; health probes no longer overwrite in-flight counts.
+
+### 中文
+
+- 「当前并发」改为按代理正在处理的请求实时加减；`/api/live` 与侧边栏几秒内跟上，健康探活不再覆盖进行中的并发数。

frontend/src/components/layout/AppSidebar.tsx

@@ -8,7 +8,7 @@ import { SkeletonBlock } from '@/components/ui/PageSkeletons'
 import { pressScale } from '@/hooks/useGsapReveal'
 import { useI18n } from '@/hooks/useI18n'
 import { useOverview } from '@/hooks/useOverview'
-import { fetchOverviewSummary } from '@/api/overview'
+import { fetchLive } from '@/api/live'
 
 const nav = [
   { to: '/', key: 'navOverview', icon: Gauge, section: 'workspace', end: true },
@@ -114,12 +114,12 @@ function SidebarStatusSkeleton({ label }: { label: string }) {
   )
 }
 
-const IN_FLIGHT_POLL_MS = 3000
+const IN_FLIGHT_POLL_MS = 2500
 
 /**
- * Live in-flight request count (sum of account in_flight from
- * /api/overview/summary). Polls every 3s only while the tab is visible;
- * the dot is muted at 0 and pulses green while requests are running.
+ * Live in-flight request count from /api/live (same source as the overview
+ * strip). Polls every 2.5s only while the tab is visible; the dot is muted
+ * at 0 and pulses green while requests are running.
  */
 function InFlightIndicator({ initial, collapsed }: { initial: number; collapsed: boolean }) {
   const { t } = useI18n()
@@ -130,8 +130,8 @@ function InFlightIndicator({ initial, collapsed }: { initial: number; collapsed:
     let cancelled = false
     let timer: number | undefined
     const poll = () => {
-      void fetchOverviewSummary()
-        .then((data) => { if (!cancelled) setCount(data.worker?.in_flight ?? 0) })
+      void fetchLive()
+        .then((data) => { if (!cancelled) setCount(data.in_flight ?? 0) })
         .catch(() => { /* keep last value */ })
     }
     const start = () => {

internal/app/regression_test.go

@@ -154,7 +154,10 @@ func TestSaturatedPoolKeepsFiveSecondRetryAfter(t *testing.T) {
 		ID: account.ID, Provider: "qoder", Runtime: "child_process", URL: worker.URL,
 		Models: []string{"glm-5.2"}, MaxInFlight: 1,
 	})
-	a.Pool.MergeHealth(account.ID, true, true, 1, 0, "")
+	a.Pool.MergeHealth(account.ID, true, true, 0, 0, "")
+	if !a.Pool.Acquire(account.ID) {
+		t.Fatal("acquire saturated account")
+	}
 	a.Gateway.Catalogs = nil
 	for _, path := range []string{"/v1/chat/completions", "/api/chat", "/v1/messages", "/v1/responses"} {
 		for _, stream := range []bool{false, true} {

internal/console/live_test.go

@@ -8,6 +8,8 @@ import (
 )
 
 func TestBuildLiveSnapshot(t *testing.T) {
+	// AccountView.InFlight is the pool's current counter (Acquire/Release),
+	// not a probe snapshot; /api/live just sums those live values.
 	views := []accounts.AccountView{
 		{Account: accounts.Account{ID: "a", Enabled: true, MaxInFlight: 4}, InFlight: 4},
 		{Account: accounts.Account{ID: "b", Enabled: true}, InFlight: 1}, // default 4

internal/executor/chat.go

@@ -11,6 +11,7 @@ import (
 	"net/http"
 	"sort"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/caigee-cmd/cli2api/internal/accounts"
@@ -834,6 +835,46 @@ func (l routeLoop) pickFailure(err error) (int, string, string, error) {
 	return 0, "", "", err
 }
 
+// releaseCloser runs release exactly once when the body is closed, so a
+// successful stream keeps its Acquire until the caller finishes reading.
+type releaseCloser struct {
+	io.ReadCloser
+	once    sync.Once
+	release func()
+}
+
+func (c *releaseCloser) Close() error {
+	err := c.ReadCloser.Close()
+	if c.release != nil {
+		c.once.Do(c.release)
+	}
+	return err
+}
+
+func (e ChatExecutor) acquireInFlight(id string) bool {
+	if e.Pool == nil {
+		return true
+	}
+	return e.Pool.Acquire(id)
+}
+
+func (e ChatExecutor) releaseInFlight(id string) {
+	if e.Pool != nil {
+		e.Pool.Release(id)
+	}
+}
+
+// attachInFlightRelease transfers an acquired slot to the response body.
+// Close releases the pool id that was Acquired (not a rewritten account id).
+func (e ChatExecutor) attachInFlightRelease(resp *http.Response, poolID string) *http.Response {
+	if resp == nil || resp.Body == nil {
+		e.releaseInFlight(poolID)
+		return resp
+	}
+	resp.Body = &releaseCloser{ReadCloser: resp.Body, release: func() { e.releaseInFlight(poolID) }}
+	return resp
+}
+
 func (e ChatExecutor) ChatNonStream(ctx context.Context, req translate.ChatRequest, prefer, providerFilter string) (result ChatResult, returnErr error) {
 	loop := e.newRouteLoop(ctx, prefer, providerFilter, req)
 	defer func() { result.Routing = loop.routing.Source }()
@@ -846,31 +887,44 @@ func (e ChatExecutor) ChatNonStream(ctx context.Context, req translate.ChatReque
 			}
 			return ChatResult{}, pickErr
 		}
+		// Acquire with the pool id before any rewrite of item.ID (X-Qoder-Account).
+		poolID := item.ID
+		if !e.acquireInFlight(poolID) {
+			loop.exclude(item)
+			continue
+		}
+		release := func() { e.releaseInFlight(poolID) }
 		if isInProcessItem(item) || e.qoderDirect(item) {
 			result, classified, err := e.chatInProcessNonStreamAttempt(ctx, item, req, i)
 			if err == nil {
 				result.AttemptCount = i + 1
 				e.observeRouting(&loop.routing, result.AccountID)
 				e.bindSession(loop.routing, result.AccountID)
+				release()
 				return result, nil
 			}
 			loop.lastErr = err
 			if requestContextDone(ctx, err) {
+				release()
 				return ChatResult{AttemptCount: i + 1, AccountID: item.ID, Provider: item.Provider}, err
 			}
 			if loop.canFailover(classified) {
+				release()
 				loop.exclude(item)
 				continue
 			}
+			release()
 			return ChatResult{AttemptCount: i + 1, AccountID: item.ID, Provider: item.Provider}, err
 		}
 		payload, err := json.Marshal(buildWorkerChatPayload(req, false))
 		if err != nil {
+			release()
 			return ChatResult{}, err
 		}
 		headerAccount := loop.headerAccount(item, i)
 		httpReq, err := e.newWorkerRequest(ctx, item, payload, headerAccount)
 		if err != nil {
+			release()
 			return ChatResult{}, err
 		}
 		started := time.Now()
@@ -882,6 +936,7 @@ func (e ChatExecutor) ChatNonStream(ctx context.Context, req translate.ChatReque
 					AttemptIndex: i, AccountID: item.ID, StartedAt: started, FinishedAt: ptrTime(time.Now().UTC()),
 					Status: accounts.AttemptStatusError, ErrorKind: accounts.KindUnavailable, ErrorMessage: err.Error(), LatencyMs: &latency,
 				})
+				release()
 				return ChatResult{AttemptCount: i + 1, AccountID: item.ID, Provider: item.Provider}, err
 			}
 			classified := Classify(0, err.Error(), "", accounts.KindUnavailable, "")
@@ -892,6 +947,7 @@ func (e ChatExecutor) ChatNonStream(ctx context.Context, req translate.ChatReque
 				AttemptIndex: i, AccountID: item.ID, StartedAt: started, FinishedAt: ptrTime(time.Now().UTC()),
 				Status: accounts.AttemptStatusFailover, ErrorKind: classified.Kind, ErrorMessage: loop.lastErr.Error(), LatencyMs: &latency,
 			})
+			release()
 			loop.exclude(item)
 			continue
 		}
@@ -921,13 +977,15 @@ func (e ChatExecutor) ChatNonStream(ctx context.Context, req translate.ChatReque
 					Status: status, HTTPStatus: ptrInt(resp.StatusCode), ErrorKind: classified.Kind,
 					ErrorMessage: truncateErr(msg), LatencyMs: &latency,
 				})
+				release()
 				continue
 			}
 			e.recordAttempt(ctx, accounts.RequestAttempt{
 				AttemptIndex: i, AccountID: item.ID, StartedAt: started, FinishedAt: &finished,
 				Status: status, HTTPStatus: ptrInt(resp.StatusCode), ErrorKind: classified.Kind,
 				ErrorMessage: truncateErr(msg), LatencyMs: &latency,
 			})
+			release()
 			return ChatResult{AttemptCount: i + 1, AccountID: item.ID, Provider: loop.resultProvider(item)}, loop.lastErr
 		}
 		result, err := decodeChatResult(req, body)
@@ -937,6 +995,7 @@ func (e ChatExecutor) ChatNonStream(ctx context.Context, req translate.ChatReque
 				Status: accounts.AttemptStatusError, HTTPStatus: ptrInt(resp.StatusCode),
 				ErrorKind: accounts.KindUnavailable, ErrorMessage: truncateErr(err.Error()), LatencyMs: &latency,
 			})
+			release()
 			return ChatResult{AttemptCount: i + 1, AccountID: item.ID, Provider: loop.resultProvider(item)}, err
 		}
 		result.AccountID = item.ID
@@ -955,6 +1014,7 @@ func (e ChatExecutor) ChatNonStream(ctx context.Context, req translate.ChatReque
 			PromptTokens: ptrInt(result.PromptTokens), CompletionTokens: ptrInt(result.CompletionTokens),
 			UsageSource: result.UsageSource,
 		})
+		release()
 		return result, nil
 	}
 	if loop.lastErr == nil {
@@ -1276,31 +1336,43 @@ func (e ChatExecutor) chatStreamProxy(ctx context.Context, req translate.ChatReq
 			}
 			return StreamResult{}, pickErr
 		}
+		poolID := item.ID
+		if !e.acquireInFlight(poolID) {
+			loop.exclude(item)
+			continue
+		}
+		release := func() { e.releaseInFlight(poolID) }
 		if isInProcessItem(item) || (e.qoderDirect(item) && !(preferNativeResponses && native != nil)) {
 			result, classified, err := e.chatInProcessStreamAttempt(ctx, item, req, native, i, preferNativeResponses)
 
 			if err == nil {
 				result.AttemptCount = i + 1
 				e.observeRouting(&loop.routing, result.AccountID)
+				result.Response = e.attachInFlightRelease(result.Response, poolID)
 				return result, nil
 			}
 			loop.lastErr = err
 			if requestContextDone(ctx, err) {
+				release()
 				return StreamResult{AttemptCount: i + 1, AccountID: item.ID, Provider: item.Provider}, err
 			}
 			if loop.canFailover(classified) {
+				release()
 				loop.exclude(item)
 				continue
 			}
+			release()
 			return StreamResult{AttemptCount: i + 1, AccountID: item.ID, Provider: item.Provider}, err
 		}
 		payload, err := json.Marshal(buildWorkerChatPayload(req, true))
 		if err != nil {
+			release()
 			return StreamResult{}, err
 		}
 		headerAccount := loop.headerAccount(item, i)
 		httpReq, err := e.newWorkerRequest(ctx, item, payload, headerAccount)
 		if err != nil {
+			release()
 			return StreamResult{}, err
 		}
 		started := time.Now()
@@ -1312,6 +1384,7 @@ func (e ChatExecutor) chatStreamProxy(ctx context.Context, req translate.ChatReq
 					AttemptIndex: i, AccountID: item.ID, StartedAt: started, FinishedAt: ptrTime(time.Now().UTC()),
 					Status: accounts.AttemptStatusError, ErrorKind: accounts.KindUnavailable, ErrorMessage: err.Error(), LatencyMs: &latency,
 				})
+				release()
 				return StreamResult{AttemptCount: i + 1, AccountID: item.ID, Provider: item.Provider}, err
 			}
 			classified := Classify(0, err.Error(), "", accounts.KindUnavailable, "")
@@ -1322,6 +1395,7 @@ func (e ChatExecutor) chatStreamProxy(ctx context.Context, req translate.ChatReq
 				AttemptIndex: i, AccountID: item.ID, StartedAt: started, FinishedAt: ptrTime(time.Now().UTC()),
 				Status: accounts.AttemptStatusFailover, ErrorKind: classified.Kind, ErrorMessage: loop.lastErr.Error(), LatencyMs: &latency,
 			})
+			release()
 			loop.exclude(item)
 			continue
 		}
@@ -1349,13 +1423,15 @@ func (e ChatExecutor) chatStreamProxy(ctx context.Context, req translate.ChatReq
 					Status: status, HTTPStatus: ptrInt(resp.StatusCode), ErrorKind: classified.Kind,
 					ErrorMessage: truncateErr(msg), LatencyMs: &latency,
 				})
+				release()
 				continue
 			}
 			e.recordAttempt(ctx, accounts.RequestAttempt{
 				AttemptIndex: i, AccountID: item.ID, StartedAt: started, FinishedAt: &finished,
 				Status: status, HTTPStatus: ptrInt(resp.StatusCode), ErrorKind: classified.Kind,
 				ErrorMessage: truncateErr(msg), LatencyMs: &latency,
 			})
+			release()
 			return StreamResult{AttemptCount: i + 1, AccountID: item.ID, Provider: loop.resultProvider(item)}, loop.lastErr
 		}
 		e.markOK(item.ID, req.Model)
@@ -1370,6 +1446,7 @@ func (e ChatExecutor) chatStreamProxy(ctx context.Context, req translate.ChatReq
 		if resolved != "" {
 			logResolvedReasoning(ctx, "chat_stream", item, req.Model, resolved)
 		}
+		resp = e.attachInFlightRelease(resp, poolID)
 		return StreamResult{
 			Response:       resp,
 			AccountID:      item.ID,

internal/executor/chat_test.go

@@ -340,8 +340,11 @@ func TestChatNonStreamAllSaturatedReturnsRateLimit(t *testing.T) {
 	pool := NewPool([]string{"http://127.0.0.1:1", "http://127.0.0.1:2"}, []string{"a", "b"})
 	pool.Upsert(Item{ID: "a", URL: "http://127.0.0.1:1", MaxInFlight: 1})
 	pool.Upsert(Item{ID: "b", URL: "http://127.0.0.1:2", MaxInFlight: 1})
-	pool.MergeHealth("a", true, false, 1, 0, "")
-	pool.MergeHealth("b", true, false, 1, 0, "")
+	pool.MergeHealth("a", true, false, 0, 0, "")
+	pool.MergeHealth("b", true, false, 0, 0, "")
+	if !pool.Acquire("a") || !pool.Acquire("b") {
+		t.Fatal("acquire a/b")
+	}
 	ex := NewChatExecutor(pool, "")
 	_, err := ex.ChatNonStream(context.Background(), translate.ChatRequest{
 		Model: "glm-5.3", Messages: []translate.ChatMessage{{Role: "user", Content: "hi"}},