fix: 修复 CPA 用量与请求终态关联
This commit is contained in:
+132
@@ -0,0 +1,132 @@
|
||||
# CLIProxyAPI 宿主补丁
|
||||
|
||||
这里保存 billing 在 CLIProxyAPI `v7.2.132` 上需要的三项宿主修复。三项补丁都不改变计费规则或数据库结构,必须按下列顺序应用:
|
||||
|
||||
1. [`cli-proxy-api-usage-context.patch`](cli-proxy-api-usage-context.patch)
|
||||
2. [`cli-proxy-api-usage-identity.patch`](cli-proxy-api-usage-identity.patch)
|
||||
3. [`cli-proxy-api-request-lifecycle-cancel.patch`](cli-proxy-api-request-lifecycle-cancel.patch)
|
||||
|
||||
适用基线:
|
||||
|
||||
- CLIProxyAPI:`v7.2.132`
|
||||
- 提交:`78f0c4079e3e6273d65d03b5549cffc898703264`
|
||||
- Native ABI:`1`
|
||||
- 第二项补丁把 RPC schema 从 `3` 提升到 `4`
|
||||
|
||||
升级 CLIProxyAPI 后不要直接假设补丁仍然需要或仍能应用。应先在新版本复测,再根据新版本的宿主契约重新核对。
|
||||
|
||||
## 1. 已取消上下文导致动态插件漏记
|
||||
|
||||
使用 `/v1/responses` 且请求体为 `"stream": false` 时,上游请求能够成功,CLIProxyAPI 内置用量统计也会增加,但动态用量插件可能完全收不到 `usage.handle`:
|
||||
|
||||
- billing 没有新增用量事实;
|
||||
- 用户余额没有扣减;
|
||||
- 内置统计和动态插件统计不一致。
|
||||
|
||||
CLIProxyAPI 会异步分发最终用量记录,但分发任务沿用了原始 HTTP 请求上下文。非流式响应很快结束后,该上下文会被取消;动态插件加载器在调用 C ABI 前检查 `ctx.Done()`,于是跳过插件调用。内置插件不经过动态 C ABI 检查,所以仍能收到记录。
|
||||
|
||||
第一项补丁在动态用量适配器入口使用 `context.WithoutCancel`:
|
||||
|
||||
- 保留上下文值;
|
||||
- 不再继承已经结束的 HTTP 请求的取消信号和截止时间;
|
||||
- 每个最终用量仍只调用一次插件,不增加逐块回调;
|
||||
- `nil` 上下文退回 `context.Background()`。
|
||||
|
||||
## 2. Usage 缺少请求身份导致并发拆行
|
||||
|
||||
CLIProxyAPI 的请求生命周期会生成唯一 Request ID,并通过 `request.complete` 发送给插件,但 schema v3 的 `usage.handle` 没有携带该身份。billing 只能按用户、模型和时间窗口关联两类事实。
|
||||
|
||||
串行请求通常可以唯一匹配;同用户、同模型高并发时会出现多个等价候选。billing 为避免错配,保守地保留一条 request-only 和一条 usage-only,因此请求明细数量可能翻倍,但底层用量和扣费并未重复。
|
||||
|
||||
第二项补丁引入 RPC schema v4:
|
||||
|
||||
- 请求生命周期创建后,把 Request ID 和父 Trace ID 写入实际执行上下文;
|
||||
- `usage.handle` 增加可选的 `RequestID` 与 `TraceID`;
|
||||
- 只有协商到 schema v4 的插件收到新增字段;
|
||||
- schema v1–v3 插件保持原有载荷,能够继续加载;
|
||||
- billing v4 按 Request ID 精确关联,旧宿主和历史数据仍走保守兼容逻辑。
|
||||
|
||||
补丁不虚构 Execution ID。一次生命周期内可能存在多个上游尝试时,只有宿主能够提供真实的逐尝试身份;当前修复只传递已经存在且语义明确的生命周期 Request ID。
|
||||
|
||||
## 3. SSE 启动阶段取消可能缺少请求终态
|
||||
|
||||
billing 在请求拦截阶段创建并发占用,依赖 CPA 后续发送唯一的 `request.complete` 释放占用。原实现只在同步执行启动完成后,由流消费协程观察执行上下文的取消并发送终态。
|
||||
|
||||
如果执行上下文恰好在上游连接、鉴权选择或首段流初始化期间取消,流消费协程可能尚未建立。此时 `request.complete` 必须等待同步启动调用返回;若自定义执行器或异常上游没有及时响应取消,就会表现为并发槽位长期占用。内置 Codex 执行器通常会随 context 立即返回,所以这是生命周期层的防御性加固,不是内置执行器的稳定必现缺陷。
|
||||
|
||||
第三项补丁在请求生命周期创建时直接注册上下文取消回调:
|
||||
|
||||
- 不依赖执行器或流转发协程是否已经启动;
|
||||
- 取消后立即发送 `outcome=canceled`、`status_code=0` 的终态;
|
||||
- 正常完成时主动撤销取消回调;
|
||||
- 与原有完成路径共用 `sync.Once`,两条路径竞态时仍只发送一个 `request.complete`;
|
||||
- 不增加轮询、定时器或逐流块处理。
|
||||
|
||||
该补丁只能处理已经到达 CPA 执行上下文的取消信号。Windows 客户端通过 WSL localhost 转发访问 WSL 内 CPA 时,客户端断开可能不会立即传递到后端连接;这种情况下 CPA 看不到取消,任何生命周期回调都无法提前触发。生产 Linux 直连和 WSL 内直连不经过该转发层。
|
||||
|
||||
## 应用
|
||||
|
||||
在干净的 CLIProxyAPI 工作区执行:
|
||||
|
||||
```bash
|
||||
git checkout 78f0c4079e3e6273d65d03b5549cffc898703264
|
||||
git apply --check /path/to/billing/patch/cli-proxy-api-usage-context.patch
|
||||
git apply /path/to/billing/patch/cli-proxy-api-usage-context.patch
|
||||
git apply --check /path/to/billing/patch/cli-proxy-api-usage-identity.patch
|
||||
git apply /path/to/billing/patch/cli-proxy-api-usage-identity.patch
|
||||
git apply --check /path/to/billing/patch/cli-proxy-api-request-lifecycle-cancel.patch
|
||||
git apply /path/to/billing/patch/cli-proxy-api-request-lifecycle-cancel.patch
|
||||
gofmt -w internal/pluginhost sdk/api/handlers sdk/cliproxy/usage sdk/pluginabi sdk/pluginapi
|
||||
go test -race ./internal/pluginhost ./sdk/api/handlers ./sdk/cliproxy/usage -count=1
|
||||
CGO_ENABLED=1 go build -o test-output ./cmd/server
|
||||
rm test-output
|
||||
```
|
||||
|
||||
billing 必须使用 RPC schema v4 重新构建;旧 billing 动态库会向下协商到旧 schema,无法获得 Usage 请求身份。
|
||||
|
||||
## 验证标准
|
||||
|
||||
运行时至少交叉核对:
|
||||
|
||||
1. 分别发送一次非流式和一次 SSE 请求;
|
||||
2. 两次请求均只产生一条明细和一次扣费;
|
||||
3. 并发发送相同用户、相同模型的混合请求;
|
||||
4. N 个 HTTP 200 最终对应 N 条明细、N 个唯一 Request ID、N 条底层 Usage 和 N 次扣费;
|
||||
5. CLIProxyAPI 内置队列、billing 和独立 keeper 的增量一致;
|
||||
6. 等待异步任务稳定后没有迟到重复;
|
||||
7. 请求正常结束后并发占用回到零;
|
||||
8. 在上游启动阶段和已开始传输后分别强制中断 SSE,两次都只产生一个 canceled 终态且并发占用回到零;
|
||||
9. 日志中没有插件错误或 panic。
|
||||
|
||||
## 已验证结果
|
||||
|
||||
三个补丁与 billing schema v4 使用真实 DeepSeek 上游完成了本地验证:
|
||||
|
||||
- `cpads` 有 Key 成功、无 Key 返回 401;
|
||||
- 非流式和 SSE 各一次:2 条明细、2 次扣费,等待 4 秒无重复;
|
||||
- 12 个混合并发请求:12 次 HTTP 200、12 条合并明细、12 个唯一 Request ID、12 次扣费;
|
||||
- 流式与非流式各 6 条,终态、状态码、Token 和 Trace ID 完整;
|
||||
- keeper 同步增加 12 条,等待 5 秒没有迟到重复;
|
||||
- WSL 内直连分别在 50ms(首字节前)和 1s(流传输中)强制断开:两次均只有一条 canceled 明细,并发占用立即回到零;
|
||||
- 仅前两项补丁与三项补丁在内置 Codex 执行器上的直连取消 A/B 都能正常终止;第三项补丁额外覆盖执行器启动阻塞且未响应 context 的单元场景;
|
||||
- 请求正常结束后 `active_requests=0`,空闲 5 秒没有持续 CPU 占用;
|
||||
- billing 全部测试、billing race 测试及 CLIProxyAPI 相关包 race 测试通过;
|
||||
- schema v3 对照插件仍能正常注册,新增字段不会发送给旧 schema。
|
||||
|
||||
CLIProxyAPI 全仓 `go test ./...` 在该基线自身仍有 5 个无关失败:3 个 Claude OS 指纹用例在 WSL 中得到 `MacOS` 而期望 `Linux`,以及 2 个 `reviewedInPlaceByteWrites` 清单陈旧项。补丁涉及的包及其 race 测试均通过。
|
||||
|
||||
## 性能影响
|
||||
|
||||
三项修复都不增加队列、重试或流式分块回调。每个最终用量和请求终态仍最多调用插件一次;新增工作是创建轻量上下文包装、在 schema v4 JSON 中增加两个短字符串,以及为活跃请求注册一个随正常完成立即撤销的标准库取消回调。并发验证中没有观察到持续 CPU 或内存增长。
|
||||
|
||||
## 回滚
|
||||
|
||||
源码必须按应用顺序的反方向回滚:
|
||||
|
||||
```bash
|
||||
git apply -R /path/to/billing/patch/cli-proxy-api-request-lifecycle-cancel.patch
|
||||
git apply -R /path/to/billing/patch/cli-proxy-api-usage-identity.patch
|
||||
git apply -R /path/to/billing/patch/cli-proxy-api-usage-context.patch
|
||||
```
|
||||
|
||||
部署回滚应恢复升级前保存的 CLIProxyAPI 和 billing 动态库后重启宿主。数据库不需要回滚;schema v4 只让新记录获得精确身份,不改变历史账目。
|
||||
@@ -0,0 +1,160 @@
|
||||
diff --git a/sdk/api/handlers/handlers_interceptors.go b/sdk/api/handlers/handlers_interceptors.go
|
||||
--- a/sdk/api/handlers/handlers_interceptors.go
|
||||
+++ b/sdk/api/handlers/handlers_interceptors.go
|
||||
@@ -1,6 +1,7 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
+ stdcontext "context"
|
||||
"net/http"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -66,6 +67,9 @@ type requestLifecycleSkipHost interface {
|
||||
|
||||
type requestLifecycleTracker struct {
|
||||
once sync.Once
|
||||
+ cancelMu sync.Mutex
|
||||
+ cancelStop func() bool
|
||||
+ completed bool
|
||||
ctx context.Context
|
||||
host PluginInterceptorHost
|
||||
skipPluginID string
|
||||
@@ -76,7 +80,7 @@ func (h *BaseAPIHandler) newRequestLifecycleTracker(ctx context.Context, sourceF
|
||||
requestID := uuid.NewString()
|
||||
traceID := logging.GetRequestID(ctx)
|
||||
ctx = coreusage.WithRequestIdentity(ctx, requestID, traceID)
|
||||
- return &requestLifecycleTracker{
|
||||
+ tracker := &requestLifecycleTracker{
|
||||
ctx: ctx,
|
||||
host: h.interceptorHost(),
|
||||
skipPluginID: skipPluginID,
|
||||
@@ -91,6 +95,39 @@ func (h *BaseAPIHandler) newRequestLifecycleTracker(ctx context.Context, sourceF
|
||||
Metadata: metadata,
|
||||
},
|
||||
}
|
||||
+ tracker.watchCancellation(ctx)
|
||||
+ return tracker
|
||||
+}
|
||||
+
|
||||
+func (t *requestLifecycleTracker) watchCancellation(ctx context.Context) {
|
||||
+ if t == nil || ctx == nil || ctx.Done() == nil {
|
||||
+ return
|
||||
+ }
|
||||
+ stop := stdcontext.AfterFunc(ctx, func() {
|
||||
+ t.complete(pluginapi.RequestCompletionCanceled, 0, ctx.Err())
|
||||
+ })
|
||||
+ t.cancelMu.Lock()
|
||||
+ if t.completed {
|
||||
+ t.cancelMu.Unlock()
|
||||
+ stop()
|
||||
+ return
|
||||
+ }
|
||||
+ t.cancelStop = stop
|
||||
+ t.cancelMu.Unlock()
|
||||
+}
|
||||
+
|
||||
+func (t *requestLifecycleTracker) stopCancellationWatch() {
|
||||
+ if t == nil {
|
||||
+ return
|
||||
+ }
|
||||
+ t.cancelMu.Lock()
|
||||
+ t.completed = true
|
||||
+ stop := t.cancelStop
|
||||
+ t.cancelStop = nil
|
||||
+ t.cancelMu.Unlock()
|
||||
+ if stop != nil {
|
||||
+ stop()
|
||||
+ }
|
||||
}
|
||||
|
||||
func (t *requestLifecycleTracker) executionContext() context.Context {
|
||||
@@ -112,6 +149,7 @@ func (t *requestLifecycleTracker) complete(outcome pluginapi.RequestCompletionOu
|
||||
return
|
||||
}
|
||||
t.once.Do(func() {
|
||||
+ t.stopCancellationWatch()
|
||||
completion := t.completion
|
||||
completion.Outcome = outcome
|
||||
completion.StatusCode = statusCode
|
||||
diff --git a/sdk/api/handlers/handlers_interceptors_test.go b/sdk/api/handlers/handlers_interceptors_test.go
|
||||
--- a/sdk/api/handlers/handlers_interceptors_test.go
|
||||
+++ b/sdk/api/handlers/handlers_interceptors_test.go
|
||||
@@ -516,6 +516,78 @@ func TestHandlerLifecycleCompletesCanceledStream(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
+func TestHandlerLifecycleCompletesCanceledStreamWhileExecutorIsStarting(t *testing.T) {
|
||||
+ model := "handler-interceptor-lifecycle-canceled-before-stream-start"
|
||||
+ executorStarted := make(chan struct{})
|
||||
+ releaseExecutor := make(chan struct{})
|
||||
+ executor := &interceptorCaptureExecutor{
|
||||
+ stream: func(context.Context, *coreauth.Auth, coreexecutor.Request, coreexecutor.Options) (*coreexecutor.StreamResult, error) {
|
||||
+ close(executorStarted)
|
||||
+ <-releaseExecutor
|
||||
+ chunks := make(chan coreexecutor.StreamChunk)
|
||||
+ close(chunks)
|
||||
+ return &coreexecutor.StreamResult{Chunks: chunks}, nil
|
||||
+ },
|
||||
+ }
|
||||
+ handler := newInterceptorHandler(t, model, executor, &sdkconfig.SDKConfig{})
|
||||
+ completions := make(chan pluginapi.RequestCompletion, 2)
|
||||
+ handler.SetPluginHost(&handlerInterceptorTestHost{
|
||||
+ completeRequest: func(_ context.Context, completion pluginapi.RequestCompletion) {
|
||||
+ completions <- completion
|
||||
+ },
|
||||
+ })
|
||||
+
|
||||
+ ctx, cancel := context.WithCancel(context.Background())
|
||||
+ type streamResult struct {
|
||||
+ data <-chan []byte
|
||||
+ errs <-chan *interfaces.ErrorMessage
|
||||
+ }
|
||||
+ result := make(chan streamResult, 1)
|
||||
+ go func() {
|
||||
+ data, _, errs := handler.ExecuteStreamWithAuthManager(ctx, "openai", model, []byte(`{"model":"`+model+`","stream":true}`), "")
|
||||
+ result <- streamResult{data: data, errs: errs}
|
||||
+ }()
|
||||
+
|
||||
+ select {
|
||||
+ case <-executorStarted:
|
||||
+ case <-time.After(time.Second):
|
||||
+ close(releaseExecutor)
|
||||
+ t.Fatal("executor did not start")
|
||||
+ }
|
||||
+ cancel()
|
||||
+
|
||||
+ select {
|
||||
+ case completion := <-completions:
|
||||
+ if completion.Outcome != pluginapi.RequestCompletionCanceled || completion.StatusCode != 0 || completion.Error == "" {
|
||||
+ close(releaseExecutor)
|
||||
+ t.Fatalf("completion = %#v", completion)
|
||||
+ }
|
||||
+ case <-time.After(time.Second):
|
||||
+ close(releaseExecutor)
|
||||
+ t.Fatal("missing cancellation completion while executor startup was blocked")
|
||||
+ }
|
||||
+
|
||||
+ close(releaseExecutor)
|
||||
+ channels := <-result
|
||||
+ for channels.data != nil || channels.errs != nil {
|
||||
+ select {
|
||||
+ case _, ok := <-channels.data:
|
||||
+ if !ok {
|
||||
+ channels.data = nil
|
||||
+ }
|
||||
+ case _, ok := <-channels.errs:
|
||||
+ if !ok {
|
||||
+ channels.errs = nil
|
||||
+ }
|
||||
+ }
|
||||
+ }
|
||||
+ select {
|
||||
+ case duplicate := <-completions:
|
||||
+ t.Fatalf("duplicate stream completion = %#v", duplicate)
|
||||
+ default:
|
||||
+ }
|
||||
+}
|
||||
+
|
||||
func TestHandlerRequestInterceptorRewritesExecutorRequest(t *testing.T) {
|
||||
model := "handler-interceptor-request-model"
|
||||
executor := &interceptorCaptureExecutor{}
|
||||
@@ -0,0 +1,60 @@
|
||||
diff --git a/internal/pluginhost/adapters_test.go b/internal/pluginhost/adapters_test.go
|
||||
index de62918e..302b1f38 100644
|
||||
--- a/internal/pluginhost/adapters_test.go
|
||||
+++ b/internal/pluginhost/adapters_test.go
|
||||
@@ -2234,6 +2234,39 @@ func TestUsageAdapterPanicFusesPlugin(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
+func TestUsageAdapterDetachesCanceledRequestContext(t *testing.T) {
|
||||
+ type contextKey struct{}
|
||||
+ const contextValue = "usage-context-value"
|
||||
+
|
||||
+ called := 0
|
||||
+ plugin := usagePluginFunc(func(ctx context.Context, record pluginapi.UsageRecord) {
|
||||
+ called++
|
||||
+ if errContext := ctx.Err(); errContext != nil {
|
||||
+ t.Fatalf("usage context error = %v, want nil", errContext)
|
||||
+ }
|
||||
+ if value := ctx.Value(contextKey{}); value != contextValue {
|
||||
+ t.Fatalf("usage context value = %v, want %q", value, contextValue)
|
||||
+ }
|
||||
+ })
|
||||
+ host := newHostWithRecords(capabilityRecord{
|
||||
+ id: "usage-canceled-context",
|
||||
+ plugin: pluginapi.Plugin{Capabilities: pluginapi.Capabilities{
|
||||
+ UsagePlugin: plugin,
|
||||
+ }},
|
||||
+ })
|
||||
+ adapter := &usageAdapter{
|
||||
+ host: host,
|
||||
+ pluginID: "usage-canceled-context",
|
||||
+ }
|
||||
+
|
||||
+ ctx, cancel := context.WithCancel(context.WithValue(context.Background(), contextKey{}, contextValue))
|
||||
+ cancel()
|
||||
+ adapter.HandleUsage(ctx, coreusage.Record{Provider: "plugin-provider"})
|
||||
+ if called != 1 {
|
||||
+ t.Fatalf("usage plugin calls = %d, want 1", called)
|
||||
+ }
|
||||
+}
|
||||
+
|
||||
func TestUsageAdapterNormalizesOmittedGenerateToTrue(t *testing.T) {
|
||||
var gotGenerate bool
|
||||
plugin := usagePluginFunc(func(ctx context.Context, record pluginapi.UsageRecord) {
|
||||
diff --git a/internal/pluginhost/adapters_usage_translation.go b/internal/pluginhost/adapters_usage_translation.go
|
||||
index 2201eb6c..93a4c8a5 100644
|
||||
--- a/internal/pluginhost/adapters_usage_translation.go
|
||||
+++ b/internal/pluginhost/adapters_usage_translation.go
|
||||
@@ -132,6 +132,11 @@ func (a *usageAdapter) HandleUsage(ctx context.Context, record coreusage.Record)
|
||||
if a == nil {
|
||||
return
|
||||
}
|
||||
+ if ctx == nil {
|
||||
+ ctx = context.Background()
|
||||
+ } else {
|
||||
+ ctx = context.WithoutCancel(ctx)
|
||||
+ }
|
||||
plugin := a.host.currentUsagePlugin(a.pluginID)
|
||||
if plugin == nil {
|
||||
return
|
||||
@@ -0,0 +1,348 @@
|
||||
diff --git a/internal/pluginhost/adapters_test.go b/internal/pluginhost/adapters_test.go
|
||||
index 302b1f38..610a3753 100644
|
||||
--- a/internal/pluginhost/adapters_test.go
|
||||
+++ b/internal/pluginhost/adapters_test.go
|
||||
@@ -2266,6 +2266,54 @@ func TestUsageAdapterDetachesCanceledRequestContext(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
+func TestUsageAdapterIncludesRequestIdentityForSchemaFour(t *testing.T) {
|
||||
+ var got pluginapi.UsageRecord
|
||||
+ plugin := usagePluginFunc(func(ctx context.Context, record pluginapi.UsageRecord) {
|
||||
+ got = record
|
||||
+ })
|
||||
+ host := newHostWithRecords(capabilityRecord{
|
||||
+ id: "usage-identity",
|
||||
+ plugin: pluginapi.Plugin{SchemaVersion: pluginabi.SchemaVersionUsageIdentity, Capabilities: pluginapi.Capabilities{
|
||||
+ UsagePlugin: plugin,
|
||||
+ }},
|
||||
+ })
|
||||
+ adapter := &usageAdapter{
|
||||
+ host: host,
|
||||
+ pluginID: "usage-identity",
|
||||
+ schemaVersion: pluginabi.SchemaVersionUsageIdentity,
|
||||
+ }
|
||||
+
|
||||
+ ctx := coreusage.WithRequestIdentity(context.Background(), "request-1", "trace-1")
|
||||
+ adapter.HandleUsage(ctx, coreusage.Record{Provider: "provider"})
|
||||
+ if got.RequestID != "request-1" || got.TraceID != "trace-1" {
|
||||
+ t.Fatalf("usage identity = %q, %q", got.RequestID, got.TraceID)
|
||||
+ }
|
||||
+}
|
||||
+
|
||||
+func TestUsageAdapterOmitsRequestIdentityForLegacySchema(t *testing.T) {
|
||||
+ var got pluginapi.UsageRecord
|
||||
+ plugin := usagePluginFunc(func(ctx context.Context, record pluginapi.UsageRecord) {
|
||||
+ got = record
|
||||
+ })
|
||||
+ host := newHostWithRecords(capabilityRecord{
|
||||
+ id: "usage-identity-legacy",
|
||||
+ plugin: pluginapi.Plugin{SchemaVersion: pluginabi.SchemaVersionUsageIdentity - 1, Capabilities: pluginapi.Capabilities{
|
||||
+ UsagePlugin: plugin,
|
||||
+ }},
|
||||
+ })
|
||||
+ adapter := &usageAdapter{
|
||||
+ host: host,
|
||||
+ pluginID: "usage-identity-legacy",
|
||||
+ schemaVersion: pluginabi.SchemaVersionUsageIdentity - 1,
|
||||
+ }
|
||||
+
|
||||
+ ctx := coreusage.WithRequestIdentity(context.Background(), "request-1", "trace-1")
|
||||
+ adapter.HandleUsage(ctx, coreusage.Record{Provider: "provider"})
|
||||
+ if got.RequestID != "" || got.TraceID != "" {
|
||||
+ t.Fatalf("legacy usage identity = %q, %q", got.RequestID, got.TraceID)
|
||||
+ }
|
||||
+}
|
||||
+
|
||||
func TestUsageAdapterNormalizesOmittedGenerateToTrue(t *testing.T) {
|
||||
var gotGenerate bool
|
||||
plugin := usagePluginFunc(func(ctx context.Context, record pluginapi.UsageRecord) {
|
||||
diff --git a/internal/pluginhost/adapters_usage_translation.go b/internal/pluginhost/adapters_usage_translation.go
|
||||
index 93a4c8a5..ad046780 100644
|
||||
--- a/internal/pluginhost/adapters_usage_translation.go
|
||||
+++ b/internal/pluginhost/adapters_usage_translation.go
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"github.com/router-for-me/CLIProxyAPI/v7/internal/registry"
|
||||
"github.com/router-for-me/CLIProxyAPI/v7/internal/thinking"
|
||||
coreusage "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/usage"
|
||||
+ "github.com/router-for-me/CLIProxyAPI/v7/sdk/pluginabi"
|
||||
"github.com/router-for-me/CLIProxyAPI/v7/sdk/pluginapi"
|
||||
sdktranslator "github.com/router-for-me/CLIProxyAPI/v7/sdk/translator"
|
||||
log "github.com/sirupsen/logrus"
|
||||
@@ -26,9 +27,10 @@ func (h *Host) RegisterUsagePlugins() {
|
||||
continue
|
||||
}
|
||||
coreusage.RegisterNamedPlugin("plugin:"+record.id, &usageAdapter{
|
||||
- host: h,
|
||||
- pluginID: record.id,
|
||||
- plugin: plugin,
|
||||
+ host: h,
|
||||
+ pluginID: record.id,
|
||||
+ plugin: plugin,
|
||||
+ schemaVersion: record.plugin.SchemaVersion,
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -114,9 +116,10 @@ func (h *Host) isPluginFused(id string) bool {
|
||||
}
|
||||
|
||||
type usageAdapter struct {
|
||||
- host *Host
|
||||
- pluginID string
|
||||
- plugin pluginapi.UsagePlugin
|
||||
+ host *Host
|
||||
+ pluginID string
|
||||
+ plugin pluginapi.UsagePlugin
|
||||
+ schemaVersion uint32
|
||||
}
|
||||
|
||||
type thinkingAdapter struct {
|
||||
@@ -146,7 +149,13 @@ func (a *usageAdapter) HandleUsage(ctx context.Context, record coreusage.Record)
|
||||
a.host.fusePlugin(a.pluginID, "UsagePlugin.HandleUsage", recovered)
|
||||
}
|
||||
}()
|
||||
+ requestID, traceID := "", ""
|
||||
+ if a.schemaVersion >= pluginabi.SchemaVersionUsageIdentity {
|
||||
+ requestID, traceID = coreusage.RequestIdentityFromContext(ctx)
|
||||
+ }
|
||||
plugin.HandleUsage(ctx, pluginapi.UsageRecord{
|
||||
+ RequestID: requestID,
|
||||
+ TraceID: traceID,
|
||||
Provider: record.Provider,
|
||||
ExecutorType: record.ExecutorType,
|
||||
Model: record.Model,
|
||||
diff --git a/sdk/api/handlers/handlers_execution.go b/sdk/api/handlers/handlers_execution.go
|
||||
index 994bd33a..3cc7743e 100644
|
||||
--- a/sdk/api/handlers/handlers_execution.go
|
||||
+++ b/sdk/api/handlers/handlers_execution.go
|
||||
@@ -71,6 +71,7 @@ func (h *BaseAPIHandler) executeWithAuthManagerFormats(ctx context.Context, entr
|
||||
}
|
||||
afterAuthCapture := &requestAfterAuthCapture{}
|
||||
lifecycle := h.newRequestLifecycleTracker(ctx, entryProtocol, normalizedModel, originalRequestedModel, false, reqMeta, execOptions.SkipInterceptorPluginID)
|
||||
+ ctx = lifecycle.executionContext()
|
||||
opts := coreexecutor.Options{
|
||||
Stream: false,
|
||||
Alt: alt,
|
||||
@@ -136,6 +137,7 @@ func (h *BaseAPIHandler) executeCountWithAuthManager(ctx context.Context, handle
|
||||
}
|
||||
afterAuthCapture := &requestAfterAuthCapture{}
|
||||
lifecycle := h.newRequestLifecycleTracker(ctx, handlerType, normalizedModel, originalRequestedModel, false, reqMeta, execOptions.SkipInterceptorPluginID)
|
||||
+ ctx = lifecycle.executionContext()
|
||||
opts := coreexecutor.Options{
|
||||
Stream: false,
|
||||
Alt: alt,
|
||||
@@ -177,6 +179,7 @@ func (h *BaseAPIHandler) executeWithPluginExecutor(ctx context.Context, entryPro
|
||||
}
|
||||
req, opts := h.pluginExecutorRequest(ctx, entryProtocol, responseProtocol, modelName, originalRequestedModel, rawJSON, alt, false, execOptions)
|
||||
lifecycle := h.newRequestLifecycleTracker(ctx, entryProtocol, modelName, originalRequestedModel, false, opts.Metadata, execOptions.SkipInterceptorPluginID)
|
||||
+ ctx = lifecycle.executionContext()
|
||||
var interceptErr *interfaces.ErrorMessage
|
||||
req, opts, interceptErr = h.applyRequestInterceptorsBeforeAuth(ctx, entryProtocol, originalRequestedModel, lifecycle.requestID(), req, opts, execOptions.SkipInterceptorPluginID)
|
||||
if interceptErr != nil {
|
||||
@@ -211,6 +214,7 @@ func (h *BaseAPIHandler) countWithPluginExecutor(ctx context.Context, handlerTyp
|
||||
}
|
||||
req, opts := h.pluginExecutorRequest(ctx, handlerType, handlerType, modelName, originalRequestedModel, rawJSON, alt, false, execOptions)
|
||||
lifecycle := h.newRequestLifecycleTracker(ctx, handlerType, modelName, originalRequestedModel, false, opts.Metadata, execOptions.SkipInterceptorPluginID)
|
||||
+ ctx = lifecycle.executionContext()
|
||||
var interceptErr *interfaces.ErrorMessage
|
||||
req, opts, interceptErr = h.applyRequestInterceptorsBeforeAuth(ctx, handlerType, originalRequestedModel, lifecycle.requestID(), req, opts, execOptions.SkipInterceptorPluginID)
|
||||
if interceptErr != nil {
|
||||
diff --git a/sdk/api/handlers/handlers_interceptors.go b/sdk/api/handlers/handlers_interceptors.go
|
||||
index a8b35604..d27198c9 100644
|
||||
--- a/sdk/api/handlers/handlers_interceptors.go
|
||||
+++ b/sdk/api/handlers/handlers_interceptors.go
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"github.com/router-for-me/CLIProxyAPI/v7/internal/interfaces"
|
||||
"github.com/router-for-me/CLIProxyAPI/v7/internal/logging"
|
||||
coreexecutor "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/executor"
|
||||
+ coreusage "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/usage"
|
||||
"github.com/router-for-me/CLIProxyAPI/v7/sdk/pluginapi"
|
||||
"golang.org/x/net/context"
|
||||
)
|
||||
@@ -74,6 +75,7 @@ type requestLifecycleTracker struct {
|
||||
func (h *BaseAPIHandler) newRequestLifecycleTracker(ctx context.Context, sourceFormat, model, requestedModel string, stream bool, metadata map[string]any, skipPluginID string) *requestLifecycleTracker {
|
||||
requestID := uuid.NewString()
|
||||
traceID := logging.GetRequestID(ctx)
|
||||
+ ctx = coreusage.WithRequestIdentity(ctx, requestID, traceID)
|
||||
return &requestLifecycleTracker{
|
||||
ctx: ctx,
|
||||
host: h.interceptorHost(),
|
||||
@@ -91,6 +93,13 @@ func (h *BaseAPIHandler) newRequestLifecycleTracker(ctx context.Context, sourceF
|
||||
}
|
||||
}
|
||||
|
||||
+func (t *requestLifecycleTracker) executionContext() context.Context {
|
||||
+ if t == nil || t.ctx == nil {
|
||||
+ return context.Background()
|
||||
+ }
|
||||
+ return t.ctx
|
||||
+}
|
||||
+
|
||||
func (t *requestLifecycleTracker) requestID() string {
|
||||
if t == nil {
|
||||
return ""
|
||||
diff --git a/sdk/api/handlers/handlers_interceptors_test.go b/sdk/api/handlers/handlers_interceptors_test.go
|
||||
index 0b328f3e..1ad85527 100644
|
||||
--- a/sdk/api/handlers/handlers_interceptors_test.go
|
||||
+++ b/sdk/api/handlers/handlers_interceptors_test.go
|
||||
@@ -16,6 +16,7 @@ import (
|
||||
"github.com/router-for-me/CLIProxyAPI/v7/internal/registry"
|
||||
coreauth "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/auth"
|
||||
coreexecutor "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/executor"
|
||||
+ coreusage "github.com/router-for-me/CLIProxyAPI/v7/sdk/cliproxy/usage"
|
||||
sdkconfig "github.com/router-for-me/CLIProxyAPI/v7/sdk/config"
|
||||
"github.com/router-for-me/CLIProxyAPI/v7/sdk/pluginapi"
|
||||
)
|
||||
@@ -230,6 +231,11 @@ func TestRequestLifecycleTrackerUsesUniqueExecutionIDs(t *testing.T) {
|
||||
if first.completion.TraceID != "trace-1" || second.completion.TraceID != "trace-1" {
|
||||
t.Fatalf("trace IDs = %q and %q", first.completion.TraceID, second.completion.TraceID)
|
||||
}
|
||||
+ firstRequestID, firstTraceID := coreusage.RequestIdentityFromContext(first.executionContext())
|
||||
+ secondRequestID, secondTraceID := coreusage.RequestIdentityFromContext(second.executionContext())
|
||||
+ if firstRequestID != first.requestID() || secondRequestID != second.requestID() || firstTraceID != "trace-1" || secondTraceID != "trace-1" {
|
||||
+ t.Fatalf("usage identities = (%q, %q) and (%q, %q)", firstRequestID, firstTraceID, secondRequestID, secondTraceID)
|
||||
+ }
|
||||
}
|
||||
|
||||
func TestHandlerRequestInterceptorTerminatesBeforeAuth(t *testing.T) {
|
||||
diff --git a/sdk/api/handlers/handlers_stream.go b/sdk/api/handlers/handlers_stream.go
|
||||
index 9cceec54..dcdc08df 100644
|
||||
--- a/sdk/api/handlers/handlers_stream.go
|
||||
+++ b/sdk/api/handlers/handlers_stream.go
|
||||
@@ -41,6 +41,7 @@ func (h *BaseAPIHandler) streamWithPluginExecutor(ctx context.Context, entryProt
|
||||
}
|
||||
req, opts := h.pluginExecutorRequest(ctx, entryProtocol, responseProtocol, modelName, originalRequestedModel, rawJSON, alt, true, execOptions)
|
||||
lifecycle := h.newRequestLifecycleTracker(ctx, entryProtocol, modelName, originalRequestedModel, true, opts.Metadata, execOptions.SkipInterceptorPluginID)
|
||||
+ ctx = lifecycle.executionContext()
|
||||
var interceptErr *interfaces.ErrorMessage
|
||||
req, opts, interceptErr = h.applyRequestInterceptorsBeforeAuth(ctx, entryProtocol, originalRequestedModel, lifecycle.requestID(), req, opts, execOptions.SkipInterceptorPluginID)
|
||||
if interceptErr != nil {
|
||||
@@ -305,6 +306,7 @@ func (h *BaseAPIHandler) executeStreamWithAuthManagerFormats(ctx context.Context
|
||||
}
|
||||
afterAuthCapture := &requestAfterAuthCapture{}
|
||||
lifecycle := h.newRequestLifecycleTracker(ctx, entryProtocol, normalizedModel, originalRequestedModel, true, reqMeta, execOptions.SkipInterceptorPluginID)
|
||||
+ ctx = lifecycle.executionContext()
|
||||
opts := coreexecutor.Options{
|
||||
Stream: true,
|
||||
Alt: alt,
|
||||
diff --git a/sdk/cliproxy/usage/manager.go b/sdk/cliproxy/usage/manager.go
|
||||
index ca36dc55..13506fbb 100644
|
||||
--- a/sdk/cliproxy/usage/manager.go
|
||||
+++ b/sdk/cliproxy/usage/manager.go
|
||||
@@ -78,6 +78,34 @@ type requestedModelAliasContextKey struct{}
|
||||
type reasoningEffortContextKey struct{}
|
||||
type serviceTierContextKey struct{}
|
||||
type generateContextKey struct{}
|
||||
+type requestIdentityContextKey struct{}
|
||||
+
|
||||
+type requestIdentity struct {
|
||||
+ requestID string
|
||||
+ traceID string
|
||||
+}
|
||||
+
|
||||
+// WithRequestIdentity stores the model execution and parent request IDs for usage sinks.
|
||||
+func WithRequestIdentity(ctx context.Context, requestID, traceID string) context.Context {
|
||||
+ if ctx == nil {
|
||||
+ ctx = context.Background()
|
||||
+ }
|
||||
+ requestID = strings.TrimSpace(requestID)
|
||||
+ traceID = strings.TrimSpace(traceID)
|
||||
+ if requestID == "" && traceID == "" {
|
||||
+ return ctx
|
||||
+ }
|
||||
+ return context.WithValue(ctx, requestIdentityContextKey{}, requestIdentity{requestID: requestID, traceID: traceID})
|
||||
+}
|
||||
+
|
||||
+// RequestIdentityFromContext returns the model execution and parent request IDs stored in ctx.
|
||||
+func RequestIdentityFromContext(ctx context.Context) (requestID, traceID string) {
|
||||
+ if ctx == nil {
|
||||
+ return "", ""
|
||||
+ }
|
||||
+ identity, _ := ctx.Value(requestIdentityContextKey{}).(requestIdentity)
|
||||
+ return strings.TrimSpace(identity.requestID), strings.TrimSpace(identity.traceID)
|
||||
+}
|
||||
|
||||
// WithRequestedModelAlias stores the client-requested model name for usage sinks.
|
||||
func WithRequestedModelAlias(ctx context.Context, alias string) context.Context {
|
||||
diff --git a/sdk/cliproxy/usage/manager_test.go b/sdk/cliproxy/usage/manager_test.go
|
||||
index 6f7b1fbb..fc2b3120 100644
|
||||
--- a/sdk/cliproxy/usage/manager_test.go
|
||||
+++ b/sdk/cliproxy/usage/manager_test.go
|
||||
@@ -36,6 +36,21 @@ func TestGenerateFromContextHonorsExplicitFalse(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
+func TestRequestIdentityContextRoundTrip(t *testing.T) {
|
||||
+ ctx := WithRequestIdentity(context.Background(), " request-1 ", " trace-1 ")
|
||||
+ requestID, traceID := RequestIdentityFromContext(ctx)
|
||||
+ if requestID != "request-1" || traceID != "trace-1" {
|
||||
+ t.Fatalf("request identity = %q, %q", requestID, traceID)
|
||||
+ }
|
||||
+}
|
||||
+
|
||||
+func TestRequestIdentityContextDefaultsEmpty(t *testing.T) {
|
||||
+ requestID, traceID := RequestIdentityFromContext(nil)
|
||||
+ if requestID != "" || traceID != "" {
|
||||
+ t.Fatalf("empty request identity = %q, %q", requestID, traceID)
|
||||
+ }
|
||||
+}
|
||||
+
|
||||
func TestRecordOmittedGenerateIsEnabled(t *testing.T) {
|
||||
// Existing callers construct Record without setting Generate.
|
||||
// Omission must remain distinguishable from explicit false and default to true.
|
||||
diff --git a/sdk/pluginabi/types.go b/sdk/pluginabi/types.go
|
||||
index 97c41a13..a8a3e470 100644
|
||||
--- a/sdk/pluginabi/types.go
|
||||
+++ b/sdk/pluginabi/types.go
|
||||
@@ -10,10 +10,13 @@ const (
|
||||
// Version 3 omits OriginalRequest/RequestBody on payload stream chunks
|
||||
// (ChunkIndex >= 0); those fields remain on StreamChunkHeaderInitIndex only.
|
||||
// Plugins that still need per-chunk request bodies should keep schema_version < 3.
|
||||
- SchemaVersion uint32 = 3
|
||||
+ SchemaVersion uint32 = 4
|
||||
// SchemaVersionStreamChunkOmitRequestBody is the first schema version that omits
|
||||
// request bodies on payload stream-chunk interceptor calls.
|
||||
SchemaVersionStreamChunkOmitRequestBody uint32 = 3
|
||||
+ // SchemaVersionUsageIdentity is the first schema version that attaches
|
||||
+ // request lifecycle identity to usage records.
|
||||
+ SchemaVersionUsageIdentity uint32 = 4
|
||||
)
|
||||
|
||||
const (
|
||||
diff --git a/sdk/pluginabi/types_test.go b/sdk/pluginabi/types_test.go
|
||||
index 8fa63542..86253848 100644
|
||||
--- a/sdk/pluginabi/types_test.go
|
||||
+++ b/sdk/pluginabi/types_test.go
|
||||
@@ -27,12 +27,15 @@ func TestEnvelopeRoundTrip(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestMethodNamesAreStable(t *testing.T) {
|
||||
- if SchemaVersion != 3 {
|
||||
- t.Fatalf("SchemaVersion = %d, want 3", SchemaVersion)
|
||||
+ if SchemaVersion != 4 {
|
||||
+ t.Fatalf("SchemaVersion = %d, want 4", SchemaVersion)
|
||||
}
|
||||
if SchemaVersionStreamChunkOmitRequestBody != 3 {
|
||||
t.Fatalf("SchemaVersionStreamChunkOmitRequestBody = %d, want 3", SchemaVersionStreamChunkOmitRequestBody)
|
||||
}
|
||||
+ if SchemaVersionUsageIdentity != 4 {
|
||||
+ t.Fatalf("SchemaVersionUsageIdentity = %d, want 4", SchemaVersionUsageIdentity)
|
||||
+ }
|
||||
if MethodPluginRegister != "plugin.register" {
|
||||
t.Fatalf("MethodPluginRegister = %q", MethodPluginRegister)
|
||||
}
|
||||
diff --git a/sdk/pluginapi/types.go b/sdk/pluginapi/types.go
|
||||
index 6add5d69..894e3b77 100644
|
||||
--- a/sdk/pluginapi/types.go
|
||||
+++ b/sdk/pluginapi/types.go
|
||||
@@ -1316,6 +1316,10 @@ type ManagementResponse struct {
|
||||
|
||||
// UsageRecord describes request usage and billing metadata.
|
||||
type UsageRecord struct {
|
||||
+ // RequestID identifies the request lifecycle associated with this usage record.
|
||||
+ RequestID string
|
||||
+ // TraceID identifies the parent inbound HTTP request when available.
|
||||
+ TraceID string
|
||||
// Provider identifies the upstream provider.
|
||||
Provider string
|
||||
// ExecutorType identifies the executor implementation.
|
||||
Reference in New Issue
Block a user