Merge remote-tracking branch 'origin/main' into codex/project-resume-reuse

internal/database/model_token_usage.go

@@ -30,6 +30,7 @@ type ModelTokenUsage struct {
 	PromptTokens     int64     `json:"promptTokens"`
 	CompletionTokens int64     `json:"completionTokens"`
 	TotalTokens      int64     `json:"totalTokens"`
+	CacheWriteTokens int64     `json:"cacheWriteTokens"`
 	CachedTokens     int64     `json:"cachedTokens"`
 	ReasoningTokens  int64     `json:"reasoningTokens"`
 	CreatedAt        time.Time `json:"createdAt"`
@@ -43,6 +44,7 @@ type ModelTokenUsageSummary struct {
 	PromptTokens     int64 `json:"promptTokens"`
 	CompletionTokens int64 `json:"completionTokens"`
 	TotalTokens      int64 `json:"totalTokens"`
+	CacheWriteTokens int64 `json:"cacheWriteTokens"`
 	CachedTokens     int64 `json:"cachedTokens"`
 	ReasoningTokens  int64 `json:"reasoningTokens"`
 }
@@ -56,6 +58,7 @@ type ModelTokenUsageBreakdown struct {
 	PromptTokens     int64  `json:"promptTokens"`
 	CompletionTokens int64  `json:"completionTokens"`
 	TotalTokens      int64  `json:"totalTokens"`
+	CacheWriteTokens int64  `json:"cacheWriteTokens"`
 	CachedTokens     int64  `json:"cachedTokens"`
 	ReasoningTokens  int64  `json:"reasoningTokens"`
 }
@@ -100,6 +103,7 @@ func modelTokenUsageFromProcessDetail(messageID, conversationID, processDetailID
 		CompletionTokens: usageInt64(m["completionTokens"]),
 		TotalTokens:      usageInt64(m["totalTokens"]),
 		CachedTokens:     usageInt64(m["cachedTokens"]),
+		CacheWriteTokens: usageInt64(m["cacheWriteTokens"]),
 		ReasoningTokens:  usageInt64(m["reasoningTokens"]),
 	}
 	if usage.TotalTokens == 0 && (usage.PromptTokens > 0 || usage.CompletionTokens > 0) {
@@ -108,7 +112,7 @@ func modelTokenUsageFromProcessDetail(messageID, conversationID, processDetailID
 	if usage.ProcessDetailID == "" || usage.MessageID == "" || usage.ConversationID == "" {
 		return ModelTokenUsage{}, false
 	}
-	if usage.ModelCalls == 0 && usage.TotalTokens == 0 && usage.PromptTokens == 0 && usage.CompletionTokens == 0 && usage.CachedTokens == 0 && usage.ReasoningTokens == 0 {
+	if usage.ModelCalls == 0 && usage.TotalTokens == 0 && usage.PromptTokens == 0 && usage.CompletionTokens == 0 && usage.CachedTokens == 0 && usage.CacheWriteTokens == 0 && usage.ReasoningTokens == 0 {
 		return ModelTokenUsage{}, false
 	}
 	return usage, true
@@ -224,9 +228,9 @@ func (db *DB) UpsertModelTokenUsage(usage ModelTokenUsage) error {
 INSERT INTO model_token_usage (
 	id, process_detail_id, message_id, conversation_id, project_id,
 	source, orchestration, reason, model, model_calls,
-	prompt_tokens, completion_tokens, total_tokens, cached_tokens, reasoning_tokens,
+	prompt_tokens, completion_tokens, total_tokens, cached_tokens, cache_write_tokens, reasoning_tokens,
 	created_at, updated_at
-) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
 ON CONFLICT(process_detail_id) DO UPDATE SET
 	message_id = excluded.message_id,
 	conversation_id = excluded.conversation_id,
@@ -240,12 +244,13 @@ ON CONFLICT(process_detail_id) DO UPDATE SET
 	completion_tokens = excluded.completion_tokens,
 	total_tokens = excluded.total_tokens,
 	cached_tokens = excluded.cached_tokens,
+	cache_write_tokens = excluded.cache_write_tokens,
 	reasoning_tokens = excluded.reasoning_tokens,
 	created_at = excluded.created_at,
 	updated_at = excluded.updated_at`,
 		usage.ID, usage.ProcessDetailID, usage.MessageID, usage.ConversationID, projectValue,
 		usage.Source, usage.Orchestration, usage.Reason, usage.Model, usage.ModelCalls,
-		usage.PromptTokens, usage.CompletionTokens, usage.TotalTokens, usage.CachedTokens, usage.ReasoningTokens,
+		usage.PromptTokens, usage.CompletionTokens, usage.TotalTokens, usage.CachedTokens, usage.CacheWriteTokens, usage.ReasoningTokens,
 		createdAt, now,
 	)
 	if err != nil {
@@ -318,7 +323,7 @@ func (db *DB) GetModelTokenUsageStats(filter ModelTokenUsageFilter) (*ModelToken
 		return nil, err
 	}
 	byDay, err := db.queryModelTokenUsageBreakdown(
-		"SELECT date(mtu.created_at) AS k, date(mtu.created_at) AS label, "+modelTokenUsageSummarySelect("mtu")+" FROM model_token_usage mtu JOIN conversations c ON c.id = mtu.conversation_id"+where+" GROUP BY date(mtu.created_at) ORDER BY k DESC LIMIT ?",
+		"SELECT date(mtu.created_at, 'localtime') AS k, date(mtu.created_at, 'localtime') AS label, "+modelTokenUsageSummarySelect("mtu")+" FROM model_token_usage mtu JOIN conversations c ON c.id = mtu.conversation_id"+where+" GROUP BY date(mtu.created_at, 'localtime') ORDER BY k DESC LIMIT ?",
 		append(args, filter.Days)...,
 	)
 	if err != nil {
@@ -363,7 +368,8 @@ COALESCE(SUM(%sprompt_tokens), 0),
 COALESCE(SUM(%scompletion_tokens), 0),
 COALESCE(SUM(%stotal_tokens), 0),
 COALESCE(SUM(%scached_tokens), 0),
-COALESCE(SUM(%sreasoning_tokens), 0)`, p, p, p, p, p, p, p)
+COALESCE(SUM(%scache_write_tokens), 0),
+COALESCE(SUM(%sreasoning_tokens), 0)`, p, p, p, p, p, p, p, p)
 }
 
 func buildModelTokenUsageWhere(filter ModelTokenUsageFilter, usageAlias, convAlias string) (string, []interface{}) {
@@ -394,7 +400,7 @@ func (db *DB) queryModelTokenUsageSummary(query string, args ...interface{}) (Mo
 	var s ModelTokenUsageSummary
 	err := db.QueryRow(query, args...).Scan(
 		&s.Events, &s.ModelCalls, &s.PromptTokens, &s.CompletionTokens,
-		&s.TotalTokens, &s.CachedTokens, &s.ReasoningTokens,
+		&s.TotalTokens, &s.CachedTokens, &s.CacheWriteTokens, &s.ReasoningTokens,
 	)
 	if err != nil {
 		return s, fmt.Errorf("查询模型Token用量汇总失败: %w", err)
@@ -413,7 +419,7 @@ func (db *DB) queryModelTokenUsageBreakdown(query string, args ...interface{}) (
 		var row ModelTokenUsageBreakdown
 		if err := rows.Scan(
 			&row.Key, &row.Label, &row.Events, &row.ModelCalls, &row.PromptTokens,
-			&row.CompletionTokens, &row.TotalTokens, &row.CachedTokens, &row.ReasoningTokens,
+			&row.CompletionTokens, &row.TotalTokens, &row.CachedTokens, &row.CacheWriteTokens, &row.ReasoningTokens,
 		); err != nil {
 			return nil, fmt.Errorf("扫描模型Token用量分组失败: %w", err)
 		}
@@ -438,7 +444,7 @@ func (db *DB) ListModelTokenUsage(filter ModelTokenUsageFilter) ([]ModelTokenUsa
 SELECT mtu.id, mtu.process_detail_id, mtu.message_id, mtu.conversation_id,
 	COALESCE(mtu.project_id, ''), mtu.source, mtu.orchestration, mtu.reason, mtu.model,
 	mtu.model_calls, mtu.prompt_tokens, mtu.completion_tokens, mtu.total_tokens,
-	mtu.cached_tokens, mtu.reasoning_tokens, mtu.created_at, mtu.updated_at
+	mtu.cached_tokens, mtu.cache_write_tokens, mtu.reasoning_tokens, mtu.created_at, mtu.updated_at
 FROM model_token_usage mtu
 JOIN conversations c ON c.id = mtu.conversation_id`+where+`
 ORDER BY mtu.created_at DESC, mtu.rowid DESC
@@ -455,7 +461,7 @@ LIMIT ?`, args...)
 			&u.ID, &u.ProcessDetailID, &u.MessageID, &u.ConversationID, &u.ProjectID,
 			&u.Source, &u.Orchestration, &u.Reason, &u.Model, &u.ModelCalls,
 			&u.PromptTokens, &u.CompletionTokens, &u.TotalTokens, &u.CachedTokens,
-			&u.ReasoningTokens, &createdAt, &updatedAt,
+			&u.CacheWriteTokens, &u.ReasoningTokens, &createdAt, &updatedAt,
 		); err != nil {
 			return nil, fmt.Errorf("扫描模型Token用量明细失败: %w", err)
 		}

internal/database/model_token_usage_test.go

@@ -28,6 +28,7 @@ func TestModelTokenUsagePersistsFromUsageProcessDetail(t *testing.T) {
 		"totalTokens":      13,
 		"cachedTokens":     4,
 		"reasoningTokens":  1,
+		"cacheWriteTokens": 2,
 	}); err != nil {
 		t.Fatalf("AddProcessDetail: %v", err)
 	}
@@ -36,12 +37,15 @@ func TestModelTokenUsagePersistsFromUsageProcessDetail(t *testing.T) {
 	if err != nil {
 		t.Fatalf("GetModelTokenUsageStats: %v", err)
 	}
-	if stats.Summary.Events != 1 || stats.Summary.ModelCalls != 2 || stats.Summary.TotalTokens != 13 || stats.Summary.CachedTokens != 4 || stats.Summary.ReasoningTokens != 1 {
+	if stats.Summary.Events != 1 || stats.Summary.ModelCalls != 2 || stats.Summary.TotalTokens != 13 || stats.Summary.CachedTokens != 4 || stats.Summary.ReasoningTokens != 1 || stats.Summary.CacheWriteTokens != 2 {
 		t.Fatalf("summary = %#v", stats.Summary)
 	}
-	if len(stats.ByModel) != 1 || stats.ByModel[0].Key != "gpt-test" || stats.ByModel[0].TotalTokens != 13 {
+	if len(stats.ByModel) != 1 || stats.ByModel[0].Key != "gpt-test" || stats.ByModel[0].TotalTokens != 13 || stats.ByModel[0].CacheWriteTokens != 2 {
 		t.Fatalf("by model = %#v", stats.ByModel)
 	}
+	if len(stats.Recent) != 1 || stats.Recent[0].CacheWriteTokens != 2 || stats.Today.CacheWriteTokens != 2 || len(stats.ByDay) != 1 || stats.ByDay[0].CacheWriteTokens != 2 {
+		t.Fatalf("cache write usage lost: %#v", stats)
+	}
 }
 
 func TestModelTokenUsageBackfillIsIdempotent(t *testing.T) {
@@ -83,3 +87,36 @@ func newModelTokenUsageTestDB(t *testing.T) *DB {
 	t.Cleanup(func() { _ = db.Close() })
 	return db
 }
+
+func TestModelTokenUsageStatsRespectConversationAccess(t *testing.T) {
+	db := newModelTokenUsageTestDB(t)
+	var ownID string
+	for _, owner := range []string{"alice", "bob"} {
+		conv, err := db.CreateConversation(owner, ConversationCreateMeta{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if owner == "alice" {
+			ownID = conv.ID
+		}
+		if _, err := db.Exec(`UPDATE conversations SET owner_user_id = ? WHERE id = ?`, owner, conv.ID); err != nil {
+			t.Fatal(err)
+		}
+		msg, err := db.AddMessage(conv.ID, "assistant", "done", nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if err := db.AddProcessDetail(msg.ID, conv.ID, modelTokenUsageEventType, "usage", map[string]interface{}{
+			"model": owner, "modelCalls": 1, "promptTokens": 100, "completionTokens": 20, "cacheWriteTokens": 30,
+		}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	stats, err := db.GetModelTokenUsageStats(ModelTokenUsageFilter{Access: RBACListAccess{UserID: "alice", Scope: "own"}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if stats.Summary.Events != 1 || stats.Summary.TotalTokens != 120 || stats.Summary.CacheWriteTokens != 30 || len(stats.Recent) != 1 || stats.Recent[0].ConversationID != ownID || len(stats.ByModel) != 1 || stats.ByModel[0].Key != "alice" {
+		t.Fatalf("usage leaks across users: %#v", stats)
+	}
+}

internal/database/schema_reconcile_test.go

@@ -115,6 +115,7 @@ func TestStartupRepairsTablesAndUnlistedColumnsPreservesData(t *testing.T) {
  ALTER TABLE hitl_interrupts DROP COLUMN payload;
  ALTER TABLE hitl_conversation_configs DROP COLUMN reviewer;
  ALTER TABLE model_token_usage DROP COLUMN cached_tokens;
+ ALTER TABLE model_token_usage DROP COLUMN cache_write_tokens;
  ALTER TABLE rbac_users DROP COLUMN display_name;
  ALTER TABLE messages DROP COLUMN reasoning_content;`)
 	db.Close()

internal/database/schemas/application.sql

@@ -321,6 +321,7 @@ CREATE TABLE model_token_usage (
     completion_tokens INTEGER NOT NULL DEFAULT 0,
     total_tokens INTEGER NOT NULL DEFAULT 0,
     cached_tokens INTEGER NOT NULL DEFAULT 0,
+    cache_write_tokens INTEGER NOT NULL DEFAULT 0,
     reasoning_tokens INTEGER NOT NULL DEFAULT 0,
     created_at DATETIME NOT NULL,
     updated_at DATETIME NOT NULL,

internal/multiagent/eino_run_usage_accumulator.go

@@ -13,6 +13,7 @@ type einoRunUsageSummary struct {
 	CompletionTokens int
 	TotalTokens      int
 	CachedTokens     int
+	CacheWriteTokens int
 	ReasoningTokens  int
 }
 
@@ -42,8 +43,13 @@ func (a *einoRunUsageAccumulator) AddUsage(usage *schema.TokenUsage) bool {
 	a.summary.ModelCalls++
 	a.summary.PromptTokens += usage.PromptTokens
 	a.summary.CompletionTokens += usage.CompletionTokens
-	a.summary.TotalTokens += usage.TotalTokens
+	total := usage.TotalTokens
+	if total == 0 {
+		total = usage.PromptTokens + usage.CompletionTokens
+	}
+	a.summary.TotalTokens += total
 	a.summary.CachedTokens += usage.PromptTokenDetails.CachedTokens
+	a.summary.CacheWriteTokens += usage.PromptTokenDetails.CacheWriteTokens
 	a.summary.ReasoningTokens += usage.CompletionTokensDetails.ReasoningTokens
 	return true
 }
@@ -88,6 +94,7 @@ func (a *einoRunUsageAccumulator) EmitOnce(
 		"completionTokens": s.CompletionTokens,
 		"totalTokens":      s.TotalTokens,
 		"cachedTokens":     s.CachedTokens,
+		"cacheWriteTokens": s.CacheWriteTokens,
 		"reasoningTokens":  s.ReasoningTokens,
 	}
 	if progress != nil {
@@ -104,6 +111,7 @@ func (a *einoRunUsageAccumulator) EmitOnce(
 			zap.Int("completionTokens", s.CompletionTokens),
 			zap.Int("totalTokens", s.TotalTokens),
 			zap.Int("cachedTokens", s.CachedTokens),
+			zap.Int("cacheWriteTokens", s.CacheWriteTokens),
 			zap.Int("reasoningTokens", s.ReasoningTokens),
 		)
 	}
@@ -129,6 +137,9 @@ func maxEinoTokenUsage(dst *schema.TokenUsage, src *schema.TokenUsage) *schema.T
 	if src.PromptTokenDetails.CachedTokens > dst.PromptTokenDetails.CachedTokens {
 		dst.PromptTokenDetails.CachedTokens = src.PromptTokenDetails.CachedTokens
 	}
+	if src.PromptTokenDetails.CacheWriteTokens > dst.PromptTokenDetails.CacheWriteTokens {
+		dst.PromptTokenDetails.CacheWriteTokens = src.PromptTokenDetails.CacheWriteTokens
+	}
 	if src.CompletionTokensDetails.ReasoningTokens > dst.CompletionTokensDetails.ReasoningTokens {
 		dst.CompletionTokensDetails.ReasoningTokens = src.CompletionTokensDetails.ReasoningTokens
 	}
@@ -149,5 +160,6 @@ func tokenUsageEmpty(u *schema.TokenUsage) bool {
 			u.CompletionTokens == 0 &&
 			u.TotalTokens == 0 &&
 			u.PromptTokenDetails.CachedTokens == 0 &&
+			u.PromptTokenDetails.CacheWriteTokens == 0 &&
 			u.CompletionTokensDetails.ReasoningTokens == 0)
 }

internal/multiagent/eino_run_usage_accumulator_test.go

@@ -79,3 +79,25 @@ func TestMaxEinoTokenUsageUsesLargestStreamChunkValues(t *testing.T) {
 		t.Fatalf("usage = %#v", got)
 	}
 }
+
+func TestEinoRunUsageCacheWriteDoesNotInflateTotal(t *testing.T) {
+	acc := newEinoRunUsageAccumulator()
+	var stream *schema.TokenUsage
+	for i := 0; i < 2; i++ {
+		stream = maxEinoTokenUsage(stream, &schema.TokenUsage{
+			PromptTokens: 100, CompletionTokens: 20,
+			PromptTokenDetails:      schema.PromptTokenDetails{CachedTokens: 40, CacheWriteTokens: 30},
+			CompletionTokensDetails: schema.CompletionTokensDetails{ReasoningTokens: 10},
+		})
+	}
+	acc.AddUsage(stream)
+	got := acc.Summary()
+	if got.ModelCalls != 1 || got.TotalTokens != 120 || got.CacheWriteTokens != 30 || got.CachedTokens != 40 || got.ReasoningTokens != 10 {
+		t.Fatalf("summary = %#v", got)
+	}
+	var emitted map[string]interface{}
+	acc.EmitOnce("conv", "deep", "final", "claude-test", func(_, _ string, data interface{}) { emitted = data.(map[string]interface{}) }, nil)
+	if emitted["cacheWriteTokens"] != 30 || emitted["totalTokens"] != 120 {
+		t.Fatalf("event = %#v", emitted)
+	}
+}

internal/openai/eino_http.go

@@ -21,6 +21,7 @@ func NewEinoHTTPClient(cfg *config.OpenAIConfig, base *http.Client) *http.Client
 	transport = &reasoningToolChoiceCompatRoundTripper{base: transport, cfg: cfg}
 	transport = &einoSSEErrorRoundTripper{base: transport}
 	transport = &einoSSESanitizingRoundTripper{base: transport}
+	transport = &einoUsageNormalizingRoundTripper{base: transport}
 	cloned.Transport = transport
 	return &cloned
 }

internal/openai/eino_sse_sanitizer.go

@@ -19,7 +19,7 @@ package openai
 //
 // 该层对调用方完全透明:
 //   - 仅当响应 Content-Type 是 text/event-stream 时介入；普通 JSON 响应原样透传
-//   - data: payload (含 [DONE] 与 {"error":...}) 一字节不改
+//   - data: payload 仅补齐 DeepSeek 缓存用量字段，其余内容原样透传
 //   - 上游真断流 (EOF / connection reset / context cancel) 原样透传
 
 import (
@@ -103,6 +103,11 @@ func (b *einoSSESanitizingBody) Read(p []byte) (int, error) {
 				if line[len(line)-1] != '\n' {
 					line = append(line, '\n')
 				}
+				payload := bytes.TrimSpace(bytes.TrimLeft(line, " \t")[5:])
+				normalized := normalizeEinoUsagePayload(payload)
+				if !bytes.Equal(payload, normalized) {
+					line = append(append([]byte("data: "), normalized...), '\n')
+				}
 				b.pending = line
 				if err != nil {
 					b.err = err

internal/openai/eino_usage_normalizer.go

@@ -0,0 +1,90 @@
+package openai
+
+import (
+	"bytes"
+	"encoding/json"
+	"io"
+	"net/http"
+	"strings"
+)
+
+// normalizeEinoUsagePayload bridges DeepSeek's cache counters to the OpenAI
+// details understood by Eino. Standard provider fields always take precedence.
+func normalizeEinoUsagePayload(payload []byte) []byte {
+	if !bytes.Contains(payload, []byte(`"prompt_cache_hit_tokens"`)) {
+		return payload
+	}
+	var root map[string]json.RawMessage
+	if json.Unmarshal(payload, &root) != nil {
+		return payload
+	}
+	var usage map[string]json.RawMessage
+	if json.Unmarshal(root["usage"], &usage) != nil {
+		return payload
+	}
+	hit, ok := usage["prompt_cache_hit_tokens"]
+	if !ok {
+		return payload
+	}
+	var count int64
+	if json.Unmarshal(hit, &count) != nil || count < 0 {
+		return payload
+	}
+	details := map[string]json.RawMessage{}
+	if raw := usage["prompt_tokens_details"]; len(raw) > 0 && string(raw) != "null" {
+		if json.Unmarshal(raw, &details) != nil {
+			return payload
+		}
+	}
+	if _, exists := details["cached_tokens"]; exists {
+		return payload
+	}
+	details["cached_tokens"] = hit
+	usage["prompt_tokens_details"], _ = json.Marshal(details)
+	root["usage"], _ = json.Marshal(usage)
+	out, err := json.Marshal(root)
+	if err != nil {
+		return payload
+	}
+	return out
+}
+
+// JSON responses are normalized lazily so RoundTrip retains its normal timing.
+// Streaming responses are normalized one data line at a time by the SSE reader.
+type einoUsageNormalizingRoundTripper struct{ base http.RoundTripper }
+
+func (rt *einoUsageNormalizingRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
+	resp, err := rt.base.RoundTrip(req)
+	if err == nil && resp != nil && resp.Body != nil && resp.StatusCode == http.StatusOK && strings.HasPrefix(strings.ToLower(resp.Header.Get("Content-Type")), "application/json") {
+		resp.Body = &einoUsageJSONBody{upstream: resp.Body}
+		resp.ContentLength = -1
+		resp.Header.Del("Content-Length")
+	}
+	return resp, err
+}
+
+type einoUsageJSONBody struct {
+	upstream io.ReadCloser
+	reader   *bytes.Reader
+	err      error
+}
+
+func (b *einoUsageJSONBody) Read(p []byte) (int, error) {
+	if len(p) == 0 {
+		return 0, nil
+	}
+	if b.reader == nil {
+		raw, err := io.ReadAll(b.upstream)
+		b.err = err
+		if err == nil {
+			raw = normalizeEinoUsagePayload(raw)
+		}
+		b.reader = bytes.NewReader(raw)
+	}
+	n, err := b.reader.Read(p)
+	if err == io.EOF && b.err != nil {
+		return n, b.err
+	}
+	return n, err
+}
+func (b *einoUsageJSONBody) Close() error { return b.upstream.Close() }

internal/openai/eino_usage_normalizer_test.go

@@ -0,0 +1,70 @@
+package openai
+
+import (
+	"encoding/json"
+	"io"
+	"net/http"
+	"strings"
+	"testing"
+)
+
+func TestNormalizeEinoDeepSeekCacheUsage(t *testing.T) {
+	for _, raw := range []string{
+		`{"id":9007199254740993,"usage":{"prompt_tokens":100,"prompt_cache_hit_tokens":80,"prompt_cache_miss_tokens":20}}`,
+		`{"usage":{"prompt_tokens_details":null,"prompt_cache_hit_tokens":80}}`,
+	} {
+		out := normalizeEinoUsagePayload([]byte(raw))
+		var root map[string]json.RawMessage
+		if err := json.Unmarshal(out, &root); err != nil {
+			t.Fatal(err)
+		}
+		var usage struct {
+			Details struct {
+				Cached int `json:"cached_tokens"`
+			} `json:"prompt_tokens_details"`
+		}
+		if err := json.Unmarshal(root["usage"], &usage); err != nil || usage.Details.Cached != 80 {
+			t.Fatalf("normalized = %s, err = %v", out, err)
+		}
+		if id := root["id"]; id != nil && string(id) != "9007199254740993" {
+			t.Fatalf("ID precision lost: %s", id)
+		}
+	}
+	for _, raw := range []string{
+		`[DONE]`, `{"error":{"message":"failed"}}`, `{"usage":{"prompt_cache_hit_tokens":-1}}`,
+		`{"usage":{"prompt_cache_hit_tokens":80,"prompt_tokens_details":{"cached_tokens":0}}}`,
+	} {
+		if out := string(normalizeEinoUsagePayload([]byte(raw))); out != raw {
+			t.Fatalf("unexpected change: %s", out)
+		}
+	}
+}
+
+func TestEinoDeepSeekCacheUsageJSONAndStream(t *testing.T) {
+	payload := `{"usage":{"prompt_tokens":100,"completion_tokens":5,"total_tokens":105,"prompt_cache_hit_tokens":80}}`
+	for _, stream := range []bool{false, true} {
+		contentType, body := "application/json", payload
+		if stream {
+			contentType, body = "text/event-stream", ": ping\ndata: "+payload+"\n\ndata: [DONE]\n"
+		}
+		client := NewEinoHTTPClient(nil, &http.Client{Transport: usageTestTransport(func(req *http.Request) (*http.Response, error) {
+			return &http.Response{StatusCode: 200, Header: http.Header{"Content-Type": []string{contentType}}, Body: io.NopCloser(strings.NewReader(body))}, nil
+		})})
+		resp, err := client.Get("http://example.invalid/v1/chat/completions")
+		if err != nil {
+			t.Fatal(err)
+		}
+		out, err := io.ReadAll(resp.Body)
+		resp.Body.Close()
+		if err != nil || !strings.Contains(string(out), `"cached_tokens":80`) {
+			t.Fatalf("stream=%v body=%s err=%v", stream, out, err)
+		}
+		if stream && !strings.Contains(string(out), "data: [DONE]") {
+			t.Fatal("lost stream terminator")
+		}
+	}
+}
+
+type usageTestTransport func(*http.Request) (*http.Response, error)
+
+func (f usageTestTransport) RoundTrip(req *http.Request) (*http.Response, error) { return f(req) }