From a2beb8230afe0ea01b0d981f0f6f41d43e6bdaed Mon Sep 17 00:00:00 2001 From: qian Date: Mon, 31 Aug 2026 22:19:35 +0800 Subject: [PATCH] =?UTF-8?q?fix(relay):=E4=BF=AE=E5=A4=8D=E8=BD=AC=E6=8D=A2?= =?UTF-8?q?=E5=B1=82=E5=B1=82=E6=95=B0=E6=8D=AE=E6=96=AD=E5=B1=82=E5=AF=BC?= =?UTF-8?q?=E8=87=B4=E8=BF=9E=E6=8E=A5=E4=B8=AD=E6=96=AD=E6=B2=A1=E6=9C=89?= =?UTF-8?q?=E6=AD=A3=E7=A1=AE=E8=AE=A1=E8=B4=B9=E7=9A=84=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/logic/admin/dashboard.go | 19 ++- internal/logic/admin/tenant.go | 23 ++-- relay/channel/claude/response.go | 6 +- relay/channel/coze/adaptor.go | 10 +- relay/channel/dify/adaptor.go | 10 +- relay/channel/gemini/response.go | 6 +- relay/channel/ollama/adaptor.go | 4 + relay/channel/openai/converter.go | 4 + relay/common/stream_status.go | 14 ++- relay/common/stream_status_test.go | 40 ++++++ relay/handler/relay_handler.go | 17 ++- relay/relaykit_bridge/stream.go | 23 +++- relay/relaykit_bridge/stream_test.go | 73 +++++++++++ .../src/components/ModelPricingModal.vue | 118 ++++++++++++------ 14 files changed, 295 insertions(+), 72 deletions(-) create mode 100644 relay/common/stream_status_test.go diff --git a/internal/logic/admin/dashboard.go b/internal/logic/admin/dashboard.go index bb2e7f9..b7d7742 100644 --- a/internal/logic/admin/dashboard.go +++ b/internal/logic/admin/dashboard.go @@ -798,7 +798,9 @@ func (s *sAdmin) OfflineRecharge(ctx context.Context, req *v1.AdminWalletOffline }, nil } -// GetWalletInfo 获取租户钱包信息（管理后台） +// GetWalletInfo 获取租户钱包信息（管理后台）。 +// balance/frozen_balance 以 Redis 权威值为准（Redis 是资金提交点，调整余额/线下入账后立即可见）， +// DB 侧余额是物化器周期覆盖的滞后副本，仅作钱包行存在性校验与 Redis 不可用时的降级展示。 func (s *sAdmin) GetWalletInfo(ctx context.Context, req *v1.AdminWalletInfoReq) (*v1.AdminWalletInfoRes, error) { type walletRow struct { ID int64 `json:"id"` @@ -815,11 +817,19 @@ func (s *sAdmin) GetWalletInfo(ctx context.Context, req *v1.AdminWalletInfoReq) return nil, common.NewNotFoundError("钱包") } - return &v1.AdminWalletInfoRes{ + res := &v1.AdminWalletInfoRes{ Balance: w.Balance, FrozenBalance: w.FrozenBalance, WarningThreshold: w.WarningThreshold, - }, nil + } + + // 覆盖为 Redis 实时余额（钱包 hash 缺失时 GetWallet 内部触发灾难恢复重建） + if wallet, err := billing.GetWallet(ctx, req.TenantID); err == nil { + res.Balance = wallet.Balance + res.FrozenBalance = wallet.FrozenBalance + } + + return res, nil } // GetWalletTransactions 获取租户钱包交易流水（管理后台） @@ -905,6 +915,9 @@ func (s *sAdmin) SetWarningThreshold(ctx context.Context, req *v1.AdminWalletSet return nil, err } + // 清除钱包静态字段缓存，让 GetWallet 读到新阈值（对齐租户端 UpdateWarningThreshold） + billing.InvalidateWalletStaticCache(ctx, req.TenantID) + // 阈值变更后重置预警标记，使新阈值能触发新的预警 billing.ResetLowBalanceNotified(ctx, req.TenantID) diff --git a/internal/logic/admin/tenant.go b/internal/logic/admin/tenant.go index d265f8b..38308f8 100644 --- a/internal/logic/admin/tenant.go +++ b/internal/logic/admin/tenant.go @@ -3,6 +3,7 @@ package admin import ( "context" "encoding/json" + "strconv" "strings" "github.com/gogf/gf/v2/database/gdb" @@ -424,16 +425,20 @@ func (s *sAdmin) GetTenant(ctx context.Context, req *v1.TenantGetReq) (*v1.Tenan memberCount, _ := dao.TntUsers.Ctx(ctx). Where("tenant_id", req.Id).Count() - // Get wallet balance - var wallet *struct { - Balance string `json:"balance"` - } - _ = dao.BilWallets.Ctx(ctx). - Where("tenant_id", req.Id).Scan(&wallet) - + // Get wallet balance：优先读 Redis 权威值（调整余额/入账后详情页立即可见），DB 物化值兜底 walletBalance := "0" - if wallet != nil && wallet.Balance != "" { - walletBalance = wallet.Balance + if w, err := billing.GetWallet(ctx, req.Id); err == nil { + walletBalance = strconv.FormatFloat(w.Balance, 'f', -1, 64) + } else { + var wallet *struct { + Balance string `json:"balance"` + } + _ = dao.BilWallets.Ctx(ctx). + Where("tenant_id", req.Id).Scan(&wallet) + + if wallet != nil && wallet.Balance != "" { + walletBalance = wallet.Balance + } } ownerName := "" if owner != nil { diff --git a/relay/channel/claude/response.go b/relay/channel/claude/response.go index 649d2cd..ebf4b24 100644 --- a/relay/channel/claude/response.go +++ b/relay/channel/claude/response.go @@ -93,8 +93,12 @@ func (a *Adaptor) handleStreamToOpenAI(ctx context.Context, resp *http.Response, return nil, constant.NewUpstreamError(resp.StatusCode, string(body), nil).WithRetryAfter(constant.RetryAfterFromHeader(resp.Header)) } - // relaykit 流式转换（特性开关控制，默认关闭）。未启用/无匹配回退旧路径。 + // relaykit 流式转换（常开）。同格式/无匹配转换器时回退旧路径。 if usage, ok := relaykit_bridge.TryConvertStreamViaRelaykit(ctx, info, resp.Body, writer); ok { + // 流中断（客户端断开/写失败）：桥接层已完成 usage 兜底，透传中断信号供上层按中断结算 + if info.StreamStatus != nil && info.StreamStatus.IsPartialStreamEnd() { + return usage, common.ErrStreamInterrupted + } return usage, nil } diff --git a/relay/channel/coze/adaptor.go b/relay/channel/coze/adaptor.go index cd0c447..d49b005 100644 --- a/relay/channel/coze/adaptor.go +++ b/relay/channel/coze/adaptor.go @@ -133,8 +133,12 @@ func (a *Adaptor) DoResponse(ctx context.Context, resp *http.Response, info *com // handleStreamResponse 流式模式：逐事件读取 Coze SSE，转换为 OpenAI SSE 格式输出 func (a *Adaptor) handleStreamResponse(ctx context.Context, resp *http.Response, info *common.RelayInfo, writer http.ResponseWriter) (*common.Usage, error) { - // relaykit 流式转换（特性开关控制，默认关闭）。未启用/无匹配回退旧路径。 + // relaykit 流式转换（常开）。同格式/无匹配转换器时回退旧路径。 if usage, ok := relaykit_bridge.TryConvertStreamViaRelaykit(ctx, info, resp.Body, writer); ok { + // 流中断（客户端断开/写失败）：桥接层已完成 usage 兜底，透传中断信号供上层按中断结算 + if info.StreamStatus != nil && info.StreamStatus.IsPartialStreamEnd() { + return usage, common.ErrStreamInterrupted + } return usage, nil } @@ -164,7 +168,7 @@ func (a *Adaptor) handleStreamResponse(ctx context.Context, resp *http.Response, // 流中断计费兜底：输出按已转发文本 2 字符/token 估算，输入用请求侧估算值补齐 interruptedUsage := &common.Usage{} helper.ApplyInterruptedUsageFallback(info, interruptedUsage, transferredTextLen) - return interruptedUsage, nil + return interruptedUsage, common.ErrStreamInterrupted default: } @@ -207,7 +211,7 @@ func (a *Adaptor) handleStreamResponse(ctx context.Context, resp *http.Response, // 流中断计费兜底：输出按已转发文本 2 字符/token 估算，输入用请求侧估算值补齐 interruptedUsage := &common.Usage{} helper.ApplyInterruptedUsageFallback(info, interruptedUsage, transferredTextLen) - return interruptedUsage, nil + return interruptedUsage, common.ErrStreamInterrupted } case "conversation.message.completed": diff --git a/relay/channel/dify/adaptor.go b/relay/channel/dify/adaptor.go index 105511a..c5501f4 100644 --- a/relay/channel/dify/adaptor.go +++ b/relay/channel/dify/adaptor.go @@ -199,8 +199,12 @@ func (a *Adaptor) handleNonStreamResponse(ctx context.Context, resp *http.Respon // handleStreamResponse 处理 Dify streaming 模式 SSE 响应 func (a *Adaptor) handleStreamResponse(ctx context.Context, resp *http.Response, info *common.RelayInfo, writer http.ResponseWriter) (*common.Usage, error) { - // relaykit 流式转换（特性开关控制，默认关闭）。未启用/无匹配回退旧路径。 + // relaykit 流式转换（常开）。同格式/无匹配转换器时回退旧路径。 if usage, ok := relaykit_bridge.TryConvertStreamViaRelaykit(ctx, info, resp.Body, writer); ok { + // 流中断（客户端断开/写失败）：桥接层已完成 usage 兜底，透传中断信号供上层按中断结算 + if info.StreamStatus != nil && info.StreamStatus.IsPartialStreamEnd() { + return usage, common.ErrStreamInterrupted + } return usage, nil } @@ -228,7 +232,7 @@ func (a *Adaptor) handleStreamResponse(ctx context.Context, resp *http.Response, streamStatus.SetEndReason(common.StreamEndReasonClientGone, common.ErrStreamInterrupted) // 流中断计费兜底：输出缺失按已转发文本 2 字符/token 估算，输入用请求侧估算值补齐 helper.ApplyInterruptedUsageFallback(info, &usage, transferredTextLen) - return &usage, nil + return &usage, common.ErrStreamInterrupted default: } @@ -267,7 +271,7 @@ func (a *Adaptor) handleStreamResponse(ctx context.Context, resp *http.Response, streamStatus.SetEndReason(common.StreamEndReasonClientGone, common.ErrStreamInterrupted) // 流中断计费兜底：输出缺失按已转发文本 2 字符/token 估算，输入用请求侧估算值补齐 helper.ApplyInterruptedUsageFallback(info, &usage, transferredTextLen) - return &usage, nil + return &usage, common.ErrStreamInterrupted } case "message_end": diff --git a/relay/channel/gemini/response.go b/relay/channel/gemini/response.go index 28ecad0..61d6ab4 100644 --- a/relay/channel/gemini/response.go +++ b/relay/channel/gemini/response.go @@ -240,8 +240,12 @@ func (a *Adaptor) handleStreamToOpenAI(ctx context.Context, resp *http.Response, return nil, buildGeminiUpstreamError(body, resp.StatusCode) } - // relaykit 流式转换（特性开关控制，默认关闭）。未启用/无匹配回退旧路径。 + // relaykit 流式转换（常开）。同格式/无匹配转换器时回退旧路径。 if usage, ok := relaykit_bridge.TryConvertStreamViaRelaykit(ctx, info, resp.Body, writer); ok { + // 流中断（客户端断开/写失败）：桥接层已完成 usage 兜底，透传中断信号供上层按中断结算 + if info.StreamStatus != nil && info.StreamStatus.IsPartialStreamEnd() { + return usage, common.ErrStreamInterrupted + } return usage, nil } diff --git a/relay/channel/ollama/adaptor.go b/relay/channel/ollama/adaptor.go index ab6bfc0..7e7919e 100644 --- a/relay/channel/ollama/adaptor.go +++ b/relay/channel/ollama/adaptor.go @@ -240,6 +240,10 @@ func (a *Adaptor) handleChatStreamResponse(ctx context.Context, resp *http.Respo // relaykit 流式转换（仅 chat；generate/embedding 不迁移，走旧路径） if usage, ok := relaykit_bridge.TryConvertStreamViaRelaykit(ctx, info, resp.Body, writer); ok { + // 流中断（客户端断开/写失败）：桥接层已完成 usage 兜底，透传中断信号供上层按中断结算 + if info.StreamStatus != nil && info.StreamStatus.IsPartialStreamEnd() { + return usage, common.ErrStreamInterrupted + } return usage, nil } diff --git a/relay/channel/openai/converter.go b/relay/channel/openai/converter.go index 705ae24..645959c 100644 --- a/relay/channel/openai/converter.go +++ b/relay/channel/openai/converter.go @@ -1573,6 +1573,10 @@ func HandleResponsesStreamToChat(ctx context.Context, resp *http.Response, info helper.WriteSSEData(writer, "[DONE]") g.Log().Infof(ctx, "[HandleResponsesStreamToChat] completed: usage=%+v, endReason=%s", totalUsage, info.StreamStatus.GetEndReason()) + // 流中断（客户端断开/写失败/上游 error 主动终止）：透传中断信号，上层按已传输部分结算 + if info.StreamStatus != nil && info.StreamStatus.IsPartialStreamEnd() { + return &totalUsage, common.ErrStreamInterrupted + } return &totalUsage, nil } diff --git a/relay/common/stream_status.go b/relay/common/stream_status.go index e6443d2..ada6279 100644 --- a/relay/common/stream_status.go +++ b/relay/common/stream_status.go @@ -13,7 +13,7 @@ const ( StreamEndReasonClientGone StreamEndReason = "client_gone" StreamEndReasonError StreamEndReason = "error" StreamEndReasonPanic StreamEndReason = "panic" - StreamEndReasonHandlerStop StreamEndReason = "handler_stop" // dataHandler 调用了 Stop() + StreamEndReasonHandlerStop StreamEndReason = "handler_stop" // dataHandler 调用了 Stop()（写客户端失败/上游 error 主动终止，按中断处理） StreamEndReasonPingFail StreamEndReason = "ping_fail" // ping 写入失败 StreamEndReasonScannerErr StreamEndReason = "scanner_err" // bufio.Scanner 错误 StreamEndReasonEOF StreamEndReason = "eof" // scanner EOF 未收到 [DONE] @@ -71,17 +71,23 @@ func (s *StreamStatus) ErrorCount() int { } // IsPartialStreamEnd 流已开始传输但未正常结束，应按已消费 usage 部分结算。 +// handler_stop（dataHandler 调用 Stop）纳入中断判定：正常结束由 sr.Done()/scanner 的 +// done 标记，实际走到 Stop 的只有「写客户端失败」（客户端不可达）与「上游 error 事件 +// 主动终止」两种，均属已向客户端写出部分字节后的非正常结束，应按中断部分结算。 +// 此前 handler_stop 不算中断，客户端断开时写失败若先于 ctx 取消被观察到（first-writer-wins +// 竞态），流会被误判为正常结束、跳过中断计费兜底，导致 0 token / 0 费用结算。 func (s *StreamStatus) IsPartialStreamEnd() bool { r := s.GetEndReason() return r == StreamEndReasonClientGone || r == StreamEndReasonScannerErr || - r == StreamEndReasonTimeout || r == StreamEndReasonPingFail + r == StreamEndReasonTimeout || r == StreamEndReasonPingFail || + r == StreamEndReasonHandlerStop } -// IsNormalEnd 是否正常结束 +// IsNormalEnd 是否正常结束（handler_stop 属异常终止，不算正常结束） func (s *StreamStatus) IsNormalEnd() bool { s.mu.RLock() defer s.mu.RUnlock() - return s.reason == StreamEndReasonDone || s.reason == StreamEndReasonHandlerStop || s.reason == "" + return s.reason == StreamEndReasonDone || s.reason == "" } // HasErrors 是否有错误 diff --git a/relay/common/stream_status_test.go b/relay/common/stream_status_test.go new file mode 100644 index 0000000..1dcc1be --- /dev/null +++ b/relay/common/stream_status_test.go @@ -0,0 +1,40 @@ +package common + +import ( + "errors" + "testing" +) + +// TestIsPartialStreamEnd_ReasonCoverage 流式结束原因的中断判定覆盖。 +// handler_stop（写客户端失败/上游 error 主动终止）必须算部分流中断： +// 客户端断开时写失败常先于 ctx 取消被观察到（SetEndReason 为 first-writer-wins）， +// 漏判会跳过中断计费兜底，按 0 token 成功结算。 +func TestIsPartialStreamEnd_ReasonCoverage(t *testing.T) { + partialReasons := []StreamEndReason{ + StreamEndReasonClientGone, + StreamEndReasonScannerErr, + StreamEndReasonTimeout, + StreamEndReasonPingFail, + StreamEndReasonHandlerStop, + } + for _, reason := range partialReasons { + s := NewStreamStatus() + s.SetEndReason(reason, errors.New("boom")) + if !s.IsPartialStreamEnd() { + t.Errorf("IsPartialStreamEnd(%s) = false, want true", reason) + } + if s.IsNormalEnd() { + t.Errorf("IsNormalEnd(%s) = true, want false", reason) + } + } + + // 正常结束不算中断 + s := NewStreamStatus() + s.SetEndReason(StreamEndReasonDone, nil) + if s.IsPartialStreamEnd() { + t.Error("IsPartialStreamEnd(done) = true, want false") + } + if !s.IsNormalEnd() { + t.Error("IsNormalEnd(done) = false, want true") + } +} diff --git a/relay/handler/relay_handler.go b/relay/handler/relay_handler.go index 9cd9362..cdccf07 100644 --- a/relay/handler/relay_handler.go +++ b/relay/handler/relay_handler.go @@ -718,6 +718,7 @@ func RelayHandler(ctx context.Context, body []byte, path string, headers http.He if streamUsage == nil { streamUsage = &common.Usage{} } + var interruptSettle *common.SettlementResult if billing != nil && preDeductAmount > 0 { settleResult, err := billing.SettleStreamInterrupted(settleCtx, rc.TenantID, rc.UserID, rc.ApiKeyID, selection.ChannelID, v.modelName, rc.RequestID, v.relayModeStr, streamUsage, preDeductAmount, rc.ProjectID, info.StartTime) @@ -734,8 +735,9 @@ func RelayHandler(ctx context.Context, body []byte, path string, headers http.He billing.IncrMemberQuotaUsed(settleCtx, rc.TenantID, rc.UserID, settleResult.ActualCost) billing.IncrApiKeyQuotaUsed(settleCtx, rc.ApiKeyID, settleResult.ActualCost) } + interruptSettle = settleResult } - recordFailedUsageWithTokens(provider, rc, selection, v.modelName, v.relayMode, v.isStream, err, streamUsage) + recordFailedUsageWithTokens(provider, rc, selection, v.modelName, v.relayMode, v.isStream, err, streamUsage, interruptSettle) finalizeTrace(trace, rc, hop, false, attempt, selection, err.Error(), info.LatencyMs()) dbgAttempt.MarkFinal(err) return usage, v.billingResult, err @@ -1056,7 +1058,7 @@ func recordFailedUsage(provider common.DataProvider, rc *RelayContext, selection // recordFailedUsageWithTokens 记录失败用量（含 token 明细，用于流中断等已有部分 usage 的场景）。 // 与 recordFailedUsage 的区别：此函数会填充 token 字段，避免报表中流中断记录的 token 全为 0。 -func recordFailedUsageWithTokens(provider common.DataProvider, rc *RelayContext, selection *common.ChannelSelection, modelName string, relayMode constant.RelayMode, isStream bool, err error, usage *common.Usage) { +func recordFailedUsageWithTokens(provider common.DataProvider, rc *RelayContext, selection *common.ChannelSelection, modelName string, relayMode constant.RelayMode, isStream bool, err error, usage *common.Usage, settleResult *common.SettlementResult) { record := &common.UsageRecord{ TenantID: rc.TenantID, UserID: rc.UserID, @@ -1082,6 +1084,17 @@ func recordFailedUsageWithTokens(provider common.DataProvider, rc *RelayContext, record.CompletionTokens = usage.CompletionTokens record.TotalTokens = usage.TotalTokens } + // 流中断结算成功时补充费用字段：中断已按已传输部分实际扣费，用量记录须与账本一致， + // 否则后台看到的中断请求永远是 0 费用（结算与用量记录是两条独立写入链路） + if settleResult != nil { + record.TotalCost = settleResult.BaseCost + record.ActualCost = settleResult.ActualCost + record.PreDeductAmount = settleResult.PreDeductAmount + record.RefundAmount = settleResult.RefundAmount + record.SupplementAmount = settleResult.SupplementAmount + record.BillingMode = settleResult.BillingMode + record.BillingSource = settleResult.BillingSource + } provider.RecordUsage(context.Background(), record) } diff --git a/relay/relaykit_bridge/stream.go b/relay/relaykit_bridge/stream.go index c5da8ff..78d8d7e 100644 --- a/relay/relaykit_bridge/stream.go +++ b/relay/relaykit_bridge/stream.go @@ -69,7 +69,8 @@ func convertStreamViaRelaykit(ctx context.Context, info *common.RelayInfo, upstr var ( gotFinish bool // 转换器是否已产出带 finish_reason 的结束 chunk firstChunk bool - transferredTextLen int // 已转发的文本/思考内容长度，供流中断输出估算 + writeFailed bool // 写客户端已失败（连接不可达），客户端断开的可靠信号 + transferredTextLen int // 已转发的文本/思考内容长度，供流中断输出估算 ) // chunkWriter：将转换器产出的 *dto.ChatCompletionStreamResponse 序列化为 SSE 写出， @@ -106,7 +107,11 @@ func convertStreamViaRelaykit(ctx context.Context, info *common.RelayInfo, upstr if err != nil { return err } - return helper.WriteSSEData(safeWriter, string(data)) + if err := helper.WriteSSEData(safeWriter, string(data)); err != nil { + writeFailed = true + return err + } + return nil } // writeTerminal 在转换器未产出结束 chunk 时补发一个终止 chunk，保证客户端正常收尾。 @@ -134,9 +139,17 @@ func convertStreamViaRelaykit(ctx context.Context, info *common.RelayInfo, upstr monitor.TrackConverterCall(converterID, string(upstream), string(clientFormat), duration, err) if err != nil { - if ctx.Err() != nil { - // 客户端断开 / 上下文取消：客户端已不可达，不写 [DONE] - setEndReason(common.StreamEndReasonClientGone, ctx.Err()) + // 客户端断开判定：ctx 已取消，或已出现过写客户端失败。后者兜住「写失败先于 + // ctx 取消被观察到」的竞态与断开信号经代理链路延迟传导的场景——此前仅凭 + // ctx.Err() 区分，这类情况会被误判为转换失败：跳过中断计费兜底、向死连接 + // 补写 [DONE]，最终按 0 token 成功结算。 + if writeFailed || ctx.Err() != nil { + endErr := ctx.Err() + if endErr == nil { + endErr = err + } + // 客户端已不可达，不写 terminal/[DONE] + setEndReason(common.StreamEndReasonClientGone, endErr) // 流中断计费兜底：输出缺失按已转发文本 2 字符/token 估算，输入用请求侧估算值补齐 helper.ApplyInterruptedUsageFallback(info, capturedUsage, transferredTextLen) return capturedUsage, true diff --git a/relay/relaykit_bridge/stream_test.go b/relay/relaykit_bridge/stream_test.go index 4367b39..c5f1ced 100644 --- a/relay/relaykit_bridge/stream_test.go +++ b/relay/relaykit_bridge/stream_test.go @@ -2,6 +2,8 @@ package relaykit_bridge import ( "context" + "errors" + "net/http" "net/http/httptest" "strings" "testing" @@ -192,3 +194,74 @@ func tail(s string, n int) string { } return s[len(s)-n:] } + +// failOnSubstringWriter 写入载荷包含 marker 时返回错误（模拟客户端已断开后的写失败）， +// 其余写入透传到内层 recorder。 +type failOnSubstringWriter struct { + rec *httptest.ResponseRecorder + marker string +} + +func (w *failOnSubstringWriter) Header() http.Header { return w.rec.Header() } +func (w *failOnSubstringWriter) WriteHeader(code int) { w.rec.WriteHeader(code) } +func (w *failOnSubstringWriter) Write(p []byte) (int, error) { + if strings.Contains(string(p), w.marker) { + return 0, errors.New("write failed: connection reset by peer") + } + return w.rec.Body.Write(p) +} + +// TestConvertStreamViaRelaykit_ClientWriteFailure 复现中断误判竞态：客户端断开由「写失败」 +// 首先暴露而 ctx 尚未取消（断开信号经代理传导延迟/first-writer-wins 场景）。 +// 此时必须按流中断处理：end reason=client_gone、usage 走中断兜底（已转发文本 2 字符/token +// 估算），且不向死连接补写 [DONE]——而非误判为转换失败导致 0 token / 0 费用结算。 +func TestConvertStreamViaRelaykit_ClientWriteFailure(t *testing.T) { + claudeStream := `data: {"type":"message_start","message":{"id":"msg_123","type":"message","role":"assistant","model":"claude-3-opus-20240229","usage":{"input_tokens":10,"output_tokens":0,"cache_read_input_tokens":4,"cache_creation_input_tokens":3}}} + +data: {"type":"content_block_start","index":0,"content_block":{"type":"text"}} + +data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello"}} + +data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":", how can I help you?"}} + +data: {"type":"content_block_stop","index":0} + +data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":7}} + +data: {"type":"message_stop"} + +` + info := newStreamTestRelayInfo(constant.ProviderClaude, constant.RelayFormatOpenAI) + // 与生产一致（relay_handler buildRelayInfo 设置）：流中断时上游 usage 缺失， + // 输入 token 按请求侧估算值补齐计费 + info.SetEstimatePromptTokens(42) + // ctx 用 Background（永不取消），精确模拟「写失败先于 ctx 取消被观察到」的竞态 + w := &failOnSubstringWriter{rec: httptest.NewRecorder(), marker: "how can I help you"} + + usage, ok := convertStreamViaRelaykit(context.Background(), info, strings.NewReader(claudeStream), w) + if !ok { + t.Fatal("expected ok=true (handled), got false") + } + if usage == nil { + t.Fatal("expected non-nil usage") + } + if reason := info.StreamStatus.GetEndReason(); reason != common.StreamEndReasonClientGone { + t.Fatalf("end reason = %q, want %q (client write failure misclassified)", reason, common.StreamEndReasonClientGone) + } + if !info.StreamStatus.IsPartialStreamEnd() { + t.Fatal("IsPartialStreamEnd = false, want true") + } + // 转换器的 usage chunk 只在流末尾产出，中断时尚未捕获——prompt 按请求侧估算值(42)补齐； + // 两个 delta 文本（"Hello"+", how can I help you?" = 5+21 = 26 字节）均在写失败前计入 + // transferredTextLen（累计发生在写之前），中断估算 completion = (26+1)/2 = 13 + if usage.PromptTokens != 42 { + t.Errorf("PromptTokens = %d, want 42 (request-side estimate fallback)", usage.PromptTokens) + } + if usage.CompletionTokens != 13 { + t.Errorf("CompletionTokens = %d, want 13 (interrupted estimate, got 0 => fallback skipped)", usage.CompletionTokens) + } + // 中断后不得向死连接补写 [DONE] + if strings.Contains(w.rec.Body.String(), "[DONE]") { + t.Error("output should not contain [DONE] after client disconnect") + } +} diff --git a/web/admin/src/components/ModelPricingModal.vue b/web/admin/src/components/ModelPricingModal.vue index fe25de8..bea081e 100644 --- a/web/admin/src/components/ModelPricingModal.vue +++ b/web/admin/src/components/ModelPricingModal.vue @@ -3,7 +3,7 @@ import { ref, reactive, watch, computed } from 'vue' import { Message } from '@arco-design/web-vue' import { IconCloudDownload, IconRefresh } from '@arco-design/web-vue/es/icon' import request from '@/utils/request' -import { displayCurrency, formatBilling } from '@/composables/useCurrency' +import { displayCurrency, formatBilling, cnyToUsd } from '@/composables/useCurrency' // 本位币符号：定价输入控件后缀跟随本位币，输入值仍为 bil 层存储原值不折算 const currencySymbol = computed(() => (displayCurrency.value === 'CNY' ? '¥' : '$')) @@ -80,6 +80,27 @@ function getOfficialSourceMeta(source: string) { return officialSourceMeta[source] || { label: source, color: 'gray' } } +// 官方数据源价格恒为美元；本位币=CNY 时按汇率倒数折算展示与填入（汇率恒为 CNY→USD 单向配置，USD→CNY 取倒数） +const usdToBaseRatio = computed(() => { + if (displayCurrency.value !== 'CNY') return 1 + return 1 / cnyToUsd.value +}) + +// USD 原值 → 本位币值（4 位小数，与定价输入控件精度对齐） +function toBasePrice(v: number | null | undefined): number { + const usd = Number(v ?? 0) + if (usd <= 0) return 0 + return Math.round(usd * usdToBaseRatio.value * 10000) / 10000 +} + +// 官方参考价展示：折算为本位币直显；本位币=CNY 时括注美元原值便于核对 +function formatOfficialPrice(v: number | null | undefined): string { + const converted = formatBilling(toBasePrice(v), 2) + if (displayCurrency.value !== 'CNY') return converted + const usd = Number(Number(v ?? 0).toFixed(2)) + return usd > 0 ? `${converted} ($${usd})` : converted +} + function resetOfficialData() { officialData.value = null officialError.value = '' @@ -111,10 +132,11 @@ async function fetchOfficialPricing() { function applyOfficialPricing(pricing: any) { if (!pricing) return if (editorItems.length > 0) { - editorItems[0].input_price = pricing.input_price - editorItems[0].output_price = pricing.output_price - editorItems[0].cache_read_price = pricing.cache_read_price - editorItems[0].cache_creation_price = pricing.cache_creation_price + // 编辑表单按本位币原值输入，官方 USD 价格须先折算为本位币 + editorItems[0].input_price = toBasePrice(pricing.input_price) + editorItems[0].output_price = toBasePrice(pricing.output_price) + editorItems[0].cache_read_price = toBasePrice(pricing.cache_read_price) + editorItems[0].cache_creation_price = toBasePrice(pricing.cache_creation_price) } if (pricing.billing_mode && pricing.billing_mode !== editorBillingMode.value) { editorBillingMode.value = pricing.billing_mode @@ -485,47 +507,52 @@ watch(() => props.visible, (val) => { -
-
-
- - {{ getOfficialSourceMeta(src.source).label }} - +
@@ -988,6 +1015,15 @@ watch(() => props.visible, (val) => { min-height: 180px; } +.official-reference-hint { + margin-bottom: 10px; + padding: 6px 10px; + font-size: 12px; + color: var(--ta-text-secondary); + background: var(--color-fill-2); + border-radius: 4px; +} + .official-source-grid { display: grid; grid-template-columns: repeat(2, minmax(0, 1fr));