diff --git a/api/admin/v1/workbench.go b/api/admin/v1/workbench.go
index 5755530..9b84109 100644
--- a/api/admin/v1/workbench.go
+++ b/api/admin/v1/workbench.go
@@ -97,13 +97,10 @@ type AdminWorkbenchSummaryRes struct {
// AdminWorkbenchBadgeReq 轻量计数接口,供左侧菜单红点轮询。
// 与 summary 共用同一份缓存,不会额外压库。
type AdminWorkbenchBadgeReq struct {
- g.Meta `path:"/workbench/badges" method:"get" mime:"json" tags:"管理后台-工作台" summary:"菜单红点计数"`
+ g.Meta `path:"/workbench/badges" method:"get" mime:"json" tags:"管理后台-工作台" summary:"工作台菜单角标计数"`
}
type AdminWorkbenchBadgeRes struct {
- Total int `json:"total" dc:"待办总数"`
- Urgent int `json:"urgent" dc:"P0 数量"`
- Domains map[string]int `json:"domains" dc:"各域待办数,键为域标识"`
- // Menus 前端菜单路由名 → 待办数,直接挂角标,前端不需要自己映射域到菜单。
- Menus map[string]int `json:"menus" dc:"菜单路由名 → 待办数"`
+ Total int `json:"total" dc:"待办总数,仅挂在工作台菜单角标"`
+ Urgent int `json:"urgent" dc:"P0 数量,控制角标转红"`
}
diff --git a/internal/dispatchadapter/maintenance.go b/internal/dispatchadapter/maintenance.go
index d4a2fae..2b0441b 100644
--- a/internal/dispatchadapter/maintenance.go
+++ b/internal/dispatchadapter/maintenance.go
@@ -148,13 +148,13 @@ func runMaintenance(ctx context.Context) {
// 覆盖摘要:目录只装载 active × 有启用能力行的渠道,未覆盖渠道的健康分不会落盘
// (探测却正常的渠道长期停在历史分数时,先看这里是否被跳过)
- g.Log().Infof(ctx, "[ChannelHealth] 健康快照维护轮次完成: 覆盖渠道 %d 个 %v", len(covered), covered)
+ g.Log().Debugf(ctx, "[ChannelHealth] 健康快照维护轮次完成: 覆盖渠道 %d 个 %v", len(covered), covered)
if len(skippedDegraded) > 0 {
g.Log().Warningf(ctx, "[ChannelHealth] 本轮 %d 个渠道因 Redis 健康读取失败跳过落盘(保留库中旧值): %v",
len(skippedDegraded), skippedDegraded)
}
if len(skippedNoData) > 0 {
- g.Log().Infof(ctx, "[ChannelHealth] 本轮 %d 个渠道全部模型无真实上报,保留库中旧值: %v",
+ g.Log().Debugf(ctx, "[ChannelHealth] 本轮 %d 个渠道全部模型无真实上报,保留库中旧值: %v",
len(skippedNoData), skippedNoData)
}
}
@@ -258,7 +258,7 @@ func snapshotHealthScore(ctx context.Context, channelID int64, succ, latMs, alph
// 分数有实际变化才记 Info;稳态渠道降为 Debug(避免每 5 分钟 × 渠道数的固定噪音)。
// 两种情况都带各模型 succ 明细:排查"健康分不符合预期"时可直接定位是哪个模型拉低了均值。
if math.Abs(healthScore-prevScore) >= 0.05 {
- g.Log().Infof(ctx, "[ChannelHealth] 渠道 %d 健康分落盘: %.1f → %.1f | avg_succ %.4f | 模型: %s",
+ g.Log().Debugf(ctx, "[ChannelHealth] 渠道 %d 健康分落盘: %.1f → %.1f | avg_succ %.4f | 模型: %s",
channelID, prevScore, healthScore, succ, modelDetail)
} else {
g.Log().Debugf(ctx, "[ChannelHealth] 渠道 %d 健康快照无变化: %.1f | avg_succ %.4f | 模型: %s",
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/internal/logic/admin/workbench.go b/internal/logic/admin/workbench.go
index 32ff739..6161299 100644
--- a/internal/logic/admin/workbench.go
+++ b/internal/logic/admin/workbench.go
@@ -39,11 +39,14 @@ const (
wbOrderUnfulfilledMin = 15 // 已支付超过 15 分钟仍未履约视为卡单
wbOAuthExpiryHours = 24 // OAuth 令牌 24 小时内过期即预警
wbHealthDropPoints = 30 // 健康分 1 小时内跌幅超过 30 分视为骤降
- wbCostSpikeMultiple = 8 // 单笔消费超过近期均值 8 倍视为异常
wbEmailFailThreshold = 5 // 1 小时内邮件失败超过 5 封视为链路异常
wbPlanExpiringDays = 7 // 套餐 7 天内到期
wbTicketSLAHours = 2 // 工单首次响应 SLA
wbWalletDriftRatio = 0.02 // Redis↔DB 余额偏差超过 2% 且绝对额可观时告警
+
+ // wbCronStallFallbackInterval 停摆判定的调度周期兜底值:任务不在注册表中
+ // (改名/下线后的遗留行)或 cron 表达式解析失败时按「日任务」保守处理。
+ wbCronStallFallbackInterval = 24 * time.Hour
)
// wbItem 内部待办结构:比 API 的 v1.WorkbenchItem 多一个 perm 字段。
@@ -63,14 +66,6 @@ type wbCollected struct {
GeneratedAt string `json:"generated_at"`
}
-// 域 → 前端菜单路由名,供菜单红点直接挂角标。
-var wbDomainMenus = map[string][]string{
- v1.WorkbenchDomainAvailability: {"AdminChannels"},
- v1.WorkbenchDomainMoney: {"AdminOrders", "AdminTransactions"},
- v1.WorkbenchDomainCustomer: {"AdminTickets", "AdminFeedback"},
- v1.WorkbenchDomainSystem: {"AdminAlertEvents", "AdminErrorLogs", "AdminCronJobs"},
-}
-
// ============================================================
// 对外接口
// ============================================================
@@ -126,30 +121,21 @@ func (s *sAdmin) GetWorkbenchSummary(ctx context.Context, _ *v1.AdminWorkbenchSu
return res, nil
}
-// GetWorkbenchBadges 菜单红点计数(与 summary 共用缓存,不额外压库)。
+// GetWorkbenchBadges 工作台菜单角标计数(与 summary 共用缓存,不额外压库)。
+//
+// 角标只挂工作台菜单一项:待办的排查线索只存在于工作台的描述文案里,
+// 业务菜单里没有对应的定位入口,往各业务菜单挂数字只会带来
+// 「进去了却找不到问题」的困惑(红点引路却无路可走)。
func (s *sAdmin) GetWorkbenchBadges(ctx context.Context, _ *v1.AdminWorkbenchBadgeReq) (*v1.AdminWorkbenchBadgeRes, error) {
col := s.collectWorkbenchCached(ctx)
items := s.filterWorkbenchItems(ctx, col.Items)
- res := &v1.AdminWorkbenchBadgeRes{
- Domains: map[string]int{},
- Menus: map[string]int{},
- }
+ res := &v1.AdminWorkbenchBadgeRes{}
for _, it := range items {
res.Total++
if it.Severity == v1.WorkbenchSeverityP0 {
res.Urgent++
}
- res.Domains[it.Domain]++
- }
- for domain, menus := range wbDomainMenus {
- n := res.Domains[domain]
- if n == 0 {
- continue
- }
- for _, m := range menus {
- res.Menus[m] += n
- }
}
return res, nil
}
diff --git a/internal/logic/admin/workbench_collectors.go b/internal/logic/admin/workbench_collectors.go
index 187489c..ca7e166 100644
--- a/internal/logic/admin/workbench_collectors.go
+++ b/internal/logic/admin/workbench_collectors.go
@@ -12,6 +12,7 @@ import (
v1 "github.com/qianfree/team-api/api/admin/v1"
"github.com/qianfree/team-api/internal/dispatchadapter"
"github.com/qianfree/team-api/internal/logic/billing"
+ "github.com/qianfree/team-api/internal/logic/common"
)
// ============================================================
@@ -235,7 +236,6 @@ func (s *sAdmin) wbCollectMoney(ctx context.Context) []wbItem {
items = append(items, s.wbWalletDrift(ctx)...)
items = append(items, s.wbFrozenStale(ctx)...)
items = append(items, s.wbOrderUnfulfilled(ctx)...)
- items = append(items, s.wbCostSpike(ctx)...)
items = append(items, s.wbLowBalanceTenants(ctx)...)
return items
}
@@ -383,58 +383,6 @@ func (s *sAdmin) wbOrderUnfulfilled(ctx context.Context) []wbItem {
return items
}
-// wbCostSpike 单笔消费异常。
-// 用途有三:防刷、防客户被恶意调用、防定价配置写错一位(乘数多个 0 的杀伤力最大)。
-// 判据是「相对该租户自身近 7 天均值的倍数」,而非全局绝对值 —— 大客户的正常单笔
-// 对小客户就是天价,全局阈值必然两头不讨好。
-func (s *sAdmin) wbCostSpike(ctx context.Context) []wbItem {
- rows, err := g.DB().Ctx(ctx).Query(ctx,
- `WITH recent AS (
- SELECT tenant_id, id, total_cost, model_name, created_at
- FROM bil_usage_logs
- WHERE created_at >= NOW() - INTERVAL '2 hours' AND total_cost > 0
- ), baseline AS (
- SELECT tenant_id, AVG(total_cost) AS avg_cost
- FROM bil_usage_logs
- WHERE created_at >= NOW() - INTERVAL '7 days' AND total_cost > 0
- GROUP BY tenant_id
- HAVING COUNT(*) >= 50
- )
- SELECT r.id, r.tenant_id, r.total_cost, r.model_name, r.created_at,
- b.avg_cost, t.name AS tenant_name
- FROM recent r
- JOIN baseline b ON b.tenant_id = r.tenant_id
- LEFT JOIN tnt_tenants t ON t.id = r.tenant_id
- WHERE r.total_cost >= b.avg_cost * ?
- ORDER BY r.total_cost DESC
- LIMIT 5`, wbCostSpikeMultiple)
- if err != nil {
- g.Log().Warningf(ctx, "workbench: 查询消费尖刺失败: %v", err)
- return nil
- }
-
- items := make([]wbItem, 0, len(rows))
- for _, r := range rows {
- cost := r["total_cost"].Float64()
- avg := r["avg_cost"].Float64()
- mult := 0.0
- if avg > 0 {
- mult = cost / avg
- }
- items = append(items, wbNew(
- fmt.Sprintf("usage_cost_spike:%d", r["id"].Int64()),
- v1.WorkbenchSeverityP1, v1.WorkbenchDomainMoney, "billing:view",
- fmt.Sprintf("租户「%s」单笔消费异常", r["tenant_name"].String()),
- fmt.Sprintf("模型 %s 单笔 %.6f,为该租户近 7 天均值的 %.1f 倍。"+
- "需确认是正常业务、刷量,还是定价配置写错",
- r["model_name"].String(), cost, mult),
- "查看用量日志", "AdminUsageLogs",
- map[string]string{"tenant_id": r["tenant_id"].String()}, r["created_at"].GTime(),
- ))
- }
- return items
-}
-
// wbLowBalanceTenants 余额低于预警阈值的租户。
// 一个信号两个用途:流失预警 + 催充窗口。聚合成一条,不逐租户刷屏。
func (s *sAdmin) wbLowBalanceTenants(ctx context.Context) []wbItem {
@@ -606,38 +554,53 @@ func (s *sAdmin) wbCronStalled(ctx context.Context) []wbItem {
rows, err := g.DB().Ctx(ctx).Query(ctx,
`SELECT job_name, last_status, last_started_at, last_error_message
FROM sys_cron_jobs
- WHERE last_status = 'failed'
- OR last_started_at IS NULL
- OR last_started_at <= NOW() - INTERVAL '1 hour'
- ORDER BY last_started_at ASC NULLS FIRST
- LIMIT 10`)
+ ORDER BY last_started_at ASC NULLS FIRST`)
if err != nil {
g.Log().Warningf(ctx, "workbench: 查询定时任务状态失败: %v", err)
return nil
}
+ // 注册表快照:任务名 → 调度表达式。停摆判定必须按各任务自身的调度周期换算,
+ // 固定 1 小时停摆线会把日任务(日对账、日清理等)一天里 23 个小时都误报成停摆。
+ schedules := make(map[string]string, 16)
+ for _, j := range common.GetCronScheduler().ListJobs() {
+ schedules[j.Name] = j.Schedule
+ }
+
items := make([]wbItem, 0, len(rows))
for _, r := range rows {
name := r["job_name"].String()
startedAt := r["last_started_at"].GTime()
failed := r["last_status"].String() == "failed"
+ // 停摆 = 距上次执行超过「正常间隔 × 2」,下限 1 小时——
+ // 下限避免分钟级任务被一次调度抖动(锁等待、实例重启)误报
+ interval := wbCronExpectedInterval(schedules, name)
+ stallAfter := max(2*interval, time.Hour)
+ stalled := startedAt != nil && time.Since(startedAt.Time) > stallAfter
+
+ if !failed && !stalled && startedAt != nil {
+ continue
+ }
+
// 资金相关任务提级为 P0:这些停摆会累积资损,不能等到明天
sev := v1.WorkbenchSeverityP1
if isMoneyCriticalJob(name) {
sev = v1.WorkbenchSeverityP0
}
- desc := ""
- if failed {
+ var desc string
+ switch {
+ case failed:
desc = "最近一次执行失败"
if e := strings.TrimSpace(r["last_error_message"].String()); e != "" {
desc += ":" + truncate(e, 150)
}
- } else if startedAt == nil {
+ case startedAt == nil:
desc = "自服务启动以来从未执行过,调度器可能未正常注册该任务"
- } else {
- desc = fmt.Sprintf("已 %s 未执行,任务可能已停摆", humanizeSince(startedAt))
+ default:
+ desc = fmt.Sprintf("正常约 %s 执行一次,已 %s 未执行,任务可能已停摆",
+ humanizeDuration(interval), humanizeSince(startedAt))
}
if isMoneyCriticalJob(name) {
desc += "。该任务涉及资金结算,停摆会持续累积差错"
@@ -654,6 +617,19 @@ func (s *sAdmin) wbCronStalled(ctx context.Context) []wbItem {
return items
}
+// wbCronExpectedInterval 从任务注册表查正常执行间隔。
+// 任务不在注册表中(改名/下线后的遗留行)或表达式解析失败时退回保守兜底值:
+// 宁可晚报一条真正死掉的任务,也不要每天误报一串健康任务。
+func wbCronExpectedInterval(schedules map[string]string, name string) time.Duration {
+ interval := wbCronStallFallbackInterval
+ if schedule, ok := schedules[name]; ok {
+ if d, err := common.ScheduleInterval(schedule); err == nil {
+ interval = d
+ }
+ }
+ return interval
+}
+
// isMoneyCriticalJob 判断任务停摆是否会造成资损。
func isMoneyCriticalJob(name string) bool {
n := strings.ToLower(name)
@@ -922,7 +898,11 @@ func humanizeSince(t *gtime.Time) string {
if t == nil {
return "未知时长"
}
- d := time.Since(t.Time)
+ return humanizeDuration(time.Since(t.Time))
+}
+
+// humanizeDuration 把时长写成人话(分钟/小时/天)。
+func humanizeDuration(d time.Duration) string {
if d < time.Minute {
return "不到 1 分钟"
}
diff --git a/internal/logic/common/cron.go b/internal/logic/common/cron.go
index 8ca7098..8f79250 100644
--- a/internal/logic/common/cron.go
+++ b/internal/logic/common/cron.go
@@ -370,6 +370,21 @@ func (cs *CronScheduler) Reschedule(ctx context.Context, name, schedule string)
return nil
}
+// ScheduleInterval 解析 cron 表达式,估算两次触发之间的正常间隔。
+// 取「接下来两次触发时刻之差」,对分钟/小时/日级表达式及 @every 描述符均适用
+// (@every 的触发时刻相对计算起点平移,间隔不受影响)。解析失败(非法表达式)返回错误,
+// 由调用方决定兜底策略——工作台的停摆判定靠它把「固定 1 小时」换成「按任务自身周期」。
+func ScheduleInterval(schedule string) (time.Duration, error) {
+ sched, err := cron.ParseStandard(schedule)
+ if err != nil {
+ return 0, gerror.Wrapf(err, "invalid cron schedule %q", schedule)
+ }
+ now := time.Now()
+ next1 := sched.Next(now)
+ next2 := sched.Next(next1)
+ return next2.Sub(next1), nil
+}
+
// cronLockTTL 分布式锁过期时间:仅用于实例崩溃时的兜底自动释放(正常完成会主动释放)。
// 取值需大于绝大多数 job 的执行时长;过长则崩溃后会阻塞该 job 至多这么久。
const cronLockTTL = 10 * time.Minute
diff --git a/internal/logic/common/cron_test.go b/internal/logic/common/cron_test.go
index 369d212..988788e 100644
--- a/internal/logic/common/cron_test.go
+++ b/internal/logic/common/cron_test.go
@@ -3,6 +3,7 @@ package common
import (
"context"
"testing"
+ "time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -93,6 +94,32 @@ func TestReschedule_ValidExprReplacesEntry(t *testing.T) {
assert.NotEqual(t, oldID, cs.entryIDs["job_c"], "条目 ID 应已更新")
}
+// TestScheduleInterval 停摆判定靠它把「固定 1 小时」换成「按任务自身周期」,
+// 各类表达式的间隔必须算准——尤其日任务(24h),算错就会退回误报。
+func TestScheduleInterval(t *testing.T) {
+ cases := []struct {
+ schedule string
+ want time.Duration
+ }{
+ {"* * * * *", time.Minute}, // 每分钟
+ {"*/5 * * * *", 5 * time.Minute}, // 每 5 分钟
+ {"*/10 * * * *", 10 * time.Minute}, // 每 10 分钟
+ {"0 */6 * * *", 6 * time.Hour}, // 每 6 小时(update_check)
+ {"0 3 * * *", 24 * time.Hour}, // 每日 3 点(各类日清理)
+ {"20 5 * * *", 24 * time.Hour}, // 每日 5:20(计费日对账)
+ {"@every 5m", 5 * time.Minute}, // @every 描述符(渠道自动探测默认)
+ {"@every 90m", 90 * time.Minute}, // @every 大间隔(*/N 在 N>=60 时会失真,故用 @every)
+ }
+ for _, c := range cases {
+ got, err := ScheduleInterval(c.schedule)
+ require.NoError(t, err, "schedule %q 应可解析", c.schedule)
+ assert.Equal(t, c.want, got, "schedule %q 的正常间隔不符", c.schedule)
+ }
+
+ _, err := ScheduleInterval("不是合法表达式")
+ assert.Error(t, err, "非法表达式应返回错误,由调用方兜底")
+}
+
// TestOnSettingsChanged_CallbackPanicDoesNotAffectOthers 回调在订阅 goroutine 内串行执行,
// 一个回调 panic 不能带崩订阅循环或阻断后续回调。
func TestOnSettingsChanged_CallbackPanicDoesNotAffectOthers(t *testing.T) {
diff --git a/internal/logic/task/async_polling.go b/internal/logic/task/async_polling.go
index 7fe06ab..c2a0e53 100644
--- a/internal/logic/task/async_polling.go
+++ b/internal/logic/task/async_polling.go
@@ -15,6 +15,7 @@ import (
"github.com/qianfree/team-api/internal/logic/relay"
"github.com/qianfree/team-api/relay/common"
"github.com/qianfree/team-api/relay/constant"
+ "github.com/qianfree/team-api/relay/helper"
"github.com/qianfree/team-api/relay/taskchannel"
)
@@ -492,21 +493,22 @@ func recordTaskUsage(task *common.AsyncTask, channel *common.ChannelBasicInfo, s
}
record := &common.UsageRecord{
- TenantID: task.TenantID,
- UserID: task.UserID,
- ApiKeyID: task.ApiKeyID,
- ChannelID: task.ChannelID,
- ChannelName: channelName,
- ChannelType: channelType,
- ModelName: task.ModelName,
- RelayMode: int(constant.RelayModeVideoGenerations),
- RequestType: 3, // async
- LatencyMs: float64(latencyMs),
- IsStream: false,
- Success: success,
- RequestID: task.RequestID,
- Status: status,
- ErrorMessage: errMsg,
+ TenantID: task.TenantID,
+ UserID: task.UserID,
+ ApiKeyID: task.ApiKeyID,
+ ChannelID: task.ChannelID,
+ ChannelName: channelName,
+ ChannelType: channelType,
+ ModelName: task.ModelName,
+ RelayMode: int(constant.RelayModeVideoGenerations),
+ RequestType: 3, // async
+ LatencyMs: float64(latencyMs),
+ IsStream: false,
+ Success: success,
+ RequestID: task.RequestID,
+ Status: status,
+ // errMsg 可能来自上游 FailReason 原文,租户端用量日志可见,抹除其中的 URL/IP
+ ErrorMessage: helper.RedactMessage(errMsg),
PromptTokens: task.PromptTokens,
CompletionTokens: task.CompletionTokens,
TotalTokens: task.TotalTokens,
diff --git a/migrations/000019_v0_2_16_version.sql b/migrations/000019_v0_2_16_version.sql
new file mode 100644
index 0000000..d3f1e71
--- /dev/null
+++ b/migrations/000019_v0_2_16_version.sql
@@ -0,0 +1,38 @@
+-- +goose Up
+-- 回填 bil_usage_logs.input_tokens 的「含缓存总输入」口径(恢复被丢失的 974633a8 配套回填)。
+--
+-- 背景:974633a8 将写侧 input_tokens 统一为「含缓存总输入」(Claude 口径入库时补加
+-- cache_read + cache_creation),缓存命中率公式随之改为 cache_read / input_tokens。
+-- 当时配套的历史数据回填 UPDATE 在 3676cd4d 中被误删、d01dbc79 合并迁移记录时彻底丢失,
+-- 导致存量 Claude 渠道行(channel_type=2)仍是「不含缓存」旧口径:
+-- 这些行 cache_read_tokens 可大于 input_tokens,个人工作台按月聚合后缓存命中率超过 100%。
+--
+-- 本迁移将 8/17 之前的存量 Claude 渠道行补加缓存 token,使全表口径一致:
+-- input_tokens = 普通输入 + cache_read + cache_creation
+-- OpenAI / Gemini 渠道行的 prompt 原生含缓存(cached 为其子集),不受影响;
+-- 经协议转换已按含缓存口径入库的行(如 openai 适配器解析 Claude 风格 usage,channel_type≠2)同样不触碰。
+--
+-- 注意:本迁移假设 channel_type=2 的历史行均为「不含缓存」口径(与 974633a8 原迁移一致)。
+-- 若曾在 974633a8 合入后、3676cd4d 之前(2026-08-17 10:01 ~ 11:25)对同一库执行过 goose up,
+-- 原回填已生效,禁止重复执行本迁移(会造成缓存 token 双重累加)。
+--
+-- 回填后需重跑用量日聚合刷新 bil_usage_daily(ON CONFLICT 幂等,可重复执行):
+-- team-api aggregate-usage --from 2026-08-01 --to <今天>
+
+-- +goose StatementBegin
+UPDATE bil_usage_logs
+SET input_tokens = input_tokens + COALESCE(cache_read_tokens, 0) + COALESCE(cache_creation_tokens, 0)
+WHERE channel_type = 2
+ AND (COALESCE(cache_read_tokens, 0) + COALESCE(cache_creation_tokens, 0)) > 0;
+-- +goose StatementEnd
+
+-- +goose Down
+-- 逆转回填:Claude 渠道行恢复「不含缓存」口径(OpenAI/Gemini 行本就含缓存,不受影响)。
+-- Down 后需再次执行 aggregate-usage 刷新 bil_usage_daily。
+
+-- +goose StatementBegin
+UPDATE bil_usage_logs
+SET input_tokens = input_tokens - COALESCE(cache_read_tokens, 0) - COALESCE(cache_creation_tokens, 0)
+WHERE channel_type = 2
+ AND (COALESCE(cache_read_tokens, 0) + COALESCE(cache_creation_tokens, 0)) > 0;
+-- +goose StatementEnd
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..e48167a 100644
--- a/relay/handler/relay_handler.go
+++ b/relay/handler/relay_handler.go
@@ -374,6 +374,12 @@ func settleSuccessfulRequest(
RetryIndex: info.RetryIndex,
}
+ // 流式追踪:正常结束也落库(done / eof)。eof(上游未发 [DONE] 即结束)按正常结束
+ // 结算但属上游侧异常信号,记录后可在用量日志直接区分,无需翻应用日志。
+ if info.StreamStatus != nil {
+ usageRecord.StreamEndReason = string(info.StreamStatus.GetEndReason())
+ }
+
// 填充结算费用数据
if settleResult != nil {
usageRecord.TotalCost = settleResult.BaseCost
@@ -692,7 +698,7 @@ func RelayHandler(ctx context.Context, body []byte, path string, headers http.He
if billing != nil && preDeductAmount > 0 {
_ = billing.SettleFailed(settleCtx, rc.TenantID, rc.RequestID, preDeductAmount)
}
- recordFailedUsage(provider, rc, selection, v.modelName, v.relayMode, v.isStream, err)
+ recordFailedUsage(provider, rc, selection, v.modelName, v.relayMode, v.isStream, err, info.StreamStatus)
recordChannelError(rc, selection, v.modelName, attempt, true, err, info.LatencyMs())
finalizeTrace(trace, rc, hop, false, attempt, selection, err.Error(), info.LatencyMs())
dbgAttempt.MarkFinal(err)
@@ -710,14 +716,16 @@ func RelayHandler(ctx context.Context, body []byte, path string, headers http.He
err = helper.RemapStatusCode(err, info.ChannelMeta.Settings.StatusCodeMapping)
if info.StreamStatus != nil && info.StreamStatus.IsPartialStreamEnd() {
- g.Log().Warningf(ctx, "[RelayHandler] Stream interrupted: adaptor=%s, model=%s, reason=%s",
- adaptor.GetChannelName(), v.modelName, info.StreamStatus.Summary())
+ g.Log().Warningf(ctx, "[RelayHandler] Stream interrupted: request=%s tenant=%d channel=%d(%s) adaptor=%s model=%s attempt=%d reason=%s ctx_err=%v usage=%v pre_deduct=%.6f",
+ rc.RequestID, rc.TenantID, selection.ChannelID, selection.ChannelName, adaptor.GetChannelName(), v.modelName,
+ attempt, info.StreamStatus.Summary(), ctx.Err(), usage, preDeductAmount)
// 客户端已收到部分流:不可重试(响应已污染),上报健康后按流中断结算
_, _ = sess.Report(settleCtx, dispatchStatusCode(err), err, dispatch.DeliveryResponseStarted, info.LatencyMs(), 0)
streamUsage := usage
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 +742,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, info.StreamStatus)
finalizeTrace(trace, rc, hop, false, attempt, selection, err.Error(), info.LatencyMs())
dbgAttempt.MarkFinal(err)
return usage, v.billingResult, err
@@ -783,7 +792,7 @@ func RelayHandler(ctx context.Context, body []byte, path string, headers http.He
if billing != nil && preDeductAmount > 0 {
_ = billing.SettleFailed(settleCtx, rc.TenantID, rc.RequestID, preDeductAmount)
}
- recordFailedUsage(provider, rc, selection, v.modelName, v.relayMode, v.isStream, err)
+ recordFailedUsage(provider, rc, selection, v.modelName, v.relayMode, v.isStream, err, info.StreamStatus)
recordChannelError(rc, selection, v.modelName, attempt, true, err, info.LatencyMs())
finalizeTrace(trace, rc, hop, false, attempt, selection, err.Error(), info.LatencyMs())
dbgAttempt.MarkFinal(err)
@@ -952,7 +961,7 @@ func handleChannelUnavailable(
"当前模型暂时不可用",
constant.ErrAllChannelsFailed,
)
- recordFailedUsage(provider, rc, nil, v.modelName, v.relayMode, v.isStream, allFailedErr)
+ recordFailedUsage(provider, rc, nil, v.modelName, v.relayMode, v.isStream, allFailedErr, nil)
return &channelUnavailableResult{nil, v.billingResult, allFailedErr}
}
@@ -1027,10 +1036,34 @@ func (e *RelayErrorWithRateLimit) Error() string {
return e.Message
}
+// applyStreamEndDiag 把流式结束原因落进失败用量记录,替代无归因价值的哨兵文案。
+// stream_end_reason 区分 client_gone / handler_stop / scanner_err / timeout / ping_fail / eof,
+// error_message 保留真实底层错误(ctx 取消原因、写客户端失败的 TCP 错误等);
+// 哨兵错误(ErrStreamInterrupted)或状态里已带底层错误时直接用摘要,其余追加在原始错误后。
+// error_message 对租户端可见,全部经 RedactMessage/SafeUpstreamErrorMessage 脱敏:
+// 摘要中的本地连接地址抹 IP,外层错误按传输层归一化,禁止上游域名/URL/IP 落入用量日志。
+func applyStreamEndDiag(record *common.UsageRecord, streamStatus *common.StreamStatus, err error) {
+ if streamStatus == nil {
+ return
+ }
+ reason := streamStatus.GetEndReason()
+ if reason == "" {
+ return
+ }
+ record.StreamEndReason = string(reason)
+ summary := helper.RedactMessage(streamStatus.Summary())
+ if err == nil || errors.Is(err, common.ErrStreamInterrupted) || streamStatus.Error() != nil {
+ record.ErrorMessage = summary
+ return
+ }
+ record.ErrorMessage = helper.SafeUpstreamErrorMessage(err) + " [stream: " + summary + "]"
+}
+
// recordFailedUsage 记录失败用量。
// selection 非 nil 时记录具体失败渠道(ID/名称/类型/上游模型),便于在用量日志定位失败渠道;
// nil 表示无单一渠道(全部渠道失败),渠道字段留空,失败详情见 error_message。
-func recordFailedUsage(provider common.DataProvider, rc *RelayContext, selection *common.ChannelSelection, modelName string, relayMode constant.RelayMode, isStream bool, err error) {
+// streamStatus 非 nil 且已有结束原因时,同步落 stream_end_reason 并修正 error_message。
+func recordFailedUsage(provider common.DataProvider, rc *RelayContext, selection *common.ChannelSelection, modelName string, relayMode constant.RelayMode, isStream bool, err error, streamStatus *common.StreamStatus) {
record := &common.UsageRecord{
TenantID: rc.TenantID,
UserID: rc.UserID,
@@ -1043,8 +1076,11 @@ func recordFailedUsage(provider common.DataProvider, rc *RelayContext, selection
Success: false,
RequestID: rc.RequestID,
Status: "error",
- ErrorMessage: err.Error(),
+ // 租户端用量日志可见 error_message,须经脱敏(上游域名/URL/IP 禁止落库);
+ // 完整原始错误保留在应用日志与审计 ForwardingTrace(仅管理员可见)
+ ErrorMessage: helper.SafeUpstreamErrorMessage(err),
}
+ applyStreamEndDiag(record, streamStatus, err)
if selection != nil {
record.ChannelID = selection.ChannelID
record.ChannelName = selection.ChannelName
@@ -1056,7 +1092,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, streamStatus *common.StreamStatus) {
record := &common.UsageRecord{
TenantID: rc.TenantID,
UserID: rc.UserID,
@@ -1069,8 +1105,11 @@ func recordFailedUsageWithTokens(provider common.DataProvider, rc *RelayContext,
Success: false,
RequestID: rc.RequestID,
Status: "error",
- ErrorMessage: err.Error(),
+ // 租户端用量日志可见 error_message,须经脱敏(上游域名/URL/IP 禁止落库);
+ // 完整原始错误保留在应用日志与审计 ForwardingTrace(仅管理员可见)
+ ErrorMessage: helper.SafeUpstreamErrorMessage(err),
}
+ applyStreamEndDiag(record, streamStatus, err)
if selection != nil {
record.ChannelID = selection.ChannelID
record.ChannelName = selection.ChannelName
@@ -1082,6 +1121,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/handler/stream_diag_test.go b/relay/handler/stream_diag_test.go
new file mode 100644
index 0000000..51d20f5
--- /dev/null
+++ b/relay/handler/stream_diag_test.go
@@ -0,0 +1,131 @@
+package handler
+
+import (
+ "context"
+ "errors"
+ "net"
+ "net/url"
+ "strings"
+ "testing"
+
+ "github.com/qianfree/team-api/relay/common"
+ "github.com/qianfree/team-api/relay/constant"
+)
+
+// TestApplyStreamEndDiag_SentinelReplacedBySummary 哨兵错误(ErrStreamInterrupted)本身无归因价值,
+// error_message 必须换成 StreamStatus 摘要,让用量日志能区分 context canceled / TCP 写失败等真实原因。
+func TestApplyStreamEndDiag_SentinelReplacedBySummary(t *testing.T) {
+ status := common.NewStreamStatus()
+ status.SetEndReason(common.StreamEndReasonClientGone, context.Canceled)
+
+ record := &common.UsageRecord{ErrorMessage: common.ErrStreamInterrupted.Error()}
+ applyStreamEndDiag(record, status, common.ErrStreamInterrupted)
+
+ if record.StreamEndReason != string(common.StreamEndReasonClientGone) {
+ t.Errorf("StreamEndReason = %q, want %q", record.StreamEndReason, common.StreamEndReasonClientGone)
+ }
+ want := "client_gone: context canceled"
+ if record.ErrorMessage != want {
+ t.Errorf("ErrorMessage = %q, want %q", record.ErrorMessage, want)
+ }
+}
+
+// TestApplyStreamEndDiag_StatusCarriesErrorSummaryWins 状态本身带底层错误(如 handler_stop 的
+// 写客户端失败)时摘要信息量高于外层错误,直接以摘要作为 error_message。
+func TestApplyStreamEndDiag_StatusCarriesErrorSummaryWins(t *testing.T) {
+ status := common.NewStreamStatus()
+ writeErr := errors.New("write tcp 127.0.0.1:18888->127.0.0.1:6396: wsasend: connection forcibly closed")
+ status.SetEndReason(common.StreamEndReasonHandlerStop, writeErr)
+
+ record := &common.UsageRecord{ErrorMessage: "upstream error"}
+ applyStreamEndDiag(record, status, errors.New("upstream error"))
+
+ if !strings.Contains(record.ErrorMessage, "write tcp") {
+ t.Errorf("ErrorMessage should keep underlying write error, got %q", record.ErrorMessage)
+ }
+ if record.StreamEndReason != string(common.StreamEndReasonHandlerStop) {
+ t.Errorf("StreamEndReason = %q, want %q", record.StreamEndReason, common.StreamEndReasonHandlerStop)
+ }
+}
+
+// TestApplyStreamEndDiag_RealErrorKeptWithSummary 非哨兵外层错误(如上游 5xx)且状态无底层错误时
+// (如 eof),保留原始错误并把流结束原因追加在后,两条线索都不丢。
+func TestApplyStreamEndDiag_RealErrorKeptWithSummary(t *testing.T) {
+ status := common.NewStreamStatus()
+ status.SetEndReason(common.StreamEndReasonEOF, nil)
+
+ upstreamErr := errors.New("upstream returned HTTP 502")
+ record := &common.UsageRecord{ErrorMessage: upstreamErr.Error()}
+ applyStreamEndDiag(record, status, upstreamErr)
+
+ if !strings.HasPrefix(record.ErrorMessage, upstreamErr.Error()) {
+ t.Errorf("ErrorMessage should keep original error, got %q", record.ErrorMessage)
+ }
+ if !strings.Contains(record.ErrorMessage, string(common.StreamEndReasonEOF)) {
+ t.Errorf("ErrorMessage should append stream summary, got %q", record.ErrorMessage)
+ }
+ if record.StreamEndReason != string(common.StreamEndReasonEOF) {
+ t.Errorf("StreamEndReason = %q, want %q", record.StreamEndReason, common.StreamEndReasonEOF)
+ }
+}
+
+// TestApplyStreamEndDiag_NoopCases nil 状态与未设置结束原因的状态不得改动记录。
+func TestApplyStreamEndDiag_NoopCases(t *testing.T) {
+ record := &common.UsageRecord{ErrorMessage: "raw"}
+ applyStreamEndDiag(record, nil, errors.New("x"))
+ if record.StreamEndReason != "" || record.ErrorMessage != "raw" {
+ t.Errorf("nil status should be no-op, got reason=%q message=%q", record.StreamEndReason, record.ErrorMessage)
+ }
+
+ empty := common.NewStreamStatus()
+ applyStreamEndDiag(record, empty, errors.New("x"))
+ if record.StreamEndReason != "" || record.ErrorMessage != "raw" {
+ t.Errorf("empty reason should be no-op, got reason=%q message=%q", record.StreamEndReason, record.ErrorMessage)
+ }
+}
+
+// TestApplyStreamEndDiag_UpstreamURLNeverLeaks 用量日志 error_message 对租户可见,
+// 编排层包装的上游传输错误(*url.Error 含完整上游域名/IP)必须归一化,禁止原样落库。
+func TestApplyStreamEndDiag_UpstreamURLNeverLeaks(t *testing.T) {
+ // 还原线上泄露形态:NewUpstreamError(502, "请求处理失败", Post "https://api.deepseek.com/...": dial tcp ...)
+ dialErr := &net.OpError{
+ Op: "dial",
+ Net: "tcp",
+ Err: errors.New("connectex: A connection attempt failed because the connected party did not properly respond"),
+ }
+ upstreamErr := &url.Error{Op: "Post", URL: "https://api.deepseek.com/anthropic/v1/messages", Err: dialErr}
+ relayErr := constant.NewUpstreamError(502, "请求处理失败", upstreamErr)
+
+ status := common.NewStreamStatus()
+ status.SetEndReason(common.StreamEndReasonEOF, nil)
+
+ record := &common.UsageRecord{ErrorMessage: relayErr.Error()}
+ applyStreamEndDiag(record, status, relayErr)
+
+ for _, leak := range []string{"api.deepseek.com", "171.105.220.186", "https://", "/anthropic/"} {
+ if strings.Contains(record.ErrorMessage, leak) {
+ t.Errorf("error_message 泄露上游信息 %q: %q", leak, record.ErrorMessage)
+ }
+ }
+}
+
+// TestApplyStreamEndDiag_SummaryLocalAddrRedacted 摘要中的写客户端 TCP 错误含本地地址,
+// IP 必须抹除(保留 wsasend 等错误类别供排查)。
+func TestApplyStreamEndDiag_SummaryLocalAddrRedacted(t *testing.T) {
+ status := common.NewStreamStatus()
+ status.SetEndReason(common.StreamEndReasonHandlerStop,
+ errors.New("write tcp 127.0.0.1:18888->127.0.0.1:6396: wsasend: An existing connection was forcibly closed"))
+
+ record := &common.UsageRecord{ErrorMessage: "x"}
+ applyStreamEndDiag(record, status, common.ErrStreamInterrupted)
+
+ if strings.Contains(record.ErrorMessage, "127.0.0.1") {
+ t.Errorf("摘要未抹除本地 IP: %q", record.ErrorMessage)
+ }
+ if !strings.Contains(record.ErrorMessage, "wsasend") {
+ t.Errorf("应保留错误类别 wsasend: %q", record.ErrorMessage)
+ }
+ if !strings.Contains(record.ErrorMessage, string(common.StreamEndReasonHandlerStop)) {
+ t.Errorf("应保留结束原因 handler_stop: %q", record.ErrorMessage)
+ }
+}
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) => {
{{ authStore.tenant?.name || '租户控制台' }}
{{ authStore.tenant?.name || '个人工作台' }}
平均响应
-{{ formatMs(overviewData.latency.avg_ms) }}
-首 Token
-{{ formatMs(overviewData.latency.avg_first_token_ms) }}
-平均响应
+{{ formatMs(overviewData.latency.avg_ms) }}
+首 Token
+{{ formatMs(overviewData.latency.avg_first_token_ms) }}
{{ item.label }}
-{{ item.value }}
-{{ item.label }}
+{{ item.value }}
+已命中 {{ formatNumber(overviewData.cache.cache_read_tokens) }} Token,约省下 {{ formatCost(overviewData.cache.saved_cost) }}
@@ -913,30 +912,6 @@ onMounted(refreshAll) padding: 0.7rem 0.8rem; } -.latency-details { - margin-top: 0.7rem; -} - -.latency-details > summary { - color: #0d9488; - cursor: pointer; - font-size: 0.75rem; - font-weight: 620; - list-style: none; -} - -.latency-details > summary::-webkit-details-marker { - display: none; -} - -.latency-details > summary::after { - content: ' ▾'; -} - -.latency-details[open] > summary::after { - content: ' ▴'; -} - .mini-progress { height: 0.35rem; overflow: hidden; @@ -951,14 +926,10 @@ onMounted(refreshAll) background: linear-gradient(90deg, #2dd4bf, #0d9488); } -.cache-strip { - display: flex; - align-items: center; - justify-content: space-between; - gap: 1rem; - border-radius: 0.8rem; - background: rgba(6, 182, 212, 0.08); - padding: 0.7rem 0.85rem; +/* 缓存命中小卡片:在次级指标带里用青色调区分身份 */ +.quality-metric-cache { + border-color: rgba(6, 182, 212, 0.2); + background: rgba(6, 182, 212, 0.07); } /* ── 失败列表 ── */