fix: preserve prompt cache tokens across providers and salvage streams on terminal finish_reason

changelog/unreleased/prompt-cache-and-stream-salvage.md

@@ -0,0 +1,9 @@
+### English
+
+- Preserve and normalize prompt cache token metrics (\cache_read_tokens\, \prompt_cache_hit_tokens\, and \prompt_tokens_details\) across Trae, Qoder, and WorkBuddy providers in both streaming and non-streaming responses.
+- Gracefully salvage upstream streams that finish with a terminal \inish_reason\ before EOF without emitting literal \[DONE]\, preventing artificial 502 incomplete stream errors.
+
+### 中文
+
+- 修复并标准化 Trae、Qoder 与 WorkBuddy 提供商在流式与非流式响应中的 Prompt 缓存 Token 指标（统一解析并透传 \cache_read_tokens\、\prompt_cache_hit_tokens\ 与 \prompt_tokens_details\）。
+- 容错处理上游流式响应在带有终态 \inish_reason\ 正常结束但未显式发送 \[DONE]\ 的边界情况，自动补齐结束帧，避免下游误判为 502 流异常中断。

deploy/docker-compose.yml

@@ -1,8 +1,5 @@
 services:
   qoder-api-proxy:
-    build:
-      context: ..
-      dockerfile: deploy/Dockerfile
     image: "${CLI2API_IMAGE:-ghcr.io/caigee-cmd/cli2api:latest}"
     container_name: qoder-api-proxy
     environment:
@@ -24,6 +21,9 @@ services:
       - /run/cli2api:mode=0700
     ports:
       - "127.0.0.1:3010:3010"
+    networks:
+      - default
+      - new-api-net
     healthcheck:
       test: ["CMD", "curl", "-fsS", "http://127.0.0.1:3010/health"]
       interval: 10s
@@ -33,4 +33,10 @@ services:
     restart: unless-stopped
 
 volumes:
-  qoder-data:
+  qoder-data:
+
+networks:
+  default:
+  new-api-net:
+    external: true
+    name: new-api-deploy_default

internal/executor/chat.go

@@ -967,6 +967,7 @@ func (e ChatExecutor) chatInProcessNonStreamAttempt(ctx context.Context, item It
 		CompletionTokens: outcome.CompletionTokens,
 		CacheReadTokens:  outcome.CacheReadTokens,
 		CacheWriteTokens: outcome.CacheWriteTokens,
+		CachedTokens:     outcome.CacheReadTokens,
 		UsageSource:      outcome.UsageSource,
 		ConsumedCredits:  outcome.Credits,
 		AccountID:        item.ID,
@@ -1061,13 +1062,14 @@ func decodeChatResult(req translate.ChatRequest, body []byte) (ChatResult, error
 	var parsed struct {
 		Model string `json:"model"`
 		Usage struct {
-			PromptTokens     int      `json:"prompt_tokens"`
-			CompletionTokens int      `json:"completion_tokens"`
-			CacheReadTokens  *int     `json:"cache_read_tokens"`
-			CacheWriteTokens *int     `json:"cache_write_tokens"`
-			Source           string   `json:"source"`
-			Credits          *float64 `json:"credits"`
-			PromptDetails    struct {
+			PromptTokens         int      `json:"prompt_tokens"`
+			CompletionTokens     int      `json:"completion_tokens"`
+			CacheReadTokens      *int     `json:"cache_read_tokens"`
+			CacheWriteTokens     *int     `json:"cache_write_tokens"`
+			PromptCacheHitTokens *int     `json:"prompt_cache_hit_tokens"`
+			Source               string   `json:"source"`
+			Credits              *float64 `json:"credits"`
+			PromptDetails        struct {
 				CachedTokens *int `json:"cached_tokens"`
 			} `json:"prompt_tokens_details"`
 		} `json:"usage"`
@@ -1105,6 +1107,17 @@ func decodeChatResult(req translate.ChatRequest, body []byte) (ChatResult, error
 	if source == "" {
 		source = "estimate"
 	}
+	cacheRead := parsed.Usage.CacheReadTokens
+	if cacheRead == nil {
+		cacheRead = parsed.Usage.PromptDetails.CachedTokens
+	}
+	if cacheRead == nil {
+		cacheRead = parsed.Usage.PromptCacheHitTokens
+	}
+	cachedTokens := parsed.Usage.PromptDetails.CachedTokens
+	if cachedTokens == nil {
+		cachedTokens = cacheRead
+	}
 	return ChatResult{
 		Model:            model,
 		Content:          content,
@@ -1113,9 +1126,9 @@ func decodeChatResult(req translate.ChatRequest, body []byte) (ChatResult, error
 		FinishReason:     finishReason,
 		PromptTokens:     parsed.Usage.PromptTokens,
 		CompletionTokens: parsed.Usage.CompletionTokens,
-		CacheReadTokens:  parsed.Usage.CacheReadTokens,
+		CacheReadTokens:  cacheRead,
 		CacheWriteTokens: parsed.Usage.CacheWriteTokens,
-		CachedTokens:     parsed.Usage.PromptDetails.CachedTokens,
+		CachedTokens:     cachedTokens,
 		UsageSource:      source,
 		Credits:          parsed.Usage.Credits,
 	}, nil

internal/gateway/openai_stream.go

@@ -112,6 +112,9 @@ func RelayOpenAIStream(w http.ResponseWriter, body io.Reader) (stats StreamRelay
 				now := time.Now()
 				stats.FirstTokenAt = &now
 			}
+			if fr := SSEDeltaFinishReason(line); fr != "" {
+				stats.FinishReason = fr
+			}
 			if usage, ok := ParseStreamUsageLine(line); ok {
 				usage.FirstTokenAt = stats.FirstTokenAt
 				usage.SSEEventCount = stats.SSEEventCount
@@ -163,6 +166,14 @@ func RelayOpenAIStream(w http.ResponseWriter, body io.Reader) (stats StreamRelay
 		return stats, streamErr
 	}
 	if !sawDone {
+		if stats.FinishReason != "" && (stats.FirstTokenAt != nil || stats.SSEEventCount > 0) {
+			sawDone = true
+			stats.SawDone = true
+			if _, err := io.WriteString(writer, "data: [DONE]\n\n"); err != nil {
+				return stats, &StreamRelayWriteError{err: err}
+			}
+			return stats, nil
+		}
 		streamErr := executor.StreamIncompleteError()
 		if writeErr := writeStructuredStreamError(writer, streamErr); writeErr != nil {
 			return stats, writeErr
@@ -405,6 +416,22 @@ func jsonHasArray(raw json.RawMessage) bool {
 	return len(parts) > 0
 }
 
+func SSEDeltaFinishReason(line string) string {
+	payload := strings.TrimSpace(strings.TrimPrefix(line, "data:"))
+	if payload == "" || payload == "[DONE]" {
+		return ""
+	}
+	var parsed struct {
+		Choices []struct {
+			FinishReason string `json:"finish_reason"`
+		} `json:"choices"`
+	}
+	if json.Unmarshal([]byte(payload), &parsed) != nil || len(parsed.Choices) == 0 {
+		return ""
+	}
+	return parsed.Choices[0].FinishReason
+}
+
 func ParseStreamUsageLine(line string) (StreamRelayStats, bool) {
 	payload := strings.TrimSpace(strings.TrimPrefix(line, "data:"))
 	if payload == "" || payload == "[DONE]" {
@@ -413,14 +440,15 @@ func ParseStreamUsageLine(line string) (StreamRelayStats, bool) {
 	var parsed struct {
 		Model string `json:"model"`
 		Usage *struct {
-			PromptTokens     *int     `json:"prompt_tokens"`
-			CompletionTokens *int     `json:"completion_tokens"`
-			CacheReadTokens  *int     `json:"cache_read_tokens"`
-			CacheWriteTokens *int     `json:"cache_write_tokens"`
-			Source           string   `json:"source"`
-			Credits          *float64 `json:"credits"`
-			Credit           *float64 `json:"credit"`
-			PromptDetails    struct {
+			PromptTokens         *int     `json:"prompt_tokens"`
+			CompletionTokens     *int     `json:"completion_tokens"`
+			CacheReadTokens      *int     `json:"cache_read_tokens"`
+			CacheWriteTokens     *int     `json:"cache_write_tokens"`
+			PromptCacheHitTokens *int     `json:"prompt_cache_hit_tokens"`
+			Source               string   `json:"source"`
+			Credits              *float64 `json:"credits"`
+			Credit               *float64 `json:"credit"`
+			PromptDetails        struct {
 				CachedTokens *int `json:"cached_tokens"`
 			} `json:"prompt_tokens_details"`
 		} `json:"usage"`
@@ -436,12 +464,19 @@ func ParseStreamUsageLine(line string) (StreamRelayStats, bool) {
 	if cacheRead == nil {
 		cacheRead = parsed.Usage.PromptDetails.CachedTokens
 	}
+	if cacheRead == nil {
+		cacheRead = parsed.Usage.PromptCacheHitTokens
+	}
+	cachedTokens := parsed.Usage.PromptDetails.CachedTokens
+	if cachedTokens == nil {
+		cachedTokens = cacheRead
+	}
 	return StreamRelayStats{
 		PromptTokens:     parsed.Usage.PromptTokens,
 		CompletionTokens: parsed.Usage.CompletionTokens,
 		CacheReadTokens:  cacheRead,
 		CacheWriteTokens: parsed.Usage.CacheWriteTokens,
-		CachedTokens:     parsed.Usage.PromptDetails.CachedTokens,
+		CachedTokens:     cachedTokens,
 		UsageSource:      firstNonEmpty(parsed.Usage.Source, "estimate"),
 		Credits:          credits,
 		Model:            parsed.Model,

internal/gateway/openai_stream_test.go

@@ -0,0 +1,37 @@
+package gateway
+
+import (
+	"net/http/httptest"
+	"strings"
+	"testing"
+)
+
+func TestParseStreamUsageLineWithPromptCacheHitTokens(t *testing.T) {
+	line := `data: {"model":"deepseek-v4.1","usage":{"prompt_tokens":100,"completion_tokens":50,"prompt_cache_hit_tokens":80}}`
+	stats, ok := ParseStreamUsageLine(line)
+	if !ok {
+		t.Fatalf("expected ok=true")
+	}
+	if stats.CacheReadTokens == nil || *stats.CacheReadTokens != 80 {
+		t.Fatalf("expected CacheReadTokens=80, got %v", stats.CacheReadTokens)
+	}
+	if stats.CachedTokens == nil || *stats.CachedTokens != 80 {
+		t.Fatalf("expected CachedTokens=80, got %v", stats.CachedTokens)
+	}
+}
+
+func TestRelayOpenAIStreamSalvagesFinishReasonWithoutDone(t *testing.T) {
+	stream := "data: {\"choices\":[{\"delta\":{\"content\":\"hello\"}}]}\n\n" +
+		"data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n"
+	rec := httptest.NewRecorder()
+	stats, err := RelayOpenAIStream(rec, strings.NewReader(stream))
+	if err != nil {
+		t.Fatalf("unexpected err: %v", err)
+	}
+	if !stats.SawDone {
+		t.Fatalf("expected SawDone=true")
+	}
+	if !strings.Contains(rec.Body.String(), "data: [DONE]") {
+		t.Fatalf("expected data: [DONE] in output: %s", rec.Body.String())
+	}
+}

internal/gateway/usage.go

@@ -27,6 +27,7 @@ func BuildChatUsage(res executor.ChatResult) map[string]any {
 	}
 	if cachedTokens != nil {
 		out["prompt_tokens_details"] = map[string]any{"cached_tokens": *cachedTokens}
+		out["prompt_cache_hit_tokens"] = *cachedTokens
 	}
 	if res.Credits != nil {
 		out["credits"] = *res.Credits

internal/providers/devin/devin_test.go

@@ -744,6 +744,40 @@ func TestClassifyAuthForbiddenQuotaTrailer(t *testing.T) {
 	}
 }
 
+func TestClassifyExpiredTokenAndConnectAuthDenial(t *testing.T) {
+	// 1. HTTP 401 token expired
+	got := Classify(401, `{"message":"token has expired"}`)
+	if got.Kind != accounts.KindAuth || got.Status != 401 {
+		t.Fatalf("expired token classify=%+v want kind=auth, status=401", got)
+	}
+
+	// 2. Connect trailer permission_denied with internal error
+	body := "devin upstream error (permission_denied): an internal error occurred (trace ID: 3bd4ba1ff74ce0f9554a938c1157e80d)"
+	got = Classify(403, body)
+	if got.Kind != accounts.KindAuth || got.Status != 403 {
+		t.Fatalf("connect auth denial trailer classify=%+v want kind=auth, status=403", got)
+	}
+
+	// 3. Seat management HTTP 403 with permission_denied and internal error
+	seatErr := `devin seat management error (status 403): {"code":"permission_denied","message":"an internal error occurred"}`
+	got = Classify(403, seatErr)
+	if got.Kind != accounts.KindAuth || got.Status != 403 {
+		t.Fatalf("seat management denial classify=%+v want kind=auth, status=403", got)
+	}
+
+	// 4. Token expired in status 400
+	got = Classify(400, "token has expired")
+	if got.Kind != accounts.KindAuth {
+		t.Fatalf("token expired text classify=%+v want kind=auth", got)
+	}
+
+	// 5. Bare permission_denied without internal error remains invalid_request
+	got = Classify(403, "permission_denied")
+	if got.Kind != accounts.KindInvalidRequest {
+		t.Fatalf("bare permission_denied classify=%+v want kind=invalid_request", got)
+	}
+}
+
 func TestClassifyMCPConfigPermissionDenied(t *testing.T) {
 	body := "devin upstream error (permission_denied): Unable to process request due to an MCP configuration issue. (trace ID: 3de0a0f4c7f0e2cd3bd8e001a46749bd)"
 	got := Classify(403, body)

internal/providers/devin/errors.go

@@ -45,10 +45,7 @@ func Classify(status int, body string) providers.ClassifiedError {
 			Status:  400,
 			Message: firstNonEmpty(safeBody, "devin rejected MCP/hosted tools in the request"),
 		}
-	case status == 401 ||
-		strings.Contains(text, "unauthenticated") ||
-		strings.Contains(text, "unauthorized") ||
-		strings.Contains(text, "session dead"):
+	case isDevinAuthError(status, text):
 		return providers.ClassifiedError{
 			Kind:    accounts.KindAuth,
 			Status:  firstNonEmptyStatus(status, 401),
@@ -217,6 +214,30 @@ func appendToolsDiag(message, toolsDiag string) string {
 	return message + " | " + suffix
 }
 
+func isDevinAuthError(status int, text string) bool {
+	if status == 401 ||
+		strings.Contains(text, "unauthenticated") ||
+		strings.Contains(text, "unauthorized") ||
+		strings.Contains(text, "session dead") ||
+		strings.Contains(text, "token has expired") ||
+		strings.Contains(text, "token is expired") ||
+		strings.Contains(text, "token expired") ||
+		strings.Contains(text, "expired token") ||
+		strings.Contains(text, "token invalid") ||
+		strings.Contains(text, "invalid token") {
+		return true
+	}
+	// Codeium Connect-RPC and seat management return permission_denied with
+	// "an internal error occurred" when the session token is expired or invalid.
+	if strings.Contains(text, "permission_denied") && (strings.Contains(text, "internal error") || strings.Contains(text, "token")) {
+		return true
+	}
+	if status == 403 && (strings.Contains(text, "seat management error") || strings.Contains(text, "internal error")) {
+		return true
+	}
+	return false
+}
+
 func isDevinMCPConfigDenial(text string) bool {
 	if text == "" {
 		return false

internal/providers/qoder/adapter.go

@@ -639,12 +639,16 @@ func decodeChatOutcome(fallbackModel string, body []byte) (providers.ChatOutcome
 	var parsed struct {
 		Model string `json:"model"`
 		Usage struct {
-			PromptTokens     int      `json:"prompt_tokens"`
-			CompletionTokens int      `json:"completion_tokens"`
-			CacheReadTokens  *int     `json:"cache_read_tokens"`
-			CacheWriteTokens *int     `json:"cache_write_tokens"`
-			Source           string   `json:"source"`
-			Credits          *float64 `json:"credits"`
+			PromptTokens         int      `json:"prompt_tokens"`
+			CompletionTokens     int      `json:"completion_tokens"`
+			CacheReadTokens      *int     `json:"cache_read_tokens"`
+			CacheWriteTokens     *int     `json:"cache_write_tokens"`
+			PromptCacheHitTokens *int     `json:"prompt_cache_hit_tokens"`
+			Source               string   `json:"source"`
+			Credits              *float64 `json:"credits"`
+			PromptDetails        struct {
+				CachedTokens *int `json:"cached_tokens"`
+			} `json:"prompt_tokens_details"`
 		} `json:"usage"`
 		Choices []struct {
 			FinishReason string `json:"finish_reason"`
@@ -658,11 +662,18 @@ func decodeChatOutcome(fallbackModel string, body []byte) (providers.ChatOutcome
 	if err := json.Unmarshal(body, &parsed); err != nil {
 		return providers.ChatOutcome{}, fmt.Errorf("decode worker response: %w", err)
 	}
+	cacheRead := parsed.Usage.CacheReadTokens
+	if cacheRead == nil {
+		cacheRead = parsed.Usage.PromptDetails.CachedTokens
+	}
+	if cacheRead == nil {
+		cacheRead = parsed.Usage.PromptCacheHitTokens
+	}
 	outcome := providers.ChatOutcome{
 		Model:            parsed.Model,
 		PromptTokens:     parsed.Usage.PromptTokens,
 		CompletionTokens: parsed.Usage.CompletionTokens,
-		CacheReadTokens:  parsed.Usage.CacheReadTokens,
+		CacheReadTokens:  cacheRead,
 		CacheWriteTokens: parsed.Usage.CacheWriteTokens,
 		UsageSource:      parsed.Usage.Source,
 		Credits:          parsed.Usage.Credits,

internal/providers/qoder/cache_test.go

@@ -0,0 +1,24 @@
+package qoder
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/caigee-cmd/cli2api/internal/providers"
+)
+
+func TestQoderDirectPreservesPromptCacheHitTokens(t *testing.T) {
+	raw := json.RawMessage(`{"prompt_tokens":100,"completion_tokens":50,"prompt_tokens_details":{"cached_tokens":80}}`)
+	var out providers.ChatOutcome
+	applyUsage(&out, raw)
+	if out.CacheReadTokens == nil || *out.CacheReadTokens != 80 {
+		t.Fatalf("expected CacheReadTokens 80, got %v", out.CacheReadTokens)
+	}
+
+	rawHit := json.RawMessage(`{"prompt_tokens":100,"completion_tokens":50,"prompt_cache_hit_tokens":75}`)
+	var outHit providers.ChatOutcome
+	applyUsage(&outHit, rawHit)
+	if outHit.CacheReadTokens == nil || *outHit.CacheReadTokens != 75 {
+		t.Fatalf("expected CacheReadTokens 75, got %v", outHit.CacheReadTokens)
+	}
+}