fix: 修正访问控制与插件契约
This commit is contained in:
+1
-4
@@ -9,7 +9,4 @@ coverage.out
|
||||
.cache/
|
||||
/.runtime/
|
||||
/.home/
|
||||
|
||||
/CLIProxyAPI/
|
||||
/cpa-plugin-key-billing/
|
||||
/cpa-usage-keeper/
|
||||
/.externals/
|
||||
|
||||
@@ -66,13 +66,13 @@ CLIProxyAPI 负责 HTTP 接入、协议转换、上游凭证和实际请求执
|
||||
|
||||
## 当前兼容目标
|
||||
|
||||
- CLIProxyAPI 源码:`CLIProxyAPI/`,检查时 revision 为 `f43aad7637ad813745bf7d341acb5663617570c5`
|
||||
- CLIProxyAPI 源码:`.externals/CLIProxyAPI/`,检查时 revision 为 `f43aad7637ad813745bf7d341acb5663617570c5`
|
||||
- Native ABI:`1`
|
||||
- RPC schema:最高 `3`,注册时按宿主版本向下协商
|
||||
- 插件 ID / 动态库文件名:`cpa-ext`
|
||||
- 已声明能力:`frontend_auth_provider`(独占)、`scheduler`、`request_interceptor`、`request_lifecycle_plugin`、`usage_plugin`、`management_api`
|
||||
|
||||
契约来源以本仓库内 `CLIProxyAPI/sdk/pluginabi/types.go`、`CLIProxyAPI/sdk/pluginapi/types.go`、`CLIProxyAPI/internal/pluginhost/rpc_schema.go` 为准。
|
||||
契约来源以本仓库内 `.externals/CLIProxyAPI/sdk/pluginabi/types.go`、`.externals/CLIProxyAPI/sdk/pluginapi/types.go`、`.externals/CLIProxyAPI/internal/pluginhost/rpc_schema.go` 为准。
|
||||
|
||||
## 环境与构建(WSL/Linux)
|
||||
|
||||
@@ -91,10 +91,10 @@ sudo apt-get install -y build-essential
|
||||
./scripts/build.ps1
|
||||
```
|
||||
|
||||
产物为 `bin/cpa-ext.so`,适用于运行在 WSL/Linux 的 CLIProxyAPI。普通开发测试不需要 CGO:
|
||||
产物为 `bin/cpa-ext.so`,适用于运行在 WSL/Linux 的 CLIProxyAPI。SQLite 驱动依赖 CGO,普通开发测试也应在已经安装 GCC 的 WSL/Linux 环境执行:
|
||||
|
||||
```powershell
|
||||
go test ./...
|
||||
```bash
|
||||
CGO_ENABLED=1 go test ./...
|
||||
```
|
||||
|
||||
## 加载到 CLIProxyAPI
|
||||
@@ -128,11 +128,11 @@ go test ./...
|
||||
|
||||
请求明细由服务端分页,每页 100 条,默认按请求时间和稳定 ID 倒序展示。管理接口支持时间范围、用户 Key、模型、结果、上游 Auth ID、端点和完整 Request ID 筛选;连续翻页使用与筛选条件绑定的游标,数字页码跳转使用 SQLite 索引定位。用户页的今日汇总、各用户用量和近 7 日 Token 由数据库独立聚合,不受当前明细页影响。
|
||||
|
||||
升级已有数据库时,插件会从原始 Usage 和请求终态事实自动建立轻量查询投影。事实表、计费账目和历史统计保持不变;重试执行继续分别展示,没有 Usage 的取消或失败请求仍然可见,缺少 Request ID 的旧回调沿用双向唯一时间匹配规则。
|
||||
升级已有数据库时,插件会从原始 Usage 和请求终态事实自动建立轻量查询投影。事实表、计费账目和历史统计保持不变;没有 Usage 的取消或失败请求仍然可见。当前 CPA Usage 契约不提供 Request ID 或 Execution ID,因此只在模型和时间足够接近且匹配关系唯一时关联请求终态;存在并发歧义时保留为独立记录,不强行合并。
|
||||
|
||||
## 工程布局
|
||||
|
||||
- `cmd/cpa-ext`:仅负责 C ABI、请求字节复制和 C 内存释放。
|
||||
- `internal/plugin`:RPC dispatcher、契约 DTO、原子配置与 Usage 接入。
|
||||
- `scripts`:环境检查与可复现构建。
|
||||
- `CLIProxyAPI`:上游契约参考,不属于插件实现。
|
||||
- `.externals/CLIProxyAPI`:上游契约参考,不属于插件实现。
|
||||
|
||||
@@ -132,7 +132,7 @@
|
||||
| `quota_change` | 管理员修改本周期额度 | 增加为正,减少为负 |
|
||||
| `cycle_reset` | 自动或手动开始新周期 | 新周期完整额度 |
|
||||
|
||||
每条账目保存用户 Key、额度周期、变动金额、变动后余额和发生时间。请求扣费还会保存 Request ID、Execution ID 和模型。
|
||||
每条账目保存用户 Key、额度周期、变动金额、变动后余额和发生时间,请求扣费还会保存模型。账目结构预留 Request ID 和 Execution ID,但当前目标 CPA 的 Usage 契约不提供这两个字段,因此实际扣费账目通常为空,不使用时间推测结果改写不可变账目。
|
||||
|
||||
账目只追加、不修改、不删除,并通过唯一事件标识避免重复写入。额度状态用于快速读取当前余额,账目用于解释余额变化过程。
|
||||
|
||||
|
||||
@@ -26,19 +26,20 @@ CLIProxyAPI 会通过两个独立回调提供请求信息:
|
||||
| 事实 | 来源 | 主要内容 |
|
||||
| --- | --- | --- |
|
||||
| 请求终态 | `request.complete` | Request ID、开始与结束时间、成功/失败/拒绝/取消、状态码和错误 |
|
||||
| 最终用量 | `usage.handle` | Execution ID、实际上游、模型、Token、延迟和 Usage 结果 |
|
||||
| 最终用量 | `usage.handle` | 实际上游、模型、Token、延迟和 Usage 结果 |
|
||||
|
||||
这两个回调可能乱序、重复或只到达其中一个,因此数据库分别保存原始事实,再建立请求明细查询投影。管理台看到的一行是查询结果,不会为了合并展示而修改原始事实或计费账目。
|
||||
|
||||
## 请求与重试
|
||||
## 请求关联与独立用量
|
||||
|
||||
- Request ID 用于关联一次下游请求的生命周期。
|
||||
- Execution ID 用于区分该请求的具体上游执行。
|
||||
- 同一次请求发生上游重试时,每次真实执行继续分别展示,不会把多个执行的 Token 或成本错误相加为一次执行。
|
||||
- 当前目标 CPA 的 Usage 契约不提供 Request ID 或 Execution ID;Request ID 和 Trace ID 来自独立的请求终态。
|
||||
- 每条可区分的 Usage 事实分别保存,不会为了得到一条整齐记录而把多个用量相加。
|
||||
- 当前契约无法保证把每次上游重试稳定标记为某个 Execution ID,因此管理台不声明这种保证。
|
||||
- 没有 Usage 的拒绝、取消或失败请求仍然显示请求终态。
|
||||
- 只有 Usage、暂时没有终态的记录也可以单独显示,终态到达后投影会自动更新。
|
||||
|
||||
部分旧版回调可能缺少 Request ID。系统仅在请求时间足够接近且匹配关系唯一时,将 Usage 与终态合并;存在并发歧义时宁可保留为两条,也不会错误关联到其他用户的请求。
|
||||
系统仅在模型和请求时间足够接近且匹配关系唯一时,将 Usage 与终态合并;存在并发歧义时宁可保留为两条,也不会错误关联到其他用户的请求。
|
||||
|
||||
## 请求结果
|
||||
|
||||
@@ -55,19 +56,18 @@ CLIProxyAPI 会通过两个独立回调提供请求信息:
|
||||
|
||||
请求明细当前可以展示:
|
||||
|
||||
- 请求时间、Request ID、Execution ID、Trace ID;
|
||||
- 请求时间,以及请求终态能够提供的 Request ID 和 Trace ID;
|
||||
- 用户 Key 名称;
|
||||
- 客户端请求模型和实际计费模型;
|
||||
- 推理强度、服务档位和速度模式;
|
||||
- 推理强度和服务档位;
|
||||
- 请求结果、HTTP 状态码和错误;
|
||||
- 请求类型与端点;
|
||||
- 实际上游 Auth ID、Auth Index 和认证类型;
|
||||
- 首字延迟、生成速度;
|
||||
- 输入、输出、推理、缓存读取、缓存写入和总 Token;
|
||||
- 缓存率、成本、价格档位和 Fast 计价结果;
|
||||
- 客户端 IP。
|
||||
- 缓存率、成本、价格档位和 Fast 计价结果。
|
||||
|
||||
并非每种端点都能提供所有字段。字段无法由 CPA 回调可靠获得时保留为空,不使用猜测值。一次性 JSON/Compact 响应不展示不可比较的首字延迟和生成速度。
|
||||
并非每种端点都能提供所有字段。管理接口为兼容历史数据保留 Execution ID、速度模式和客户端 IP 等可选字段,但当前目标 CPA 不提供这些值;字段无法可靠获得时保留为空,不使用猜测值。一次性 JSON/Compact 响应不展示不可比较的首字延迟和生成速度。
|
||||
|
||||
## 服务端分页
|
||||
|
||||
@@ -113,7 +113,7 @@ CLIProxyAPI 会通过两个独立回调提供请求信息:
|
||||
|
||||
## 数据一致性
|
||||
|
||||
- Usage 插入使用稳定事件标识,重复回调不会重复记录或扣费。
|
||||
- Usage 插入使用内容事件标识,完全相同的重复回调不会重复记录或扣费。
|
||||
- 请求终态按 Request ID 幂等更新。
|
||||
- 回调乱序时,后到达的事实会重新同步查询投影。
|
||||
- 失败、取消和孤立 Usage 不会为了界面整齐而被删除。
|
||||
|
||||
@@ -75,11 +75,10 @@ Key 值允许 1–256 个非空白、非控制字符。未手动填写时,系
|
||||
出现以下情况时,请求返回 HTTP 503 和错误码 `bound_upstream_unavailable`:
|
||||
|
||||
- 绑定的账号已经不存在;
|
||||
- 绑定的账号被禁用或处于不可用状态;
|
||||
- 绑定账号不在本次请求的可用候选中;
|
||||
- 绑定账号不在本次 CPA 实时可用候选中;
|
||||
- 调度完成后的实际上游与绑定账号不一致。
|
||||
|
||||
上游账号信息来自 CLIProxyAPI 的账号列表和实际调度候选,插件只保存路由所需的账号标识、提供商、显示名称和状态,不读取或保存上游 Token、Cookie 等凭证内容。
|
||||
上游账号信息来自 CLIProxyAPI 的账号列表和实际调度候选。持久化状态用于管理台展示,本次调度候选才是请求时的可用性依据,避免账号恢复后被旧状态继续拦截。插件只保存路由所需的账号标识、提供商、显示名称和状态,不读取或保存上游 Token、Cookie 等凭证内容。
|
||||
|
||||
## 默认 Key 与新建 Key
|
||||
|
||||
@@ -93,7 +92,7 @@ Key 值允许 1–256 个非空白、非控制字符。未手动填写时,系
|
||||
|
||||
该行为用于让已有 Codex/CLIProxyAPI 调用配置无需修改即可迁移到插件管理。
|
||||
|
||||
之后创建新 Key 时,系统先复制 `default` 当时的模型权限和上游路由,再应用管理员本次明确填写的设置。复制只发生在创建时,后续修改 `default` 不会影响已经创建的 Key。
|
||||
之后创建新 Key 时,系统按稳定 ID `key_default` 复制默认 Key 当时的模型权限和上游路由,再应用管理员本次明确填写的设置。管理员重命名默认 Key 不影响继承。复制只发生在创建时,后续修改默认 Key 不会影响已经创建的 Key。
|
||||
|
||||
如果数据库中已经存在 Key,启动配置不会覆盖或重新创建 `default`。
|
||||
|
||||
|
||||
@@ -1,144 +0,0 @@
|
||||
// Package data 定义 cpa-ext 各模块共享且与 Provider 无关的数据事实。
|
||||
package data
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"math"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// UsageQuality 表示 Provider Token 数据经过规范化后的可信程度。
|
||||
// 使用方必须检查该值,不能把所有零值都当成真实测量结果。
|
||||
type UsageQuality string
|
||||
|
||||
const (
|
||||
UsageQualityComplete UsageQuality = "complete"
|
||||
UsageQualityNormalized UsageQuality = "normalized"
|
||||
UsageQualityPartial UsageQuality = "partial"
|
||||
UsageQualityInconsistent UsageQuality = "inconsistent"
|
||||
UsageQualityUnclassified UsageQuality = "unclassified"
|
||||
UsageQualityMissing UsageQuality = "missing"
|
||||
)
|
||||
|
||||
// TokenUsage 保存互不重叠的 Token 分项。每个 Token 只能进入一个分项,
|
||||
// 防止计价和统计重复计算。
|
||||
type TokenUsage struct {
|
||||
UncachedInputTokens int64
|
||||
CacheReadTokens int64
|
||||
CacheCreationTokens int64
|
||||
NonReasoningTokens int64
|
||||
ReasoningTokens int64
|
||||
UnclassifiedTokens int64
|
||||
TotalTokens int64
|
||||
}
|
||||
|
||||
// Usage 是与 Provider 无关的最小用量事实。该类型不保存凭证、认证请求头、
|
||||
// 请求体或响应内容。
|
||||
type Usage struct {
|
||||
Provider string
|
||||
Model string
|
||||
RequestedAt time.Time
|
||||
Generated bool
|
||||
Failed bool
|
||||
Quality UsageQuality
|
||||
Tokens TokenUsage
|
||||
}
|
||||
|
||||
// Validate 检查 Usage 的结构约束,但不会修改观测到的数据。
|
||||
// inconsistent 类型允许保留互相矛盾的总数和分项,供后续诊断使用。
|
||||
func (u Usage) Validate() error {
|
||||
if strings.TrimSpace(u.Provider) == "" {
|
||||
return errors.New("usage provider is required")
|
||||
}
|
||||
if strings.TrimSpace(u.Model) == "" {
|
||||
return errors.New("usage model is required")
|
||||
}
|
||||
if u.RequestedAt.IsZero() {
|
||||
return errors.New("usage requested time is required")
|
||||
}
|
||||
if !u.Quality.valid() {
|
||||
return fmt.Errorf("unknown usage quality %q", u.Quality)
|
||||
}
|
||||
return u.Tokens.validate(u.Quality)
|
||||
}
|
||||
|
||||
func (q UsageQuality) valid() bool {
|
||||
switch q {
|
||||
case UsageQualityComplete,
|
||||
UsageQualityNormalized,
|
||||
UsageQualityPartial,
|
||||
UsageQualityInconsistent,
|
||||
UsageQualityUnclassified,
|
||||
UsageQualityMissing:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func (t TokenUsage) validate(quality UsageQuality) error {
|
||||
counts := []struct {
|
||||
name string
|
||||
value int64
|
||||
}{
|
||||
{"uncached input tokens", t.UncachedInputTokens},
|
||||
{"cache read tokens", t.CacheReadTokens},
|
||||
{"cache creation tokens", t.CacheCreationTokens},
|
||||
{"non-reasoning output tokens", t.NonReasoningTokens},
|
||||
{"reasoning tokens", t.ReasoningTokens},
|
||||
{"unclassified tokens", t.UnclassifiedTokens},
|
||||
{"total tokens", t.TotalTokens},
|
||||
}
|
||||
for _, count := range counts {
|
||||
if count.value < 0 {
|
||||
return fmt.Errorf("%s cannot be negative", count.name)
|
||||
}
|
||||
}
|
||||
|
||||
sum, ok := t.bucketSum()
|
||||
if !ok {
|
||||
return errors.New("token bucket sum overflows int64")
|
||||
}
|
||||
|
||||
switch quality {
|
||||
case UsageQualityMissing:
|
||||
if sum != 0 || t.TotalTokens != 0 {
|
||||
return errors.New("missing usage cannot contain token counts")
|
||||
}
|
||||
case UsageQualityComplete, UsageQualityNormalized:
|
||||
if t.UnclassifiedTokens != 0 {
|
||||
return errors.New("classified usage cannot contain unclassified tokens")
|
||||
}
|
||||
if sum != t.TotalTokens {
|
||||
return errors.New("classified token buckets do not equal total tokens")
|
||||
}
|
||||
case UsageQualityPartial, UsageQualityUnclassified:
|
||||
if sum != t.TotalTokens {
|
||||
return errors.New("token buckets do not equal total tokens")
|
||||
}
|
||||
case UsageQualityInconsistent:
|
||||
// 矛盾的计数需要作为明确的诊断事实保留下来。
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (t TokenUsage) bucketSum() (int64, bool) {
|
||||
values := [...]int64{
|
||||
t.UncachedInputTokens,
|
||||
t.CacheReadTokens,
|
||||
t.CacheCreationTokens,
|
||||
t.NonReasoningTokens,
|
||||
t.ReasoningTokens,
|
||||
t.UnclassifiedTokens,
|
||||
}
|
||||
var sum int64
|
||||
for _, value := range values {
|
||||
if value > math.MaxInt64-sum {
|
||||
return 0, false
|
||||
}
|
||||
sum += value
|
||||
}
|
||||
return sum, true
|
||||
}
|
||||
@@ -1,146 +0,0 @@
|
||||
package data_test
|
||||
|
||||
import (
|
||||
"math"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"cpa-ext/internal/data"
|
||||
)
|
||||
|
||||
func validUsage() data.Usage {
|
||||
return data.Usage{
|
||||
Provider: "codex",
|
||||
Model: "gpt-5.5",
|
||||
RequestedAt: time.Date(2026, time.August, 14, 8, 0, 0, 0, time.UTC),
|
||||
Generated: true,
|
||||
Quality: data.UsageQualityComplete,
|
||||
Tokens: data.TokenUsage{
|
||||
UncachedInputTokens: 10,
|
||||
CacheReadTokens: 20,
|
||||
CacheCreationTokens: 30,
|
||||
NonReasoningTokens: 40,
|
||||
ReasoningTokens: 50,
|
||||
TotalTokens: 150,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func TestUsageValidateAcceptsCompleteBreakdown(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
if err := validUsage().Validate(); err != nil {
|
||||
t.Fatalf("Validate() error = %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestUsageDistinguishesMissingFromMeasuredZero(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
measured := validUsage()
|
||||
measured.Tokens = data.TokenUsage{}
|
||||
if err := measured.Validate(); err != nil {
|
||||
t.Fatalf("measured zero Validate() error = %v", err)
|
||||
}
|
||||
|
||||
missing := measured
|
||||
missing.Quality = data.UsageQualityMissing
|
||||
if err := missing.Validate(); err != nil {
|
||||
t.Fatalf("missing Validate() error = %v", err)
|
||||
}
|
||||
|
||||
missing.Tokens.TotalTokens = 1
|
||||
if err := missing.Validate(); err == nil {
|
||||
t.Fatal("missing usage with tokens unexpectedly passed validation")
|
||||
}
|
||||
}
|
||||
|
||||
func TestUsageValidateRejectsInvalidIdentityAndQuality(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
mutate func(*data.Usage)
|
||||
}{
|
||||
{"missing provider", func(u *data.Usage) { u.Provider = " " }},
|
||||
{"missing model", func(u *data.Usage) { u.Model = "" }},
|
||||
{"missing requested time", func(u *data.Usage) { u.RequestedAt = time.Time{} }},
|
||||
{"unknown quality", func(u *data.Usage) { u.Quality = "other" }},
|
||||
}
|
||||
|
||||
for _, test := range tests {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
usage := validUsage()
|
||||
test.mutate(&usage)
|
||||
if err := usage.Validate(); err == nil {
|
||||
t.Fatal("Validate() unexpectedly succeeded")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestUsageValidateRejectsInvalidTokenBreakdown(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
tests := []struct {
|
||||
name string
|
||||
mutate func(*data.Usage)
|
||||
}{
|
||||
{"negative token", func(u *data.Usage) { u.Tokens.CacheReadTokens = -1 }},
|
||||
{"wrong total", func(u *data.Usage) { u.Tokens.TotalTokens++ }},
|
||||
{"unclassified complete", func(u *data.Usage) {
|
||||
u.Tokens.UnclassifiedTokens = 1
|
||||
u.Tokens.TotalTokens++
|
||||
}},
|
||||
{"overflow", func(u *data.Usage) {
|
||||
u.Tokens = data.TokenUsage{
|
||||
UncachedInputTokens: math.MaxInt64,
|
||||
CacheReadTokens: 1,
|
||||
TotalTokens: math.MaxInt64,
|
||||
}
|
||||
}},
|
||||
}
|
||||
|
||||
for _, test := range tests {
|
||||
t.Run(test.name, func(t *testing.T) {
|
||||
t.Parallel()
|
||||
usage := validUsage()
|
||||
test.mutate(&usage)
|
||||
if err := usage.Validate(); err == nil {
|
||||
t.Fatal("Validate() unexpectedly succeeded")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestUsageValidatePreservesExplicitQualityStates(t *testing.T) {
|
||||
t.Parallel()
|
||||
|
||||
normalized := validUsage()
|
||||
normalized.Quality = data.UsageQualityNormalized
|
||||
if err := normalized.Validate(); err != nil {
|
||||
t.Fatalf("normalized Validate() error = %v", err)
|
||||
}
|
||||
|
||||
partial := validUsage()
|
||||
partial.Quality = data.UsageQualityPartial
|
||||
if err := partial.Validate(); err != nil {
|
||||
t.Fatalf("partial Validate() error = %v", err)
|
||||
}
|
||||
|
||||
unclassified := validUsage()
|
||||
unclassified.Quality = data.UsageQualityUnclassified
|
||||
unclassified.Tokens.UnclassifiedTokens = 5
|
||||
unclassified.Tokens.TotalTokens += 5
|
||||
if err := unclassified.Validate(); err != nil {
|
||||
t.Fatalf("unclassified Validate() error = %v", err)
|
||||
}
|
||||
|
||||
inconsistent := validUsage()
|
||||
inconsistent.Quality = data.UsageQualityInconsistent
|
||||
inconsistent.Tokens.TotalTokens++
|
||||
if err := inconsistent.Validate(); err != nil {
|
||||
t.Fatalf("inconsistent Validate() error = %v", err)
|
||||
}
|
||||
}
|
||||
@@ -150,7 +150,7 @@ func (a *App) pickScheduler(raw []byte) ([]byte, error) {
|
||||
return OKEnvelope(SchedulerPickResponse{Handled: false})
|
||||
}
|
||||
account, err := a.upstreamAccount(key.UpstreamAccountID)
|
||||
if err != nil || account.Disabled || account.Unavailable {
|
||||
if err != nil {
|
||||
return ErrorEnvelope("bound_upstream_unavailable", "指定的上游账号不可用", http.StatusServiceUnavailable), nil
|
||||
}
|
||||
for _, candidate := range req.Candidates {
|
||||
|
||||
@@ -31,6 +31,9 @@ func TestManagedKeyAuthenticationAndLifecycle(t *testing.T) {
|
||||
if got := auth("000000"); !got.Authenticated || got.Principal != "key_default" {
|
||||
t.Fatalf("default auth = %+v", got)
|
||||
}
|
||||
if got := auth("key_default"); got.Authenticated {
|
||||
t.Fatalf("internal Key ID authenticated as a credential: %+v", got)
|
||||
}
|
||||
xAPIKeyRaw, _ := json.Marshal(FrontendAuthRequest{Headers: http.Header{"X-Api-Key": []string{"000000"}}})
|
||||
xAPIKeyResponse, _ := app.HandleMethod(MethodFrontendAuthenticate, xAPIKeyRaw)
|
||||
var xAPIKeyEnvelope Envelope
|
||||
@@ -69,7 +72,8 @@ func TestModelAllowlistAndStrictUpstreamRouting(t *testing.T) {
|
||||
}
|
||||
store, _ := app.currentStore()
|
||||
if err := store.SyncUpstreamAccounts(context.Background(), []managedaccess.UpstreamAccount{{
|
||||
CPAAuthID: "oauth-1", CPAAuthIndex: "index-1", Provider: "codex", DisplayName: "OAuth One", LastSeenAt: time.Now(),
|
||||
CPAAuthID: "oauth-1", CPAAuthIndex: "index-1", Provider: "codex", DisplayName: "OAuth One",
|
||||
Disabled: true, Unavailable: true, LastSeenAt: time.Now(),
|
||||
}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
@@ -157,6 +161,18 @@ func TestCreateKeyCopiesDefaultRuleSnapshot(t *testing.T) {
|
||||
if _, err := app.HandleMethod(MethodPluginRegister, lifecycleRequest(t, SchemaVersion, testConfig(t, "enabled: true\n"))); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
store, _ := app.currentStore()
|
||||
template, err := store.ManagedKeyByID(context.Background(), "key_default")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
template.Name = "renamed-default"
|
||||
template.AllModels = false
|
||||
template.Models = []string{"deepseek-*"}
|
||||
if err := store.UpdateManagedKey(context.Background(), template); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
response := managementCallBody(t, app, http.MethodPost, managementBase+routeKeys, []byte(`{"name":"alice","secret":"alice-000000"}`))
|
||||
if response.StatusCode != http.StatusCreated {
|
||||
t.Fatalf("create status=%d body=%s", response.StatusCode, response.Body)
|
||||
@@ -165,7 +181,7 @@ func TestCreateKeyCopiesDefaultRuleSnapshot(t *testing.T) {
|
||||
if err := json.Unmarshal(response.Body, &created); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if created.Secret != "alice-000000" || created.RouteMode != managedaccess.RouteAuto || !created.AllModels {
|
||||
if created.Secret != "alice-000000" || created.RouteMode != managedaccess.RouteAuto || created.AllModels || len(created.Models) != 1 || created.Models[0] != "deepseek-*" {
|
||||
t.Fatalf("created = %+v", created)
|
||||
}
|
||||
}
|
||||
@@ -223,7 +239,7 @@ func TestBillingAdmissionConcurrencyAndUsageSettlement(t *testing.T) {
|
||||
if _, err := app.HandleMethod(MethodRequestComplete, completionRaw); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
usage := UsageRecord{ExecutionID: "settled-once", APIKey: "000000", Model: "deepseek-v4-flash", RequestedAt: time.Now(), Detail: UsageDetail{OutputTokens: 10_000, TotalTokens: 10_000}}
|
||||
usage := UsageRecord{APIKey: "000000", Model: "deepseek-v4-flash", RequestedAt: time.Now(), Detail: UsageDetail{OutputTokens: 10_000, TotalTokens: 10_000}}
|
||||
raw, _ := json.Marshal(usage)
|
||||
if _, err := app.HandleMethod(MethodUsageHandle, raw); err != nil {
|
||||
t.Fatal(err)
|
||||
|
||||
+3
-28
@@ -104,9 +104,9 @@ func (a *App) configure(raw []byte) ([]byte, error) {
|
||||
nextPrices[normalizeModelName(policy.Model)] = policy
|
||||
}
|
||||
a.mu.RLock()
|
||||
firstInitialization := a.usage == nil
|
||||
releaseStaleAdmissions := a.usage == nil || a.config.DatabasePath != cfg.DatabasePath
|
||||
a.mu.RUnlock()
|
||||
if firstInitialization {
|
||||
if releaseStaleAdmissions {
|
||||
if err := usageRepository.ReleaseStaleAdmissions(context.Background(), time.Now().UTC()); err != nil {
|
||||
_ = nextUsage.Close()
|
||||
return nil, fmt.Errorf("释放遗留并发占用: %w", err)
|
||||
@@ -170,9 +170,6 @@ func (a *App) handleUsage(raw []byte) ([]byte, error) {
|
||||
}
|
||||
// CPA wire 类型只存在于适配层,采集模块接收与协议无关的观察值。
|
||||
observed := collection.Record{
|
||||
RequestID: record.RequestID,
|
||||
ExecutionID: record.ExecutionID,
|
||||
TraceID: record.TraceID,
|
||||
RequestedAt: record.RequestedAt,
|
||||
APIKey: record.APIKey,
|
||||
AuthID: record.AuthID,
|
||||
@@ -181,11 +178,8 @@ func (a *App) handleUsage(raw []byte) ([]byte, error) {
|
||||
Model: record.Model,
|
||||
ReasoningEffort: record.ReasoningEffort,
|
||||
ServiceTier: record.ServiceTier,
|
||||
Speed: record.Speed,
|
||||
Failed: record.Failed,
|
||||
ExecutorType: record.ExecutorType,
|
||||
RequestType: usageRequestType(record.Endpoint),
|
||||
Endpoint: record.Endpoint,
|
||||
InputTokens: record.Detail.InputTokens,
|
||||
OutputTokens: record.Detail.OutputTokens,
|
||||
TotalTokens: record.Detail.TotalTokens,
|
||||
@@ -195,7 +189,6 @@ func (a *App) handleUsage(raw []byte) ([]byte, error) {
|
||||
ReasoningTokens: record.Detail.ReasoningTokens,
|
||||
TTFT: record.TTFT,
|
||||
Latency: record.Latency,
|
||||
ClientIP: record.ClientIP,
|
||||
}
|
||||
if key, keyErr := a.store.ManagedKeyByCredential(context.Background(), record.APIKey); keyErr == nil {
|
||||
observed.ManagedKeyID = key.ID
|
||||
@@ -205,7 +198,7 @@ func (a *App) handleUsage(raw []byte) ([]byte, error) {
|
||||
cost, calculateErr := pricing.Calculate(policy, pricing.Usage{
|
||||
InputTokens: record.Detail.InputTokens, CacheReadTokens: record.Detail.CacheReadTokens,
|
||||
CacheWriteTokens: record.Detail.CacheCreationTokens, OutputTokens: record.Detail.OutputTokens,
|
||||
ServiceTier: record.ServiceTier, Speed: record.Speed,
|
||||
ServiceTier: record.ServiceTier,
|
||||
})
|
||||
if calculateErr == nil {
|
||||
observed.CostMicros = &cost.CostMicros
|
||||
@@ -261,24 +254,6 @@ func (a *App) handleRequestComplete(raw []byte) ([]byte, error) {
|
||||
return OKEnvelope(struct{}{})
|
||||
}
|
||||
|
||||
func usageRequestType(endpoint string) string {
|
||||
if isCompactEndpoint(endpoint) {
|
||||
return "JSON"
|
||||
}
|
||||
parts := strings.Fields(endpoint)
|
||||
if len(parts) == 0 {
|
||||
return ""
|
||||
}
|
||||
switch strings.ToUpper(parts[0]) {
|
||||
case http.MethodPost:
|
||||
return "SSE"
|
||||
case http.MethodGet:
|
||||
return "WS"
|
||||
default:
|
||||
return ""
|
||||
}
|
||||
}
|
||||
|
||||
func isCompactEndpoint(endpoint string) bool {
|
||||
path := strings.ToLower(strings.TrimSpace(endpoint))
|
||||
if fields := strings.Fields(path); len(fields) > 1 {
|
||||
|
||||
+80
-17
@@ -1,12 +1,17 @@
|
||||
package plugin
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
managedaccess "cpa-ext/internal/access"
|
||||
"cpa-ext/internal/repository"
|
||||
)
|
||||
|
||||
func lifecycleRequest(t *testing.T, schema uint32, config string) []byte {
|
||||
@@ -60,6 +65,30 @@ func TestLifecycleConfigYAMLUsesBase64WireEncoding(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestUsageRecordWireFieldsMatchTargetContract(t *testing.T) {
|
||||
raw, err := json.Marshal(UsageRecord{})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var fields map[string]json.RawMessage
|
||||
if err := json.Unmarshal(raw, &fields); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
want := []string{
|
||||
"Provider", "ExecutorType", "Model", "Alias", "APIKey", "AuthID", "AuthIndex", "AuthType",
|
||||
"Source", "ReasoningEffort", "ServiceTier", "Generate", "RequestedAt", "Latency", "TTFT",
|
||||
"Failed", "Failure", "Detail", "ResponseHeaders",
|
||||
}
|
||||
if len(fields) != len(want) {
|
||||
t.Fatalf("UsageRecord fields=%v", fields)
|
||||
}
|
||||
for _, name := range want {
|
||||
if _, ok := fields[name]; !ok {
|
||||
t.Fatalf("UsageRecord missing target field %s", name)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestReconfigureIsAtomicAndFiltersUsage(t *testing.T) {
|
||||
app := NewApp()
|
||||
if _, err := app.HandleMethod(MethodPluginRegister, lifecycleRequest(t, SchemaVersion, testConfig(t, "enabled: true\ncodex_only: true\n"))); err != nil {
|
||||
@@ -84,6 +113,57 @@ func TestReconfigureIsAtomicAndFiltersUsage(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestReconfigureReleasesAdmissionsWhenDatabaseChanges(t *testing.T) {
|
||||
directory := t.TempDir()
|
||||
firstPath := filepath.Join(directory, "first.db")
|
||||
secondPath := filepath.Join(directory, "second.db")
|
||||
store, err := repository.OpenSQLiteUsage(secondPath)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := store.BootstrapManagedKey(context.Background(), "default", "000000"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := store.UpdateBilling(context.Background(), "key_default", managedaccess.BillingSettings{
|
||||
QuotaMicros: 1_000_000, ResetPeriod: managedaccess.ResetNone, MaxConcurrency: 4,
|
||||
}, time.Now()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := store.AuthorizeBilling(context.Background(), "key_default", "stale-request", time.Now()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := store.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
app := NewApp()
|
||||
if _, err := app.HandleMethod(MethodPluginRegister, lifecycleRequest(t, SchemaVersion, fmt.Sprintf("database_path: %q\n", firstPath))); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := app.store.UpdateBilling(context.Background(), "key_default", managedaccess.BillingSettings{
|
||||
QuotaMicros: 1_000_000, ResetPeriod: managedaccess.ResetNone, MaxConcurrency: 4,
|
||||
}, time.Now()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := app.store.AuthorizeBilling(context.Background(), "key_default", "live-request", time.Now()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := app.HandleMethod(MethodPluginReconfigure, lifecycleRequest(t, SchemaVersion, fmt.Sprintf("database_path: %q\n", firstPath))); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
state, err := app.store.BillingState(context.Background(), "key_default", time.Now())
|
||||
if err != nil || state.ActiveRequests != 1 {
|
||||
t.Fatalf("same-database reconfiguration released live admission: state=%+v err=%v", state, err)
|
||||
}
|
||||
if _, err := app.HandleMethod(MethodPluginReconfigure, lifecycleRequest(t, SchemaVersion, fmt.Sprintf("database_path: %q\n", secondPath))); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
state, err = app.store.BillingState(context.Background(), "key_default", time.Now())
|
||||
if err != nil || state.ActiveRequests != 0 {
|
||||
t.Fatalf("reconfigured billing state=%+v err=%v", state, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestConcurrentUsage(t *testing.T) {
|
||||
app := NewApp()
|
||||
if _, err := app.HandleMethod(MethodPluginRegister, lifecycleRequest(t, SchemaVersion, testConfig(t, "enabled: true\ncodex_only: true\n"))); err != nil {
|
||||
@@ -119,20 +199,3 @@ func TestUnknownMethodReturnsErrorEnvelope(t *testing.T) {
|
||||
t.Fatalf("unexpected envelope: %s", raw)
|
||||
}
|
||||
}
|
||||
|
||||
func TestUsageRequestTypeMatchesKeeperEndpointSemantics(t *testing.T) {
|
||||
tests := []struct {
|
||||
endpoint string
|
||||
want string
|
||||
}{
|
||||
{endpoint: "POST /v1/responses", want: "SSE"},
|
||||
{endpoint: "POST /v1/chat/completions", want: "SSE"},
|
||||
{endpoint: "GET /v1/responses", want: "WS"},
|
||||
{endpoint: "/v1/responses", want: ""},
|
||||
}
|
||||
for _, test := range tests {
|
||||
if got := usageRequestType(test.endpoint); got != test.want {
|
||||
t.Fatalf("usageRequestType(%q) = %q, want %q", test.endpoint, got, test.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -134,10 +134,7 @@ func (a *App) createManagedKey(body []byte) ManagementResponse {
|
||||
ID: keyID, Name: name, Secret: secret,
|
||||
Status: managedaccess.StatusActive, RouteMode: managedaccess.RouteAuto, AllModels: true,
|
||||
}
|
||||
a.mu.RLock()
|
||||
defaultName := a.config.BootstrapName
|
||||
a.mu.RUnlock()
|
||||
if template, err := store.ManagedKeyByName(context.Background(), defaultName); err == nil {
|
||||
if template, err := store.ManagedKeyByID(context.Background(), "key_default"); err == nil {
|
||||
key.RouteMode = template.RouteMode
|
||||
key.UpstreamAccountID = template.UpstreamAccountID
|
||||
key.AllModels = template.AllModels
|
||||
|
||||
@@ -67,7 +67,7 @@ func TestUsageManagementSupportsServerPaginationFiltersAndSummary(t *testing.T)
|
||||
}
|
||||
base := time.Date(2026, 8, 15, 8, 0, 0, 0, time.UTC)
|
||||
for index := 0; index < 205; index++ {
|
||||
record := UsageRecord{RequestID: fmt.Sprintf("page-request-%03d", index), ExecutionID: fmt.Sprintf("page-execution-%03d", index), Provider: "openai", Model: fmt.Sprintf("model-%d", index%2), AuthID: fmt.Sprintf("auth-%d", index%2), RequestedAt: base.Add(time.Duration(index) * time.Second), Endpoint: "POST /v1/responses", Failed: index%10 == 0, Detail: UsageDetail{TotalTokens: int64(index + 1)}}
|
||||
record := UsageRecord{Provider: "openai", Model: fmt.Sprintf("model-%d", index%2), AuthID: fmt.Sprintf("auth-%d", index%2), RequestedAt: base.Add(time.Duration(index) * time.Second), Failed: index%10 == 0, Detail: UsageDetail{TotalTokens: int64(index + 1)}}
|
||||
raw, _ := json.Marshal(record)
|
||||
if _, err := app.HandleMethod(MethodUsageHandle, raw); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -153,7 +153,7 @@ func TestFastRequestCanUseStandardPricing(t *testing.T) {
|
||||
if response := managementCallBody(t, app, http.MethodPut, managementBase+routePrices, priceBody); response.StatusCode != http.StatusOK {
|
||||
t.Fatalf("put price status = %d, body = %s", response.StatusCode, response.Body)
|
||||
}
|
||||
record := UsageRecord{Provider: "codex", Model: "gpt-5.6-sol", Speed: "fast", RequestedAt: time.Now(), Detail: UsageDetail{InputTokens: 100_000, TotalTokens: 100_000}}
|
||||
record := UsageRecord{Provider: "codex", Model: "gpt-5.6-sol", ServiceTier: "fast", RequestedAt: time.Now(), Detail: UsageDetail{InputTokens: 100_000, TotalTokens: 100_000}}
|
||||
raw, _ := json.Marshal(record)
|
||||
if _, err := app.HandleMethod(MethodUsageHandle, raw); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -209,9 +209,9 @@ func TestCompactUsageIsPricedAndUsesJSONMetrics(t *testing.T) {
|
||||
if response := managementCallBody(t, app, http.MethodPut, managementBase+routePrices, priceBody); response.StatusCode != http.StatusOK {
|
||||
t.Fatalf("put price status = %d, body = %s", response.StatusCode, response.Body)
|
||||
}
|
||||
startedAt := time.Now()
|
||||
record := UsageRecord{
|
||||
RequestID: "compact-success", ExecutionID: "compact-attempt", Provider: "openai",
|
||||
Model: "deepseek-v4-flash", Endpoint: "POST /v1/responses/compact", RequestedAt: time.Now(),
|
||||
Provider: "openai", Model: "deepseek-v4-flash", RequestedAt: startedAt,
|
||||
Latency: time.Second, TTFT: 200 * time.Millisecond,
|
||||
Detail: UsageDetail{InputTokens: 1_000, OutputTokens: 100, TotalTokens: 1_100},
|
||||
}
|
||||
@@ -219,6 +219,14 @@ func TestCompactUsageIsPricedAndUsesJSONMetrics(t *testing.T) {
|
||||
if _, err := app.HandleMethod(MethodUsageHandle, raw); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
completionRaw, _ := json.Marshal(RequestCompletion{
|
||||
RequestID: "compact-success", Model: "deepseek-v4-flash", Outcome: RequestCompletionSucceeded,
|
||||
StartedAt: startedAt, CompletedAt: startedAt.Add(time.Second),
|
||||
Metadata: map[string]any{"request_path": "/v1/responses/compact"},
|
||||
})
|
||||
if _, err := app.HandleMethod(MethodRequestComplete, completionRaw); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
response := managementCall(t, app, http.MethodGet, managementBase+routeUsage)
|
||||
var payload usageListResponse
|
||||
if err := json.Unmarshal(response.Body, &payload); err != nil {
|
||||
@@ -228,7 +236,7 @@ func TestCompactUsageIsPricedAndUsesJSONMetrics(t *testing.T) {
|
||||
t.Fatalf("records = %d, want 1", len(payload.Records))
|
||||
}
|
||||
got := payload.Records[0]
|
||||
if got.RequestType != "JSON" || got.Endpoint != "POST /v1/responses/compact" || got.TTFTMilliseconds != 0 || got.SpeedTPS != nil {
|
||||
if got.RequestType != "JSON" || got.Endpoint != "/v1/responses/compact" || got.TTFTMilliseconds != 0 || got.SpeedTPS != nil {
|
||||
t.Fatalf("unexpected compact metadata: %+v", got)
|
||||
}
|
||||
if !got.CostAvailable || got.CostUSD == nil || *got.CostUSD != 0.004 {
|
||||
@@ -277,18 +285,15 @@ func TestCompactFailureAndCancellationRemainVisible(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestUsageAndLifecycleMergeInEitherOrderAndDeduplicateExecution(t *testing.T) {
|
||||
func TestUsageAndLifecycleMergeInEitherOrderAndDeduplicateUsage(t *testing.T) {
|
||||
for _, usageFirst := range []bool{true, false} {
|
||||
app := NewApp()
|
||||
if _, err := app.HandleMethod(MethodPluginRegister, lifecycleRequest(t, SchemaVersion, testConfig(t, "enabled: true\ncodex_only: false\n"))); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
startedAt := time.Date(2026, 8, 14, 12, 0, 0, 0, time.UTC)
|
||||
usage := UsageRecord{
|
||||
RequestID: "request-1", ExecutionID: "attempt-1", TraceID: "trace-1",
|
||||
Provider: "openai", Model: "deepseek-v4-flash", RequestedAt: startedAt,
|
||||
Endpoint: "POST /v1/responses", Detail: UsageDetail{InputTokens: 10, TotalTokens: 10},
|
||||
}
|
||||
usage := UsageRecord{Provider: "openai", Model: "deepseek-v4-flash", RequestedAt: startedAt,
|
||||
Detail: UsageDetail{InputTokens: 10, TotalTokens: 10}}
|
||||
completion := RequestCompletion{
|
||||
RequestID: "request-1", TraceID: "trace-1", Model: "deepseek-v4-flash",
|
||||
Outcome: RequestCompletionFailed, StatusCode: 500, Error: "upstream failed",
|
||||
@@ -317,13 +322,13 @@ func TestUsageAndLifecycleMergeInEitherOrderAndDeduplicateExecution(t *testing.T
|
||||
if err := json.Unmarshal(response.Body, &payload); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(payload.Records) != 1 || payload.Records[0].ExecutionID != "attempt-1" || payload.Records[0].Outcome != "failed" || payload.Records[0].StatusCode != 500 {
|
||||
if len(payload.Records) != 1 || payload.Records[0].RequestID != "request-1" || payload.Records[0].ExecutionID != "" || payload.Records[0].Outcome != "failed" || payload.Records[0].StatusCode != 500 {
|
||||
t.Fatalf("usageFirst=%v records=%+v", usageFirst, payload.Records)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestRetriedExecutionsRemainSeparate(t *testing.T) {
|
||||
func TestDistinctContractUsageEventsRemainSeparate(t *testing.T) {
|
||||
app := NewApp()
|
||||
if _, err := app.HandleMethod(MethodPluginRegister, lifecycleRequest(t, SchemaVersion, testConfig(t, "enabled: true\ncodex_only: false\n"))); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -331,7 +336,6 @@ func TestRetriedExecutionsRemainSeparate(t *testing.T) {
|
||||
startedAt := time.Date(2026, 8, 14, 12, 0, 0, 0, time.UTC)
|
||||
for index, failed := range []bool{true, false} {
|
||||
usage := UsageRecord{
|
||||
RequestID: "request-retry", ExecutionID: fmt.Sprintf("attempt-%d", index+1),
|
||||
Provider: "openai", Model: "deepseek-v4-flash", RequestedAt: startedAt.Add(time.Duration(index) * time.Second),
|
||||
Failed: failed, Detail: UsageDetail{InputTokens: int64(10 + index), TotalTokens: int64(10 + index)},
|
||||
}
|
||||
@@ -340,7 +344,7 @@ func TestRetriedExecutionsRemainSeparate(t *testing.T) {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
completion := RequestCompletion{RequestID: "request-retry", Outcome: RequestCompletionSucceeded, StartedAt: startedAt, CompletedAt: startedAt.Add(2 * time.Second)}
|
||||
completion := RequestCompletion{RequestID: "request-retry", Model: "deepseek-v4-flash", Outcome: RequestCompletionSucceeded, StartedAt: startedAt, CompletedAt: startedAt.Add(2 * time.Second)}
|
||||
raw, _ := json.Marshal(completion)
|
||||
if _, err := app.HandleMethod(MethodRequestComplete, raw); err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -350,8 +354,8 @@ func TestRetriedExecutionsRemainSeparate(t *testing.T) {
|
||||
if err := json.Unmarshal(response.Body, &payload); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(payload.Records) != 2 || payload.Records[0].ExecutionID != "attempt-2" || payload.Records[0].Failed || payload.Records[1].ExecutionID != "attempt-1" || !payload.Records[1].Failed {
|
||||
t.Fatalf("unexpected retry records: %+v", payload.Records)
|
||||
if len(payload.Records) != 2 || payload.Records[0].ExecutionID != "" || payload.Records[0].RequestID != "" || payload.Records[0].Failed || payload.Records[1].ExecutionID != "" || payload.Records[1].RequestID != "request-retry" || !payload.Records[1].Failed {
|
||||
t.Fatalf("unexpected usage records: %+v", payload.Records)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -360,6 +364,7 @@ func TestUsageManagementResponseContainsDisplayFields(t *testing.T) {
|
||||
if _, err := app.HandleMethod(MethodPluginRegister, lifecycleRequest(t, SchemaVersion, testConfig(t, "enabled: true\ncodex_only: true\n"))); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
startedAt := time.Date(2026, 8, 14, 12, 0, 0, 0, time.UTC)
|
||||
record := UsageRecord{
|
||||
Provider: "codex",
|
||||
APIKey: "test-key",
|
||||
@@ -367,9 +372,7 @@ func TestUsageManagementResponseContainsDisplayFields(t *testing.T) {
|
||||
ReasoningEffort: "high",
|
||||
ServiceTier: "priority",
|
||||
ExecutorType: "CodexExecutor",
|
||||
Endpoint: "POST /v1/responses",
|
||||
ClientIP: "192.0.2.10",
|
||||
RequestedAt: time.Date(2026, 8, 14, 12, 0, 0, 0, time.UTC),
|
||||
RequestedAt: startedAt,
|
||||
Latency: 1500 * time.Millisecond,
|
||||
TTFT: 250 * time.Millisecond,
|
||||
Detail: UsageDetail{
|
||||
@@ -386,6 +389,14 @@ func TestUsageManagementResponseContainsDisplayFields(t *testing.T) {
|
||||
if _, err := app.HandleMethod(MethodUsageHandle, raw); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
completionRaw, _ := json.Marshal(RequestCompletion{
|
||||
RequestID: "request-display", TraceID: "trace-display", Model: "gpt-5.5", Stream: true,
|
||||
Outcome: RequestCompletionSucceeded, StartedAt: startedAt, CompletedAt: startedAt.Add(1500 * time.Millisecond),
|
||||
Metadata: map[string]any{"request_path": "/v1/responses"},
|
||||
})
|
||||
if _, err := app.HandleMethod(MethodRequestComplete, completionRaw); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
response := managementCall(t, app, http.MethodGet, managementBase+routeUsage)
|
||||
if response.StatusCode != http.StatusOK {
|
||||
@@ -402,7 +413,7 @@ func TestUsageManagementResponseContainsDisplayFields(t *testing.T) {
|
||||
if got.APIKey != "test-key" || got.Model != "gpt-5.5" || got.ReasoningEffort != "high" || got.ServiceTier != "priority" || got.ExecutorType != "CodexExecutor" {
|
||||
t.Fatalf("unexpected usage identity fields: %+v", got)
|
||||
}
|
||||
if got.RequestType != "SSE" || got.Endpoint != "POST /v1/responses" || got.ClientIP != "192.0.2.10" {
|
||||
if got.RequestID != "request-display" || got.ExecutionID != "" || got.TraceID != "trace-display" || got.RequestType != "SSE" || got.Endpoint != "/v1/responses" || got.ClientIP != "" {
|
||||
t.Fatalf("unexpected request metadata: %+v", got)
|
||||
}
|
||||
if got.TotalTokens != 15 || got.TTFTMilliseconds != 250 || got.CacheReadTokens != 4 || got.CacheWriteTokens != 1 {
|
||||
|
||||
@@ -227,12 +227,7 @@ type ManagementResponse struct {
|
||||
// target host because exported field names are part of the JSON wire contract.
|
||||
type UsageRecord struct {
|
||||
Provider string
|
||||
RequestID string
|
||||
ExecutionID string
|
||||
TraceID string
|
||||
ExecutorType string
|
||||
Endpoint string
|
||||
ClientIP string
|
||||
Model string
|
||||
Alias string
|
||||
APIKey string
|
||||
@@ -242,7 +237,6 @@ type UsageRecord struct {
|
||||
Source string
|
||||
ReasoningEffort string
|
||||
ServiceTier string
|
||||
Speed string
|
||||
Generate bool
|
||||
RequestedAt time.Time
|
||||
Latency time.Duration
|
||||
|
||||
@@ -150,7 +150,7 @@ func replaceManagedKeyModels(ctx context.Context, tx *sql.Tx, keyID string, mode
|
||||
}
|
||||
|
||||
func (r *SQLiteUsageRepository) ManagedKeyByCredential(ctx context.Context, credential string) (managedaccess.ManagedKey, error) {
|
||||
return r.scanManagedKey(ctx, `WHERE id=? OR secret=? LIMIT 1`, credential, credential)
|
||||
return r.scanManagedKey(ctx, `WHERE secret=? LIMIT 1`, credential)
|
||||
}
|
||||
|
||||
func (r *SQLiteUsageRepository) ManagedKeyByID(ctx context.Context, id string) (managedaccess.ManagedKey, error) {
|
||||
@@ -173,10 +173,6 @@ func (r *SQLiteUsageRepository) ManagedKeyByReference(ctx context.Context, refer
|
||||
return managedaccess.ManagedKey{}, ErrManagedKeyNotFound
|
||||
}
|
||||
|
||||
func (r *SQLiteUsageRepository) ManagedKeyByName(ctx context.Context, name string) (managedaccess.ManagedKey, error) {
|
||||
return r.scanManagedKey(ctx, `WHERE name=? COLLATE NOCASE LIMIT 1`, name)
|
||||
}
|
||||
|
||||
func (r *SQLiteUsageRepository) scanManagedKey(ctx context.Context, where string, args ...any) (managedaccess.ManagedKey, error) {
|
||||
var key managedaccess.ManagedKey
|
||||
var createdAt, updatedAt string
|
||||
|
||||
+8
-3
@@ -6,9 +6,14 @@ bash "$root/scripts/check-env.sh"
|
||||
|
||||
cd "$root"
|
||||
mkdir -p bin
|
||||
gofmt -w cmd internal
|
||||
go mod tidy
|
||||
go test ./...
|
||||
unformatted="$(gofmt -l cmd internal)"
|
||||
if [[ -n "$unformatted" ]]; then
|
||||
echo '以下 Go 文件尚未格式化:' >&2
|
||||
echo "$unformatted" >&2
|
||||
exit 1
|
||||
fi
|
||||
go list -mod=readonly ./... >/dev/null
|
||||
CGO_ENABLED=1 go test ./...
|
||||
CGO_ENABLED=1 go build -tags cshared -buildmode=c-shared -o bin/cpa-ext.so ./cmd/cpa-ext
|
||||
|
||||
echo "Built: $root/bin/cpa-ext.so"
|
||||
|
||||
@@ -3,8 +3,9 @@ set -euo pipefail
|
||||
|
||||
root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
|
||||
runtime="$root/.runtime"
|
||||
pid_file="$runtime/cliproxy.pid"
|
||||
log_file="$runtime/cliproxy.log"
|
||||
binary="$runtime/bin/cli-proxy-api"
|
||||
pid_file="$runtime/cpa.pid"
|
||||
log_file="$runtime/logs/cpa-console.log"
|
||||
|
||||
if [[ -f "$pid_file" ]]; then
|
||||
old_pid="$(cat "$pid_file" || true)"
|
||||
@@ -14,9 +15,15 @@ if [[ -f "$pid_file" ]]; then
|
||||
fi
|
||||
fi
|
||||
|
||||
mkdir -p "$runtime/auths" "$runtime/plugins"
|
||||
if [[ ! -x "$binary" ]]; then
|
||||
echo "CPA test host binary not found: $binary" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
mkdir -p "$runtime/auths" "$runtime/plugins" "$runtime/logs"
|
||||
: >"$log_file"
|
||||
nohup "$runtime/cliproxy" -config "$runtime/config.yaml" \
|
||||
cd "$runtime"
|
||||
nohup "$binary" --config "$runtime/config.yaml" \
|
||||
>"$log_file" 2>&1 </dev/null &
|
||||
pid=$!
|
||||
echo "$pid" >"$pid_file"
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
set -euo pipefail
|
||||
|
||||
root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
|
||||
pid_file="$root/.runtime/cliproxy.pid"
|
||||
pid_file="$root/.runtime/cpa.pid"
|
||||
|
||||
if [[ ! -f "$pid_file" ]]; then
|
||||
echo 'CPA test host is not running.'
|
||||
|
||||
Reference in New Issue
Block a user